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 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
* const res
= mono_runtime_try_invoke (mono_defaults
.threadpool_perform_wait_callback_method
, NULL
, NULL
, exc
, error
);
285 HANDLE_FUNCTION_RETURN_VAL (res
);
289 set_thread_name (MonoInternalThread
*thread
)
291 return mono_thread_set_name_constant_ignore_error (thread
, "Thread Pool Worker", MonoSetThreadNameFlag_Reset
);
295 worker_callback (void)
297 ThreadPoolDomain
*tpdomain
, *previous_tpdomain
;
298 ThreadPoolCounter counter
;
299 MonoInternalThread
*thread
;
301 if (!mono_refcount_tryinc (&threadpool
))
304 thread
= mono_thread_internal_current ();
306 COUNTER_ATOMIC (counter
, {
307 if (!(counter
._
.working
< 32767 /* G_MAXINT16 */))
308 g_error ("%s: counter._.working = %d, but should be < 32767", __func__
, counter
._
.working
);
310 counter
._
.starting
--;
311 counter
._
.working
++;
314 if (mono_runtime_is_shutting_down ()) {
315 COUNTER_ATOMIC (counter
, {
316 counter
._
.working
--;
319 mono_refcount_dec (&threadpool
);
324 * This is needed so there is always an lmf frame in the runtime invoke call below,
325 * so ThreadAbortExceptions are caught even if the thread is in native code.
327 mono_defaults
.threadpool_perform_wait_callback_method
->save_lmf
= TRUE
;
329 gsize name_generation
= thread
->name
.generation
;
330 /* Set the name if this is the first call to worker_callback on this thread */
331 if (name_generation
== 0)
332 name_generation
= set_thread_name (thread
);
336 previous_tpdomain
= NULL
;
338 while (!mono_runtime_is_shutting_down ()) {
339 gboolean retire
= FALSE
;
341 if (thread
->state
& (ThreadState_AbortRequested
| ThreadState_SuspendRequested
)) {
343 if (mono_thread_interruption_checkpoint_bool ()) {
350 tpdomain
= tpdomain_get_next (previous_tpdomain
);
354 tpdomain
->outstanding_request
--;
355 g_assert (tpdomain
->outstanding_request
>= 0);
357 mono_trace (G_LOG_LEVEL_DEBUG
, MONO_TRACE_THREADPOOL
, "[%p] worker running in domain %p (outstanding requests %d)",
358 GUINT_TO_POINTER (MONO_NATIVE_THREAD_ID_TO_UINT (mono_native_thread_id_get ())), tpdomain
->domain
, tpdomain
->outstanding_request
);
360 g_assert (tpdomain
->threadpool_jobs
>= 0);
361 tpdomain
->threadpool_jobs
++;
365 // Any thread can set any other thread name at any time.
366 // So this is unavoidably racy.
367 // This only partly fights against that -- i.e. not atomic and not a loop.
368 // It is reliable against the thread setting its own name, and somewhat
369 // reliable against other threads setting this thread's name.
370 if (name_generation
!= thread
->name
.generation
)
371 name_generation
= set_thread_name (thread
);
373 mono_thread_clear_and_set_state (thread
,
374 (MonoThreadState
)~ThreadState_Background
,
375 ThreadState_Background
);
377 mono_thread_push_appdomain_ref (tpdomain
->domain
);
378 if (mono_domain_set_fast (tpdomain
->domain
, FALSE
)) {
379 MonoObject
*exc
= NULL
, *res
;
383 res
= try_invoke_perform_wait_callback (&exc
, error
);
384 if (exc
|| !is_ok(error
)) {
386 exc
= (MonoObject
*) mono_error_convert_to_exception (error
);
388 mono_error_cleanup (error
);
389 mono_thread_internal_unhandled_exception (exc
);
390 } else if (res
&& *(MonoBoolean
*) mono_object_unbox_internal (res
) == FALSE
) {
394 mono_domain_set_fast (mono_get_root_domain (), TRUE
);
396 mono_thread_pop_appdomain_ref ();
398 /* Reset name after every callback */
399 if (name_generation
!= thread
->name
.generation
)
400 name_generation
= set_thread_name (thread
);
404 tpdomain
->threadpool_jobs
--;
405 g_assert (tpdomain
->threadpool_jobs
>= 0);
407 if (tpdomain
->outstanding_request
+ tpdomain
->threadpool_jobs
== 0 && mono_domain_is_unloading (tpdomain
->domain
)) {
410 removed
= tpdomain_remove (tpdomain
);
413 mono_coop_cond_signal (&tpdomain
->cleanup_cond
);
420 previous_tpdomain
= tpdomain
;
425 COUNTER_ATOMIC (counter
, {
426 counter
._
.working
--;
429 mono_refcount_dec (&threadpool
);
433 mono_threadpool_cleanup (void)
435 #ifndef DISABLE_SOCKETS
436 mono_threadpool_io_cleanup ();
438 mono_lazy_cleanup (&status
, cleanup
);
442 mono_threadpool_begin_invoke (MonoDomain
*domain
, MonoObject
*target
, MonoMethod
*method
, gpointer
*params
, MonoError
*error
)
444 static MonoClass
*async_call_klass
= NULL
;
445 MonoMethodMessage
*message
;
446 MonoAsyncResult
*async_result
;
447 MonoAsyncCall
*async_call
;
448 MonoDelegate
*async_callback
= NULL
;
449 MonoObject
*state
= NULL
;
451 if (!async_call_klass
)
452 async_call_klass
= mono_class_load_from_name (mono_defaults
.corlib
, "System", "MonoAsyncCall");
456 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
);
457 return_val_if_nok (error
, NULL
);
459 async_call
= (MonoAsyncCall
*) mono_object_new_checked (domain
, async_call_klass
, error
);
460 return_val_if_nok (error
, NULL
);
462 MONO_OBJECT_SETREF_INTERNAL (async_call
, msg
, message
);
463 MONO_OBJECT_SETREF_INTERNAL (async_call
, state
, state
);
465 if (async_callback
) {
466 MONO_OBJECT_SETREF_INTERNAL (async_call
, cb_method
, mono_get_delegate_invoke_internal (((MonoObject
*) async_callback
)->vtable
->klass
));
467 MONO_OBJECT_SETREF_INTERNAL (async_call
, cb_target
, async_callback
);
470 async_result
= mono_async_result_new (domain
, NULL
, async_call
->state
, NULL
, (MonoObject
*) async_call
, error
);
471 return_val_if_nok (error
, NULL
);
472 MONO_OBJECT_SETREF_INTERNAL (async_result
, async_delegate
, target
);
474 mono_threadpool_enqueue_work_item (domain
, (MonoObject
*) async_result
, error
);
475 return_val_if_nok (error
, NULL
);
481 mono_threadpool_end_invoke (MonoAsyncResult
*ares
, MonoArray
**out_args
, MonoObject
**exc
, MonoError
*error
)
492 /* check if already finished */
493 mono_monitor_enter_internal ((MonoObject
*) ares
);
495 if (ares
->endinvoke_called
) {
496 mono_error_set_invalid_operation(error
, "Delegate EndInvoke method called more than once");
497 mono_monitor_exit_internal ((MonoObject
*) ares
);
501 ares
->endinvoke_called
= 1;
503 /* wait until we are really finished */
504 if (ares
->completed
) {
505 mono_monitor_exit_internal ((MonoObject
*) ares
);
509 wait_event
= mono_wait_handle_get_handle ((MonoWaitHandle
*) ares
->handle
);
511 wait_event
= mono_w32event_create (TRUE
, FALSE
);
512 g_assert(wait_event
);
513 MonoWaitHandle
*wait_handle
= mono_wait_handle_new (mono_object_domain (ares
), wait_event
, error
);
514 if (!is_ok (error
)) {
515 mono_w32event_close (wait_event
);
518 MONO_OBJECT_SETREF_INTERNAL (ares
, handle
, (MonoObject
*) wait_handle
);
520 mono_monitor_exit_internal ((MonoObject
*) ares
);
521 mono_w32handle_wait_one (wait_event
, MONO_INFINITE_WAIT
, TRUE
);
524 ac
= (MonoAsyncCall
*) ares
->object_data
;
527 *exc
= ac
->msg
->exc
; /* FIXME: GC add write barrier */
528 *out_args
= ac
->out_args
;
533 mono_threadpool_remove_domain_jobs (MonoDomain
*domain
, int timeout
)
536 ThreadPoolDomain
*tpdomain
;
540 g_assert (timeout
>= -1);
542 g_assert (mono_domain_is_unloading (domain
));
545 end
= mono_msec_ticks () + timeout
;
547 #ifndef DISABLE_SOCKETS
548 mono_threadpool_io_remove_domain_jobs (domain
);
550 if (mono_msec_ticks () > end
)
556 * Wait for all threads which execute jobs in the domain to exit.
557 * The is_unloading () check in worker_request () ensures that
558 * no new jobs are added after we enter the lock below.
561 if (!mono_lazy_is_initialized (&status
))
564 mono_refcount_inc (&threadpool
);
568 tpdomain
= tpdomain_get (domain
);
571 mono_refcount_dec (&threadpool
);
577 while (tpdomain
->outstanding_request
+ tpdomain
->threadpool_jobs
> 0) {
579 mono_coop_cond_wait (&tpdomain
->cleanup_cond
, &threadpool
.domains_lock
);
584 now
= mono_msec_ticks();
590 res
= mono_coop_cond_timedwait (&tpdomain
->cleanup_cond
, &threadpool
.domains_lock
, end
- now
);
598 /* Remove from the list the worker threads look at */
599 tpdomain_remove (tpdomain
);
603 mono_coop_cond_destroy (&tpdomain
->cleanup_cond
);
604 tpdomain_free (tpdomain
);
606 mono_refcount_dec (&threadpool
);
612 mono_threadpool_suspend (void)
614 if (mono_lazy_is_initialized (&status
))
615 mono_threadpool_worker_set_suspended (TRUE
);
619 mono_threadpool_resume (void)
621 if (mono_lazy_is_initialized (&status
))
622 mono_threadpool_worker_set_suspended (FALSE
);
626 ves_icall_System_Threading_ThreadPool_GetAvailableThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
628 ThreadPoolCounter counter
;
630 if (!worker_threads
|| !completion_port_threads
)
633 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
635 *completion_port_threads
= 0;
639 counter
= COUNTER_READ ();
641 *worker_threads
= MAX (0, mono_threadpool_worker_get_max () - counter
._
.working
);
642 *completion_port_threads
= threadpool
.limit_io_max
;
644 mono_refcount_dec (&threadpool
);
648 ves_icall_System_Threading_ThreadPool_GetMinThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
650 if (!worker_threads
|| !completion_port_threads
)
653 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
655 *completion_port_threads
= 0;
659 *worker_threads
= mono_threadpool_worker_get_min ();
660 *completion_port_threads
= threadpool
.limit_io_min
;
662 mono_refcount_dec (&threadpool
);
666 ves_icall_System_Threading_ThreadPool_GetMaxThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
668 if (!worker_threads
|| !completion_port_threads
)
671 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
673 *completion_port_threads
= 0;
677 *worker_threads
= mono_threadpool_worker_get_max ();
678 *completion_port_threads
= threadpool
.limit_io_max
;
680 mono_refcount_dec (&threadpool
);
684 ves_icall_System_Threading_ThreadPool_SetMinThreadsNative (gint32 worker_threads
, gint32 completion_port_threads
, MonoError
*error
)
686 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
))
689 if (completion_port_threads
<= 0 || completion_port_threads
> threadpool
.limit_io_max
)
692 if (!mono_threadpool_worker_set_min (worker_threads
)) {
693 mono_refcount_dec (&threadpool
);
697 threadpool
.limit_io_min
= completion_port_threads
;
699 mono_refcount_dec (&threadpool
);
704 ves_icall_System_Threading_ThreadPool_SetMaxThreadsNative (gint32 worker_threads
, gint32 completion_port_threads
, MonoError
*error
)
706 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
))
709 worker_threads
= MIN (worker_threads
, MAX_POSSIBLE_THREADS
);
710 completion_port_threads
= MIN (completion_port_threads
, MAX_POSSIBLE_THREADS
);
712 gint cpu_count
= mono_cpu_count ();
714 if (completion_port_threads
< threadpool
.limit_io_min
|| completion_port_threads
< cpu_count
)
717 if (!mono_threadpool_worker_set_max (worker_threads
)) {
718 mono_refcount_dec (&threadpool
);
722 threadpool
.limit_io_max
= completion_port_threads
;
724 mono_refcount_dec (&threadpool
);
728 #ifdef ENABLE_NETCORE
730 ves_icall_System_Threading_ThreadPool_GetThreadCount (MonoError
*error
)
732 return mono_threadpool_worker_get_threads_count ();
736 ves_icall_System_Threading_ThreadPool_GetCompletedWorkItemCount (MonoError
*error
)
738 return mono_threadpool_worker_get_completed_threads_count ();
743 ves_icall_System_Threading_ThreadPool_InitializeVMTp (MonoBoolean
*enable_worker_tracking
, MonoError
*error
)
745 if (enable_worker_tracking
) {
746 // TODO implement some kind of switch to have the possibily to use it
747 *enable_worker_tracking
= FALSE
;
750 mono_lazy_initialize (&status
, initialize
);
754 ves_icall_System_Threading_ThreadPool_NotifyWorkItemComplete (MonoError
*error
)
756 if (mono_domain_is_unloading (mono_domain_get ()) || mono_runtime_is_shutting_down ())
759 return mono_threadpool_worker_notify_completed ();
763 ves_icall_System_Threading_ThreadPool_NotifyWorkItemProgressNative (MonoError
*error
)
765 mono_threadpool_worker_notify_completed ();
769 ves_icall_System_Threading_ThreadPool_NotifyWorkItemQueued (MonoError
*error
)
770 // FIXME Move to managed.
772 #ifndef DISABLE_PERFCOUNTERS
773 mono_atomic_inc_i64 (&mono_perfcounters
->threadpool_workitems
);
778 ves_icall_System_Threading_ThreadPool_ReportThreadStatus (MonoBoolean is_working
, MonoError
*error
)
781 mono_error_set_not_implemented (error
, "");
785 ves_icall_System_Threading_ThreadPool_RequestWorkerThread (MonoError
*error
)
788 ThreadPoolDomain
*tpdomain
;
789 ThreadPoolCounter counter
;
791 domain
= mono_domain_get ();
792 if (mono_domain_is_unloading (domain
))
795 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
796 /* threadpool has been destroyed, we are shutting down */
802 tpdomain
= tpdomain_get (domain
);
804 /* synchronize with mono_threadpool_remove_domain_jobs */
805 if (mono_domain_is_unloading (domain
)) {
807 mono_refcount_dec (&threadpool
);
811 tpdomain
= tpdomain_create (domain
);
816 tpdomain
->outstanding_request
++;
817 g_assert (tpdomain
->outstanding_request
>= 1);
821 COUNTER_ATOMIC (counter
, {
822 if (counter
._
.starting
== 16) {
823 mono_refcount_dec (&threadpool
);
827 counter
._
.starting
++;
830 mono_threadpool_worker_request ();
832 mono_refcount_dec (&threadpool
);