mono/metadata/threadpool-ms.c

   1 /*
   2  * threadpool-ms.c: Microsoft threadpool runtime support
   3  *
   4  * Author:
   5  *      Ludovic Henry (ludovic.henry@xamarin.com)
   6  *
   7  * Copyright 2015 Xamarin, Inc (http://www.xamarin.com)
   8  */
   9
  10 //
  11 // Copyright (c) Microsoft. All rights reserved.
  12 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
  13 //
  14 // Files:
  15 //  - src/vm/comthreadpool.cpp
  16 //  - src/vm/win32threadpoolcpp
  17 //  - src/vm/threadpoolrequest.cpp
  18 //  - src/vm/hillclimbing.cpp
  19 //
  20 // Ported from C++ to C and adjusted to Mono runtime
  21
  22 #include <stdlib.h>
  23 #define _USE_MATH_DEFINES // needed by MSVC to define math constants
  24 #include <math.h>
  25 #include <config.h>
  26 #include <glib.h>
  27
  28 #include <mono/metadata/class-internals.h>
  29 #include <mono/metadata/exception.h>
  30 #include <mono/metadata/gc-internal.h>
  31 #include <mono/metadata/object.h>
  32 #include <mono/metadata/object-internals.h>
  33 #include <mono/metadata/threadpool-ms.h>
  34 #include <mono/metadata/threadpool-ms-io.h>
  35 #include <mono/utils/atomic.h>
  36 #include <mono/utils/mono-compiler.h>
  37 #include <mono/utils/mono-complex.h>
  38 #include <mono/utils/mono-proclib.h>
  39 #include <mono/utils/mono-threads.h>
  40 #include <mono/utils/mono-time.h>
  41 #include <mono/utils/mono-rand.h>
  42
  43 #define CPU_USAGE_LOW 80
  44 #define CPU_USAGE_HIGH 95
  45
  46 #define MONITOR_INTERVAL 100 // ms
  47
  48 /* The exponent to apply to the gain. 1.0 means to use linear gain,
  49  * higher values will enhance large moves and damp small ones.
  50  * default: 2.0 */
  51 #define HILL_CLIMBING_GAIN_EXPONENT 2.0
  52
  53 /* The 'cost' of a thread. 0 means drive for increased throughput regardless
  54  * of thread count, higher values bias more against higher thread counts.
  55  * default: 0.15 */
  56 #define HILL_CLIMBING_BIAS 0.15
  57
  58 #define HILL_CLIMBING_WAVE_PERIOD 4
  59 #define HILL_CLIMBING_MAX_WAVE_MAGNITUDE 20
  60 #define HILL_CLIMBING_WAVE_MAGNITUDE_MULTIPLIER 1.0
  61 #define HILL_CLIMBING_WAVE_HISTORY_SIZE 8
  62 #define HILL_CLIMBING_TARGET_SIGNAL_TO_NOISE_RATIO 3.0
  63 #define HILL_CLIMBING_MAX_CHANGE_PER_SECOND 4
  64 #define HILL_CLIMBING_MAX_CHANGE_PER_SAMPLE 20
  65 #define HILL_CLIMBING_SAMPLE_INTERVAL_LOW 10
  66 #define HILL_CLIMBING_SAMPLE_INTERVAL_HIGH 200
  67 #define HILL_CLIMBING_ERROR_SMOOTHING_FACTOR 0.01
  68 #define HILL_CLIMBING_MAX_SAMPLE_ERROR_PERCENT 0.15
  69
  70 typedef union {
  71         struct {
  72                 gint16 max_working; /* determined by heuristic */
  73                 gint16 active; /* executing worker_thread */
  74                 gint16 working; /* actively executing worker_thread, not parked */
  75                 gint16 parked; /* parked */
  76         } _;
  77         gint64 as_gint64;
  78 } ThreadPoolCounter;
  79
  80 typedef struct {
  81         MonoDomain *domain;
  82         gint32 outstanding_request;
  83 } ThreadPoolDomain;
  84
  85 typedef MonoInternalThread ThreadPoolWorkingThread;
  86 typedef mono_cond_t ThreadPoolParkedThread;
  87
  88 typedef struct {
  89         gint32 wave_period;
  90         gint32 samples_to_measure;
  91         gdouble target_throughput_ratio;
  92         gdouble target_signal_to_noise_ratio;
  93         gdouble max_change_per_second;
  94         gdouble max_change_per_sample;
  95         gint32 max_thread_wave_magnitude;
  96         gint32 sample_interval_low;
  97         gdouble thread_magnitude_multiplier;
  98         gint32 sample_interval_high;
  99         gdouble throughput_error_smoothing_factor;
 100         gdouble gain_exponent;
 101         gdouble max_sample_error;
 102
 103         gdouble current_control_setting;
 104         gint64 total_samples;
 105         gint16 last_thread_count;
 106         gdouble elapsed_since_last_change;
 107         gdouble completions_since_last_change;
 108
 109         gdouble average_throughput_noise;
 110
 111         gdouble *samples;
 112         gdouble *thread_counts;
 113
 114         guint32 current_sample_interval;
 115         gpointer random_interval_generator;
 116
 117         gint32 accumulated_completion_count;
 118         gdouble accumulated_sample_duration;
 119 } ThreadPoolHillClimbing;
 120
 121 typedef struct {
 122         ThreadPoolCounter counters;
 123
 124         GPtrArray *domains; // ThreadPoolDomain* []
 125         mono_mutex_t domains_lock;
 126
 127         GPtrArray *working_threads; // ThreadPoolWorkingThread* []
 128         GPtrArray *parked_threads; // ThreadPoolParkedThread* []
 129         mono_mutex_t active_threads_lock; /* protect access to working_threads and parked_threads */
 130
 131         gint32 heuristic_completions;
 132         guint32 heuristic_sample_start;
 133         guint32 heuristic_last_dequeue; // ms
 134         guint32 heuristic_last_adjustment; // ms
 135         guint32 heuristic_adjustment_interval; // ms
 136         ThreadPoolHillClimbing heuristic_hill_climbing;
 137         mono_mutex_t heuristic_lock;
 138
 139         gint32 limit_worker_min;
 140         gint32 limit_worker_max;
 141         gint32 limit_io_min;
 142         gint32 limit_io_max;
 143
 144         MonoCpuUsageState *cpu_usage_state;
 145         gint32 cpu_usage;
 146
 147         /* suspended by the debugger */
 148         gboolean suspended;
 149 } ThreadPool;
 150
 151 typedef enum {
 152         TRANSITION_WARMUP,
 153         TRANSITION_INITIALIZING,
 154         TRANSITION_RANDOM_MOVE,
 155         TRANSITION_CLIMBING_MOVE,
 156         TRANSITION_CHANGE_POINT,
 157         TRANSITION_STABILIZING,
 158         TRANSITION_STARVATION,
 159         TRANSITION_THREAD_TIMED_OUT,
 160         TRANSITION_UNDEFINED,
 161 } ThreadPoolHeuristicStateTransition;
 162
 163 enum {
 164         MONITOR_STATUS_REQUESTED,
 165         MONITOR_STATUS_WAITING_FOR_REQUEST,
 166         MONITOR_STATUS_NOT_RUNNING,
 167 };
 168
 169 static gint32 status = STATUS_NOT_INITIALIZED;
 170 static gint32 monitor_status = MONITOR_STATUS_NOT_RUNNING;
 171
 172 static ThreadPool* threadpool;
 173
 174 #define COUNTER_CHECK(counter) \
 175         do { \
 176                 g_assert (counter._.max_working > 0); \
 177                 g_assert (counter._.working >= 0); \
 178                 g_assert (counter._.active >= 0); \
 179         } while (0)
 180
 181 #define COUNTER_READ() (InterlockedRead64 (&threadpool->counters.as_gint64))
 182
 183 #define COUNTER_ATOMIC(var,block) \
 184         do { \
 185                 ThreadPoolCounter __old; \
 186                 do { \
 187                         g_assert (threadpool); \
 188                         __old.as_gint64 = COUNTER_READ (); \
 189                         (var) = __old; \
 190                         { block; } \
 191                         COUNTER_CHECK (var); \
 192                 } while (InterlockedCompareExchange64 (&threadpool->counters.as_gint64, (var).as_gint64, __old.as_gint64) != __old.as_gint64); \
 193         } while (0)
 194
 195 #define COUNTER_TRY_ATOMIC(res,var,block) \
 196         do { \
 197                 ThreadPoolCounter __old; \
 198                 do { \
 199                         g_assert (threadpool); \
 200                         __old.as_gint64 = COUNTER_READ (); \
 201                         (var) = __old; \
 202                         (res) = FALSE; \
 203                         { block; } \
 204                         COUNTER_CHECK (var); \
 205                         (res) = InterlockedCompareExchange64 (&threadpool->counters.as_gint64, (var).as_gint64, __old.as_gint64) == __old.as_gint64; \
 206                 } while (0); \
 207         } while (0)
 208
 209 static gpointer
 210 rand_create (void)
 211 {
 212         mono_rand_open ();
 213         return mono_rand_init (NULL, 0);
 214 }
 215
 216 static guint32
 217 rand_next (gpointer *handle, guint32 min, guint32 max)
 218 {
 219         guint32 val;
 220         if (!mono_rand_try_get_uint32 (handle, &val, min, max)) {
 221                 // FIXME handle error
 222                 g_assert_not_reached ();
 223         }
 224         return val;
 225 }
 226
 227 static void
 228 rand_free (gpointer handle)
 229 {
 230         mono_rand_close (handle);
 231 }
 232
 233 static void
 234 ensure_initialized (MonoBoolean *enable_worker_tracking)
 235 {
 236         ThreadPoolHillClimbing *hc;
 237         const char *threads_per_cpu_env;
 238         gint threads_per_cpu;
 239         gint threads_count;
 240
 241         if (enable_worker_tracking) {
 242                 // TODO implement some kind of switch to have the possibily to use it
 243                 *enable_worker_tracking = FALSE;
 244         }
 245
 246         if (status >= STATUS_INITIALIZED)
 247                 return;
 248         if (status == STATUS_INITIALIZING || InterlockedCompareExchange (&status, STATUS_INITIALIZING, STATUS_NOT_INITIALIZED) != STATUS_NOT_INITIALIZED) {
 249                 while (status == STATUS_INITIALIZING)
 250                         mono_thread_info_yield ();
 251                 g_assert (status >= STATUS_INITIALIZED);
 252                 return;
 253         }
 254
 255         g_assert (!threadpool);
 256         threadpool = g_new0 (ThreadPool, 1);
 257         g_assert (threadpool);
 258
 259         threadpool->domains = g_ptr_array_new ();
 260         mono_mutex_init_recursive (&threadpool->domains_lock);
 261
 262         threadpool->parked_threads = g_ptr_array_new ();
 263         threadpool->working_threads = g_ptr_array_new ();
 264         mono_mutex_init (&threadpool->active_threads_lock);
 265
 266         threadpool->heuristic_adjustment_interval = 10;
 267         mono_mutex_init (&threadpool->heuristic_lock);
 268
 269         mono_rand_open ();
 270
 271         hc = &threadpool->heuristic_hill_climbing;
 272
 273         hc->wave_period = HILL_CLIMBING_WAVE_PERIOD;
 274         hc->max_thread_wave_magnitude = HILL_CLIMBING_MAX_WAVE_MAGNITUDE;
 275         hc->thread_magnitude_multiplier = (gdouble) HILL_CLIMBING_WAVE_MAGNITUDE_MULTIPLIER;
 276         hc->samples_to_measure = hc->wave_period * HILL_CLIMBING_WAVE_HISTORY_SIZE;
 277         hc->target_throughput_ratio = (gdouble) HILL_CLIMBING_BIAS;
 278         hc->target_signal_to_noise_ratio = (gdouble) HILL_CLIMBING_TARGET_SIGNAL_TO_NOISE_RATIO;
 279         hc->max_change_per_second = (gdouble) HILL_CLIMBING_MAX_CHANGE_PER_SECOND;
 280         hc->max_change_per_sample = (gdouble) HILL_CLIMBING_MAX_CHANGE_PER_SAMPLE;
 281         hc->sample_interval_low = HILL_CLIMBING_SAMPLE_INTERVAL_LOW;
 282         hc->sample_interval_high = HILL_CLIMBING_SAMPLE_INTERVAL_HIGH;
 283         hc->throughput_error_smoothing_factor = (gdouble) HILL_CLIMBING_ERROR_SMOOTHING_FACTOR;
 284         hc->gain_exponent = (gdouble) HILL_CLIMBING_GAIN_EXPONENT;
 285         hc->max_sample_error = (gdouble) HILL_CLIMBING_MAX_SAMPLE_ERROR_PERCENT;
 286         hc->current_control_setting = 0;
 287         hc->total_samples = 0;
 288         hc->last_thread_count = 0;
 289         hc->average_throughput_noise = 0;
 290         hc->elapsed_since_last_change = 0;
 291         hc->accumulated_completion_count = 0;
 292         hc->accumulated_sample_duration = 0;
 293         hc->samples = g_new0 (gdouble, hc->samples_to_measure);
 294         hc->thread_counts = g_new0 (gdouble, hc->samples_to_measure);
 295         hc->random_interval_generator = rand_create ();
 296         hc->current_sample_interval = rand_next (&hc->random_interval_generator, hc->sample_interval_low, hc->sample_interval_high);
 297
 298         if (!(threads_per_cpu_env = g_getenv ("MONO_THREADS_PER_CPU")))
 299                 threads_per_cpu = 1;
 300         else
 301                 threads_per_cpu = CLAMP (atoi (threads_per_cpu_env), 1, 50);
 302
 303         threads_count = mono_cpu_count () * threads_per_cpu;
 304
 305         threadpool->limit_worker_min = threadpool->limit_io_min = threads_count;
 306         threadpool->limit_worker_max = threadpool->limit_io_max = threads_count * 100;
 307
 308         threadpool->counters._.max_working = threadpool->limit_worker_min;
 309
 310         threadpool->cpu_usage_state = g_new0 (MonoCpuUsageState, 1);
 311
 312         threadpool->suspended = FALSE;
 313
 314         status = STATUS_INITIALIZED;
 315 }
 316
 317 static void worker_unpark (ThreadPoolParkedThread *thread);
 318 static void worker_kill (ThreadPoolWorkingThread *thread);
 319
 320 static void
 321 ensure_cleanedup (void)
 322 {
 323         guint i;
 324
 325         if (status == STATUS_NOT_INITIALIZED && InterlockedCompareExchange (&status, STATUS_CLEANED_UP, STATUS_NOT_INITIALIZED) == STATUS_NOT_INITIALIZED)
 326                 return;
 327         if (status == STATUS_INITIALIZING) {
 328                 while (status == STATUS_INITIALIZING)
 329                         mono_thread_info_yield ();
 330         }
 331         if (status == STATUS_CLEANED_UP)
 332                 return;
 333         if (status == STATUS_CLEANING_UP || InterlockedCompareExchange (&status, STATUS_CLEANING_UP, STATUS_INITIALIZED) != STATUS_INITIALIZED) {
 334                 while (status == STATUS_CLEANING_UP)
 335                         mono_thread_info_yield ();
 336                 g_assert (status == STATUS_CLEANED_UP);
 337                 return;
 338         }
 339
 340         /* we make the assumption along the code that we are
 341          * cleaning up only if the runtime is shutting down */
 342         g_assert (mono_runtime_is_shutting_down ());
 343
 344         while (monitor_status != MONITOR_STATUS_NOT_RUNNING)
 345                 g_usleep (1000);
 346
 347         mono_mutex_lock (&threadpool->active_threads_lock);
 348
 349         /* stop all threadpool->working_threads */
 350         for (i = 0; i < threadpool->working_threads->len; ++i)
 351                 worker_kill ((ThreadPoolWorkingThread*) g_ptr_array_index (threadpool->working_threads, i));
 352
 353         /* unpark all threadpool->parked_threads */
 354         for (i = 0; i < threadpool->parked_threads->len; ++i)
 355                 worker_unpark ((ThreadPoolParkedThread*) g_ptr_array_index (threadpool->parked_threads, i));
 356
 357         mono_mutex_unlock (&threadpool->active_threads_lock);
 358
 359         status = STATUS_CLEANED_UP;
 360 }
 361
 362 void
 363 mono_threadpool_ms_enqueue_work_item (MonoDomain *domain, MonoObject *work_item)
 364 {
 365         static MonoClass *threadpool_class = NULL;
 366         static MonoMethod *unsafe_queue_custom_work_item_method = NULL;
 367         MonoDomain *current_domain;
 368         MonoBoolean f;
 369         gpointer args [2];
 370
 371         g_assert (work_item);
 372
 373         if (!threadpool_class)
 374                 threadpool_class = mono_class_from_name (mono_defaults.corlib, "System.Threading", "ThreadPool");
 375         g_assert (threadpool_class);
 376
 377         if (!unsafe_queue_custom_work_item_method)
 378                 unsafe_queue_custom_work_item_method = mono_class_get_method_from_name (threadpool_class, "UnsafeQueueCustomWorkItem", 2);
 379         g_assert (unsafe_queue_custom_work_item_method);
 380
 381         f = FALSE;
 382
 383         args [0] = (gpointer) work_item;
 384         args [1] = (gpointer) &f;
 385
 386         current_domain = mono_domain_get ();
 387         if (current_domain == domain) {
 388                 mono_runtime_invoke (unsafe_queue_custom_work_item_method, NULL, args, NULL);
 389         } else {
 390                 mono_thread_push_appdomain_ref (domain);
 391                 if (mono_domain_set (domain, FALSE)) {
 392                         mono_runtime_invoke (unsafe_queue_custom_work_item_method, NULL, args, NULL);
 393                         mono_domain_set (current_domain, TRUE);
 394                 }
 395                 mono_thread_pop_appdomain_ref ();
 396         }
 397 }
 398
 399 static void
 400 domain_add (ThreadPoolDomain *tpdomain)
 401 {
 402         guint i, len;
 403
 404         g_assert (tpdomain);
 405
 406         mono_mutex_lock (&threadpool->domains_lock);
 407         len = threadpool->domains->len;
 408         for (i = 0; i < len; ++i) {
 409                 if (g_ptr_array_index (threadpool->domains, i) == tpdomain)
 410                         break;
 411         }
 412         if (i == len)
 413                 g_ptr_array_add (threadpool->domains, tpdomain);
 414         mono_mutex_unlock (&threadpool->domains_lock);
 415 }
 416
 417 static gboolean
 418 domain_remove (ThreadPoolDomain *tpdomain)
 419 {
 420         gboolean res;
 421
 422         g_assert (tpdomain);
 423
 424         mono_mutex_lock (&threadpool->domains_lock);
 425         res = g_ptr_array_remove (threadpool->domains, tpdomain);
 426         mono_mutex_unlock (&threadpool->domains_lock);
 427
 428         return res;
 429 }
 430
 431 static ThreadPoolDomain *
 432 domain_get (MonoDomain *domain, gboolean create)
 433 {
 434         ThreadPoolDomain *tpdomain = NULL;
 435         guint i;
 436
 437         g_assert (domain);
 438
 439         mono_mutex_lock (&threadpool->domains_lock);
 440         for (i = 0; i < threadpool->domains->len; ++i) {
 441                 ThreadPoolDomain *tmp = g_ptr_array_index (threadpool->domains, i);
 442                 if (tmp->domain == domain) {
 443                         tpdomain = tmp;
 444                         break;
 445                 }
 446         }
 447         if (!tpdomain && create) {
 448                 tpdomain = g_new0 (ThreadPoolDomain, 1);
 449                 tpdomain->domain = domain;
 450                 domain_add (tpdomain);
 451         }
 452         mono_mutex_unlock (&threadpool->domains_lock);
 453         return tpdomain;
 454 }
 455
 456 static void
 457 domain_free (ThreadPoolDomain *tpdomain)
 458 {
 459         g_free (tpdomain);
 460 }
 461
 462 static gboolean
 463 domain_any_has_request (void)
 464 {
 465         gboolean res = FALSE;
 466         guint i;
 467
 468         mono_mutex_lock (&threadpool->domains_lock);
 469         for (i = 0; i < threadpool->domains->len; ++i) {
 470                 ThreadPoolDomain *tmp = g_ptr_array_index (threadpool->domains, i);
 471                 if (tmp->outstanding_request > 0) {
 472                         res = TRUE;
 473                         break;
 474                 }
 475         }
 476         mono_mutex_unlock (&threadpool->domains_lock);
 477         return res;
 478 }
 479
 480 static ThreadPoolDomain *
 481 domain_get_next (ThreadPoolDomain *current)
 482 {
 483         ThreadPoolDomain *tpdomain = NULL;
 484         guint len;
 485
 486         mono_mutex_lock (&threadpool->domains_lock);
 487         len = threadpool->domains->len;
 488         if (len > 0) {
 489                 guint i, current_idx = -1;
 490                 if (current) {
 491                         for (i = 0; i < len; ++i) {
 492                                 if (current == g_ptr_array_index (threadpool->domains, i)) {
 493                                         current_idx = i;
 494                                         break;
 495                                 }
 496                         }
 497                         g_assert (current_idx >= 0);
 498                 }
 499                 for (i = current_idx + 1; i < len + current_idx + 1; ++i) {
 500                         ThreadPoolDomain *tmp = g_ptr_array_index (threadpool->domains, i % len);
 501                         if (tmp->outstanding_request > 0) {
 502                                 tpdomain = tmp;
 503                                 break;
 504                         }
 505                 }
 506         }
 507         mono_mutex_unlock (&threadpool->domains_lock);
 508         return tpdomain;
 509 }
 510
 511 static void
 512 worker_park (void)
 513 {
 514         mono_cond_t cond;
 515         MonoInternalThread *thread = mono_thread_internal_current ();
 516
 517         mono_cond_init (&cond, NULL);
 518
 519         mono_gc_set_skip_thread (TRUE);
 520
 521         mono_mutex_lock (&threadpool->active_threads_lock);
 522
 523         if (!mono_runtime_is_shutting_down ()) {
 524                 g_ptr_array_add (threadpool->parked_threads, &cond);
 525                 g_ptr_array_remove_fast (threadpool->working_threads, thread);
 526
 527                 mono_cond_wait (&cond, &threadpool->active_threads_lock);
 528
 529                 g_ptr_array_add (threadpool->working_threads, thread);
 530                 g_ptr_array_remove (threadpool->parked_threads, &cond);
 531         }
 532
 533         mono_mutex_unlock (&threadpool->active_threads_lock);
 534
 535         mono_gc_set_skip_thread (FALSE);
 536
 537         mono_cond_destroy (&cond);
 538 }
 539
 540 static gboolean
 541 worker_try_unpark (void)
 542 {
 543         gboolean res = FALSE;
 544         guint len;
 545
 546         mono_mutex_lock (&threadpool->active_threads_lock);
 547         len = threadpool->parked_threads->len;
 548         if (len > 0) {
 549                 mono_cond_t *cond = (mono_cond_t*) g_ptr_array_index (threadpool->parked_threads, len - 1);
 550                 mono_cond_signal (cond);
 551                 res = TRUE;
 552         }
 553         mono_mutex_unlock (&threadpool->active_threads_lock);
 554         return res;
 555 }
 556
 557 static void
 558 worker_unpark (ThreadPoolParkedThread *thread)
 559 {
 560         mono_cond_signal ((mono_cond_t*) thread);
 561 }
 562
 563 static void
 564 worker_kill (ThreadPoolWorkingThread *thread)
 565 {
 566         if (thread == mono_thread_internal_current ())
 567                 return;
 568
 569         mono_thread_internal_stop ((MonoInternalThread*) thread);
 570 }
 571
 572 static void
 573 worker_thread (gpointer data)
 574 {
 575         static MonoClass *threadpool_wait_callback_class = NULL;
 576         static MonoMethod *perform_wait_callback_method = NULL;
 577         MonoInternalThread *thread;
 578         ThreadPoolDomain *tpdomain, *previous_tpdomain;
 579         ThreadPoolCounter counter;
 580         gboolean retire = FALSE;
 581
 582         g_assert (status >= STATUS_INITIALIZED);
 583
 584         if (!threadpool_wait_callback_class)
 585                 threadpool_wait_callback_class = mono_class_from_name (mono_defaults.corlib, "System.Threading", "_ThreadPoolWaitCallback");
 586         g_assert (threadpool_wait_callback_class);
 587
 588         if (!perform_wait_callback_method)
 589                 perform_wait_callback_method = mono_class_get_method_from_name (threadpool_wait_callback_class, "PerformWaitCallback", 0);
 590         g_assert (perform_wait_callback_method);
 591
 592         g_assert (threadpool);
 593
 594         thread = mono_thread_internal_current ();
 595         g_assert (thread);
 596
 597         mono_thread_set_name_internal (thread, mono_string_new (mono_domain_get (), "Threadpool worker"), FALSE);
 598
 599         mono_mutex_lock (&threadpool->active_threads_lock);
 600         g_ptr_array_add (threadpool->working_threads, thread);
 601         mono_mutex_unlock (&threadpool->active_threads_lock);
 602
 603         previous_tpdomain = NULL;
 604
 605         mono_mutex_lock (&threadpool->domains_lock);
 606
 607         while (!mono_runtime_is_shutting_down ()) {
 608                 tpdomain = NULL;
 609
 610                 if ((thread->state & (ThreadState_StopRequested | ThreadState_SuspendRequested)) != 0) {
 611                         mono_mutex_unlock (&threadpool->domains_lock);
 612                         mono_thread_interruption_checkpoint ();
 613                         mono_mutex_lock (&threadpool->domains_lock);
 614                 }
 615
 616                 if (retire || !(tpdomain = domain_get_next (previous_tpdomain))) {
 617                         COUNTER_ATOMIC (counter, {
 618                                 counter._.working --;
 619                                 counter._.parked ++;
 620                         });
 621
 622                         mono_mutex_unlock (&threadpool->domains_lock);
 623                         worker_park ();
 624                         mono_mutex_lock (&threadpool->domains_lock);
 625
 626                         COUNTER_ATOMIC (counter, {
 627                                 counter._.working ++;
 628                                 counter._.parked --;
 629                         });
 630
 631                         if (retire)
 632                                 retire = FALSE;
 633
 634                         continue;
 635                 }
 636
 637                 tpdomain->outstanding_request --;
 638                 g_assert (tpdomain->outstanding_request >= 0);
 639
 640                 g_assert (tpdomain->domain);
 641                 g_assert (tpdomain->domain->threadpool_jobs >= 0);
 642                 tpdomain->domain->threadpool_jobs ++;
 643
 644                 mono_mutex_unlock (&threadpool->domains_lock);
 645
 646                 mono_thread_push_appdomain_ref (tpdomain->domain);
 647                 if (mono_domain_set (tpdomain->domain, FALSE)) {
 648                         MonoObject *exc = NULL;
 649                         MonoObject *res = mono_runtime_invoke (perform_wait_callback_method, NULL, NULL, &exc);
 650                         if (exc)
 651                                 mono_thread_internal_unhandled_exception (exc);
 652                         else if (res && *(MonoBoolean*) mono_object_unbox (res) == FALSE)
 653                                 retire = TRUE;
 654
 655                         mono_thread_clr_state (thread , ~ThreadState_Background);
 656                         if (!mono_thread_test_state (thread , ThreadState_Background))
 657                                 ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
 658
 659                         mono_domain_set (mono_get_root_domain (), TRUE);
 660                 }
 661                 mono_thread_pop_appdomain_ref ();
 662
 663                 mono_mutex_lock (&threadpool->domains_lock);
 664
 665                 tpdomain->domain->threadpool_jobs --;
 666                 g_assert (tpdomain->domain->threadpool_jobs >= 0);
 667
 668                 if (tpdomain->domain->threadpool_jobs == 0 && mono_domain_is_unloading (tpdomain->domain)) {
 669                         gboolean removed = domain_remove (tpdomain);
 670                         g_assert (removed);
 671                         if (tpdomain->domain->cleanup_semaphore)
 672                                 ReleaseSemaphore (tpdomain->domain->cleanup_semaphore, 1, NULL);
 673                         domain_free (tpdomain);
 674                         tpdomain = NULL;
 675                 }
 676
 677                 previous_tpdomain = tpdomain;
 678         }
 679
 680         mono_mutex_unlock (&threadpool->domains_lock);
 681
 682         mono_mutex_lock (&threadpool->active_threads_lock);
 683         g_ptr_array_remove_fast (threadpool->working_threads, thread);
 684         mono_mutex_unlock (&threadpool->active_threads_lock);
 685
 686         COUNTER_ATOMIC (counter, {
 687                 counter._.working--;
 688                 counter._.active --;
 689         });
 690 }
 691
 692 static gboolean
 693 worker_try_create (void)
 694 {
 695         ThreadPoolCounter counter;
 696
 697         COUNTER_ATOMIC (counter, {
 698                 if (counter._.working >= counter._.max_working)
 699                         return FALSE;
 700                 counter._.working ++;
 701                 counter._.active ++;
 702         });
 703
 704         if (mono_thread_create_internal (mono_get_root_domain (), worker_thread, NULL, TRUE, 0) != NULL)
 705                 return TRUE;
 706
 707         COUNTER_ATOMIC (counter, {
 708                 counter._.working --;
 709                 counter._.active --;
 710         });
 711
 712         return FALSE;
 713 }
 714
 715 static void monitor_ensure_running (void);
 716
 717 static gboolean
 718 worker_request (MonoDomain *domain)
 719 {
 720         ThreadPoolDomain *tpdomain;
 721
 722         g_assert (domain);
 723         g_assert (threadpool);
 724
 725         if (mono_runtime_is_shutting_down ())
 726                 return FALSE;
 727
 728         mono_mutex_lock (&threadpool->domains_lock);
 729
 730         /* synchronize check with worker_thread */
 731         if (mono_domain_is_unloading (domain)) {
 732                 mono_mutex_unlock (&threadpool->domains_lock);
 733                 return FALSE;
 734         }
 735
 736         tpdomain = domain_get (domain, TRUE);
 737         g_assert (tpdomain);
 738         tpdomain->outstanding_request ++;
 739
 740         mono_mutex_unlock (&threadpool->domains_lock);
 741
 742         if (threadpool->suspended)
 743                 return FALSE;
 744
 745         monitor_ensure_running ();
 746
 747         if (worker_try_unpark ())
 748                 return TRUE;
 749
 750         if (worker_try_create ())
 751                 return TRUE;
 752
 753         return FALSE;
 754 }
 755
 756 static gboolean
 757 monitor_should_keep_running (void)
 758 {
 759         g_assert (monitor_status == MONITOR_STATUS_WAITING_FOR_REQUEST || monitor_status == MONITOR_STATUS_REQUESTED);
 760
 761         if (InterlockedExchange (&monitor_status, MONITOR_STATUS_WAITING_FOR_REQUEST) == MONITOR_STATUS_WAITING_FOR_REQUEST) {
 762                 if (mono_runtime_is_shutting_down () || !domain_any_has_request ()) {
 763                         if (InterlockedCompareExchange (&monitor_status, MONITOR_STATUS_NOT_RUNNING, MONITOR_STATUS_WAITING_FOR_REQUEST) == MONITOR_STATUS_WAITING_FOR_REQUEST)
 764                                 return FALSE;
 765                 }
 766         }
 767
 768         g_assert (monitor_status == MONITOR_STATUS_WAITING_FOR_REQUEST || monitor_status == MONITOR_STATUS_REQUESTED);
 769
 770         return TRUE;
 771 }
 772
 773 static gboolean
 774 monitor_sufficient_delay_since_last_dequeue (void)
 775 {
 776         guint32 threshold;
 777
 778         g_assert (threadpool);
 779
 780         if (threadpool->cpu_usage < CPU_USAGE_LOW) {
 781                 threshold = MONITOR_INTERVAL;
 782         } else {
 783                 ThreadPoolCounter counter;
 784                 counter.as_gint64 = COUNTER_READ();
 785                 threshold = counter._.max_working * MONITOR_INTERVAL * 2;
 786         }
 787
 788         return mono_msec_ticks () >= threadpool->heuristic_last_dequeue + threshold;
 789 }
 790
 791 static void hill_climbing_force_change (gint16 new_thread_count, ThreadPoolHeuristicStateTransition transition);
 792
 793 static void
 794 monitor_thread (void)
 795 {
 796         MonoInternalThread *current_thread = mono_thread_internal_current ();
 797         guint i;
 798
 799         mono_cpu_usage (threadpool->cpu_usage_state);
 800
 801         do {
 802                 MonoInternalThread *thread;
 803                 gboolean all_waitsleepjoin = TRUE;
 804                 gint32 interval_left = MONITOR_INTERVAL;
 805                 gint32 awake = 0; /* number of spurious awakes we tolerate before doing a round of rebalancing */
 806
 807                 g_assert (monitor_status != MONITOR_STATUS_NOT_RUNNING);
 808
 809                 mono_gc_set_skip_thread (TRUE);
 810
 811                 do {
 812                         guint32 ts;
 813
 814                         if (mono_runtime_is_shutting_down ())
 815                                 break;
 816
 817                         ts = mono_msec_ticks ();
 818                         if (SleepEx (interval_left, TRUE) == 0)
 819                                 break;
 820                         interval_left -= mono_msec_ticks () - ts;
 821
 822                         mono_gc_set_skip_thread (FALSE);
 823                         if ((current_thread->state & (ThreadState_StopRequested | ThreadState_SuspendRequested)) != 0)
 824                                 mono_thread_interruption_checkpoint ();
 825                         mono_gc_set_skip_thread (TRUE);
 826                 } while (interval_left > 0 && ++awake < 10);
 827
 828                 mono_gc_set_skip_thread (FALSE);
 829
 830                 if (threadpool->suspended)
 831                         continue;
 832
 833                 if (mono_runtime_is_shutting_down () || !domain_any_has_request ())
 834                         continue;
 835
 836                 mono_mutex_lock (&threadpool->active_threads_lock);
 837                 for (i = 0; i < threadpool->working_threads->len; ++i) {
 838                         thread = g_ptr_array_index (threadpool->working_threads, i);
 839                         if ((thread->state & ThreadState_WaitSleepJoin) == 0) {
 840                                 all_waitsleepjoin = FALSE;
 841                                 break;
 842                         }
 843                 }
 844                 mono_mutex_unlock (&threadpool->active_threads_lock);
 845
 846                 if (all_waitsleepjoin) {
 847                         ThreadPoolCounter counter;
 848                         COUNTER_ATOMIC (counter, { counter._.max_working ++; });
 849                         hill_climbing_force_change (counter._.max_working, TRANSITION_STARVATION);
 850                 }
 851
 852                 threadpool->cpu_usage = mono_cpu_usage (threadpool->cpu_usage_state);
 853
 854                 if (monitor_sufficient_delay_since_last_dequeue ()) {
 855                         for (i = 0; i < 5; ++i) {
 856                                 if (mono_runtime_is_shutting_down ())
 857                                         break;
 858
 859                                 if (worker_try_unpark ())
 860                                         break;
 861
 862                                 if (worker_try_create ())
 863                                         break;
 864                         }
 865                 }
 866         } while (monitor_should_keep_running ());
 867 }
 868
 869 static void
 870 monitor_ensure_running (void)
 871 {
 872         for (;;) {
 873                 switch (monitor_status) {
 874                 case MONITOR_STATUS_REQUESTED:
 875                         return;
 876                 case MONITOR_STATUS_WAITING_FOR_REQUEST:
 877                         InterlockedCompareExchange (&monitor_status, MONITOR_STATUS_REQUESTED, MONITOR_STATUS_WAITING_FOR_REQUEST);
 878                         break;
 879                 case MONITOR_STATUS_NOT_RUNNING:
 880                         if (mono_runtime_is_shutting_down ())
 881                                 return;
 882                         if (InterlockedCompareExchange (&monitor_status, MONITOR_STATUS_REQUESTED, MONITOR_STATUS_NOT_RUNNING) == MONITOR_STATUS_NOT_RUNNING) {
 883                                 if (!mono_thread_create_internal (mono_get_root_domain (), monitor_thread, NULL, TRUE, SMALL_STACK))
 884                                         monitor_status = MONITOR_STATUS_NOT_RUNNING;
 885                                 return;
 886                         }
 887                         break;
 888                 default: g_assert_not_reached ();
 889                 }
 890         }
 891 }
 892
 893 static void
 894 hill_climbing_change_thread_count (gint16 new_thread_count, ThreadPoolHeuristicStateTransition transition)
 895 {
 896         ThreadPoolHillClimbing *hc;
 897
 898         g_assert (threadpool);
 899
 900         hc = &threadpool->heuristic_hill_climbing;
 901
 902         hc->last_thread_count = new_thread_count;
 903         hc->current_sample_interval = rand_next (&hc->random_interval_generator, hc->sample_interval_low, hc->sample_interval_high);
 904         hc->elapsed_since_last_change = 0;
 905         hc->completions_since_last_change = 0;
 906 }
 907
 908 static void
 909 hill_climbing_force_change (gint16 new_thread_count, ThreadPoolHeuristicStateTransition transition)
 910 {
 911         ThreadPoolHillClimbing *hc;
 912
 913         g_assert (threadpool);
 914
 915         hc = &threadpool->heuristic_hill_climbing;
 916
 917         if (new_thread_count != hc->last_thread_count) {
 918                 hc->current_control_setting += new_thread_count - hc->last_thread_count;
 919                 hill_climbing_change_thread_count (new_thread_count, transition);
 920         }
 921 }
 922
 923 static double_complex
 924 hill_climbing_get_wave_component (gdouble *samples, guint sample_count, gdouble period)
 925 {
 926         ThreadPoolHillClimbing *hc;
 927         gdouble w, cosine, sine, coeff, q0, q1, q2;
 928         guint i;
 929
 930         g_assert (threadpool);
 931         g_assert (sample_count >= period);
 932         g_assert (period >= 2);
 933
 934         hc = &threadpool->heuristic_hill_climbing;
 935
 936         w = 2.0 * M_PI / period;
 937         cosine = cos (w);
 938         sine = sin (w);
 939         coeff = 2.0 * cosine;
 940         q0 = q1 = q2 = 0;
 941
 942         for (i = 0; i < sample_count; ++i) {
 943                 q0 = coeff * q1 - q2 + samples [(hc->total_samples - sample_count + i) % hc->samples_to_measure];
 944                 q2 = q1;
 945                 q1 = q0;
 946         }
 947
 948         return mono_double_complex_scalar_div (mono_double_complex_make (q1 - q2 * cosine, (q2 * sine)), ((gdouble)sample_count));
 949 }
 950
 951 static gint16
 952 hill_climbing_update (gint16 current_thread_count, guint32 sample_duration, gint32 completions, guint32 *adjustment_interval)
 953 {
 954         ThreadPoolHillClimbing *hc;
 955         ThreadPoolHeuristicStateTransition transition;
 956         gdouble throughput;
 957         gdouble throughput_error_estimate;
 958         gdouble confidence;
 959         gdouble move;
 960         gdouble gain;
 961         gint sample_index;
 962         gint sample_count;
 963         gint new_thread_wave_magnitude;
 964         gint new_thread_count;
 965         double_complex thread_wave_component;
 966         double_complex throughput_wave_component;
 967         double_complex ratio;
 968
 969         g_assert (threadpool);
 970         g_assert (adjustment_interval);
 971
 972         hc = &threadpool->heuristic_hill_climbing;
 973
 974         /* If someone changed the thread count without telling us, update our records accordingly. */
 975         if (current_thread_count != hc->last_thread_count)
 976                 hill_climbing_force_change (current_thread_count, TRANSITION_INITIALIZING);
 977
 978         /* Update the cumulative stats for this thread count */
 979         hc->elapsed_since_last_change += sample_duration;
 980         hc->completions_since_last_change += completions;
 981
 982         /* Add in any data we've already collected about this sample */
 983         sample_duration += hc->accumulated_sample_duration;
 984         completions += hc->accumulated_completion_count;
 985
 986         /* We need to make sure we're collecting reasonably accurate data. Since we're just counting the end
 987          * of each work item, we are goinng to be missing some data about what really happened during the
 988          * sample interval. The count produced by each thread includes an initial work item that may have
 989          * started well before the start of the interval, and each thread may have been running some new
 990          * work item for some time before the end of the interval, which did not yet get counted. So
 991          * our count is going to be off by +/- threadCount workitems.
 992          *
 993          * The exception is that the thread that reported to us last time definitely wasn't running any work
 994          * at that time, and the thread that's reporting now definitely isn't running a work item now. So
 995          * we really only need to consider threadCount-1 threads.
 996          *
 997          * Thus the percent error in our count is +/- (threadCount-1)/numCompletions.
 998          *
 999          * We cannot rely on the frequency-domain analysis we'll be doing later to filter out this error, because
1000          * of the way it accumulates over time. If this sample is off by, say, 33% in the negative direction,
1001          * then the next one likely will be too. The one after that will include the sum of the completions
1002          * we missed in the previous samples, and so will be 33% positive. So every three samples we'll have
1003          * two "low" samples and one "high" sample. This will appear as periodic variation right in the frequency
1004          * range we're targeting, which will not be filtered by the frequency-domain translation. */
1005         if (hc->total_samples > 0 && ((current_thread_count - 1.0) / completions) >= hc->max_sample_error) {
1006                 /* Not accurate enough yet. Let's accumulate the data so
1007                  * far, and tell the ThreadPool to collect a little more. */
1008                 hc->accumulated_sample_duration = sample_duration;
1009                 hc->accumulated_completion_count = completions;
1010                 *adjustment_interval = 10;
1011                 return current_thread_count;
1012         }
1013
1014         /* We've got enouugh data for our sample; reset our accumulators for next time. */
1015         hc->accumulated_sample_duration = 0;
1016         hc->accumulated_completion_count = 0;
1017
1018         /* Add the current thread count and throughput sample to our history. */
1019         throughput = ((gdouble) completions) / sample_duration;
1020
1021         sample_index = hc->total_samples % hc->samples_to_measure;
1022         hc->samples [sample_index] = throughput;
1023         hc->thread_counts [sample_index] = current_thread_count;
1024         hc->total_samples ++;
1025
1026         /* Set up defaults for our metrics. */
1027         thread_wave_component = mono_double_complex_make(0, 0);
1028         throughput_wave_component = mono_double_complex_make(0, 0);
1029         throughput_error_estimate = 0;
1030         ratio = mono_double_complex_make(0, 0);
1031         confidence = 0;
1032
1033         transition = TRANSITION_WARMUP;
1034
1035         /* How many samples will we use? It must be at least the three wave periods we're looking for, and it must also
1036          * be a whole multiple of the primary wave's period; otherwise the frequency we're looking for will fall between
1037          * two frequency bands in the Fourier analysis, and we won't be able to measure it accurately. */
1038         sample_count = ((gint) MIN (hc->total_samples - 1, hc->samples_to_measure) / hc->wave_period) * hc->wave_period;
1039
1040         if (sample_count > hc->wave_period) {
1041                 guint i;
1042                 gdouble average_throughput;
1043                 gdouble average_thread_count;
1044                 gdouble sample_sum = 0;
1045                 gdouble thread_sum = 0;
1046
1047                 /* Average the throughput and thread count samples, so we can scale the wave magnitudes later. */
1048                 for (i = 0; i < sample_count; ++i) {
1049                         guint j = (hc->total_samples - sample_count + i) % hc->samples_to_measure;
1050                         sample_sum += hc->samples [j];
1051                         thread_sum += hc->thread_counts [j];
1052                 }
1053
1054                 average_throughput = sample_sum / sample_count;
1055                 average_thread_count = thread_sum / sample_count;
1056
1057                 if (average_throughput > 0 && average_thread_count > 0) {
1058                         gdouble noise_for_confidence, adjacent_period_1, adjacent_period_2;
1059
1060                         /* Calculate the periods of the adjacent frequency bands we'll be using to
1061                          * measure noise levels. We want the two adjacent Fourier frequency bands. */
1062                         adjacent_period_1 = sample_count / (((gdouble) sample_count) / ((gdouble) hc->wave_period) + 1);
1063                         adjacent_period_2 = sample_count / (((gdouble) sample_count) / ((gdouble) hc->wave_period) - 1);
1064
1065                         /* Get the the three different frequency components of the throughput (scaled by average
1066                          * throughput). Our "error" estimate (the amount of noise that might be present in the
1067                          * frequency band we're really interested in) is the average of the adjacent bands. */
1068                         throughput_wave_component = mono_double_complex_scalar_div (hill_climbing_get_wave_component (hc->samples, sample_count, hc->wave_period), average_throughput);
1069                         throughput_error_estimate = cabs (mono_double_complex_scalar_div (hill_climbing_get_wave_component (hc->samples, sample_count, adjacent_period_1), average_throughput));
1070
1071                         if (adjacent_period_2 <= sample_count) {
1072                                 throughput_error_estimate = MAX (throughput_error_estimate, cabs (mono_double_complex_scalar_div (hill_climbing_get_wave_component (
1073                                         hc->samples, sample_count, adjacent_period_2), average_throughput)));
1074                         }
1075
1076                         /* Do the same for the thread counts, so we have something to compare to. We don't
1077                          * measure thread count noise, because there is none; these are exact measurements. */
1078                         thread_wave_component = mono_double_complex_scalar_div (hill_climbing_get_wave_component (hc->thread_counts, sample_count, hc->wave_period), average_thread_count);
1079
1080                         /* Update our moving average of the throughput noise. We'll use this
1081                          * later as feedback to determine the new size of the thread wave. */
1082                         if (hc->average_throughput_noise == 0) {
1083                                 hc->average_throughput_noise = throughput_error_estimate;
1084                         } else {
1085                                 hc->average_throughput_noise = (hc->throughput_error_smoothing_factor * throughput_error_estimate)
1086                                         + ((1.0 + hc->throughput_error_smoothing_factor) * hc->average_throughput_noise);
1087                         }
1088
1089                         if (cabs (thread_wave_component) > 0) {
1090                                 /* Adjust the throughput wave so it's centered around the target wave,
1091                                  * and then calculate the adjusted throughput/thread ratio. */
1092                                 ratio = mono_double_complex_div (mono_double_complex_sub (throughput_wave_component, mono_double_complex_scalar_mul(thread_wave_component, hc->target_throughput_ratio)), thread_wave_component);
1093                                 transition = TRANSITION_CLIMBING_MOVE;
1094                         } else {
1095                                 ratio = mono_double_complex_make (0, 0);
1096                                 transition = TRANSITION_STABILIZING;
1097                         }
1098
1099                         noise_for_confidence = MAX (hc->average_throughput_noise, throughput_error_estimate);
1100                         if (noise_for_confidence > 0) {
1101                                 confidence = cabs (thread_wave_component) / noise_for_confidence / hc->target_signal_to_noise_ratio;
1102                         } else {
1103                                 /* there is no noise! */
1104                                 confidence = 1.0;
1105                         }
1106                 }
1107         }
1108
1109         /* We use just the real part of the complex ratio we just calculated. If the throughput signal
1110          * is exactly in phase with the thread signal, this will be the same as taking the magnitude of
1111          * the complex move and moving that far up. If they're 180 degrees out of phase, we'll move
1112          * backward (because this indicates that our changes are having the opposite of the intended effect).
1113          * If they're 90 degrees out of phase, we won't move at all, because we can't tell wether we're
1114          * having a negative or positive effect on throughput. */
1115         move = creal (ratio);
1116         move = CLAMP (move, -1.0, 1.0);
1117
1118         /* Apply our confidence multiplier. */
1119         move *= CLAMP (confidence, -1.0, 1.0);
1120
1121         /* Now apply non-linear gain, such that values around zero are attenuated, while higher values
1122          * are enhanced. This allows us to move quickly if we're far away from the target, but more slowly
1123         * if we're getting close, giving us rapid ramp-up without wild oscillations around the target. */
1124         gain = hc->max_change_per_second * sample_duration;
1125         move = pow (fabs (move), hc->gain_exponent) * (move >= 0.0 ? 1 : -1) * gain;
1126         move = MIN (move, hc->max_change_per_sample);
1127
1128         /* If the result was positive, and CPU is > 95%, refuse the move. */
1129         if (move > 0.0 && threadpool->cpu_usage > CPU_USAGE_HIGH)
1130                 move = 0.0;
1131
1132         /* Apply the move to our control setting. */
1133         hc->current_control_setting += move;
1134
1135         /* Calculate the new thread wave magnitude, which is based on the moving average we've been keeping of the
1136          * throughput error.  This average starts at zero, so we'll start with a nice safe little wave at first. */
1137         new_thread_wave_magnitude = (gint)(0.5 + (hc->current_control_setting * hc->average_throughput_noise
1138                 * hc->target_signal_to_noise_ratio * hc->thread_magnitude_multiplier * 2.0));
1139         new_thread_wave_magnitude = CLAMP (new_thread_wave_magnitude, 1, hc->max_thread_wave_magnitude);
1140
1141         /* Make sure our control setting is within the ThreadPool's limits. */
1142         hc->current_control_setting = CLAMP (hc->current_control_setting, threadpool->limit_worker_min, threadpool->limit_worker_max - new_thread_wave_magnitude);
1143
1144         /* Calculate the new thread count (control setting + square wave). */
1145         new_thread_count = (gint)(hc->current_control_setting + new_thread_wave_magnitude * ((hc->total_samples / (hc->wave_period / 2)) % 2));
1146
1147         /* Make sure the new thread count doesn't exceed the ThreadPool's limits. */
1148         new_thread_count = CLAMP (new_thread_count, threadpool->limit_worker_min, threadpool->limit_worker_max);
1149
1150         if (new_thread_count != current_thread_count)
1151                 hill_climbing_change_thread_count (new_thread_count, transition);
1152
1153         if (creal (ratio) < 0.0 && new_thread_count == threadpool->limit_worker_min)
1154                 *adjustment_interval = (gint)(0.5 + hc->current_sample_interval * (10.0 * MAX (-1.0 * creal (ratio), 1.0)));
1155         else
1156                 *adjustment_interval = hc->current_sample_interval;
1157
1158         return new_thread_count;
1159 }
1160
1161 static void
1162 heuristic_notify_work_completed (void)
1163 {
1164         g_assert (threadpool);
1165
1166         InterlockedIncrement (&threadpool->heuristic_completions);
1167         threadpool->heuristic_last_dequeue = mono_msec_ticks ();
1168 }
1169
1170 static gboolean
1171 heuristic_should_adjust (void)
1172 {
1173         g_assert (threadpool);
1174
1175         if (threadpool->heuristic_last_dequeue > threadpool->heuristic_last_adjustment + threadpool->heuristic_adjustment_interval) {
1176                 ThreadPoolCounter counter;
1177                 counter.as_gint64 = COUNTER_READ();
1178                 if (counter._.working <= counter._.max_working)
1179                         return TRUE;
1180         }
1181
1182         return FALSE;
1183 }
1184
1185 static void
1186 heuristic_adjust (void)
1187 {
1188         g_assert (threadpool);
1189
1190         if (mono_mutex_trylock (&threadpool->heuristic_lock) == 0) {
1191                 gint32 completions = InterlockedExchange (&threadpool->heuristic_completions, 0);
1192                 guint32 sample_end = mono_msec_ticks ();
1193                 guint32 sample_duration = sample_end - threadpool->heuristic_sample_start;
1194
1195                 if (sample_duration >= threadpool->heuristic_adjustment_interval / 2) {
1196                         ThreadPoolCounter counter;
1197                         gint16 new_thread_count;
1198
1199                         counter.as_gint64 = COUNTER_READ ();
1200                         new_thread_count = hill_climbing_update (counter._.max_working, sample_duration, completions, &threadpool->heuristic_adjustment_interval);
1201
1202                         COUNTER_ATOMIC (counter, { counter._.max_working = new_thread_count; });
1203
1204                         if (new_thread_count > counter._.max_working)
1205                                 worker_request (mono_domain_get ());
1206
1207                         threadpool->heuristic_sample_start = sample_end;
1208                         threadpool->heuristic_last_adjustment = mono_msec_ticks ();
1209                 }
1210
1211                 mono_mutex_unlock (&threadpool->heuristic_lock);
1212         }
1213 }
1214
1215 void
1216 mono_threadpool_ms_cleanup (void)
1217 {
1218         #ifndef DISABLE_SOCKETS
1219                 mono_threadpool_ms_io_cleanup ();
1220         #endif
1221         ensure_cleanedup ();
1222 }
1223
1224 MonoAsyncResult *
1225 mono_threadpool_ms_begin_invoke (MonoDomain *domain, MonoObject *target, MonoMethod *method, gpointer *params)
1226 {
1227         static MonoClass *async_call_klass = NULL;
1228         MonoMethodMessage *message;
1229         MonoAsyncResult *async_result;
1230         MonoAsyncCall *async_call;
1231         MonoDelegate *async_callback = NULL;
1232         MonoObject *state = NULL;
1233
1234         if (!async_call_klass)
1235                 async_call_klass = mono_class_from_name (mono_defaults.corlib, "System", "MonoAsyncCall");
1236         g_assert (async_call_klass);
1237
1238         ensure_initialized (NULL);
1239
1240         message = mono_method_call_message_new (method, params, mono_get_delegate_invoke (method->klass), (params != NULL) ? (&async_callback) : NULL, (params != NULL) ? (&state) : NULL);
1241
1242         async_call = (MonoAsyncCall*) mono_object_new (domain, async_call_klass);
1243         MONO_OBJECT_SETREF (async_call, msg, message);
1244         MONO_OBJECT_SETREF (async_call, state, state);
1245
1246         if (async_callback) {
1247                 MONO_OBJECT_SETREF (async_call, cb_method, mono_get_delegate_invoke (((MonoObject*) async_callback)->vtable->klass));
1248                 MONO_OBJECT_SETREF (async_call, cb_target, async_callback);
1249         }
1250
1251         async_result = mono_async_result_new (domain, NULL, async_call->state, NULL, (MonoObject*) async_call);
1252         MONO_OBJECT_SETREF (async_result, async_delegate, target);
1253
1254 #ifndef DISABLE_SOCKETS
1255         if (mono_threadpool_ms_is_io (target, state))
1256                 return mono_threadpool_ms_io_add (async_result, (MonoSocketAsyncResult*) state);
1257 #endif
1258
1259         mono_threadpool_ms_enqueue_work_item (domain, (MonoObject*) async_result);
1260
1261         return async_result;
1262 }
1263
1264 MonoObject *
1265 mono_threadpool_ms_end_invoke (MonoAsyncResult *ares, MonoArray **out_args, MonoObject **exc)
1266 {
1267         MonoAsyncCall *ac;
1268
1269         g_assert (exc);
1270         g_assert (out_args);
1271
1272         *exc = NULL;
1273         *out_args = NULL;
1274
1275         /* check if already finished */
1276         mono_monitor_enter ((MonoObject*) ares);
1277
1278         if (ares->endinvoke_called) {
1279                 *exc = (MonoObject*) mono_get_exception_invalid_operation (NULL);
1280                 mono_monitor_exit ((MonoObject*) ares);
1281                 return NULL;
1282         }
1283
1284         MONO_OBJECT_SETREF (ares, endinvoke_called, 1);
1285
1286         /* wait until we are really finished */
1287         if (ares->completed) {
1288                 mono_monitor_exit ((MonoObject *) ares);
1289         } else {
1290                 gpointer wait_event;
1291                 if (ares->handle) {
1292                         wait_event = mono_wait_handle_get_handle ((MonoWaitHandle*) ares->handle);
1293                 } else {
1294                         wait_event = CreateEvent (NULL, TRUE, FALSE, NULL);
1295                         g_assert(wait_event);
1296                         MONO_OBJECT_SETREF (ares, handle, (MonoObject*) mono_wait_handle_new (mono_object_domain (ares), wait_event));
1297                 }
1298                 mono_monitor_exit ((MonoObject*) ares);
1299                 MONO_PREPARE_BLOCKING
1300                 WaitForSingleObjectEx (wait_event, INFINITE, TRUE);
1301                 MONO_FINISH_BLOCKING
1302         }
1303
1304         ac = (MonoAsyncCall*) ares->object_data;
1305         g_assert (ac);
1306
1307         *exc = ac->msg->exc; /* FIXME: GC add write barrier */
1308         *out_args = ac->out_args;
1309         return ac->res;
1310 }
1311
1312 gboolean
1313 mono_threadpool_ms_remove_domain_jobs (MonoDomain *domain, int timeout)
1314 {
1315         gboolean res = TRUE;
1316         guint32 start;
1317         gpointer sem;
1318
1319         g_assert (domain);
1320         g_assert (timeout >= -1);
1321
1322         g_assert (mono_domain_is_unloading (domain));
1323
1324         if (timeout != -1)
1325                 start = mono_msec_ticks ();
1326
1327 #ifndef DISABLE_SOCKETS
1328         mono_threadpool_ms_io_remove_domain_jobs (domain);
1329         if (timeout != -1) {
1330                 timeout -= mono_msec_ticks () - start;
1331                 if (timeout < 0)
1332                         return FALSE;
1333         }
1334 #endif
1335
1336         /*
1337          * There might be some threads out that could be about to execute stuff from the given domain.
1338          * We avoid that by setting up a semaphore to be pulsed by the thread that reaches zero.
1339          */
1340         sem = domain->cleanup_semaphore = CreateSemaphore (NULL, 0, 1, NULL);
1341
1342         /*
1343          * The memory barrier here is required to have global ordering between assigning to cleanup_semaphone
1344          * and reading threadpool_jobs. Otherwise this thread could read a stale version of threadpool_jobs
1345          * and wait forever.
1346          */
1347         mono_memory_write_barrier ();
1348
1349         while (domain->threadpool_jobs) {
1350                 MONO_PREPARE_BLOCKING
1351                 WaitForSingleObject (sem, timeout);
1352                 MONO_FINISH_BLOCKING
1353                 if (timeout != -1) {
1354                         timeout -= mono_msec_ticks () - start;
1355                         if (timeout <= 0) {
1356                                 res = FALSE;
1357                                 break;
1358                         }
1359                 }
1360         }
1361
1362         domain->cleanup_semaphore = NULL;
1363         CloseHandle (sem);
1364
1365         return res;
1366 }
1367
1368 void
1369 mono_threadpool_ms_suspend (void)
1370 {
1371         if (threadpool)
1372                 threadpool->suspended = TRUE;
1373 }
1374
1375 void
1376 mono_threadpool_ms_resume (void)
1377 {
1378         if (threadpool)
1379                 threadpool->suspended = FALSE;
1380 }
1381
1382 void
1383 ves_icall_System_Threading_ThreadPool_GetAvailableThreadsNative (gint32 *worker_threads, gint32 *completion_port_threads)
1384 {
1385         if (!worker_threads || !completion_port_threads)
1386                 return;
1387
1388         ensure_initialized (NULL);
1389
1390         *worker_threads = threadpool->limit_worker_max;
1391         *completion_port_threads = threadpool->limit_io_max;
1392 }
1393
1394 void
1395 ves_icall_System_Threading_ThreadPool_GetMinThreadsNative (gint32 *worker_threads, gint32 *completion_port_threads)
1396 {
1397         if (!worker_threads || !completion_port_threads)
1398                 return;
1399
1400         ensure_initialized (NULL);
1401
1402         *worker_threads = threadpool->limit_worker_min;
1403         *completion_port_threads = threadpool->limit_io_min;
1404 }
1405
1406 void
1407 ves_icall_System_Threading_ThreadPool_GetMaxThreadsNative (gint32 *worker_threads, gint32 *completion_port_threads)
1408 {
1409         if (!worker_threads || !completion_port_threads)
1410                 return;
1411
1412         ensure_initialized (NULL);
1413
1414         *worker_threads = threadpool->limit_worker_max;
1415         *completion_port_threads = threadpool->limit_io_max;
1416 }
1417
1418 MonoBoolean
1419 ves_icall_System_Threading_ThreadPool_SetMinThreadsNative (gint32 worker_threads, gint32 completion_port_threads)
1420 {
1421         ensure_initialized (NULL);
1422
1423         if (worker_threads <= 0 || worker_threads > threadpool->limit_worker_max)
1424                 return FALSE;
1425         if (completion_port_threads <= 0 || completion_port_threads > threadpool->limit_io_max)
1426                 return FALSE;
1427
1428         threadpool->limit_worker_max = worker_threads;
1429         threadpool->limit_io_max = completion_port_threads;
1430
1431         return TRUE;
1432 }
1433
1434 MonoBoolean
1435 ves_icall_System_Threading_ThreadPool_SetMaxThreadsNative (gint32 worker_threads, gint32 completion_port_threads)
1436 {
1437         gint cpu_count = mono_cpu_count ();
1438
1439         ensure_initialized (NULL);
1440
1441         if (worker_threads < threadpool->limit_worker_min || worker_threads < cpu_count)
1442                 return FALSE;
1443         if (completion_port_threads < threadpool->limit_io_min || completion_port_threads < cpu_count)
1444                 return FALSE;
1445
1446         threadpool->limit_worker_max = worker_threads;
1447         threadpool->limit_io_max = completion_port_threads;
1448
1449         return TRUE;
1450 }
1451
1452 void
1453 ves_icall_System_Threading_ThreadPool_InitializeVMTp (MonoBoolean *enable_worker_tracking)
1454 {
1455         ensure_initialized (enable_worker_tracking);
1456 }
1457
1458 MonoBoolean
1459 ves_icall_System_Threading_ThreadPool_NotifyWorkItemComplete (void)
1460 {
1461         ThreadPoolCounter counter;
1462
1463         if (mono_domain_is_unloading (mono_domain_get ()) || mono_runtime_is_shutting_down ())
1464                 return FALSE;
1465
1466         heuristic_notify_work_completed ();
1467
1468         if (heuristic_should_adjust ())
1469                 heuristic_adjust ();
1470
1471         counter.as_gint64 = COUNTER_READ ();
1472         return counter._.working <= counter._.max_working;
1473 }
1474
1475 void
1476 ves_icall_System_Threading_ThreadPool_NotifyWorkItemProgressNative (void)
1477 {
1478         heuristic_notify_work_completed ();
1479
1480         if (heuristic_should_adjust ())
1481                 heuristic_adjust ();
1482 }
1483
1484 void
1485 ves_icall_System_Threading_ThreadPool_ReportThreadStatus (MonoBoolean is_working)
1486 {
1487         // TODO
1488         mono_raise_exception (mono_get_exception_not_implemented (NULL));
1489 }
1490
1491 MonoBoolean
1492 ves_icall_System_Threading_ThreadPool_RequestWorkerThread (void)
1493 {
1494         return worker_request (mono_domain_get ());
1495 }
1496
1497 MonoBoolean G_GNUC_UNUSED
1498 ves_icall_System_Threading_ThreadPool_PostQueuedCompletionStatus (MonoNativeOverlapped *native_overlapped)
1499 {
1500         /* This copy the behavior of the current Mono implementation */
1501         mono_raise_exception (mono_get_exception_not_implemented (NULL));
1502         return FALSE;
1503 }
1504
1505 MonoBoolean G_GNUC_UNUSED
1506 ves_icall_System_Threading_ThreadPool_BindIOCompletionCallbackNative (gpointer file_handle)
1507 {
1508         /* This copy the behavior of the current Mono implementation */
1509         return TRUE;
1510 }
1511
1512 MonoBoolean G_GNUC_UNUSED
1513 ves_icall_System_Threading_ThreadPool_IsThreadPoolHosted (void)
1514 {
1515         return FALSE;
1516 }