Fix typo.

[mono.git] / mono / metadata / threadpool.c

diff --git a/mono/metadata/threadpool.c b/mono/metadata/threadpool.c
index e14470c0e5d2c1e294059a5ae27c2015dd72c7c8..11f86ad1fa4472a44f43458fe98cb4afc30bf308 100644 (file)
--- a/mono/metadata/threadpool.c
+++ b/mono/metadata/threadpool.c
@@ -15,6 +15,9 @@
 #ifdef PLATFORM_WIN32
 #define WINVER 0x0500
 #define _WIN32_WINNT 0x0500
+#define THREADS_PER_CPU        25
+#else
+#define THREADS_PER_CPU        50
 #endif
 
 #include <mono/metadata/domain-internals.h>
@@ -31,7 +34,7 @@
 #include "threadpool.h"
 
 /* maximum number of worker threads */
-int mono_max_worker_threads = 25; /* fixme: should be 25 per available CPU */
+int mono_max_worker_threads = THREADS_PER_CPU;
 static int mono_min_worker_threads = 0;
 
 /* current number of worker threads */
@@ -40,9 +43,14 @@ static int mono_worker_threads = 0;
 /* current number of busy threads */
 static int busy_worker_threads = 0;
 
+/* mono_thread_pool_init called */
+static int tp_inited;
+
 /* we use this to store a reference to the AsyncResult to avoid GC */
 static MonoGHashTable *ares_htable = NULL;
 
+static CRITICAL_SECTION ares_lock;
+
 /* we append a job */
 static HANDLE job_added;
 
@@ -82,8 +90,7 @@ mono_async_invoke (MonoAsyncResult *ares)
        }
 
        /* notify listeners */
-       if(!mono_monitor_enter ((MonoObject *) ares))
-               return;
+       mono_monitor_enter ((MonoObject *) ares);
        
        if (ares->handle != NULL) {
                ac->wait_event = ((MonoWaitHandle *) ares->handle)->handle;
@@ -91,7 +98,32 @@ mono_async_invoke (MonoAsyncResult *ares)
        }
        mono_monitor_exit ((MonoObject *) ares);
 
+       EnterCriticalSection (&ares_lock);
        mono_g_hash_table_remove (ares_htable, ares);
+       LeaveCriticalSection (&ares_lock);
+}
+
+void
+mono_thread_pool_init ()
+{
+       SYSTEM_INFO info;
+       int threads_per_cpu = THREADS_PER_CPU;
+
+       if ((int) InterlockedCompareExchange (&tp_inited, 1, 0) == 1)
+               return;
+
+       MONO_GC_REGISTER_ROOT (ares_htable);
+       InitializeCriticalSection (&ares_lock);
+       ares_htable = mono_g_hash_table_new (NULL, NULL);
+       job_added = CreateSemaphore (NULL, 0, 0x7fffffff, NULL);
+       GetSystemInfo (&info);
+       if (getenv ("MONO_THREADS_PER_CPU") != NULL) {
+               threads_per_cpu = atoi (getenv ("MONO_THREADS_PER_CPU"));
+               if (threads_per_cpu <= 0)
+                       threads_per_cpu = THREADS_PER_CPU;
+       }
+
+       mono_max_worker_threads = threads_per_cpu * info.dwNumberOfProcessors;
 }
 
 MonoAsyncResult *
