mono/metadata/sgen-workers.c

   1 /*
   2  * Copyright 2001-2003 Ximian, Inc
   3  * Copyright 2003-2010 Novell, Inc.
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining
   6  * a copy of this software and associated documentation files (the
   7  * "Software"), to deal in the Software without restriction, including
   8  * without limitation the rights to use, copy, modify, merge, publish,
   9  * distribute, sublicense, and/or sell copies of the Software, and to
  10  * permit persons to whom the Software is furnished to do so, subject to
  11  * the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be
  14  * included in all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  17  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  18  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  19  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  20  * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  21  * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  22  * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  23  */
  24
  25 #define STEALABLE_STACK_SIZE    512
  26
  27 typedef struct _WorkerData WorkerData;
  28 struct _WorkerData {
  29         pthread_t thread;
  30         void *major_collector_data;
  31
  32         GrayQueue private_gray_queue; /* only read/written by worker thread */
  33
  34         pthread_mutex_t stealable_stack_mutex;
  35         volatile int stealable_stack_fill;
  36         char *stealable_stack [STEALABLE_STACK_SIZE];
  37 };
  38
  39 typedef void (*JobFunc) (WorkerData *worker_data, void *job_data);
  40
  41 typedef struct _JobQueueEntry JobQueueEntry;
  42 struct _JobQueueEntry {
  43         JobFunc func;
  44         void *data;
  45
  46         volatile JobQueueEntry *next;
  47 };
  48
  49 static int workers_num;
  50 static WorkerData *workers_data;
  51 static WorkerData workers_gc_thread_data;
  52
  53 static GrayQueue workers_distribute_gray_queue;
  54 static SgenInternalAllocator workers_distribute_gray_queue_allocator;
  55
  56 #define WORKERS_DISTRIBUTE_GRAY_QUEUE (major_collector.is_parallel ? &workers_distribute_gray_queue : &gray_queue)
  57
  58 static volatile gboolean workers_gc_in_progress = FALSE;
  59 static volatile gboolean workers_marking = FALSE;
  60 static gboolean workers_started = FALSE;
  61 static volatile int workers_num_waiting = 0;
  62 static MonoSemType workers_waiting_sem;
  63 static MonoSemType workers_done_sem;
  64 static volatile int workers_done_posted = 0;
  65
  66 static volatile int workers_job_queue_num_entries = 0;
  67 static volatile JobQueueEntry *workers_job_queue = NULL;
  68 static LOCK_DECLARE (workers_job_queue_mutex);
  69
  70 static long long stat_workers_stolen_from_self_lock;
  71 static long long stat_workers_stolen_from_self_no_lock;
  72 static long long stat_workers_stolen_from_others;
  73 static long long stat_workers_num_waited;
  74
  75 static void
  76 workers_wake_up (int max)
  77 {
  78         int i;
  79
  80         for (i = 0; i < max; ++i) {
  81                 int num;
  82                 do {
  83                         num = workers_num_waiting;
  84                         if (num == 0)
  85                                 return;
  86                 } while (InterlockedCompareExchange (&workers_num_waiting, num - 1, num) != num);
  87                 MONO_SEM_POST (&workers_waiting_sem);
  88         }
  89 }
  90
  91 static void
  92 workers_wake_up_all (void)
  93 {
  94         workers_wake_up (workers_num);
  95 }
  96
  97 static void
  98 workers_wait (void)
  99 {
 100         int num;
 101         ++stat_workers_num_waited;
 102         do {
 103                 num = workers_num_waiting;
 104         } while (InterlockedCompareExchange (&workers_num_waiting, num + 1, num) != num);
 105         if (num + 1 == workers_num && !workers_gc_in_progress) {
 106                 /* Make sure the done semaphore is only posted once. */
 107                 int posted;
 108                 do {
 109                         posted = workers_done_posted;
 110                         if (posted)
 111                                 break;
 112                 } while (InterlockedCompareExchange (&workers_done_posted, 1, 0) != 0);
 113                 if (!posted)
 114                         MONO_SEM_POST (&workers_done_sem);
 115         }
 116         MONO_SEM_WAIT (&workers_waiting_sem);
 117 }
 118
 119 static void
 120 workers_enqueue_job (JobFunc func, void *data)
 121 {
 122         int num_entries;
 123         JobQueueEntry *entry = mono_sgen_alloc_internal (INTERNAL_MEM_JOB_QUEUE_ENTRY);
 124         entry->func = func;
 125         entry->data = data;
 126
 127         pthread_mutex_lock (&workers_job_queue_mutex);
 128         entry->next = workers_job_queue;
 129         workers_job_queue = entry;
 130         num_entries = ++workers_job_queue_num_entries;
 131         pthread_mutex_unlock (&workers_job_queue_mutex);
 132
 133         workers_wake_up (num_entries);
 134 }
 135
 136 static gboolean
 137 workers_dequeue_and_do_job (WorkerData *data)
 138 {
 139         JobQueueEntry *entry;
 140
 141         if (!workers_job_queue_num_entries)
 142                 return FALSE;
 143
 144         pthread_mutex_lock (&workers_job_queue_mutex);
 145         entry = (JobQueueEntry*)workers_job_queue;
 146         if (entry) {
 147                 workers_job_queue = entry->next;
 148                 --workers_job_queue_num_entries;
 149         }
 150         pthread_mutex_unlock (&workers_job_queue_mutex);
 151
 152         if (!entry)
 153                 return FALSE;
 154
 155         entry->func (data, entry->data);
 156         mono_sgen_free_internal (entry, INTERNAL_MEM_JOB_QUEUE_ENTRY);
 157         return TRUE;
 158 }
 159
 160 static gboolean
 161 workers_steal (WorkerData *data, WorkerData *victim_data, gboolean lock)
 162 {
 163         GrayQueue *queue = &data->private_gray_queue;
 164         int num, n;
 165
 166         g_assert (!queue->first);
 167
 168         if (!victim_data->stealable_stack_fill)
 169                 return FALSE;
 170
 171         if (lock && pthread_mutex_trylock (&victim_data->stealable_stack_mutex))
 172                 return FALSE;
 173
 174         n = num = (victim_data->stealable_stack_fill + 1) / 2;
 175         /* We're stealing num entries. */
 176
 177         while (n > 0) {
 178                 int m = MIN (SGEN_GRAY_QUEUE_SECTION_SIZE, n);
 179                 n -= m;
 180
 181                 gray_object_alloc_queue_section (queue);
 182                 memcpy (queue->first->objects,
 183                                 victim_data->stealable_stack + victim_data->stealable_stack_fill - num + n,
 184                                 sizeof (char*) * m);
 185                 queue->first->end = m;
 186         }
 187
 188         victim_data->stealable_stack_fill -= num;
 189
 190         if (lock)
 191                 pthread_mutex_unlock (&victim_data->stealable_stack_mutex);
 192
 193         if (data == victim_data) {
 194                 if (lock)
 195                         stat_workers_stolen_from_self_lock += num;
 196                 else
 197                         stat_workers_stolen_from_self_no_lock += num;
 198         } else {
 199                 stat_workers_stolen_from_others += num;
 200         }
 201
 202         return num != 0;
 203 }
 204
 205 static gboolean
 206 workers_get_work (WorkerData *data)
 207 {
 208         int i;
 209
 210         g_assert (gray_object_queue_is_empty (&data->private_gray_queue));
 211
 212         /* Try to steal from our own stack. */
 213         if (workers_steal (data, data, TRUE))
 214                 return TRUE;
 215
 216         /* Then from the GC thread's stack. */
 217         if (workers_steal (data, &workers_gc_thread_data, TRUE))
 218                 return TRUE;
 219
 220         /* Finally, from another worker. */
 221         for (i = 0; i < workers_num; ++i) {
 222                 WorkerData *victim_data = &workers_data [i];
 223                 if (data == victim_data)
 224                         continue;
 225                 if (workers_steal (data, victim_data, TRUE))
 226                         return TRUE;
 227         }
 228
 229         /* Nobody to steal from */
 230         g_assert (gray_object_queue_is_empty (&data->private_gray_queue));
 231         return FALSE;
 232 }
 233
 234 static void
 235 workers_gray_queue_share_redirect (GrayQueue *queue)
 236 {
 237         GrayQueueSection *section;
 238         WorkerData *data = queue->alloc_prepare_data;
 239
 240         if (data->stealable_stack_fill) {
 241                 /*
 242                  * There are still objects in the stealable stack, so
 243                  * wake up any workers that might be sleeping
 244                  */
 245                 if (workers_gc_in_progress)
 246                         workers_wake_up_all ();
 247                 return;
 248         }
 249
 250         /* The stealable stack is empty, so fill it. */
 251         pthread_mutex_lock (&data->stealable_stack_mutex);
 252
 253         while (data->stealable_stack_fill < STEALABLE_STACK_SIZE &&
 254                         (section = gray_object_dequeue_section (queue))) {
 255                 int num = MIN (section->end, STEALABLE_STACK_SIZE - data->stealable_stack_fill);
 256
 257                 memcpy (data->stealable_stack + data->stealable_stack_fill,
 258                                 section->objects + section->end - num,
 259                                 sizeof (char*) * num);
 260
 261                 section->end -= num;
 262                 data->stealable_stack_fill += num;
 263
 264                 if (section->end)
 265                         gray_object_enqueue_section (queue, section);
 266                 else
 267                         gray_object_free_queue_section (section, queue->allocator);
 268         }
 269
 270         if (data != &workers_gc_thread_data && gray_object_queue_is_empty (queue))
 271                 workers_steal (data, data, FALSE);
 272
 273         pthread_mutex_unlock (&data->stealable_stack_mutex);
 274
 275         if (workers_gc_in_progress)
 276                 workers_wake_up_all ();
 277 }
 278
 279 static void*
 280 workers_thread_func (void *data_untyped)
 281 {
 282         WorkerData *data = data_untyped;
 283         SgenInternalAllocator allocator;
 284
 285         if (major_collector.init_worker_thread)
 286                 major_collector.init_worker_thread (data->major_collector_data);
 287
 288         memset (&allocator, 0, sizeof (allocator));
 289 #ifdef SGEN_DEBUG_INTERNAL_ALLOC
 290         allocator.thread = pthread_self ();
 291 #endif
 292
 293         gray_object_queue_init_with_alloc_prepare (&data->private_gray_queue, &allocator,
 294                         workers_gray_queue_share_redirect, data);
 295
 296         for (;;) {
 297                 gboolean did_work = FALSE;
 298
 299                 while (workers_dequeue_and_do_job (data))
 300                         did_work = TRUE;
 301
 302                 if (workers_marking && workers_get_work (data)) {
 303                         g_assert (!gray_object_queue_is_empty (&data->private_gray_queue));
 304
 305                         while (!drain_gray_stack (&data->private_gray_queue, 32))
 306                                 workers_gray_queue_share_redirect (&data->private_gray_queue);
 307                         g_assert (gray_object_queue_is_empty (&data->private_gray_queue));
 308
 309                         gray_object_queue_init (&data->private_gray_queue, &allocator);
 310
 311                         did_work = TRUE;
 312                 }
 313
 314                 if (!did_work)
 315                         workers_wait ();
 316         }
 317
 318         /* dummy return to make compilers happy */
 319         return NULL;
 320 }
 321
 322 static void
 323 workers_distribute_gray_queue_sections (void)
 324 {
 325         if (!major_collector.is_parallel)
 326                 return;
 327
 328         workers_gray_queue_share_redirect (&workers_distribute_gray_queue);
 329 }
 330
 331 static void
 332 workers_init_distribute_gray_queue (void)
 333 {
 334         if (!major_collector.is_parallel)
 335                 return;
 336
 337         gray_object_queue_init (&workers_distribute_gray_queue, &workers_distribute_gray_queue_allocator);
 338 #ifdef SGEN_DEBUG_INTERNAL_ALLOC
 339         workers_distribute_gray_queue_allocator.thread = pthread_self ();
 340 #endif
 341 }
 342
 343 static void
 344 workers_init (int num_workers)
 345 {
 346         int i;
 347
 348         if (!major_collector.is_parallel)
 349                 return;
 350
 351         //g_print ("initing %d workers\n", num_workers);
 352
 353         workers_num = num_workers;
 354
 355         workers_data = mono_sgen_alloc_internal_dynamic (sizeof (WorkerData) * num_workers, INTERNAL_MEM_WORKER_DATA);
 356         memset (workers_data, 0, sizeof (WorkerData) * num_workers);
 357
 358         MONO_SEM_INIT (&workers_waiting_sem, 0);
 359         MONO_SEM_INIT (&workers_done_sem, 0);
 360
 361         gray_object_queue_init_with_alloc_prepare (&workers_distribute_gray_queue, &workers_distribute_gray_queue_allocator,
 362                         workers_gray_queue_share_redirect, &workers_gc_thread_data);
 363         pthread_mutex_init (&workers_gc_thread_data.stealable_stack_mutex, NULL);
 364         workers_gc_thread_data.stealable_stack_fill = 0;
 365
 366         if (major_collector.alloc_worker_data)
 367                 workers_gc_thread_data.major_collector_data = major_collector.alloc_worker_data ();
 368
 369         for (i = 0; i < workers_num; ++i) {
 370                 /* private gray queue is inited by the thread itself */
 371                 pthread_mutex_init (&workers_data [i].stealable_stack_mutex, NULL);
 372                 workers_data [i].stealable_stack_fill = 0;
 373
 374                 if (major_collector.alloc_worker_data)
 375                         workers_data [i].major_collector_data = major_collector.alloc_worker_data ();
 376         }
 377
 378         LOCK_INIT (workers_job_queue_mutex);
 379
 380         mono_sgen_register_fixed_internal_mem_type (INTERNAL_MEM_JOB_QUEUE_ENTRY, sizeof (JobQueueEntry));
 381
 382         mono_counters_register ("Stolen from self lock", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_workers_stolen_from_self_lock);
 383         mono_counters_register ("Stolen from self no lock", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_workers_stolen_from_self_no_lock);
 384         mono_counters_register ("Stolen from others", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_workers_stolen_from_others);
 385         mono_counters_register ("# workers waited", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_workers_num_waited);
 386 }
 387
 388 /* only the GC thread is allowed to start and join workers */
 389
 390 static void
 391 workers_start_worker (int index)
 392 {
 393         g_assert (index >= 0 && index < workers_num);
 394
 395         g_assert (!workers_data [index].thread);
 396         pthread_create (&workers_data [index].thread, NULL, workers_thread_func, &workers_data [index]);
 397 }
 398
 399 static void
 400 workers_start_all_workers (void)
 401 {
 402         int i;
 403
 404         if (!major_collector.is_parallel)
 405                 return;
 406
 407         if (major_collector.init_worker_thread)
 408                 major_collector.init_worker_thread (workers_gc_thread_data.major_collector_data);
 409
 410         g_assert (!workers_gc_in_progress);
 411         workers_gc_in_progress = TRUE;
 412         workers_marking = FALSE;
 413         workers_done_posted = 0;
 414
 415         if (workers_started) {
 416                 g_assert (workers_num_waiting == workers_num);
 417                 workers_wake_up_all ();
 418                 return;
 419         }
 420
 421         for (i = 0; i < workers_num; ++i)
 422                 workers_start_worker (i);
 423
 424         workers_started = TRUE;
 425 }
 426
 427 static void
 428 workers_start_marking (void)
 429 {
 430         g_assert (workers_started && workers_gc_in_progress);
 431         g_assert (!workers_marking);
 432
 433         workers_marking = TRUE;
 434
 435         workers_wake_up_all ();
 436 }
 437
 438 static void
 439 workers_join (void)
 440 {
 441         int i;
 442
 443         if (!major_collector.is_parallel)
 444                 return;
 445
 446         g_assert (gray_object_queue_is_empty (&workers_gc_thread_data.private_gray_queue));
 447         g_assert (gray_object_queue_is_empty (&workers_distribute_gray_queue));
 448
 449         g_assert (workers_gc_in_progress);
 450         workers_gc_in_progress = FALSE;
 451         if (workers_num_waiting == workers_num)
 452                 workers_wake_up_all ();
 453         MONO_SEM_WAIT (&workers_done_sem);
 454         workers_marking = FALSE;
 455
 456         if (major_collector.reset_worker_data) {
 457                 for (i = 0; i < workers_num; ++i)
 458                         major_collector.reset_worker_data (workers_data [i].major_collector_data);
 459         }
 460
 461         g_assert (workers_done_posted);
 462         g_assert (workers_num_waiting == workers_num);
 463
 464         g_assert (!workers_gc_thread_data.stealable_stack_fill);
 465         g_assert (gray_object_queue_is_empty (&workers_gc_thread_data.private_gray_queue));
 466         for (i = 0; i < workers_num; ++i) {
 467                 g_assert (!workers_data [i].stealable_stack_fill);
 468                 g_assert (gray_object_queue_is_empty (&workers_data [i].private_gray_queue));
 469         }
 470 }
 471
 472 gboolean
 473 mono_sgen_is_worker_thread (pthread_t thread)
 474 {
 475         int i;
 476
 477         if (major_collector.is_worker_thread && major_collector.is_worker_thread (thread))
 478                 return TRUE;
 479
 480         if (!major_collector.is_parallel)
 481                 return FALSE;
 482
 483         for (i = 0; i < workers_num; ++i) {
 484                 if (workers_data [i].thread == thread)
 485                         return TRUE;
 486         }
 487         return FALSE;
 488 }