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