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/exception.h>
32 #include <mono/metadata/gc-internals.h>
33 #include <mono/metadata/object.h>
34 #include <mono/metadata/object-internals.h>
35 #include <mono/metadata/threadpool.h>
36 #include <mono/metadata/threadpool-worker.h>
37 #include <mono/metadata/threadpool-io.h>
38 #include <mono/metadata/w32event.h>
39 #include <mono/utils/atomic.h>
40 #include <mono/utils/mono-compiler.h>
41 #include <mono/utils/mono-complex.h>
42 #include <mono/utils/mono-lazy-init.h>
43 #include <mono/utils/mono-logger.h>
44 #include <mono/utils/mono-logger-internals.h>
45 #include <mono/utils/mono-proclib.h>
46 #include <mono/utils/mono-threads.h>
47 #include <mono/utils/mono-time.h>
48 #include <mono/utils/refcount.h>
49 #include <mono/utils/mono-os-wait.h>
51 #include "icall-decl.h"
53 // consistency with coreclr https://github.com/dotnet/coreclr/blob/643b09f966e68e06d5f0930755985a01a2a2b096/src/vm/win32threadpool.h#L111
54 #define MAX_POSSIBLE_THREADS 0x7fff
58 /* Number of outstanding jobs */
59 gint32 outstanding_request
;
60 /* Number of currently executing jobs */
61 gint32 threadpool_jobs
;
62 /* Signalled when threadpool_jobs + outstanding_request is 0 */
63 /* Protected by threadpool.domains_lock */
64 MonoCoopCond cleanup_cond
;
69 gint16 starting
; /* starting, but not yet in worker_callback */
70 gint16 working
; /* executing worker_callback */
78 GPtrArray
*domains
; // ThreadPoolDomain* []
79 MonoCoopMutex domains_lock
;
81 ThreadPoolCounter counters
;
87 static mono_lazy_init_t status
= MONO_LAZY_INIT_STATUS_NOT_INITIALIZED
;
89 static ThreadPool threadpool
;
91 #define COUNTER_ATOMIC(var,block) \
93 ThreadPoolCounter __old; \
95 (var) = __old = COUNTER_READ (); \
97 if (!(counter._.starting >= 0)) \
98 g_error ("%s: counter._.starting = %d, but should be >= 0", __func__, counter._.starting); \
99 if (!(counter._.working >= 0)) \
100 g_error ("%s: counter._.working = %d, but should be >= 0", __func__, counter._.working); \
101 } while (mono_atomic_cas_i32 (&threadpool.counters.as_gint32, (var).as_gint32, __old.as_gint32) != __old.as_gint32); \
104 static inline ThreadPoolCounter
107 ThreadPoolCounter counter
;
108 counter
.as_gint32
= mono_atomic_load_i32 (&threadpool
.counters
.as_gint32
);
115 mono_coop_mutex_lock (&threadpool
.domains_lock
);
119 domains_unlock (void)
121 mono_coop_mutex_unlock (&threadpool
.domains_lock
);
125 destroy (gpointer unused
)
127 g_ptr_array_free (threadpool
.domains
, TRUE
);
128 mono_coop_mutex_destroy (&threadpool
.domains_lock
);
132 worker_callback (void);
137 g_assert (sizeof (ThreadPoolCounter
) == sizeof (gint32
));
139 mono_refcount_init (&threadpool
, destroy
);
141 threadpool
.domains
= g_ptr_array_new ();
142 mono_coop_mutex_init (&threadpool
.domains_lock
);
144 threadpool
.limit_io_min
= mono_cpu_count ();
145 threadpool
.limit_io_max
= CLAMP (threadpool
.limit_io_min
* 100, MIN (threadpool
.limit_io_min
, 200), MAX (threadpool
.limit_io_min
, 200));
147 mono_threadpool_worker_init (worker_callback
);
153 mono_threadpool_worker_cleanup ();
155 mono_refcount_dec (&threadpool
);
159 mono_threadpool_enqueue_work_item (MonoDomain
*domain
, MonoObject
*work_item
, MonoError
*error
)
161 static MonoClass
*threadpool_class
= NULL
;
162 static MonoMethod
*unsafe_queue_custom_work_item_method
= NULL
;
163 MonoDomain
*current_domain
;
168 g_assert (work_item
);
170 if (!threadpool_class
)
171 threadpool_class
= mono_class_load_from_name (mono_defaults
.corlib
, "System.Threading", "ThreadPool");
173 if (!unsafe_queue_custom_work_item_method
) {
174 unsafe_queue_custom_work_item_method
= mono_class_get_method_from_name_checked (threadpool_class
, "UnsafeQueueCustomWorkItem", 2, 0, error
);
175 mono_error_assert_ok (error
);
177 g_assert (unsafe_queue_custom_work_item_method
);
181 args
[0] = (gpointer
) work_item
;
182 args
[1] = (gpointer
) &f
;
184 current_domain
= mono_domain_get ();
185 if (current_domain
== domain
) {
186 mono_runtime_invoke_checked (unsafe_queue_custom_work_item_method
, NULL
, args
, error
);
187 return_val_if_nok (error
, FALSE
);
189 mono_thread_push_appdomain_ref (domain
);
190 if (mono_domain_set (domain
, FALSE
)) {
191 mono_runtime_invoke_checked (unsafe_queue_custom_work_item_method
, NULL
, args
, error
);
192 if (!is_ok (error
)) {
193 mono_thread_pop_appdomain_ref ();
196 mono_domain_set (current_domain
, TRUE
);
198 mono_thread_pop_appdomain_ref ();
203 /* LOCKING: domains_lock must be held. */
204 static ThreadPoolDomain
*
205 tpdomain_create (MonoDomain
*domain
)
207 ThreadPoolDomain
*tpdomain
;
209 tpdomain
= g_new0 (ThreadPoolDomain
, 1);
210 tpdomain
->domain
= domain
;
211 mono_coop_cond_init (&tpdomain
->cleanup_cond
);
213 g_ptr_array_add (threadpool
.domains
, tpdomain
);
218 /* LOCKING: domains_lock must be held. */
220 tpdomain_remove (ThreadPoolDomain
*tpdomain
)
223 return g_ptr_array_remove (threadpool
.domains
, tpdomain
);
226 /* LOCKING: domains_lock must be held */
227 static ThreadPoolDomain
*
228 tpdomain_get (MonoDomain
*domain
)
234 for (i
= 0; i
< threadpool
.domains
->len
; ++i
) {
235 ThreadPoolDomain
*tpdomain
;
237 tpdomain
= (ThreadPoolDomain
*)g_ptr_array_index (threadpool
.domains
, i
);
238 if (tpdomain
->domain
== domain
)
246 tpdomain_free (ThreadPoolDomain
*tpdomain
)
251 /* LOCKING: domains_lock must be held */
252 static ThreadPoolDomain
*
253 tpdomain_get_next (ThreadPoolDomain
*current
)
255 ThreadPoolDomain
*tpdomain
= NULL
;
258 len
= threadpool
.domains
->len
;
260 gint i
, current_idx
= -1;
262 for (i
= 0; i
< len
; ++i
) {
263 if (current
== g_ptr_array_index (threadpool
.domains
, i
)) {
269 for (i
= current_idx
+ 1; i
< len
+ current_idx
+ 1; ++i
) {
270 ThreadPoolDomain
*tmp
= (ThreadPoolDomain
*)g_ptr_array_index (threadpool
.domains
, i
% len
);
271 if (tmp
->outstanding_request
> 0) {
282 try_invoke_perform_wait_callback (MonoObject
** exc
, MonoError
*error
)
284 HANDLE_FUNCTION_ENTER ();
286 MonoObject
*res
= mono_runtime_try_invoke (mono_defaults
.threadpool_perform_wait_callback_method
, NULL
, NULL
, exc
, error
);
287 HANDLE_FUNCTION_RETURN_VAL (res
);
291 worker_callback (void)
294 ThreadPoolDomain
*tpdomain
, *previous_tpdomain
;
295 ThreadPoolCounter counter
;
296 MonoInternalThread
*thread
;
298 if (!mono_refcount_tryinc (&threadpool
))
301 thread
= mono_thread_internal_current ();
303 COUNTER_ATOMIC (counter
, {
304 if (!(counter
._
.working
< 32767 /* G_MAXINT16 */))
305 g_error ("%s: counter._.working = %d, but should be < 32767", __func__
, counter
._
.working
);
307 counter
._
.starting
--;
308 counter
._
.working
++;
311 if (mono_runtime_is_shutting_down ()) {
312 COUNTER_ATOMIC (counter
, {
313 counter
._
.working
--;
316 mono_refcount_dec (&threadpool
);
321 * This is needed so there is always an lmf frame in the runtime invoke call below,
322 * so ThreadAbortExceptions are caught even if the thread is in native code.
324 mono_defaults
.threadpool_perform_wait_callback_method
->save_lmf
= TRUE
;
328 previous_tpdomain
= NULL
;
330 while (!mono_runtime_is_shutting_down ()) {
331 gboolean retire
= FALSE
;
333 if (thread
->state
& (ThreadState_AbortRequested
| ThreadState_SuspendRequested
)) {
335 if (mono_thread_interruption_checkpoint_bool ()) {
342 tpdomain
= tpdomain_get_next (previous_tpdomain
);
346 tpdomain
->outstanding_request
--;
347 g_assert (tpdomain
->outstanding_request
>= 0);
349 mono_trace (G_LOG_LEVEL_DEBUG
, MONO_TRACE_THREADPOOL
, "[%p] worker running in domain %p (outstanding requests %d)",
350 GUINT_TO_POINTER (MONO_NATIVE_THREAD_ID_TO_UINT (mono_native_thread_id_get ())), tpdomain
->domain
, tpdomain
->outstanding_request
);
352 g_assert (tpdomain
->threadpool_jobs
>= 0);
353 tpdomain
->threadpool_jobs
++;
357 MonoString
*thread_name
= mono_string_new_checked (mono_get_root_domain (), "Thread Pool Worker", error
);
358 mono_error_assert_ok (error
);
359 mono_thread_set_name_internal (thread
, thread_name
, FALSE
, TRUE
, error
);
360 mono_error_assert_ok (error
);
362 mono_thread_clear_and_set_state (thread
,
363 (MonoThreadState
)~ThreadState_Background
,
364 ThreadState_Background
);
366 mono_thread_push_appdomain_ref (tpdomain
->domain
);
367 if (mono_domain_set (tpdomain
->domain
, FALSE
)) {
368 MonoObject
*exc
= NULL
, *res
;
370 res
= try_invoke_perform_wait_callback (&exc
, error
);
371 if (exc
|| !mono_error_ok(error
)) {
373 exc
= (MonoObject
*) mono_error_convert_to_exception (error
);
375 mono_error_cleanup (error
);
376 mono_thread_internal_unhandled_exception (exc
);
377 } else if (res
&& *(MonoBoolean
*) mono_object_unbox_internal (res
) == FALSE
) {
381 mono_domain_set (mono_get_root_domain (), TRUE
);
383 mono_thread_pop_appdomain_ref ();
387 tpdomain
->threadpool_jobs
--;
388 g_assert (tpdomain
->threadpool_jobs
>= 0);
390 if (tpdomain
->outstanding_request
+ tpdomain
->threadpool_jobs
== 0 && mono_domain_is_unloading (tpdomain
->domain
)) {
393 removed
= tpdomain_remove (tpdomain
);
396 mono_coop_cond_signal (&tpdomain
->cleanup_cond
);
403 previous_tpdomain
= tpdomain
;
408 COUNTER_ATOMIC (counter
, {
409 counter
._
.working
--;
412 mono_refcount_dec (&threadpool
);
416 mono_threadpool_cleanup (void)
418 #ifndef DISABLE_SOCKETS
419 mono_threadpool_io_cleanup ();
421 mono_lazy_cleanup (&status
, cleanup
);
425 mono_threadpool_begin_invoke (MonoDomain
*domain
, MonoObject
*target
, MonoMethod
*method
, gpointer
*params
, MonoError
*error
)
427 static MonoClass
*async_call_klass
= NULL
;
428 MonoMethodMessage
*message
;
429 MonoAsyncResult
*async_result
;
430 MonoAsyncCall
*async_call
;
431 MonoDelegate
*async_callback
= NULL
;
432 MonoObject
*state
= NULL
;
434 if (!async_call_klass
)
435 async_call_klass
= mono_class_load_from_name (mono_defaults
.corlib
, "System", "MonoAsyncCall");
439 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
);
440 return_val_if_nok (error
, NULL
);
442 async_call
= (MonoAsyncCall
*) mono_object_new_checked (domain
, async_call_klass
, error
);
443 return_val_if_nok (error
, NULL
);
445 MONO_OBJECT_SETREF_INTERNAL (async_call
, msg
, message
);
446 MONO_OBJECT_SETREF_INTERNAL (async_call
, state
, state
);
448 if (async_callback
) {
449 MONO_OBJECT_SETREF_INTERNAL (async_call
, cb_method
, mono_get_delegate_invoke_internal (((MonoObject
*) async_callback
)->vtable
->klass
));
450 MONO_OBJECT_SETREF_INTERNAL (async_call
, cb_target
, async_callback
);
453 async_result
= mono_async_result_new (domain
, NULL
, async_call
->state
, NULL
, (MonoObject
*) async_call
, error
);
454 return_val_if_nok (error
, NULL
);
455 MONO_OBJECT_SETREF_INTERNAL (async_result
, async_delegate
, target
);
457 mono_threadpool_enqueue_work_item (domain
, (MonoObject
*) async_result
, error
);
458 return_val_if_nok (error
, NULL
);
464 mono_threadpool_end_invoke (MonoAsyncResult
*ares
, MonoArray
**out_args
, MonoObject
**exc
, MonoError
*error
)
475 /* check if already finished */
476 mono_monitor_enter_internal ((MonoObject
*) ares
);
478 if (ares
->endinvoke_called
) {
479 mono_error_set_invalid_operation(error
, "Delegate EndInvoke method called more than once");
480 mono_monitor_exit_internal ((MonoObject
*) ares
);
484 ares
->endinvoke_called
= 1;
486 /* wait until we are really finished */
487 if (ares
->completed
) {
488 mono_monitor_exit_internal ((MonoObject
*) ares
);
492 wait_event
= mono_wait_handle_get_handle ((MonoWaitHandle
*) ares
->handle
);
494 wait_event
= mono_w32event_create (TRUE
, FALSE
);
495 g_assert(wait_event
);
496 MonoWaitHandle
*wait_handle
= mono_wait_handle_new (mono_object_domain (ares
), wait_event
, error
);
497 if (!is_ok (error
)) {
498 mono_w32event_close (wait_event
);
501 MONO_OBJECT_SETREF_INTERNAL (ares
, handle
, (MonoObject
*) wait_handle
);
503 mono_monitor_exit_internal ((MonoObject
*) ares
);
506 mono_win32_wait_for_single_object_ex (wait_event
, INFINITE
, TRUE
);
509 mono_w32handle_wait_one (wait_event
, MONO_INFINITE_WAIT
, TRUE
);
513 ac
= (MonoAsyncCall
*) ares
->object_data
;
516 *exc
= ac
->msg
->exc
; /* FIXME: GC add write barrier */
517 *out_args
= ac
->out_args
;
522 mono_threadpool_remove_domain_jobs (MonoDomain
*domain
, int timeout
)
525 ThreadPoolDomain
*tpdomain
;
529 g_assert (timeout
>= -1);
531 g_assert (mono_domain_is_unloading (domain
));
534 end
= mono_msec_ticks () + timeout
;
536 #ifndef DISABLE_SOCKETS
537 mono_threadpool_io_remove_domain_jobs (domain
);
539 if (mono_msec_ticks () > end
)
545 * Wait for all threads which execute jobs in the domain to exit.
546 * The is_unloading () check in worker_request () ensures that
547 * no new jobs are added after we enter the lock below.
550 if (!mono_lazy_is_initialized (&status
))
553 mono_refcount_inc (&threadpool
);
557 tpdomain
= tpdomain_get (domain
);
560 mono_refcount_dec (&threadpool
);
566 while (tpdomain
->outstanding_request
+ tpdomain
->threadpool_jobs
> 0) {
568 mono_coop_cond_wait (&tpdomain
->cleanup_cond
, &threadpool
.domains_lock
);
573 now
= mono_msec_ticks();
579 res
= mono_coop_cond_timedwait (&tpdomain
->cleanup_cond
, &threadpool
.domains_lock
, end
- now
);
587 /* Remove from the list the worker threads look at */
588 tpdomain_remove (tpdomain
);
592 mono_coop_cond_destroy (&tpdomain
->cleanup_cond
);
593 tpdomain_free (tpdomain
);
595 mono_refcount_dec (&threadpool
);
601 mono_threadpool_suspend (void)
603 if (mono_lazy_is_initialized (&status
))
604 mono_threadpool_worker_set_suspended (TRUE
);
608 mono_threadpool_resume (void)
610 if (mono_lazy_is_initialized (&status
))
611 mono_threadpool_worker_set_suspended (FALSE
);
615 ves_icall_System_Threading_ThreadPool_GetAvailableThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
617 ThreadPoolCounter counter
;
619 if (!worker_threads
|| !completion_port_threads
)
622 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
624 *completion_port_threads
= 0;
628 counter
= COUNTER_READ ();
630 *worker_threads
= MAX (0, mono_threadpool_worker_get_max () - counter
._
.working
);
631 *completion_port_threads
= threadpool
.limit_io_max
;
633 mono_refcount_dec (&threadpool
);
637 ves_icall_System_Threading_ThreadPool_GetMinThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
639 if (!worker_threads
|| !completion_port_threads
)
642 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
644 *completion_port_threads
= 0;
648 *worker_threads
= mono_threadpool_worker_get_min ();
649 *completion_port_threads
= threadpool
.limit_io_min
;
651 mono_refcount_dec (&threadpool
);
655 ves_icall_System_Threading_ThreadPool_GetMaxThreadsNative (gint32
*worker_threads
, gint32
*completion_port_threads
, MonoError
*error
)
657 if (!worker_threads
|| !completion_port_threads
)
660 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
662 *completion_port_threads
= 0;
666 *worker_threads
= mono_threadpool_worker_get_max ();
667 *completion_port_threads
= threadpool
.limit_io_max
;
669 mono_refcount_dec (&threadpool
);
673 ves_icall_System_Threading_ThreadPool_SetMinThreadsNative (gint32 worker_threads
, gint32 completion_port_threads
, MonoError
*error
)
675 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
))
678 if (completion_port_threads
<= 0 || completion_port_threads
> threadpool
.limit_io_max
)
681 if (!mono_threadpool_worker_set_min (worker_threads
)) {
682 mono_refcount_dec (&threadpool
);
686 threadpool
.limit_io_min
= completion_port_threads
;
688 mono_refcount_dec (&threadpool
);
693 ves_icall_System_Threading_ThreadPool_SetMaxThreadsNative (gint32 worker_threads
, gint32 completion_port_threads
, MonoError
*error
)
695 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
))
698 worker_threads
= MIN (worker_threads
, MAX_POSSIBLE_THREADS
);
699 completion_port_threads
= MIN (completion_port_threads
, MAX_POSSIBLE_THREADS
);
701 gint cpu_count
= mono_cpu_count ();
703 if (completion_port_threads
< threadpool
.limit_io_min
|| completion_port_threads
< cpu_count
)
706 if (!mono_threadpool_worker_set_max (worker_threads
)) {
707 mono_refcount_dec (&threadpool
);
711 threadpool
.limit_io_max
= completion_port_threads
;
713 mono_refcount_dec (&threadpool
);
718 ves_icall_System_Threading_ThreadPool_InitializeVMTp (MonoBoolean
*enable_worker_tracking
, MonoError
*error
)
720 if (enable_worker_tracking
) {
721 // TODO implement some kind of switch to have the possibily to use it
722 *enable_worker_tracking
= FALSE
;
725 mono_lazy_initialize (&status
, initialize
);
729 ves_icall_System_Threading_ThreadPool_NotifyWorkItemComplete (MonoError
*error
)
731 if (mono_domain_is_unloading (mono_domain_get ()) || mono_runtime_is_shutting_down ())
734 return mono_threadpool_worker_notify_completed ();
738 ves_icall_System_Threading_ThreadPool_NotifyWorkItemProgressNative (MonoError
*error
)
740 mono_threadpool_worker_notify_completed ();
744 ves_icall_System_Threading_ThreadPool_NotifyWorkItemQueued (MonoError
*error
)
745 // FIXME Move to managed.
747 #ifndef DISABLE_PERFCOUNTERS
748 mono_atomic_inc_i64 (&mono_perfcounters
->threadpool_workitems
);
753 ves_icall_System_Threading_ThreadPool_ReportThreadStatus (MonoBoolean is_working
, MonoError
*error
)
756 mono_error_set_not_implemented (error
, "");
760 ves_icall_System_Threading_ThreadPool_RequestWorkerThread (MonoError
*error
)
763 ThreadPoolDomain
*tpdomain
;
764 ThreadPoolCounter counter
;
766 domain
= mono_domain_get ();
767 if (mono_domain_is_unloading (domain
))
770 if (!mono_lazy_initialize (&status
, initialize
) || !mono_refcount_tryinc (&threadpool
)) {
771 /* threadpool has been destroyed, we are shutting down */
777 tpdomain
= tpdomain_get (domain
);
779 /* synchronize with mono_threadpool_remove_domain_jobs */
780 if (mono_domain_is_unloading (domain
)) {
782 mono_refcount_dec (&threadpool
);
786 tpdomain
= tpdomain_create (domain
);
791 tpdomain
->outstanding_request
++;
792 g_assert (tpdomain
->outstanding_request
>= 1);
796 COUNTER_ATOMIC (counter
, {
797 if (counter
._
.starting
== 16) {
798 mono_refcount_dec (&threadpool
);
802 counter
._
.starting
++;
805 mono_threadpool_worker_request ();
807 mono_refcount_dec (&threadpool
);
811 MonoBoolean G_GNUC_UNUSED
812 ves_icall_System_Threading_ThreadPool_PostQueuedCompletionStatus (MonoNativeOverlapped
*native_overlapped
, MonoError
*error
)
814 /* This copy the behavior of the current Mono implementation */
815 mono_error_set_not_implemented (error
, "");
819 MonoBoolean G_GNUC_UNUSED
820 ves_icall_System_Threading_ThreadPool_BindIOCompletionCallbackNative (gpointer file_handle
, MonoError
*error
)
822 /* This copy the behavior of the current Mono implementation */
826 MonoBoolean G_GNUC_UNUSED
827 ves_icall_System_Threading_ThreadPool_IsThreadPoolHosted (MonoError
*error
)