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