mono/metadata/threadpool.c

   1 /*
   2  * threadpool.c: global thread pool
   3  *
   4  * Authors:
   5  *   Dietmar Maurer (dietmar@ximian.com)
   6  *
   7  * (C) 2001 Ximian, Inc.
   8  */
   9
  10 #include <config.h>
  11 #include <glib.h>
  12
  13 #include <mono/metadata/appdomain.h>
  14 #include <mono/metadata/tabledefs.h>
  15 #include <mono/metadata/threads.h>
  16 #include <mono/metadata/exception.h>
  17 #ifndef PLATFORM_WIN32
  18 #include "mono/io-layer/wapi.h"
  19 #include "mono/io-layer/uglify.h"
  20 #endif
  21
  22 #include "threadpool.h"
  23
  24 /* FIXME:
  25  * - worker threads need to be initialized correctly.
  26  * - worker threads should be domain specific
  27  */
  28
  29 /* maximum number of worker threads */
  30 int mono_worker_threads = 1;
  31
  32 static int workers = 0;
  33
  34 typedef struct {
  35         MonoMethodMessage *msg;
  36         HANDLE             wait_semaphore;
  37         MonoMethod        *cb_method;
  38         MonoDelegate      *cb_target;
  39         MonoObject        *state;
  40         MonoObject        *res;
  41         MonoArray         *out_args;
  42 } ASyncCall;
  43
  44 static void async_invoke_thread (void);
  45
  46 static GList *async_call_queue = NULL;
  47
  48 static void
  49 mono_async_invoke (MonoAsyncResult *ares)
  50 {
  51         ASyncCall *ac = (ASyncCall *)ares->data;
  52
  53         ac->msg->exc = NULL;
  54         ac->res = mono_message_invoke (ares->async_delegate, ac->msg,
  55                                        &ac->msg->exc, &ac->out_args);
  56
  57         ares->completed = 1;
  58
  59         /* notify listeners */
  60         ReleaseSemaphore (ac->wait_semaphore, 0x7fffffff, NULL);
  61
  62         /* call async callback if cb_method != null*/
  63         if (ac->cb_method) {
  64                 MonoObject *exc = NULL;
  65                 void *pa = &ares;
  66                 mono_runtime_invoke (ac->cb_method, ac->cb_target, pa, &exc);
  67                 if (!ac->msg->exc)
  68                         ac->msg->exc = exc;
  69         }
  70 }
  71
  72 MonoAsyncResult *
  73 mono_thread_pool_add (MonoObject *target, MonoMethodMessage *msg, MonoDelegate *async_callback,
  74                       MonoObject *state)
  75 {
  76         MonoDomain *domain = mono_domain_get ();
  77         MonoAsyncResult *ares;
  78         ASyncCall *ac;
  79
  80         ac = g_new0 (ASyncCall, 1);
  81         ac->wait_semaphore = CreateSemaphore (NULL, 0, 0x7fffffff, NULL);
  82         ac->msg = msg;
  83         ac->state = state;
  84
  85         if (async_callback) {
  86                 ac->cb_method = mono_get_delegate_invoke (((MonoObject *)async_callback)->vtable->klass);
  87                 ac->cb_target = async_callback;
  88         }
  89
  90         ares = mono_async_result_new (domain, ac->wait_semaphore, ac->state, ac);
  91         ares->async_delegate = target;
  92
  93         EnterCriticalSection (&mono_delegate_section);
  94         async_call_queue = g_list_append (async_call_queue, ares);
  95         ReleaseSemaphore (mono_delegate_semaphore, 1, NULL);
  96
  97         if (workers == 0) {
  98                 MonoObject *thread;
  99                 workers++;
 100                 thread = mono_thread_create (domain, async_invoke_thread);
 101                 g_assert (thread != NULL);
 102         }
 103         LeaveCriticalSection (&mono_delegate_section);
 104
 105         return ares;
 106 }
 107
 108 MonoObject *
 109 mono_thread_pool_finish (MonoAsyncResult *ares, MonoArray **out_args, MonoObject **exc)
 110 {
 111         ASyncCall *ac;
 112         GList *l;
 113
 114         *exc = NULL;
 115         *out_args = NULL;
 116
 117         EnterCriticalSection (&mono_delegate_section);
 118         /* check if already finished */
 119         if (ares->endinvoke_called) {
 120                 *exc = (MonoObject *)mono_exception_from_name (mono_defaults.corlib, "System",
 121                                               "InvalidOperationException");
 122                 LeaveCriticalSection (&mono_delegate_section);
 123                 return NULL;
 124         }
 125
 126         ares->endinvoke_called = 1;
 127         ac = (ASyncCall *)ares->data;
 128
 129         g_assert (ac != NULL);
 130
 131         if ((l = g_list_find (async_call_queue, ares))) {
 132                 async_call_queue = g_list_remove_link (async_call_queue, l);
 133                 mono_async_invoke (ares);
 134         }
 135         LeaveCriticalSection (&mono_delegate_section);
 136
 137         /* wait until we are really finished */
 138         WaitForSingleObject (ac->wait_semaphore, INFINITE);
 139
 140         *exc = ac->msg->exc;
 141         *out_args = ac->out_args;
 142
 143         return ac->res;
 144 }
 145
 146 static void
 147 async_invoke_thread ()
 148 {
 149         MonoDomain *domain;
 150
 151         for (;;) {
 152                 MonoAsyncResult *ar;
 153                 gboolean new_worker = FALSE;
 154
 155                 if (WaitForSingleObject (mono_delegate_semaphore, 3000) == WAIT_TIMEOUT) {
 156                         EnterCriticalSection (&mono_delegate_section);
 157                         workers--;
 158                         LeaveCriticalSection (&mono_delegate_section);
 159                         ExitThread (0);
 160                 }
 161
 162                 ar = NULL;
 163                 EnterCriticalSection (&mono_delegate_section);
 164
 165                 if (async_call_queue) {
 166                         if ((g_list_length (async_call_queue) > 1) &&
 167                             (workers < mono_worker_threads)) {
 168                                 new_worker = TRUE;
 169                                 workers++;
 170                         }
 171
 172                         ar = (MonoAsyncResult *)async_call_queue->data;
 173                         async_call_queue = g_list_remove_link (async_call_queue, async_call_queue);
 174
 175                 }
 176
 177                 LeaveCriticalSection (&mono_delegate_section);
 178
 179                 if (mono_runtime_shutdown) {
 180                         EnterCriticalSection (&mono_delegate_section);
 181                         workers--;
 182                         LeaveCriticalSection (&mono_delegate_section);
 183                         ExitThread (0);
 184                 }
 185
 186                 if (!ar)
 187                         continue;
 188
 189                 /* worker threads invokes methods in different domains,
 190                  * so we need to set the right domain here */
 191                 domain = ((MonoObject *)ar)->vtable->domain;
 192                 mono_domain_set (domain);
 193
 194                 if (new_worker)
 195                         mono_thread_create (domain, async_invoke_thread);
 196
 197                 mono_async_invoke (ar);
 198         }
 199
 200         g_assert_not_reached ();
 201 }