mono/metadata/sgen-workers.c

   1 /*
   2  * sgen-workers.c: Worker threads for parallel and concurrent GC.
   3  *
   4  * Copyright 2001-2003 Ximian, Inc
   5  * Copyright 2003-2010 Novell, Inc.
   6  * Copyright (C) 2012 Xamarin Inc
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Library General Public
  10  * License 2.0 as published by the Free Software Foundation;
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License 2.0 along with this library; if not, write to the Free
  19  * Software Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22 #include "config.h"
  23 #ifdef HAVE_SGEN_GC
  24
  25 #include "metadata/sgen-gc.h"
  26 #include "metadata/sgen-workers.h"
  27 #include "utils/mono-counters.h"
  28
  29 static int workers_num;
  30 static WorkerData *workers_data;
  31 static void *workers_gc_thread_major_collector_data = NULL;
  32
  33 static SgenSectionGrayQueue workers_distribute_gray_queue;
  34 static gboolean workers_distribute_gray_queue_inited;
  35
  36 static gboolean workers_started = FALSE;
  37
  38 enum {
  39         STATE_NOT_WORKING,
  40         STATE_WORKING,
  41         STATE_NURSERY_COLLECTION
  42 } WorkersStateName;
  43
  44 /*
  45  * | state                    | num_awake | num_posted                 | post_done |
  46  * |--------------------------+-----------+----------------------------+-----------|
  47  * | STATE_NOT_WORKING        | 0         | *                          |         0 |
  48  * | STATE_WORKING            | > 0       | <= workers_num - num_awake |         * |
  49  * | STATE_NURSERY_COLLECTION | *         | <= workers_num - num_awake |         1 |
  50  * | STATE_NURSERY_COLLECTION | 0         | 0                          |         0 |
  51  */
  52 typedef union {
  53         gint32 value;
  54         struct {
  55                 guint state : 4; /* WorkersStateName */
  56                 /* Number of worker threads awake. */
  57                 guint num_awake : 8;
  58                 /* The state of the waiting semaphore. */
  59                 guint num_posted : 8;
  60                 /* Whether to post `workers_done_sem` */
  61                 guint post_done : 1;
  62         } data;
  63 } State;
  64
  65 static volatile State workers_state;
  66
  67 static MonoSemType workers_waiting_sem;
  68 static MonoSemType workers_done_sem;
  69
  70 static volatile int workers_job_queue_num_entries = 0;
  71 static volatile JobQueueEntry *workers_job_queue = NULL;
  72 static LOCK_DECLARE (workers_job_queue_mutex);
  73 static int workers_num_jobs_enqueued = 0;
  74 static volatile int workers_num_jobs_finished = 0;
  75
  76 static guint64 stat_workers_stolen_from_self_lock;
  77 static guint64 stat_workers_stolen_from_self_no_lock;
  78 static guint64 stat_workers_stolen_from_others;
  79 static guint64 stat_workers_num_waited;
  80
  81 static gboolean
  82 set_state (State old_state, State new_state)
  83 {
  84         if (old_state.data.state == STATE_NURSERY_COLLECTION)
  85                 SGEN_ASSERT (0, new_state.data.state != STATE_NOT_WORKING, "Can't go from nursery collection to not working");
  86
  87         return InterlockedCompareExchange (&workers_state.value,
  88                         new_state.value, old_state.value) == old_state.value;
  89 }
  90
  91 static void
  92 assert_not_working (State state)
  93 {
  94         SGEN_ASSERT (0, state.data.state == STATE_NOT_WORKING, "Can only signal enqueue work when in no work state");
  95         SGEN_ASSERT (0, state.data.num_awake == 0, "No workers can be awake when not working");
  96         SGEN_ASSERT (0, state.data.num_posted == 0, "Can't have posted already");
  97         SGEN_ASSERT (0, !state.data.post_done, "post_done can only be set when working");
  98
  99 }
 100
 101 static void
 102 assert_working (State state, gboolean from_worker)
 103 {
 104         SGEN_ASSERT (0, state.data.state == STATE_WORKING, "A worker can't wait without being in working state");
 105         if (from_worker)
 106                 SGEN_ASSERT (0, state.data.num_awake > 0, "How can we be awake, yet we are not counted?");
 107         else
 108                 SGEN_ASSERT (0, state.data.num_awake + state.data.num_posted > 0, "How can we be working, yet no worker threads are awake or to be awoken?");
 109         SGEN_ASSERT (0, state.data.num_awake + state.data.num_posted <= workers_num, "There are too many worker threads awake");
 110 }
 111
 112 static void
 113 assert_nursery_collection (State state, gboolean from_worker)
 114 {
 115         SGEN_ASSERT (0, state.data.state == STATE_NURSERY_COLLECTION, "Must be in the nursery collection state");
 116         if (from_worker) {
 117                 SGEN_ASSERT (0, state.data.num_awake > 0, "We're awake, but num_awake is zero");
 118                 SGEN_ASSERT (0, state.data.post_done, "post_done must be set in the nursery collection state");
 119         }
 120         SGEN_ASSERT (0, state.data.num_awake <= workers_num, "There are too many worker threads awake");
 121         if (!state.data.post_done) {
 122                 SGEN_ASSERT (0, state.data.num_awake == 0, "Once done has been posted no threads can be awake");
 123                 SGEN_ASSERT (0, state.data.num_posted == 0, "Once done has been posted no thread must be awoken");
 124         }
 125 }
 126
 127 static void
 128 assert_working_or_nursery_collection (State state)
 129 {
 130         if (state.data.state == STATE_WORKING)
 131                 assert_working (state, TRUE);
 132         else
 133                 assert_nursery_collection (state, TRUE);
 134 }
 135
 136 static void
 137 workers_signal_enqueue_work (int num_wake_up, gboolean from_nursery_collection)
 138 {
 139         State old_state = workers_state;
 140         State new_state = old_state;
 141         int i;
 142         gboolean did_set_state;
 143
 144         SGEN_ASSERT (0, num_wake_up <= workers_num, "Cannot wake up more workers than are present");
 145
 146         if (from_nursery_collection)
 147                 assert_nursery_collection (old_state, FALSE);
 148         else
 149                 assert_not_working (old_state);
 150
 151         new_state.data.state = STATE_WORKING;
 152         new_state.data.num_posted = num_wake_up;
 153
 154         did_set_state = set_state (old_state, new_state);
 155         SGEN_ASSERT (0, did_set_state, "Nobody else should be mutating the state");
 156
 157         for (i = 0; i < num_wake_up; ++i)
 158                 MONO_SEM_POST (&workers_waiting_sem);
 159 }
 160
 161 static void
 162 workers_signal_enqueue_work_if_necessary (int num_wake_up)
 163 {
 164         if (workers_state.data.state == STATE_NOT_WORKING)
 165                 workers_signal_enqueue_work (num_wake_up, FALSE);
 166 }
 167
 168 void
 169 sgen_workers_ensure_awake (void)
 170 {
 171         SGEN_ASSERT (0, workers_state.data.state != STATE_NURSERY_COLLECTION, "Can't wake workers during nursery collection");
 172         workers_signal_enqueue_work_if_necessary (workers_num);
 173 }
 174
 175 static void
 176 workers_wait (void)
 177 {
 178         State old_state, new_state;
 179         gboolean post_done;
 180
 181         ++stat_workers_num_waited;
 182
 183         do {
 184                 new_state = old_state = workers_state;
 185
 186                 assert_working_or_nursery_collection (old_state);
 187
 188                 --new_state.data.num_awake;
 189                 post_done = FALSE;
 190                 if (!new_state.data.num_awake && !new_state.data.num_posted) {
 191                         /* We are the last thread to go to sleep. */
 192                         if (old_state.data.state == STATE_WORKING)
 193                                 new_state.data.state = STATE_NOT_WORKING;
 194
 195                         new_state.data.post_done = 0;
 196                         if (old_state.data.post_done)
 197                                 post_done = TRUE;
 198                 }
 199         } while (!set_state (old_state, new_state));
 200
 201         if (post_done)
 202                 MONO_SEM_POST (&workers_done_sem);
 203
 204         MONO_SEM_WAIT (&workers_waiting_sem);
 205
 206         do {
 207                 new_state = old_state = workers_state;
 208
 209                 SGEN_ASSERT (0, old_state.data.num_posted > 0, "How can we be awake without the semaphore having been posted?");
 210                 SGEN_ASSERT (0, old_state.data.num_awake < workers_num, "There are too many worker threads awake");
 211
 212                 --new_state.data.num_posted;
 213                 ++new_state.data.num_awake;
 214
 215                 assert_working_or_nursery_collection (new_state);
 216         } while (!set_state (old_state, new_state));
 217 }
 218
 219 static gboolean
 220 collection_needs_workers (void)
 221 {
 222         return sgen_collection_is_concurrent ();
 223 }
 224
 225 void
 226 sgen_workers_enqueue_job (const char *name, JobFunc func, void *data)
 227 {
 228         int num_entries;
 229         JobQueueEntry *entry;
 230
 231         if (!collection_needs_workers ()) {
 232                 func (NULL, data);
 233                 return;
 234         }
 235
 236         entry = sgen_alloc_internal (INTERNAL_MEM_JOB_QUEUE_ENTRY);
 237         entry->name = name;
 238         entry->func = func;
 239         entry->data = data;
 240
 241         mono_mutex_lock (&workers_job_queue_mutex);
 242         entry->next = workers_job_queue;
 243         workers_job_queue = entry;
 244         num_entries = ++workers_job_queue_num_entries;
 245         ++workers_num_jobs_enqueued;
 246         mono_mutex_unlock (&workers_job_queue_mutex);
 247
 248         if (workers_state.data.state != STATE_NURSERY_COLLECTION)
 249                 workers_signal_enqueue_work_if_necessary (num_entries < workers_num ? num_entries : workers_num);
 250 }
 251
 252 void
 253 sgen_workers_wait_for_jobs_finished (void)
 254 {
 255         // FIXME: implement this properly
 256         while (workers_num_jobs_finished < workers_num_jobs_enqueued) {
 257                 workers_signal_enqueue_work_if_necessary (workers_num);
 258                 /* FIXME: sleep less? */
 259                 g_usleep (1000);
 260         }
 261 }
 262
 263 void
 264 sgen_workers_signal_start_nursery_collection_and_wait (void)
 265 {
 266         State old_state, new_state;
 267
 268         do {
 269                 new_state = old_state = workers_state;
 270
 271                 new_state.data.state = STATE_NURSERY_COLLECTION;
 272
 273                 if (old_state.data.state == STATE_NOT_WORKING) {
 274                         assert_not_working (old_state);
 275                 } else {
 276                         assert_working (old_state, FALSE);
 277                         SGEN_ASSERT (0, !old_state.data.post_done, "We are not waiting for the workers");
 278
 279                         new_state.data.post_done = 1;
 280                 }
 281         } while (!set_state (old_state, new_state));
 282
 283         if (new_state.data.post_done)
 284                 MONO_SEM_WAIT (&workers_done_sem);
 285
 286         old_state = workers_state;
 287         assert_nursery_collection (old_state, FALSE);
 288         SGEN_ASSERT (0, !old_state.data.post_done, "We got the semaphore, so it must have been posted");
 289 }
 290
 291 void
 292 sgen_workers_signal_finish_nursery_collection (void)
 293 {
 294         State old_state = workers_state;
 295
 296         assert_nursery_collection (old_state, FALSE);
 297         SGEN_ASSERT (0, !old_state.data.post_done, "We are finishing the nursery collection, so we should have waited for the semaphore earlier");
 298
 299         workers_signal_enqueue_work (workers_num, TRUE);
 300 }
 301
 302 static gboolean
 303 workers_dequeue_and_do_job (WorkerData *data)
 304 {
 305         JobQueueEntry *entry;
 306
 307         /*
 308          * At this point the GC might not be running anymore.  We
 309          * could have been woken up by a job that was then taken by
 310          * another thread, after which the collection finished, so we
 311          * first have to successfully dequeue a job before doing
 312          * anything assuming that the collection is still ongoing.
 313          */
 314
 315         if (!workers_job_queue_num_entries)
 316                 return FALSE;
 317
 318         mono_mutex_lock (&workers_job_queue_mutex);
 319         entry = (JobQueueEntry*)workers_job_queue;
 320         if (entry) {
 321                 workers_job_queue = entry->next;
 322                 --workers_job_queue_num_entries;
 323         }
 324         mono_mutex_unlock (&workers_job_queue_mutex);
 325
 326         if (!entry)
 327                 return FALSE;
 328
 329         g_assert (collection_needs_workers ());
 330
 331         entry->func (data, entry->data);
 332         sgen_free_internal (entry, INTERNAL_MEM_JOB_QUEUE_ENTRY);
 333
 334         SGEN_ATOMIC_ADD (workers_num_jobs_finished, 1);
 335
 336         return TRUE;
 337 }
 338
 339 static gboolean
 340 workers_get_work (WorkerData *data)
 341 {
 342         SgenMajorCollector *major;
 343
 344         g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
 345
 346         /* If we're concurrent, steal from the workers distribute gray queue. */
 347         major = sgen_get_major_collector ();
 348         if (major->is_concurrent) {
 349                 GrayQueueSection *section = sgen_section_gray_queue_dequeue (&workers_distribute_gray_queue);
 350                 if (section) {
 351                         sgen_gray_object_enqueue_section (&data->private_gray_queue, section);
 352                         return TRUE;
 353                 }
 354         }
 355
 356         /* Nobody to steal from */
 357         g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
 358         return FALSE;
 359 }
 360
 361 static void
 362 concurrent_enqueue_check (char *obj)
 363 {
 364         g_assert (sgen_concurrent_collection_in_progress ());
 365         g_assert (!sgen_ptr_in_nursery (obj));
 366         g_assert (SGEN_LOAD_VTABLE (obj));
 367 }
 368
 369 static void
 370 init_private_gray_queue (WorkerData *data)
 371 {
 372         sgen_gray_object_queue_init (&data->private_gray_queue,
 373                         sgen_get_major_collector ()->is_concurrent ? concurrent_enqueue_check : NULL);
 374 }
 375
 376 static mono_native_thread_return_t
 377 workers_thread_func (void *data_untyped)
 378 {
 379         WorkerData *data = data_untyped;
 380         SgenMajorCollector *major = sgen_get_major_collector ();
 381
 382         mono_thread_info_register_small_id ();
 383
 384         if (major->init_worker_thread)
 385                 major->init_worker_thread (data->major_collector_data);
 386
 387         init_private_gray_queue (data);
 388
 389         for (;;) {
 390                 gboolean did_work = FALSE;
 391
 392                 SGEN_ASSERT (0, sgen_get_current_collection_generation () != GENERATION_NURSERY, "Why are we doing work while there's a nursery collection happening?");
 393
 394                 while (workers_state.data.state == STATE_WORKING && workers_dequeue_and_do_job (data)) {
 395                         did_work = TRUE;
 396                         /* FIXME: maybe distribute the gray queue here? */
 397                 }
 398
 399                 if (!did_work && (!sgen_gray_object_queue_is_empty (&data->private_gray_queue) || workers_get_work (data))) {
 400                         SgenObjectOperations *ops = sgen_concurrent_collection_in_progress ()
 401                                 ? &major->major_concurrent_ops
 402                                 : &major->major_ops;
 403                         ScanCopyContext ctx = { ops->scan_object, NULL, &data->private_gray_queue };
 404
 405                         g_assert (!sgen_gray_object_queue_is_empty (&data->private_gray_queue));
 406
 407                         while (!sgen_drain_gray_stack (32, ctx)) {
 408                                 if (workers_state.data.state == STATE_NURSERY_COLLECTION)
 409                                         workers_wait ();
 410                         }
 411                         g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
 412
 413                         init_private_gray_queue (data);
 414
 415                         did_work = TRUE;
 416                 }
 417
 418                 if (!did_work)
 419                         workers_wait ();
 420         }
 421
 422         /* dummy return to make compilers happy */
 423         return NULL;
 424 }
 425
 426 static void
 427 init_distribute_gray_queue (gboolean locked)
 428 {
 429         if (workers_distribute_gray_queue_inited) {
 430                 g_assert (sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue));
 431                 g_assert (!workers_distribute_gray_queue.locked == !locked);
 432                 return;
 433         }
 434
 435         sgen_section_gray_queue_init (&workers_distribute_gray_queue, locked,
 436                         sgen_get_major_collector ()->is_concurrent ? concurrent_enqueue_check : NULL);
 437         workers_distribute_gray_queue_inited = TRUE;
 438 }
 439
 440 void
 441 sgen_workers_init_distribute_gray_queue (void)
 442 {
 443         if (!collection_needs_workers ())
 444                 return;
 445
 446         init_distribute_gray_queue (sgen_get_major_collector ()->is_concurrent);
 447 }
 448
 449 void
 450 sgen_workers_init (int num_workers)
 451 {
 452         int i;
 453
 454         if (!sgen_get_major_collector ()->is_concurrent)
 455                 return;
 456
 457         //g_print ("initing %d workers\n", num_workers);
 458
 459         workers_num = num_workers;
 460
 461         workers_data = sgen_alloc_internal_dynamic (sizeof (WorkerData) * num_workers, INTERNAL_MEM_WORKER_DATA, TRUE);
 462         memset (workers_data, 0, sizeof (WorkerData) * num_workers);
 463
 464         MONO_SEM_INIT (&workers_waiting_sem, 0);
 465         MONO_SEM_INIT (&workers_done_sem, 0);
 466
 467         init_distribute_gray_queue (sgen_get_major_collector ()->is_concurrent);
 468
 469         if (sgen_get_major_collector ()->alloc_worker_data)
 470                 workers_gc_thread_major_collector_data = sgen_get_major_collector ()->alloc_worker_data ();
 471
 472         for (i = 0; i < workers_num; ++i) {
 473                 workers_data [i].index = i;
 474
 475                 if (sgen_get_major_collector ()->alloc_worker_data)
 476                         workers_data [i].major_collector_data = sgen_get_major_collector ()->alloc_worker_data ();
 477         }
 478
 479         LOCK_INIT (workers_job_queue_mutex);
 480
 481         sgen_register_fixed_internal_mem_type (INTERNAL_MEM_JOB_QUEUE_ENTRY, sizeof (JobQueueEntry));
 482
 483         mono_counters_register ("Stolen from self lock", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_stolen_from_self_lock);
 484         mono_counters_register ("Stolen from self no lock", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_stolen_from_self_no_lock);
 485         mono_counters_register ("Stolen from others", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_stolen_from_others);
 486         mono_counters_register ("# workers waited", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_num_waited);
 487 }
 488
 489 /* only the GC thread is allowed to start and join workers */
 490
 491 static void
 492 workers_start_worker (int index)
 493 {
 494         g_assert (index >= 0 && index < workers_num);
 495
 496         g_assert (!workers_data [index].thread);
 497         mono_native_thread_create (&workers_data [index].thread, workers_thread_func, &workers_data [index]);
 498 }
 499
 500 void
 501 sgen_workers_start_all_workers (void)
 502 {
 503         State old_state, new_state;
 504         int i;
 505         gboolean result;
 506
 507         if (!collection_needs_workers ())
 508                 return;
 509
 510         if (sgen_get_major_collector ()->init_worker_thread)
 511                 sgen_get_major_collector ()->init_worker_thread (workers_gc_thread_major_collector_data);
 512
 513         old_state = new_state = workers_state;
 514         assert_not_working (old_state);
 515
 516         g_assert (workers_job_queue_num_entries == 0);
 517         workers_num_jobs_enqueued = 0;
 518         workers_num_jobs_finished = 0;
 519
 520         if (workers_started) {
 521                 workers_signal_enqueue_work (workers_num, FALSE);
 522                 return;
 523         }
 524
 525         new_state.data.state = STATE_WORKING;
 526         new_state.data.num_awake = workers_num;
 527         result = set_state (old_state, new_state);
 528         SGEN_ASSERT (0, result, "Nobody else should have modified the state - workers have not been started yet");
 529
 530         for (i = 0; i < workers_num; ++i)
 531                 workers_start_worker (i);
 532
 533         workers_started = TRUE;
 534 }
 535
 536 gboolean
 537 sgen_workers_have_started (void)
 538 {
 539         return workers_started;
 540 }
 541
 542 void
 543 sgen_workers_join (void)
 544 {
 545         State old_state;
 546         int i;
 547
 548         if (!collection_needs_workers ())
 549                 return;
 550
 551         for (;;) {
 552                 old_state = workers_state;
 553                 SGEN_ASSERT (0, old_state.data.state != STATE_NURSERY_COLLECTION, "Can't be in nursery collection when joining");
 554
 555                 if (old_state.data.state == STATE_WORKING) {
 556                         State new_state = old_state;
 557
 558                         SGEN_ASSERT (0, !old_state.data.post_done, "Why is post_done already set?");
 559                         new_state.data.post_done = 1;
 560                         if (!set_state (old_state, new_state))
 561                                 continue;
 562
 563                         MONO_SEM_WAIT (&workers_done_sem);
 564
 565                         old_state = workers_state;
 566                 }
 567
 568                 assert_not_working (old_state);
 569
 570                 /*
 571                  * Checking whether there is still work left and, if not, going to sleep,
 572                  * are two separate actions that are not performed atomically by the
 573                  * workers.  Therefore there's a race condition where work can be added
 574                  * after they've checked for work, and before they've gone to sleep.
 575                  */
 576                 if (!workers_job_queue_num_entries && sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue))
 577                         break;
 578
 579                 workers_signal_enqueue_work (workers_num, FALSE);
 580         }
 581
 582         /* At this point all the workers have stopped. */
 583
 584         if (sgen_get_major_collector ()->reset_worker_data) {
 585                 for (i = 0; i < workers_num; ++i)
 586                         sgen_get_major_collector ()->reset_worker_data (workers_data [i].major_collector_data);
 587         }
 588
 589         g_assert (workers_job_queue_num_entries == 0);
 590         g_assert (sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue));
 591         for (i = 0; i < workers_num; ++i)
 592                 g_assert (sgen_gray_object_queue_is_empty (&workers_data [i].private_gray_queue));
 593 }
 594
 595 gboolean
 596 sgen_workers_all_done (void)
 597 {
 598         return workers_state.data.state == STATE_NOT_WORKING;
 599 }
 600
 601 gboolean
 602 sgen_workers_are_working (void)
 603 {
 604         State state = workers_state;
 605         return state.data.num_awake > 0 || state.data.num_posted > 0;
 606 }
 607
 608 gboolean
 609 sgen_is_worker_thread (MonoNativeThreadId thread)
 610 {
 611         int i;
 612
 613         if (sgen_get_major_collector ()->is_worker_thread && sgen_get_major_collector ()->is_worker_thread (thread))
 614                 return TRUE;
 615
 616         for (i = 0; i < workers_num; ++i) {
 617                 if (workers_data [i].thread == thread)
 618                         return TRUE;
 619         }
 620         return FALSE;
 621 }
 622
 623 SgenSectionGrayQueue*
 624 sgen_workers_get_distribute_section_gray_queue (void)
 625 {
 626         return &workers_distribute_gray_queue;
 627 }
 628
 629 void
 630 sgen_workers_reset_data (void)
 631 {
 632         if (sgen_get_major_collector ()->reset_worker_data)
 633                 sgen_get_major_collector ()->reset_worker_data (workers_gc_thread_major_collector_data);
 634 }
 635 #endif