@@ -121,13 +153,10 @@ mono_thread_pool_add (MonoObject *target, MonoMethodMessage *msg, MonoDelegate *
        ares = mono_async_result_new (domain, NULL, ac->state, ac);
        ares->async_delegate = target;
 
-       if (!ares_htable) {
-               ares_htable = mono_g_hash_table_new (NULL, NULL);
-               job_added = CreateSemaphore (NULL, 0, 0x7fffffff, NULL);
-       }
-
+       EnterCriticalSection (&ares_lock);
        mono_g_hash_table_insert (ares_htable, ares, ares);
-
+       LeaveCriticalSection (&ares_lock);
+       
        busy = (int) InterlockedCompareExchange (&busy_worker_threads, 0, -1);
        worker = (int) InterlockedCompareExchange (&mono_worker_threads, 0, -1); 
        if (worker <= ++busy &&
@@ -152,9 +181,7 @@ mono_thread_pool_finish (MonoAsyncResult *ares, MonoArray **out_args, MonoObject
        *out_args = NULL;
 
        /* check if already finished */
-       if (!mono_monitor_enter ((MonoObject *) ares)) {
-               return NULL;
-       }
+       mono_monitor_enter ((MonoObject *) ares);
        
        if (ares->endinvoke_called) {
                *exc = (MonoObject *)mono_exception_from_name (mono_defaults.corlib, "System", 
@@ -203,8 +230,19 @@ mono_thread_pool_cleanup (void)
 static void
 append_job (MonoAsyncResult *ar)
 {
+       GList *tmp;
+
        EnterCriticalSection (&mono_delegate_section);
-       async_call_queue = g_list_append (async_call_queue, ar); 
+       if (async_call_queue == NULL) {
+               async_call_queue = g_list_append (async_call_queue, ar); 
+       } else {
+               for (tmp = async_call_queue; tmp && tmp->data != NULL; tmp = tmp->next);
+               if (tmp == NULL) {
+                       async_call_queue = g_list_append (async_call_queue, ar); 
+               } else {
+                       tmp->data = ar;
+               }
+       }
        LeaveCriticalSection (&mono_delegate_section);
 }
 
@@ -212,17 +250,23 @@ static MonoAsyncResult *
 dequeue_job (void)
 {
        MonoAsyncResult *ar = NULL;
-       GList *tmp = NULL;
+       GList *tmp, *tmp2;
 
        EnterCriticalSection (&mono_delegate_section);
-       if (async_call_queue) {
-               ar = (MonoAsyncResult *)async_call_queue->data;
-               tmp = async_call_queue;
-               async_call_queue = g_list_remove_link (tmp, tmp); 
+       tmp = async_call_queue;
+       if (tmp) {
+               ar = (MonoAsyncResult *) tmp->data;
+               tmp->data = NULL;
+               tmp2 = tmp;
+               for (tmp2 = tmp; tmp2->next != NULL; tmp2 = tmp2->next);
+               if (tmp2 != tmp) {
+                       async_call_queue = tmp->next;
+                       tmp->next = NULL;
+                       tmp2->next = tmp;
+                       tmp->prev = tmp2;
+               }
        }
        LeaveCriticalSection (&mono_delegate_section);
-       if (tmp)
-               g_list_free_1 (tmp);
 
        return ar;
 }
@@ -237,6 +281,7 @@ async_invoke_thread (gpointer data)
        thread = mono_thread_current ();
        thread->threadpool_thread = TRUE;
        thread->state |= ThreadState_Background;
+
        for (;;) {
                MonoAsyncResult *ar;
 
@@ -245,8 +290,11 @@ async_invoke_thread (gpointer data)
                        /* worker threads invokes methods in different domains,
                         * so we need to set the right domain here */
                        domain = ((MonoObject *)ar)->vtable->domain;
-                       if (mono_domain_set (domain, FALSE))
+                       if (mono_domain_set (domain, FALSE)) {
+                               mono_thread_push_appdomain_ref (domain);
                                mono_async_invoke (ar);
+                               mono_thread_pop_appdomain_ref ();
+                       }
                        InterlockedDecrement (&busy_worker_threads);
                }
 
@@ -259,7 +307,8 @@ async_invoke_thread (gpointer data)
                        
                        do {
                                wr = WaitForSingleObjectEx (job_added, (guint32)timeout, TRUE);
-                               mono_thread_interruption_checkpoint ();
+                               if ((thread->state & ThreadState_StopRequested)!=0)
+                                       mono_thread_interruption_checkpoint ();
                        
                                timeout -= GetTickCount () - start_time;
                        
@@ -268,14 +317,15 @@ async_invoke_thread (gpointer data)
                        }
                        while (!data && timeout > 0);
                }
-               
+
                if (!data) {
                        workers = (int) InterlockedCompareExchange (&mono_worker_threads, 0, -1); 
                        min = (int) InterlockedCompareExchange (&mono_min_worker_threads, 0, -1); 
        
                        while (!data && workers <= min) {
                                WaitForSingleObjectEx (job_added, INFINITE, TRUE);
-                               mono_thread_interruption_checkpoint ();
+                               if ((thread->state & ThreadState_StopRequested)!=0)
+                                       mono_thread_interruption_checkpoint ();
                        
                                data = dequeue_job ();
                                workers = (int) InterlockedCompareExchange (&mono_worker_threads, 0, -1);