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  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License 2.0 as published by the Free Software Foundation;
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License 2.0 along with this library; if not, write to the Free
  17  * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  18  */
  19
  20 #include "config.h"
  21 #ifdef HAVE_SGEN_GC
  22
  23 #include "mono/sgen/sgen-gc.h"
  24 #include "mono/sgen/sgen-thread-pool.h"
  25 #include "mono/sgen/sgen-pointer-queue.h"
  26 #include "mono/utils/mono-os-mutex.h"
  27 #ifndef SGEN_WITHOUT_MONO
  28 #include "mono/utils/mono-threads.h"
  29 #endif
  30
  31 static mono_mutex_t lock;
  32 static mono_cond_t work_cond;
  33 static mono_cond_t done_cond;
  34
  35 static MonoNativeThreadId thread;
  36
  37 /* Only accessed with the lock held. */
  38 static SgenPointerQueue job_queue;
  39
  40 static SgenThreadPoolThreadInitFunc thread_init_func;
  41 static SgenThreadPoolIdleJobFunc idle_job_func;
  42 static SgenThreadPoolContinueIdleJobFunc continue_idle_job_func;
  43
  44 static volatile gboolean threadpool_shutdown;
  45 static volatile gboolean thread_finished;
  46
  47 enum {
  48         STATE_WAITING,
  49         STATE_IN_PROGRESS,
  50         STATE_DONE
  51 };
  52
  53 /* Assumes that the lock is held. */
  54 static SgenThreadPoolJob*
  55 get_job_and_set_in_progress (void)
  56 {
  57         for (size_t i = 0; i < job_queue.next_slot; ++i) {
  58                 SgenThreadPoolJob *job = (SgenThreadPoolJob *)job_queue.data [i];
  59                 if (job->state == STATE_WAITING) {
  60                         job->state = STATE_IN_PROGRESS;
  61                         return job;
  62                 }
  63         }
  64         return NULL;
  65 }
  66
  67 /* Assumes that the lock is held. */
  68 static ssize_t
  69 find_job_in_queue (SgenThreadPoolJob *job)
  70 {
  71         for (ssize_t i = 0; i < job_queue.next_slot; ++i) {
  72                 if (job_queue.data [i] == job)
  73                         return i;
  74         }
  75         return -1;
  76 }
  77
  78 /* Assumes that the lock is held. */
  79 static void
  80 remove_job (SgenThreadPoolJob *job)
  81 {
  82         ssize_t index;
  83         SGEN_ASSERT (0, job->state == STATE_DONE, "Why are we removing a job that's not done?");
  84         index = find_job_in_queue (job);
  85         SGEN_ASSERT (0, index >= 0, "Why is the job we're trying to remove not in the queue?");
  86         job_queue.data [index] = NULL;
  87         sgen_pointer_queue_remove_nulls (&job_queue);
  88         sgen_thread_pool_job_free (job);
  89 }
  90
  91 static gboolean
  92 continue_idle_job (void)
  93 {
  94         if (!continue_idle_job_func)
  95                 return FALSE;
  96         return continue_idle_job_func ();
  97 }
  98
  99 static mono_native_thread_return_t
 100 thread_func (void *thread_data)
 101 {
 102         thread_init_func (thread_data);
 103
 104         mono_os_mutex_lock (&lock);
 105         for (;;) {
 106                 /*
 107                  * It's important that we check the continue idle flag with the lock held.
 108                  * Suppose we didn't check with the lock held, and the result is FALSE.  The
 109                  * main thread might then set continue idle and signal us before we can take
 110                  * the lock, and we'd lose the signal.
 111                  */
 112                 gboolean do_idle = continue_idle_job ();
 113                 SgenThreadPoolJob *job = get_job_and_set_in_progress ();
 114
 115                 if (!job && !do_idle && !threadpool_shutdown) {
 116                         /*
 117                          * pthread_cond_wait() can return successfully despite the condition
 118                          * not being signalled, so we have to run this in a loop until we
 119                          * really have work to do.
 120                          */
 121                         mono_os_cond_wait (&work_cond, &lock);
 122                         continue;
 123                 }
 124
 125                 mono_os_mutex_unlock (&lock);
 126
 127                 if (job) {
 128                         job->func (thread_data, job);
 129
 130                         mono_os_mutex_lock (&lock);
 131
 132                         SGEN_ASSERT (0, job->state == STATE_IN_PROGRESS, "The job should still be in progress.");
 133                         job->state = STATE_DONE;
 134                         remove_job (job);
 135                         /*
 136                          * Only the main GC thread will ever wait on the done condition, so we don't
 137                          * have to broadcast.
 138                          */
 139                         mono_os_cond_signal (&done_cond);
 140                 } else if (do_idle) {
 141                         SGEN_ASSERT (0, idle_job_func, "Why do we have idle work when there's no idle job function?");
 142                         do {
 143                                 idle_job_func (thread_data);
 144                                 do_idle = continue_idle_job ();
 145                         } while (do_idle && !job_queue.next_slot);
 146
 147                         mono_os_mutex_lock (&lock);
 148
 149                         if (!do_idle)
 150                                 mono_os_cond_signal (&done_cond);
 151                 } else {
 152                         SGEN_ASSERT (0, threadpool_shutdown, "Why did we unlock if no jobs and not shutting down?");
 153                         mono_os_mutex_lock (&lock);
 154                         thread_finished = TRUE;
 155                         mono_os_cond_signal (&done_cond);
 156                         mono_os_mutex_unlock (&lock);
 157                         return 0;
 158                 }
 159         }
 160
 161         return (mono_native_thread_return_t)0;
 162 }
 163
 164 void
 165 sgen_thread_pool_init (int num_threads, SgenThreadPoolThreadInitFunc init_func, SgenThreadPoolIdleJobFunc idle_func, SgenThreadPoolContinueIdleJobFunc continue_idle_func, void **thread_datas)
 166 {
 167         SGEN_ASSERT (0, num_threads == 1, "We only support 1 thread pool thread for now.");
 168
 169         mono_os_mutex_init (&lock);
 170         mono_os_cond_init (&work_cond);
 171         mono_os_cond_init (&done_cond);
 172
 173         thread_init_func = init_func;
 174         idle_job_func = idle_func;
 175         continue_idle_job_func = continue_idle_func;
 176
 177         mono_native_thread_create (&thread, thread_func, thread_datas ? thread_datas [0] : NULL);
 178 }
 179
 180 void
 181 sgen_thread_pool_shutdown (void)
 182 {
 183         if (!thread)
 184                 return;
 185
 186         mono_os_mutex_lock (&lock);
 187         threadpool_shutdown = TRUE;
 188         mono_os_cond_signal (&work_cond);
 189         while (!thread_finished)
 190                 mono_os_cond_wait (&done_cond, &lock);
 191         mono_os_mutex_unlock (&lock);
 192
 193         mono_os_mutex_destroy (&lock);
 194         mono_os_cond_destroy (&work_cond);
 195         mono_os_cond_destroy (&done_cond);
 196 }
 197
 198 SgenThreadPoolJob*
 199 sgen_thread_pool_job_alloc (const char *name, SgenThreadPoolJobFunc func, size_t size)
 200 {
 201         SgenThreadPoolJob *job = (SgenThreadPoolJob *)sgen_alloc_internal_dynamic (size, INTERNAL_MEM_THREAD_POOL_JOB, TRUE);
 202         job->name = name;
 203         job->size = size;
 204         job->state = STATE_WAITING;
 205         job->func = func;
 206         return job;
 207 }
 208
 209 void
 210 sgen_thread_pool_job_free (SgenThreadPoolJob *job)
 211 {
 212         sgen_free_internal_dynamic (job, job->size, INTERNAL_MEM_THREAD_POOL_JOB);
 213 }
 214
 215 void
 216 sgen_thread_pool_job_enqueue (SgenThreadPoolJob *job)
 217 {
 218         mono_os_mutex_lock (&lock);
 219
 220         sgen_pointer_queue_add (&job_queue, job);
 221         /*
 222          * FIXME: We could check whether there is a job in progress.  If there is, there's
 223          * no need to signal the condition, at least as long as we have only one thread.
 224          */
 225         mono_os_cond_signal (&work_cond);
 226
 227         mono_os_mutex_unlock (&lock);
 228 }
 229
 230 void
 231 sgen_thread_pool_job_wait (SgenThreadPoolJob *job)
 232 {
 233         SGEN_ASSERT (0, job, "Where's the job?");
 234
 235         mono_os_mutex_lock (&lock);
 236
 237         while (find_job_in_queue (job) >= 0)
 238                 mono_os_cond_wait (&done_cond, &lock);
 239
 240         mono_os_mutex_unlock (&lock);
 241 }
 242
 243 void
 244 sgen_thread_pool_idle_signal (void)
 245 {
 246         SGEN_ASSERT (0, idle_job_func, "Why are we signaling idle without an idle function?");
 247
 248         mono_os_mutex_lock (&lock);
 249
 250         if (continue_idle_job_func ())
 251                 mono_os_cond_signal (&work_cond);
 252
 253         mono_os_mutex_unlock (&lock);
 254 }
 255
 256 void
 257 sgen_thread_pool_idle_wait (void)
 258 {
 259         SGEN_ASSERT (0, idle_job_func, "Why are we waiting for idle without an idle function?");
 260
 261         mono_os_mutex_lock (&lock);
 262
 263         while (continue_idle_job_func ())
 264                 mono_os_cond_wait (&done_cond, &lock);
 265
 266         mono_os_mutex_unlock (&lock);
 267 }
 268
 269 void
 270 sgen_thread_pool_wait_for_all_jobs (void)
 271 {
 272         mono_os_mutex_lock (&lock);
 273
 274         while (!sgen_pointer_queue_is_empty (&job_queue))
 275                 mono_os_cond_wait (&done_cond, &lock);
 276
 277         mono_os_mutex_unlock (&lock);
 278 }
 279
 280 gboolean
 281 sgen_thread_pool_is_thread_pool_thread (MonoNativeThreadId some_thread)
 282 {
 283         return some_thread == thread;
 284 }
 285
 286 #endif