#include <mono/metadata/object-internals.h>
#include <mono/metadata/threadpool-ms.h>
#include <mono/metadata/threadpool-ms-io.h>
+#include <mono/metadata/w32event.h>
#include <mono/utils/atomic.h>
#include <mono/utils/mono-compiler.h>
#include <mono/utils/mono-complex.h>
typedef struct {
MonoDomain *domain;
+ /* Number of outstanding jobs */
gint32 outstanding_request;
+ /* Number of currently executing jobs */
+ int threadpool_jobs;
+ /* Signalled when threadpool_jobs + outstanding_request is 0 */
+ /* Protected by threadpool->domains_lock */
+ MonoCoopCond cleanup_cond;
} ThreadPoolDomain;
typedef MonoInternalThread ThreadPoolWorkingThread;
} while (0); \
} while (0)
+static inline void
+domains_lock (void)
+{
+ mono_coop_mutex_lock (&threadpool->domains_lock);
+}
+
+static inline void
+domains_unlock (void)
+{
+ mono_coop_mutex_unlock (&threadpool->domains_lock);
+}
+
static gpointer
rand_create (void)
{
return TRUE;
}
-/* LOCKING: threadpool->domains_lock must be held */
+/* LOCKING: domains_lock must be held */
static void
-domain_add (ThreadPoolDomain *tpdomain)
+tpdomain_add (ThreadPoolDomain *tpdomain)
{
guint i, len;
g_ptr_array_add (threadpool->domains, tpdomain);
}
-/* LOCKING: threadpool->domains_lock must be held */
+/* LOCKING: domains_lock must be held. */
static gboolean
-domain_remove (ThreadPoolDomain *tpdomain)
+tpdomain_remove (ThreadPoolDomain *tpdomain)
{
g_assert (tpdomain);
return g_ptr_array_remove (threadpool->domains, tpdomain);
}
-/* LOCKING: threadpool->domains_lock must be held */
+/* LOCKING: domains_lock must be held */
static ThreadPoolDomain *
-domain_get (MonoDomain *domain, gboolean create)
+tpdomain_get (MonoDomain *domain, gboolean create)
{
- ThreadPoolDomain *tpdomain = NULL;
guint i;
+ ThreadPoolDomain *tpdomain;
g_assert (domain);
for (i = 0; i < threadpool->domains->len; ++i) {
+ ThreadPoolDomain *tpdomain;
+
tpdomain = (ThreadPoolDomain *)g_ptr_array_index (threadpool->domains, i);
if (tpdomain->domain == domain)
return tpdomain;
}
- if (create) {
- tpdomain = g_new0 (ThreadPoolDomain, 1);
- tpdomain->domain = domain;
- domain_add (tpdomain);
- }
+ if (!create)
+ return NULL;
+
+ tpdomain = g_new0 (ThreadPoolDomain, 1);
+ tpdomain->domain = domain;
+ mono_coop_cond_init (&tpdomain->cleanup_cond);
+
+ tpdomain_add (tpdomain);
return tpdomain;
}
static void
-domain_free (ThreadPoolDomain *tpdomain)
+tpdomain_free (ThreadPoolDomain *tpdomain)
{
g_free (tpdomain);
}
-/* LOCKING: threadpool->domains_lock must be held */
+/* LOCKING: domains_lock must be held */
static gboolean
domain_any_has_request (void)
{
return FALSE;
}
-/* LOCKING: threadpool->domains_lock must be held */
+/* LOCKING: domains_lock must be held */
static ThreadPoolDomain *
-domain_get_next (ThreadPoolDomain *current)
+tpdomain_get_next (ThreadPoolDomain *current)
{
ThreadPoolDomain *tpdomain = NULL;
guint len;
if (thread == mono_thread_internal_current ())
return;
- mono_thread_internal_stop ((MonoInternalThread*) thread);
+ mono_thread_internal_abort ((MonoInternalThread*) thread);
}
static void
previous_tpdomain = NULL;
- mono_coop_mutex_lock (&threadpool->domains_lock);
+ domains_lock ();
while (!mono_runtime_is_shutting_down ()) {
tpdomain = NULL;
- if ((thread->state & (ThreadState_StopRequested | ThreadState_SuspendRequested)) != 0) {
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ if ((thread->state & (ThreadState_AbortRequested | ThreadState_SuspendRequested)) != 0) {
+ domains_unlock ();
mono_thread_interruption_checkpoint ();
- mono_coop_mutex_lock (&threadpool->domains_lock);
+ domains_lock ();
}
- if (retire || !(tpdomain = domain_get_next (previous_tpdomain))) {
+ if (retire || !(tpdomain = tpdomain_get_next (previous_tpdomain))) {
gboolean timeout;
COUNTER_ATOMIC (counter, {
counter._.parked ++;
});
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ domains_unlock ();
timeout = worker_park ();
- mono_coop_mutex_lock (&threadpool->domains_lock);
+ domains_lock ();
COUNTER_ATOMIC (counter, {
counter._.working ++;
if (retire)
retire = FALSE;
+ /* The tpdomain->domain might have unloaded, while this thread was parked */
+ previous_tpdomain = NULL;
+
continue;
}
tpdomain->outstanding_request --;
g_assert (tpdomain->outstanding_request >= 0);
- mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_THREADPOOL, "[%p] worker running in domain %p",
+ mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_THREADPOOL, "[%p] worker running in domain %p (outstanding requests %d) ",
mono_native_thread_id_get (), tpdomain->domain, tpdomain->outstanding_request);
g_assert (tpdomain->domain);
- g_assert (tpdomain->domain->threadpool_jobs >= 0);
- tpdomain->domain->threadpool_jobs ++;
+ g_assert (tpdomain->threadpool_jobs >= 0);
+ tpdomain->threadpool_jobs ++;
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ /*
+ * This is needed so there is always an lmf frame in the runtime invoke call below,
+ * so ThreadAbortExceptions are caught even if the thread is in native code.
+ */
+ mono_defaults.threadpool_perform_wait_callback_method->save_lmf = TRUE;
+
+ domains_unlock ();
mono_thread_push_appdomain_ref (tpdomain->domain);
if (mono_domain_set (tpdomain->domain, FALSE)) {
}
mono_thread_pop_appdomain_ref ();
- mono_coop_mutex_lock (&threadpool->domains_lock);
+ domains_lock ();
+
+ tpdomain->threadpool_jobs --;
+ g_assert (tpdomain->threadpool_jobs >= 0);
- tpdomain->domain->threadpool_jobs --;
- g_assert (tpdomain->domain->threadpool_jobs >= 0);
+ if (tpdomain->outstanding_request + tpdomain->threadpool_jobs == 0 && mono_domain_is_unloading (tpdomain->domain)) {
+ gboolean removed;
- if (tpdomain->domain->threadpool_jobs == 0 && mono_domain_is_unloading (tpdomain->domain)) {
- gboolean removed = domain_remove (tpdomain);
+ removed = tpdomain_remove (tpdomain);
g_assert (removed);
- if (tpdomain->domain->cleanup_semaphore)
- ReleaseSemaphore (tpdomain->domain->cleanup_semaphore, 1, NULL);
- domain_free (tpdomain);
+
+ mono_coop_cond_signal (&tpdomain->cleanup_cond);
tpdomain = NULL;
}
previous_tpdomain = tpdomain;
}
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ domains_unlock ();
mono_coop_mutex_lock (&threadpool->active_threads_lock);
g_ptr_array_remove_fast (threadpool->working_threads, thread);
if ((thread = mono_thread_create_internal (mono_get_root_domain (), worker_thread, NULL, TRUE, 0, &error)) != NULL) {
threadpool->worker_creation_current_count += 1;
- mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_THREADPOOL, "[%p] try create worker, created %p, now = %d count = %d", mono_native_thread_id_get (), thread->tid, now, threadpool->worker_creation_current_count);
+ mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_THREADPOOL, "[%p] try create worker, created %p, now = %d count = %d", mono_native_thread_id_get (), GUINT_TO_POINTER(thread->tid), now, threadpool->worker_creation_current_count);
mono_coop_mutex_unlock (&threadpool->worker_creation_lock);
return TRUE;
}
if (mono_runtime_is_shutting_down ())
return FALSE;
- mono_coop_mutex_lock (&threadpool->domains_lock);
+ domains_lock ();
/* synchronize check with worker_thread */
if (mono_domain_is_unloading (domain)) {
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ domains_unlock ();
return FALSE;
}
- tpdomain = domain_get (domain, TRUE);
+ tpdomain = tpdomain_get (domain, TRUE);
g_assert (tpdomain);
tpdomain->outstanding_request ++;
mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_THREADPOOL, "[%p] request worker, domain = %p, outstanding_request = %d",
mono_native_thread_id_get (), tpdomain->domain, tpdomain->outstanding_request);
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ domains_unlock ();
if (threadpool->suspended)
return FALSE;
if (mono_runtime_is_shutting_down ()) {
should_keep_running = FALSE;
} else {
- mono_coop_mutex_lock (&threadpool->domains_lock);
+ domains_lock ();
if (!domain_any_has_request ())
should_keep_running = FALSE;
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ domains_unlock ();
if (!should_keep_running) {
if (last_should_keep_running == -1 || mono_100ns_ticks () - last_should_keep_running < MONITOR_MINIMAL_LIFETIME * 1000 * 10) {
if (mono_runtime_is_shutting_down ())
continue;
- mono_coop_mutex_lock (&threadpool->domains_lock);
+ domains_lock ();
if (!domain_any_has_request ()) {
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ domains_unlock ();
continue;
}
- mono_coop_mutex_unlock (&threadpool->domains_lock);
+ domains_unlock ();
threadpool->cpu_usage = mono_cpu_usage (threadpool->cpu_usage_state);
void
mono_threadpool_ms_cleanup (void)
{
- #ifndef DISABLE_SOCKETS
- mono_threadpool_ms_io_cleanup ();
- #endif
+#ifndef DISABLE_SOCKETS
+ mono_threadpool_ms_io_cleanup ();
+#endif
mono_lazy_cleanup (&status, cleanup);
}
if (ares->handle) {
wait_event = mono_wait_handle_get_handle ((MonoWaitHandle*) ares->handle);
} else {
- wait_event = CreateEvent (NULL, TRUE, FALSE, NULL);
+ wait_event = mono_w32event_create (TRUE, FALSE);
g_assert(wait_event);
MonoWaitHandle *wait_handle = mono_wait_handle_new (mono_object_domain (ares), wait_event, error);
if (!is_ok (error)) {
}
mono_monitor_exit ((MonoObject*) ares);
MONO_ENTER_GC_SAFE;
+#ifdef HOST_WIN32
WaitForSingleObjectEx (wait_event, INFINITE, TRUE);
+#else
+ mono_w32handle_wait_one (wait_event, MONO_INFINITE_WAIT, TRUE);
+#endif
MONO_EXIT_GC_SAFE;
}
gboolean
mono_threadpool_ms_remove_domain_jobs (MonoDomain *domain, int timeout)
{
- gboolean res = TRUE;
gint64 end;
- gpointer sem;
+ ThreadPoolDomain *tpdomain;
+ gboolean ret;
g_assert (domain);
g_assert (timeout >= -1);
#endif
/*
- * There might be some threads out that could be about to execute stuff from the given domain.
- * We avoid that by setting up a semaphore to be pulsed by the thread that reaches zero.
+ * Wait for all threads which execute jobs in the domain to exit.
+ * The is_unloading () check in worker_request () ensures that
+ * no new jobs are added after we enter the lock below.
*/
- sem = domain->cleanup_semaphore = CreateSemaphore (NULL, 0, 1, NULL);
+ mono_lazy_initialize (&status, initialize);
+ domains_lock ();
- /*
- * The memory barrier here is required to have global ordering between assigning to cleanup_semaphone
- * and reading threadpool_jobs. Otherwise this thread could read a stale version of threadpool_jobs
- * and wait forever.
- */
- mono_memory_write_barrier ();
+ tpdomain = tpdomain_get (domain, FALSE);
+ if (!tpdomain) {
+ domains_unlock ();
+ return TRUE;
+ }
- while (domain->threadpool_jobs) {
- gint64 now;
+ ret = TRUE;
- if (timeout != -1) {
- now = mono_msec_ticks ();
+ while (tpdomain->outstanding_request + tpdomain->threadpool_jobs > 0) {
+ if (timeout == -1) {
+ mono_coop_cond_wait (&tpdomain->cleanup_cond, &threadpool->domains_lock);
+ } else {
+ gint64 now;
+ gint res;
+
+ now = mono_msec_ticks();
if (now > end) {
- res = FALSE;
+ ret = FALSE;
break;
}
- }
- MONO_ENTER_GC_SAFE;
- WaitForSingleObject (sem, timeout != -1 ? end - now : timeout);
- MONO_EXIT_GC_SAFE;
+ res = mono_coop_cond_timedwait (&tpdomain->cleanup_cond, &threadpool->domains_lock, end - now);
+ if (res != 0) {
+ ret = FALSE;
+ break;
+ }
+ }
}
- domain->cleanup_semaphore = NULL;
- CloseHandle (sem);
+ /* Remove from the list the worker threads look at */
+ tpdomain_remove (tpdomain);
- return res;
+ domains_unlock ();
+
+ mono_coop_cond_destroy (&tpdomain->cleanup_cond);
+ tpdomain_free (tpdomain);
+
+ return ret;
}
void
projects / mono.git / blobdiff