mono/sgen/sgen-thread-pool.c

   1 /*
   2  * sgen-thread-pool.c: Threadpool for all concurrent GC work.
   3  *
   4  * Copyright (C) 2015 Xamarin Inc
   5  *
   6  * Licensed under the MIT license. See LICENSE file in the project root for full license information.
   7  */
   8
   9 #include "config.h"
  10 #ifdef HAVE_SGEN_GC
  11
  12 #include "mono/sgen/sgen-gc.h"
  13 #include "mono/sgen/sgen-thread-pool.h"
  14 #include "mono/sgen/sgen-pointer-queue.h"
  15 #include "mono/utils/mono-os-mutex.h"
  16 #ifndef SGEN_WITHOUT_MONO
  17 #include "mono/utils/mono-threads.h"
  18 #endif
  19
  20 static mono_mutex_t lock;
  21 static mono_cond_t work_cond;
  22 static mono_cond_t done_cond;
  23
  24 static MonoNativeThreadId thread;
  25
  26 /* Only accessed with the lock held. */
  27 static SgenPointerQueue job_queue;
  28
  29 static SgenThreadPoolThreadInitFunc thread_init_func;
  30 static SgenThreadPoolIdleJobFunc idle_job_func;
  31 static SgenThreadPoolContinueIdleJobFunc continue_idle_job_func;
  32
  33 enum {
  34         STATE_WAITING,
  35         STATE_IN_PROGRESS,
  36         STATE_DONE
  37 };
  38
  39 /* Assumes that the lock is held. */
  40 static SgenThreadPoolJob*
  41 get_job_and_set_in_progress (void)
  42 {
  43         for (size_t i = 0; i < job_queue.next_slot; ++i) {
  44                 SgenThreadPoolJob *job = (SgenThreadPoolJob *)job_queue.data [i];
  45                 if (job->state == STATE_WAITING) {
  46                         job->state = STATE_IN_PROGRESS;
  47                         return job;
  48                 }
  49         }
  50         return NULL;
  51 }
  52
  53 /* Assumes that the lock is held. */
  54 static ssize_t
  55 find_job_in_queue (SgenThreadPoolJob *job)
  56 {
  57         for (ssize_t i = 0; i < job_queue.next_slot; ++i) {
  58                 if (job_queue.data [i] == job)
  59                         return i;
  60         }
  61         return -1;
  62 }
  63
  64 /* Assumes that the lock is held. */
  65 static void
  66 remove_job (SgenThreadPoolJob *job)
  67 {
  68         ssize_t index;
  69         SGEN_ASSERT (0, job->state == STATE_DONE, "Why are we removing a job that's not done?");
  70         index = find_job_in_queue (job);
  71         SGEN_ASSERT (0, index >= 0, "Why is the job we're trying to remove not in the queue?");
  72         job_queue.data [index] = NULL;
  73         sgen_pointer_queue_remove_nulls (&job_queue);
  74         sgen_thread_pool_job_free (job);
  75 }
  76
  77 static gboolean
  78 continue_idle_job (void)
  79 {
  80         if (!continue_idle_job_func)
  81                 return FALSE;
  82         return continue_idle_job_func ();
  83 }
  84
  85 static mono_native_thread_return_t
  86 thread_func (void *thread_data)
  87 {
  88         thread_init_func (thread_data);
  89
  90         mono_os_mutex_lock (&lock);
  91         for (;;) {
  92                 /*
  93                  * It's important that we check the continue idle flag with the lock held.
  94                  * Suppose we didn't check with the lock held, and the result is FALSE.  The
  95                  * main thread might then set continue idle and signal us before we can take
  96                  * the lock, and we'd lose the signal.
  97                  */
  98                 gboolean do_idle = continue_idle_job ();
  99                 SgenThreadPoolJob *job = get_job_and_set_in_progress ();
 100
 101                 if (!job && !do_idle) {
 102                         /*
 103                          * pthread_cond_wait() can return successfully despite the condition
 104                          * not being signalled, so we have to run this in a loop until we
 105                          * really have work to do.
 106                          */
 107                         mono_os_cond_wait (&work_cond, &lock);
 108                         continue;
 109                 }
 110
 111                 mono_os_mutex_unlock (&lock);
 112
 113                 if (job) {
 114                         job->func (thread_data, job);
 115
 116                         mono_os_mutex_lock (&lock);
 117
 118                         SGEN_ASSERT (0, job->state == STATE_IN_PROGRESS, "The job should still be in progress.");
 119                         job->state = STATE_DONE;
 120                         remove_job (job);
 121                         /*
 122                          * Only the main GC thread will ever wait on the done condition, so we don't
 123                          * have to broadcast.
 124                          */
 125                         mono_os_cond_signal (&done_cond);
 126                 } else {
 127                         SGEN_ASSERT (0, do_idle, "Why did we unlock if we still have to wait for idle?");
 128                         SGEN_ASSERT (0, idle_job_func, "Why do we have idle work when there's no idle job function?");
 129                         do {
 130                                 idle_job_func (thread_data);
 131                                 do_idle = continue_idle_job ();
 132                         } while (do_idle && !job_queue.next_slot);
 133
 134                         mono_os_mutex_lock (&lock);
 135
 136                         if (!do_idle)
 137                                 mono_os_cond_signal (&done_cond);
 138                 }
 139         }
 140
 141         return (mono_native_thread_return_t)0;
 142 }
 143
 144 void
 145 sgen_thread_pool_init (int num_threads, SgenThreadPoolThreadInitFunc init_func, SgenThreadPoolIdleJobFunc idle_func, SgenThreadPoolContinueIdleJobFunc continue_idle_func, void **thread_datas)
 146 {
 147         SGEN_ASSERT (0, num_threads == 1, "We only support 1 thread pool thread for now.");
 148
 149         mono_os_mutex_init (&lock);
 150         mono_os_cond_init (&work_cond);
 151         mono_os_cond_init (&done_cond);
 152
 153         thread_init_func = init_func;
 154         idle_job_func = idle_func;
 155         continue_idle_job_func = continue_idle_func;
 156
 157         mono_native_thread_create (&thread, thread_func, thread_datas ? thread_datas [0] : NULL);
 158 }
 159
 160 SgenThreadPoolJob*
 161 sgen_thread_pool_job_alloc (const char *name, SgenThreadPoolJobFunc func, size_t size)
 162 {
 163         SgenThreadPoolJob *job = (SgenThreadPoolJob *)sgen_alloc_internal_dynamic (size, INTERNAL_MEM_THREAD_POOL_JOB, TRUE);
 164         job->name = name;
 165         job->size = size;
 166         job->state = STATE_WAITING;
 167         job->func = func;
 168         return job;
 169 }
 170
 171 void
 172 sgen_thread_pool_job_free (SgenThreadPoolJob *job)
 173 {
 174         sgen_free_internal_dynamic (job, job->size, INTERNAL_MEM_THREAD_POOL_JOB);
 175 }
 176
 177 void
 178 sgen_thread_pool_job_enqueue (SgenThreadPoolJob *job)
 179 {
 180         mono_os_mutex_lock (&lock);
 181
 182         sgen_pointer_queue_add (&job_queue, job);
 183         /*
 184          * FIXME: We could check whether there is a job in progress.  If there is, there's
 185          * no need to signal the condition, at least as long as we have only one thread.
 186          */
 187         mono_os_cond_signal (&work_cond);
 188
 189         mono_os_mutex_unlock (&lock);
 190 }
 191
 192 void
 193 sgen_thread_pool_job_wait (SgenThreadPoolJob *job)
 194 {
 195         SGEN_ASSERT (0, job, "Where's the job?");
 196
 197         mono_os_mutex_lock (&lock);
 198
 199         while (find_job_in_queue (job) >= 0)
 200                 mono_os_cond_wait (&done_cond, &lock);
 201
 202         mono_os_mutex_unlock (&lock);
 203 }
 204
 205 void
 206 sgen_thread_pool_idle_signal (void)
 207 {
 208         SGEN_ASSERT (0, idle_job_func, "Why are we signaling idle without an idle function?");
 209
 210         mono_os_mutex_lock (&lock);
 211
 212         if (continue_idle_job_func ())
 213                 mono_os_cond_signal (&work_cond);
 214
 215         mono_os_mutex_unlock (&lock);
 216 }
 217
 218 void
 219 sgen_thread_pool_idle_wait (void)
 220 {
 221         SGEN_ASSERT (0, idle_job_func, "Why are we waiting for idle without an idle function?");
 222
 223         mono_os_mutex_lock (&lock);
 224
 225         while (continue_idle_job_func ())
 226                 mono_os_cond_wait (&done_cond, &lock);
 227
 228         mono_os_mutex_unlock (&lock);
 229 }
 230
 231 void
 232 sgen_thread_pool_wait_for_all_jobs (void)
 233 {
 234         mono_os_mutex_lock (&lock);
 235
 236         while (!sgen_pointer_queue_is_empty (&job_queue))
 237                 mono_os_cond_wait (&done_cond, &lock);
 238
 239         mono_os_mutex_unlock (&lock);
 240 }
 241
 242 gboolean
 243 sgen_thread_pool_is_thread_pool_thread (MonoNativeThreadId some_thread)
 244 {
 245         return some_thread == thread;
 246 }
 247
 248 #endif