3 * Microsoft threadpool runtime support
6 * Ludovic Henry (ludovic.henry@xamarin.com)
8 * Copyright 2015 Xamarin, Inc (http://www.xamarin.com)
9 * Licensed under the MIT license. See LICENSE file in the project root for full license information.
13 // Copyright (c) Microsoft. All rights reserved.
14 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
17 // - src/vm/comthreadpool.cpp
18 // - src/vm/win32threadpoolcpp
19 // - src/vm/threadpoolrequest.cpp
20 // - src/vm/hillclimbing.cpp
22 // Ported from C++ to C and adjusted to Mono runtime
25 #define _USE_MATH_DEFINES // needed by MSVC to define math constants
30 #include <mono/metadata/class-internals.h>
31 #include <mono/metadata/domain-internals.h>
32 #include <mono/metadata/exception.h>
33 #include <mono/metadata/gc-internals.h>
34 #include <mono/metadata/object.h>
35 #include <mono/metadata/object-internals.h>
36 #include <mono/metadata/threadpool.h>
37 #include <mono/metadata/threadpool-worker.h>
38 #include <mono/metadata/threadpool-io.h>
39 #include <mono/metadata/w32event.h>
40 #include <mono/utils/atomic.h>
41 #include <mono/utils/mono-compiler.h>
42 #include <mono/utils/mono-complex.h>
43 #include <mono/utils/mono-lazy-init.h>
44 #include <mono/utils/mono-logger.h>
45 #include <mono/utils/mono-logger-internals.h>
46 #include <mono/utils/mono-proclib.h>
47 #include <mono/utils/mono-threads.h>
48 #include <mono/utils/mono-time.h>
49 #include <mono/utils/refcount.h>
50 #include <mono/utils/mono-os-wait.h>
52 #include "icall-decl.h"
54 // consistency with coreclr https://github.com/dotnet/coreclr/blob/643b09f966e68e06d5f0930755985a01a2a2b096/src/vm/win32threadpool.h#L111
55 #define MAX_POSSIBLE_THREADS 0x7fff
59 /* Number of outstanding jobs */
60 gint32 outstanding_request
;
61 /* Number of currently executing jobs */
62 gint32 threadpool_jobs
;
63 /* Signalled when threadpool_jobs + outstanding_request is 0 */
64 /* Protected by threadpool.domains_lock */
65 MonoCoopCond cleanup_cond
;
70 gint16 starting
; /* starting, but not yet in worker_callback */
71 gint16 working
; /* executing worker_callback */
79 GPtrArray
*domains
; // ThreadPoolDomain* []
80 MonoCoopMutex domains_lock
;
82 ThreadPoolCounter counters
;
88 static mono_lazy_init_t status
= MONO_LAZY_INIT_STATUS_NOT_INITIALIZED
;
90 static ThreadPool threadpool
;
92 #define COUNTER_ATOMIC(var,block) \
94 ThreadPoolCounter __old; \
96 (var) = __old = COUNTER_READ (); \
98 if (!(counter._.starting >= 0)) \
99 g_error ("%s: counter._.starting = %d, but should be >= 0", __func__, counter._.starting); \
100 if (!(counter._.working >= 0)) \
101 g_error ("%s: counter._.working = %d, but should be >= 0", __func__, counter._.working); \
102 } while (mono_atomic_cas_i32 (&threadpool.counters.as_gint32, (var).as_gint32, __old.as_gint32) != __old.as_gint32); \
105 static inline ThreadPoolCounter
108 ThreadPoolCounter counter
;
109 counter
.as_gint32
= mono_atomic_load_i32 (&threadpool
.counters
.as_gint32
);
116 mono_coop_mutex_lock (&threadpool
.domains_lock
);
120 domains_unlock (void)
122 mono_coop_mutex_unlock (&threadpool
.domains_lock
);
126 destroy (gpointer unused
)
128 g_ptr_array_free (threadpool
.domains
, TRUE
);
129 mono_coop_mutex_destroy (&threadpool
.domains_lock
);
133 worker_callback (void);
138 g_assert (sizeof (ThreadPoolCounter
) == sizeof (gint32
));
140 mono_refcount_init (&threadpool
, destroy
);
142 threadpool
.domains
= g_ptr_array_new ();
143 mono_coop_mutex_init (&threadpool
.domains_lock
);
145 threadpool
.limit_io_min
= mono_cpu_count ();
146 threadpool
.limit_io_max
= CLAMP (threadpool
.limit_io_min
* 100, MIN (threadpool
.limit_io_min
, 200), MAX (threadpool
.limit_io_min
, 200));
148 mono_threadpool_worker_init (worker_callback
);
154 mono_threadpool_worker_cleanup ();
156 mono_refcount_dec (&threadpool
);
160 mono_threadpool_enqueue_work_item (MonoDomain
*domain
, MonoObject
*work_item
, MonoError
*error
)
162 static MonoClass
*threadpool_class
= NULL
;
163 static MonoMethod
*unsafe_queue_custom_work_item_method
= NULL
;
164 MonoDomain
*current_domain
;
169 g_assert (work_item
);
171 if (!threadpool_class
)
172 threadpool_class
= mono_class_load_from_name (mono_defaults
.corlib
, "System.Threading", "ThreadPool");
174 if (!unsafe_queue_custom_work_item_method
) {
175 unsafe_queue_custom_work_item_method
= mono_class_get_method_from_name_checked (threadpool_class
, "UnsafeQueueCustomWorkItem", 2, 0, error
);
176 mono_error_assert_ok (error
);
178 g_assert (unsafe_queue_custom_work_item_method
);
182 args
[0] = (gpointer
) work_item
;
183 args
[1] = (gpointer
) &f
;
185 current_domain
= mono_domain_get ();
186 if (current_domain
== domain
) {
187 mono_runtime_invoke_checked (unsafe_queue_custom_work_item_method
, NULL
, args
, error
);
189 mono_thread_push_appdomain_ref (domain
);
190 if (mono_domain_set_fast (domain
, FALSE
)) {
191 mono_runtime_invoke_checked (unsafe_queue_custom_work_item_method
, NULL
, args
, error
);
192 mono_domain_set_fast (current_domain
, TRUE
);
194 // mono_domain_set_fast failing still leads to success.
196 mono_thread_pop_appdomain_ref ();
198 return is_ok (error
);
201 /* LOCKING: domains_lock must be held. */
202 static ThreadPoolDomain
*
203 tpdomain_create (MonoDomain
*domain
)
205 ThreadPoolDomain
*tpdomain
;
207 tpdomain
= g_new0 (ThreadPoolDomain
, 1);
208 tpdomain
->domain
= domain
;
209 mono_coop_cond_init (&tpdomain
->cleanup_cond
);
211 g_ptr_array_add (threadpool
.domains
, tpdomain
);
216 /* LOCKING: domains_lock must be held. */
218 tpdomain_remove (ThreadPoolDomain
*tpdomain
)
221 return g_ptr_array_remove (threadpool
.domains
, tpdomain
);
224 /* LOCKING: domains_lock must be held */
225 static ThreadPoolDomain
*
226 tpdomain_get (MonoDomain
*domain
)
232 for (i
= 0; i
< threadpool
.domains
->len
; ++i
) {
233 ThreadPoolDomain
*tpdomain
;
235 tpdomain
= (ThreadPoolDomain
*)g_ptr_array_index (threadpool
.domains
, i
);
236 if (tpdomain
->domain
== domain
)
244 tpdomain_free (ThreadPoolDomain
*tpdomain
)
249 /* LOCKING: domains_lock must be held */
250 static ThreadPoolDomain
*
251 tpdomain_get_next (ThreadPoolDomain
*current
)
253 ThreadPoolDomain
*tpdomain
= NULL
;
256 len
= threadpool
.domains
->len
;
258 gint i
, current_idx
= -1;
260 for (i
= 0; i
< len
; ++i
) {
261 if (current
== g_ptr_array_index (threadpool
.domains
, i
)) {
267 for (i
= current_idx
+ 1; i
< len
+ current_idx
+ 1; ++i
) {
268 ThreadPoolDomain
*tmp
= (ThreadPoolDomain
*)g_ptr_array_index (threadpool
.domains
, i
% len
);
269 if (tmp
->outstanding_request
> 0) {
280 try_invoke_perform_wait_callback (MonoObject
** exc
, MonoError
*error
)
282 HANDLE_FUNCTION_ENTER ();
284 MonoObject
*res
= mono_runtime_try_invoke (mono_defaults
.threadpool_perform_wait_callback_method
, NULL
, NULL
, exc
, error
);
285 HANDLE_FUNCTION_RETURN_VAL (res
);
289 worker_callback (void)
292 ThreadPoolDomain
*tpdomain
, *previous_tpdomain
;
293 ThreadPoolCounter counter
;
294 MonoInternalThread
*thread
;
296 if (!mono_refcount_tryinc (&threadpool
))
299 thread
= mono_thread_internal_current ();
301 COUNTER_ATOMIC (counter
, {
302 if (!(counter
._
.working
< 32767 /* G_MAXINT16 */))
303 g_error ("%s: counter._.working = %d, but should be < 32767", __func__
, counter
._
.working
);
305 counter
._
.starting
--;
306 counter
._
.working
++;
309 if (mono_runtime_is_shutting_down ()) {
310 COUNTER_ATOMIC (counter
, {
311 counter
._
.working
--;
314 mono_refcount_dec (&threadpool
);
319 * This is needed so there is always an lmf frame in the runtime invoke call below,
320 * so ThreadAbortExceptions are caught even if the thread is in native code.
322 mono_defaults
.threadpool_perform_wait_callback_method
->save_lmf
= TRUE
;
326 previous_tpdomain
= NULL
;
328 while (!mono_runtime_is_shutting_down ()) {
329 gboolean retire
= FALSE
;
331 if (thread
->state
& (ThreadState_AbortRequested
| ThreadState_SuspendRequested
)) {
333 if (mono_thread_interruption_checkpoint_bool ()) {
340 tpdomain
= tpdomain_get_next (previous_tpdomain
);
344 tpdomain
->outstanding_request
--;
345 g_assert (tpdomain
->outstanding_request
>= 0);
347 mono_trace (G_LOG_LEVEL_DEBUG
, MONO_TRACE_THREADPOOL
, "[%p] worker running in domain %p (outstanding requests %d)",
348 GUINT_TO_POINTER (MONO_NATIVE_THREAD_ID_TO_UINT (mono_native_thread_id_get ())), tpdomain
->domain
, tpdomain
->outstanding_request
);
350 g_assert (tpdomain
->threadpool_jobs
>= 0);
351 tpdomain
->threadpool_jobs
++;
355 MonoString
*thread_name
= mono_string_new_checked (mono_get_root_domain (), "Thread Pool Worker", error
);
356 mono_error_assert_ok (error
);
357 mono_thread_set_name_internal (thread
, thread_name
, FALSE
, TRUE
, error
);
358 mono_error_assert_ok (error
);
360 mono_thread_clear_and_set_state (thread
,
361 (MonoThreadState
)~ThreadState_Background
,
362 ThreadState_Background
);
364 mono_thread_push_appdomain_ref (tpdomain
->domain
);
365 if (mono_domain_set_fast (tpdomain
->domain
, FALSE
)) {
366 MonoObject
*exc
= NULL
, *res
;
368 res
= try_invoke_perform_wait_callback (&exc
, error
);
369 if (exc
|| !mono_error_ok(error
)) {
371 exc
= (MonoObject
*) mono_error_convert_to_exception (error
);
373 mono_error_cleanup (error
);
374 mono_thread_internal_unhandled_exception (exc
);
375 } else if (res
&& *(MonoBoolean
*) mono_object_unbox_internal (res
) == FALSE
) {
379 mono_domain_set_fast (mono_get_root_domain (), TRUE
);
381 mono_thread_pop_appdomain_ref ();
385 tpdomain
->threadpool_jobs
--;
386 g_assert (tpdomain
->threadpool_jobs
>= 0);
388 if (tpdomain
->outstanding_request
+ tpdomain
->threadpool_jobs
== 0 && mono_domain_is_unloading (tpdomain
->domain
)) {
391 removed
= tpdomain_remove (tpdomain
);
394 mono_coop_cond_signal (&tpdomain
->cleanup_cond
);
401 previous_tpdomain
= tpdomain
;
406 COUNTER_ATOMIC (counter
, {
407 counter
._
.working
--;
410 mono_refcount_dec (&threadpool
);
414 mono_threadpool_cleanup (void)
416 #ifndef DISABLE_SOCKETS
417 mono_threadpool_io_cleanup ();
419 mono_lazy_cleanup (&status
, cleanup
);
423 mono_threadpool_begin_invoke (MonoDomain
*domain
, MonoObject
*target
, MonoMethod
*method
, gpointer
*params
, MonoError
*error
)
425 static MonoClass
*async_call_klass
= NULL
;
426 MonoMethodMessage
*message
;
427 MonoAsyncResult
*async_result
;
428 MonoAsyncCall
*async_call
;
429 MonoDelegate
*async_callback
= NULL
;
430 MonoObject
*state
= NULL
;
432 if (!async_call_klass
)
433 async_call_klass
= mono_class_load_from_name (mono_defaults
.corlib
, "System", "MonoAsyncCall");
437 message
= mono_method_call_message_new (method
, params
, mono_get_delegate_invoke_internal (method
->klass
), (params
!= NULL
) ? (&async_callback
) : NULL
, (params
!= NULL
) ? (&state
) : NULL
, error
);
438 return_val_if_nok (error
, NULL
);
440 async_call
= (MonoAsyncCall
*) mono_object_new_checked (domain
, async_call_klass
, error
);
441 return_val_if_nok (error
, NULL
);
443 MONO_OBJECT_SETREF_INTERNAL (async_call
, msg
, message
);
444 MONO_OBJECT_SETREF_INTERNAL (async_call
, state
, state
);
446 if (async_callback
) {
447 MONO_OBJECT_SETREF_INTERNAL (async_call
, cb_method
, mono_get_delegate_invoke_internal (((MonoObject
*) async_callback
)->vtable
->klass
));
448 MONO_OBJECT_SETREF_INTERNAL (async_call
, cb_target
, async_callback
);
451 async_result
= mono_async_result_new (domain
, NULL
, async_call
->state
, NULL
, (MonoObject
*) async_call
, error
);
452 return_val_if_nok (error
, NULL
);
453 MONO_OBJECT_SETREF_INTERNAL (async_result
, async_delegate
, target
);
455 mono_threadpool_enqueue_work_item (domain
, (MonoObject
*) async_result
, error
);
456 return_val_if_nok (error
, NULL
);
462 mono_threadpool_end_invoke (MonoAsyncResult
*ares
, MonoArray
**out_args
, MonoObject
**exc
, MonoError
*error
)
473 /* check if already finished */
474 mono_monitor_enter_internal ((MonoObject
*) ares
);
476 if (ares
->endinvoke_called
) {
477 mono_error_set_invalid_operation(error
, "Delegate EndInvoke method called more than once");
478 mono_monitor_exit_internal ((MonoObject
*) ares
);
482 ares
->endinvoke_called
= 1;
484 /* wait until we are really finished */
485 if (ares
->completed
) {
486 mono_monitor_exit_internal ((MonoObject
*) ares
);
490 wait_event
= mono_wait_handle_get_handle ((MonoWaitHandle
*) ares
->handle
);
492 wait_event
= mono_w32event_create (TRUE
, FALSE
);
493 g_assert(wait_event
);
494 MonoWaitHandle
*wait_handle
= mono_wait_handle_new (mono_object_domain (ares
), wait_event
, error
);
495 if (!is_ok (error
)) {
496 mono_w32event_close (wait_event
);
499 MONO_OBJECT_SETREF_INTERNAL (ares
, handle
, (MonoObject
*) wait_handle
);
501 mono_monitor_exit_internal ((MonoObject
*) ares
);
502 mono_w32handle_wait_one (wait_event
, MONO_INFINITE_WAIT
, TRUE
);
505 ac
= (MonoAsyncCall
*) ares
->object_data
;
508 *exc
= ac
->msg
->exc
; /* FIXME: GC add write barrier */
509 *out_args
= ac
->out_args
;
514 mono_threadpool_remove_domain_jobs (MonoDomain
*domain
, int timeout
)
517 ThreadPoolDomain
*tpdomain
;
521 g_assert (timeout
>= -1);
523 g_assert (mono_domain_is_unloading (domain
));
526 end
= mono_msec_ticks () + timeout
;
528 #ifndef DISABLE_SOCKETS
529 mono_threadpool_io_remove_domain_jobs (domain
);
531 if (mono_msec_ticks () > end
)
537 * Wait for all threads which execute jobs in the domain to exit.
538 * The is_unloading () check in worker_request () ensures that
539 * no new jobs are added after we enter the lock below.
542 if (!mono_lazy_is_initialized (&status
))
545 mono_refcount_inc (&threadpool
);
549 tpdomain
= tpdomain_get (domain
);
552 mono_refcount_dec (&threadpool
);
558 while (tpdomain
->outstanding_request
+ tpdomain
->threadpool_jobs
> 0) {
560 mono_coop_cond_wait (&tpdomain
->cleanup_cond
, &threadpool
.domains_lock
);
565 now
= mono_msec_ticks();
571 res
= mono_coop_cond_timedwait (&tpdomain
->cleanup_cond
, &threadpool
.domains_lock
, end
- now
);
579 /* Remove from the list the worker threads look at */
580 tpdomain_remove (tpdomain
);
584 mono_coop_cond_destroy (&tpdomain
->cleanup_cond
);
585 tpdomain_free (tpdomain
);
587 mono_refcount_dec (&threadpool
);
593 mono_threadpool_suspend (void)
595 if (mono_lazy_is_initialized (&status
))
596 mono_threadpool_worker_set_suspended (TRUE
);
600 mono_threadpool_resume (void)
602 if (mono_lazy_is_initialized (&status
))
603 mono_threadpool_worker_set_suspended (FALSE
);
607 ves_icall_System_Threading_ThreadPool_GetAvailableThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
609 ThreadPoolCounter counter
;
611 if (!worker_threads
|| !completion_port_threads
)
614 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
616 *completion_port_threads
= 0;
620 counter
= COUNTER_READ ();
622 *worker_threads
= MAX (0, mono_threadpool_worker_get_max () - counter
._
.working
);
623 *completion_port_threads
= threadpool
.limit_io_max
;
625 mono_refcount_dec (&threadpool
);
629 ves_icall_System_Threading_ThreadPool_GetMinThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
631 if (!worker_threads
|| !completion_port_threads
)
634 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
636 *completion_port_threads
= 0;
640 *worker_threads
= mono_threadpool_worker_get_min ();
641 *completion_port_threads
= threadpool
.limit_io_min
;
643 mono_refcount_dec (&threadpool
);
647 ves_icall_System_Threading_ThreadPool_GetMaxThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
649 if (!worker_threads
|| !completion_port_threads
)
652 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
654 *completion_port_threads
= 0;
658 *worker_threads
= mono_threadpool_worker_get_max ();
659 *completion_port_threads
= threadpool
.limit_io_max
;
661 mono_refcount_dec (&threadpool
);
665 ves_icall_System_Threading_ThreadPool_SetMinThreadsNative (gint32 worker_threads
, gint32 completion_port_threads
, MonoError
*error
)
667 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
))
670 if (completion_port_threads
<= 0 || completion_port_threads
> threadpool
.limit_io_max
)
673 if (!mono_threadpool_worker_set_min (worker_threads
)) {
674 mono_refcount_dec (&threadpool
);
678 threadpool
.limit_io_min
= completion_port_threads
;
680 mono_refcount_dec (&threadpool
);
685 ves_icall_System_Threading_ThreadPool_SetMaxThreadsNative (gint32 worker_threads
, gint32 completion_port_threads
, MonoError
*error
)
687 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
))
690 worker_threads
= MIN (worker_threads
, MAX_POSSIBLE_THREADS
);
691 completion_port_threads
= MIN (completion_port_threads
, MAX_POSSIBLE_THREADS
);
693 gint cpu_count
= mono_cpu_count ();
695 if (completion_port_threads
< threadpool
.limit_io_min
|| completion_port_threads
< cpu_count
)
698 if (!mono_threadpool_worker_set_max (worker_threads
)) {
699 mono_refcount_dec (&threadpool
);
703 threadpool
.limit_io_max
= completion_port_threads
;
705 mono_refcount_dec (&threadpool
);
710 ves_icall_System_Threading_ThreadPool_InitializeVMTp (MonoBoolean
*enable_worker_tracking
, MonoError
*error
)
712 if (enable_worker_tracking
) {
713 // TODO implement some kind of switch to have the possibily to use it
714 *enable_worker_tracking
= FALSE
;
717 mono_lazy_initialize (&status
, initialize
);
721 ves_icall_System_Threading_ThreadPool_NotifyWorkItemComplete (MonoError
*error
)
723 if (mono_domain_is_unloading (mono_domain_get ()) || mono_runtime_is_shutting_down ())
726 return mono_threadpool_worker_notify_completed ();
730 ves_icall_System_Threading_ThreadPool_NotifyWorkItemProgressNative (MonoError
*error
)
732 mono_threadpool_worker_notify_completed ();
736 ves_icall_System_Threading_ThreadPool_NotifyWorkItemQueued (MonoError
*error
)
737 // FIXME Move to managed.
739 #ifndef DISABLE_PERFCOUNTERS
740 mono_atomic_inc_i64 (&mono_perfcounters
->threadpool_workitems
);
745 ves_icall_System_Threading_ThreadPool_ReportThreadStatus (MonoBoolean is_working
, MonoError
*error
)
748 mono_error_set_not_implemented (error
, "");
752 ves_icall_System_Threading_ThreadPool_RequestWorkerThread (MonoError
*error
)
755 ThreadPoolDomain
*tpdomain
;
756 ThreadPoolCounter counter
;
758 domain
= mono_domain_get ();
759 if (mono_domain_is_unloading (domain
))
762 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
763 /* threadpool has been destroyed, we are shutting down */
769 tpdomain
= tpdomain_get (domain
);
771 /* synchronize with mono_threadpool_remove_domain_jobs */
772 if (mono_domain_is_unloading (domain
)) {
774 mono_refcount_dec (&threadpool
);
778 tpdomain
= tpdomain_create (domain
);
783 tpdomain
->outstanding_request
++;
784 g_assert (tpdomain
->outstanding_request
>= 1);
788 COUNTER_ATOMIC (counter
, {
789 if (counter
._
.starting
== 16) {
790 mono_refcount_dec (&threadpool
);
794 counter
._
.starting
++;
797 mono_threadpool_worker_request ();
799 mono_refcount_dec (&threadpool
);