[sgen] Workers use thread pool.
[mono.git] / mono / metadata / sgen-workers.c
index 9144fc27c28e933869b3d9b3e6f8f1cb57882f05..f54362656101a765c0a40309c9b90e301846d95e 100644 (file)
@@ -24,6 +24,7 @@
 
 #include "metadata/sgen-gc.h"
 #include "metadata/sgen-workers.h"
+#include "metadata/sgen-thread-pool.h"
 #include "utils/mono-counters.h"
 
 static int workers_num;
@@ -33,50 +34,22 @@ static void *workers_gc_thread_major_collector_data = NULL;
 static SgenSectionGrayQueue workers_distribute_gray_queue;
 static gboolean workers_distribute_gray_queue_inited;
 
-static gboolean workers_started = FALSE;
-
 enum {
        STATE_NOT_WORKING,
        STATE_WORKING,
        STATE_NURSERY_COLLECTION
 } WorkersStateName;
 
-/*
- * | state                    | num_awake | num_posted                 | post_done |
- * |--------------------------+-----------+----------------------------+-----------|
- * | STATE_NOT_WORKING        | 0         | *                          |         0 |
- * | STATE_WORKING            | > 0       | <= workers_num - num_awake |         * |
- * | STATE_NURSERY_COLLECTION | *         | <= workers_num - num_awake |         1 |
- * | STATE_NURSERY_COLLECTION | 0         | 0                          |         0 |
- */
 typedef union {
        gint32 value;
        struct {
                guint state : 4; /* WorkersStateName */
-               /* Number of worker threads awake. */
-               guint num_awake : 8;
-               /* The state of the waiting semaphore. */
-               guint num_posted : 8;
-               /* Whether to post `workers_done_sem` */
-               guint post_done : 1;
        } data;
 } State;
 
 static volatile State workers_state;
 
-static MonoSemType workers_waiting_sem;
-static MonoSemType workers_done_sem;
-
-static volatile int workers_job_queue_num_entries = 0;
-static volatile JobQueueEntry *workers_job_queue = NULL;
-static LOCK_DECLARE (workers_job_queue_mutex);
-static int workers_num_jobs_enqueued = 0;
-static volatile int workers_num_jobs_finished = 0;
-
-static guint64 stat_workers_stolen_from_self_lock;
-static guint64 stat_workers_stolen_from_self_no_lock;
-static guint64 stat_workers_stolen_from_others;
-static guint64 stat_workers_num_waited;
+static guint64 stat_workers_num_finished;
 
 static gboolean
 set_state (State old_state, State new_state)
@@ -92,9 +65,6 @@ static void
 assert_not_working (State state)
 {
        SGEN_ASSERT (0, state.data.state == STATE_NOT_WORKING, "Can only signal enqueue work when in no work state");
-       SGEN_ASSERT (0, state.data.num_awake == 0, "No workers can be awake when not working");
-       SGEN_ASSERT (0, state.data.num_posted == 0, "Can't have posted already");
-       SGEN_ASSERT (0, !state.data.post_done, "post_done can only be set when working");
 
 }
 
@@ -102,26 +72,12 @@ static void
 assert_working (State state, gboolean from_worker)
 {
        SGEN_ASSERT (0, state.data.state == STATE_WORKING, "A worker can't wait without being in working state");
-       if (from_worker)
-               SGEN_ASSERT (0, state.data.num_awake > 0, "How can we be awake, yet we are not counted?");
-       else
-               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?");
-       SGEN_ASSERT (0, state.data.num_awake + state.data.num_posted <= workers_num, "There are too many worker threads awake");
 }
 
 static void
 assert_nursery_collection (State state, gboolean from_worker)
 {
        SGEN_ASSERT (0, state.data.state == STATE_NURSERY_COLLECTION, "Must be in the nursery collection state");
-       if (from_worker) {
-               SGEN_ASSERT (0, state.data.num_awake > 0, "We're awake, but num_awake is zero");
-               SGEN_ASSERT (0, state.data.post_done, "post_done must be set in the nursery collection state");
-       }
-       SGEN_ASSERT (0, state.data.num_awake <= workers_num, "There are too many worker threads awake");
-       if (!state.data.post_done) {
-               SGEN_ASSERT (0, state.data.num_awake == 0, "Once done has been posted no threads can be awake");
-               SGEN_ASSERT (0, state.data.num_posted == 0, "Once done has been posted no thread must be awoken");
-       }
 }
 
 static void
@@ -134,85 +90,54 @@ assert_working_or_nursery_collection (State state)
 }
 
 static void
-workers_signal_enqueue_work (int num_wake_up, gboolean from_nursery_collection)
+workers_signal_enqueue_work (gboolean from_nursery_collection)
 {
        State old_state = workers_state;
        State new_state = old_state;
-       int i;
        gboolean did_set_state;
 
-       SGEN_ASSERT (0, num_wake_up <= workers_num, "Cannot wake up more workers than are present");
-
        if (from_nursery_collection)
                assert_nursery_collection (old_state, FALSE);
        else
                assert_not_working (old_state);
 
        new_state.data.state = STATE_WORKING;
-       new_state.data.num_posted = num_wake_up;
 
        did_set_state = set_state (old_state, new_state);
        SGEN_ASSERT (0, did_set_state, "Nobody else should be mutating the state");
 
-       for (i = 0; i < num_wake_up; ++i)
-               MONO_SEM_POST (&workers_waiting_sem);
+       sgen_thread_pool_idle_signal ();
 }
 
 static void
-workers_signal_enqueue_work_if_necessary (int num_wake_up)
+workers_signal_enqueue_work_if_necessary (void)
 {
        if (workers_state.data.state == STATE_NOT_WORKING)
-               workers_signal_enqueue_work (num_wake_up, FALSE);
+               workers_signal_enqueue_work (FALSE);
 }
 
 void
 sgen_workers_ensure_awake (void)
 {
        SGEN_ASSERT (0, workers_state.data.state != STATE_NURSERY_COLLECTION, "Can't wake workers during nursery collection");
-       workers_signal_enqueue_work_if_necessary (workers_num);
+       workers_signal_enqueue_work_if_necessary ();
 }
 
 static void
-workers_wait (void)
+worker_finish (void)
 {
        State old_state, new_state;
-       gboolean post_done;
 
-       ++stat_workers_num_waited;
+       ++stat_workers_num_finished;
 
        do {
                new_state = old_state = workers_state;
 
                assert_working_or_nursery_collection (old_state);
 
-               --new_state.data.num_awake;
-               post_done = FALSE;
-               if (!new_state.data.num_awake && !new_state.data.num_posted) {
-                       /* We are the last thread to go to sleep. */
-                       if (old_state.data.state == STATE_WORKING)
-                               new_state.data.state = STATE_NOT_WORKING;
-
-                       new_state.data.post_done = 0;
-                       if (old_state.data.post_done)
-                               post_done = TRUE;
-               }
-       } while (!set_state (old_state, new_state));
-
-       if (post_done)
-               MONO_SEM_POST (&workers_done_sem);
-
-       MONO_SEM_WAIT (&workers_waiting_sem);
-
-       do {
-               new_state = old_state = workers_state;
-
-               SGEN_ASSERT (0, old_state.data.num_posted > 0, "How can we be awake without the semaphore having been posted?");
-               SGEN_ASSERT (0, old_state.data.num_awake < workers_num, "There are too many worker threads awake");
-
-               --new_state.data.num_posted;
-               ++new_state.data.num_awake;
-
-               assert_working_or_nursery_collection (new_state);
+               /* We are the last thread to go to sleep. */
+               if (old_state.data.state == STATE_WORKING)
+                       new_state.data.state = STATE_NOT_WORKING;
        } while (!set_state (old_state, new_state));
 }
 
@@ -223,40 +148,21 @@ collection_needs_workers (void)
 }
 
 void
-sgen_workers_enqueue_job (JobFunc func, void *data)
+sgen_workers_enqueue_job (SgenThreadPoolJob *job)
 {
-       int num_entries;
-       JobQueueEntry *entry;
-
        if (!collection_needs_workers ()) {
-               func (NULL, data);
+               job->func (NULL, job);
+               sgen_thread_pool_job_free (job);
                return;
        }
 
-       entry = sgen_alloc_internal (INTERNAL_MEM_JOB_QUEUE_ENTRY);
-       entry->func = func;
-       entry->data = data;
-
-       mono_mutex_lock (&workers_job_queue_mutex);
-       entry->next = workers_job_queue;
-       workers_job_queue = entry;
-       num_entries = ++workers_job_queue_num_entries;
-       ++workers_num_jobs_enqueued;
-       mono_mutex_unlock (&workers_job_queue_mutex);
-
-       if (workers_state.data.state != STATE_NURSERY_COLLECTION)
-               workers_signal_enqueue_work_if_necessary (num_entries);
+       sgen_thread_pool_job_enqueue (job);
 }
 
 void
 sgen_workers_wait_for_jobs_finished (void)
 {
-       // FIXME: implement this properly
-       while (workers_num_jobs_finished < workers_num_jobs_enqueued) {
-               workers_signal_enqueue_work_if_necessary (workers_num);
-               /* FIXME: sleep less? */
-               g_usleep (1000);
-       }
+       sgen_thread_pool_wait_for_all_jobs ();
 }
 
 void
@@ -273,147 +179,30 @@ sgen_workers_signal_start_nursery_collection_and_wait (void)
                        assert_not_working (old_state);
                } else {
                        assert_working (old_state, FALSE);
-                       SGEN_ASSERT (0, !old_state.data.post_done, "We are not waiting for the workers");
-
-                       new_state.data.post_done = 1;
                }
        } while (!set_state (old_state, new_state));
 
-       if (new_state.data.post_done)
-               MONO_SEM_WAIT (&workers_done_sem);
+       sgen_thread_pool_idle_wait ();
 
        old_state = workers_state;
        assert_nursery_collection (old_state, FALSE);
-       SGEN_ASSERT (0, !old_state.data.post_done, "We got the semaphore, so it must have been posted");
 }
 
 void
 sgen_workers_signal_finish_nursery_collection (void)
 {
-       State old_state = workers_state;
-
-       assert_nursery_collection (old_state, FALSE);
-       SGEN_ASSERT (0, !old_state.data.post_done, "We are finishing the nursery collection, so we should have waited for the semaphore earlier");
-
-       workers_signal_enqueue_work (workers_num, TRUE);
-}
-
-static gboolean
-workers_dequeue_and_do_job (WorkerData *data)
-{
-       JobQueueEntry *entry;
-
-       /*
-        * At this point the GC might not be running anymore.  We
-        * could have been woken up by a job that was then taken by
-        * another thread, after which the collection finished, so we
-        * first have to successfully dequeue a job before doing
-        * anything assuming that the collection is still ongoing.
-        */
-
-       if (!workers_job_queue_num_entries)
-               return FALSE;
-
-       mono_mutex_lock (&workers_job_queue_mutex);
-       entry = (JobQueueEntry*)workers_job_queue;
-       if (entry) {
-               workers_job_queue = entry->next;
-               --workers_job_queue_num_entries;
-       }
-       mono_mutex_unlock (&workers_job_queue_mutex);
-
-       if (!entry)
-               return FALSE;
-
-       g_assert (collection_needs_workers ());
-
-       entry->func (data, entry->data);
-       sgen_free_internal (entry, INTERNAL_MEM_JOB_QUEUE_ENTRY);
-
-       SGEN_ATOMIC_ADD (workers_num_jobs_finished, 1);
-
-       return TRUE;
-}
-
-static gboolean
-workers_steal (WorkerData *data, WorkerData *victim_data, gboolean lock)
-{
-       SgenGrayQueue *queue = &data->private_gray_queue;
-       int num, n;
-
-       g_assert (!queue->first);
-
-       if (!victim_data->stealable_stack_fill)
-               return FALSE;
-
-       if (lock && mono_mutex_trylock (&victim_data->stealable_stack_mutex))
-               return FALSE;
-
-       n = num = (victim_data->stealable_stack_fill + 1) / 2;
-       /* We're stealing num entries. */
-
-       while (n > 0) {
-               int m = MIN (SGEN_GRAY_QUEUE_SECTION_SIZE, n);
-               n -= m;
-
-               sgen_gray_object_alloc_queue_section (queue);
-               memcpy (queue->first->entries,
-                               victim_data->stealable_stack + victim_data->stealable_stack_fill - num + n,
-                               sizeof (GrayQueueEntry) * m);
-               queue->first->size = m;
-
-               /*
-                * DO NOT move outside this loop
-                * Doing so trigger "assert not reached" in sgen-scan-object.h : we use the queue->cursor
-                * to compute the size of the first section during section allocation (via alloc_prepare_func
-                * -> workers_gray_queue_share_redirect -> sgen_gray_object_dequeue_section) which will be then
-                * set to 0, because queue->cursor is still pointing to queue->first->entries [-1], thus
-                * losing objects in the gray queue.
-                */
-               queue->cursor = queue->first->entries + queue->first->size - 1;
-       }
-
-       victim_data->stealable_stack_fill -= num;
-
-       if (lock)
-               mono_mutex_unlock (&victim_data->stealable_stack_mutex);
-
-       if (data == victim_data) {
-               if (lock)
-                       stat_workers_stolen_from_self_lock += num;
-               else
-                       stat_workers_stolen_from_self_no_lock += num;
-       } else {
-               stat_workers_stolen_from_others += num;
-       }
-
-       return num != 0;
+       assert_nursery_collection (workers_state, FALSE);
+       workers_signal_enqueue_work (TRUE);
 }
 
 static gboolean
 workers_get_work (WorkerData *data)
 {
        SgenMajorCollector *major;
-       int i;
 
        g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
 
-       /* Try to steal from our own stack. */
-       if (workers_steal (data, data, TRUE))
-               return TRUE;
-
-       /* From another worker. */
-       for (i = data->index + 1; i < workers_num + data->index; ++i) {
-               WorkerData *victim_data = &workers_data [i % workers_num];
-               g_assert (data != victim_data);
-               if (workers_steal (data, victim_data, TRUE))
-                       return TRUE;
-       }
-
-       /*
-        * If we're concurrent or parallel, from the workers
-        * distribute gray queue.
-        */
+       /* If we're concurrent, steal from the workers distribute gray queue. */
        major = sgen_get_major_collector ();
        if (major->is_concurrent) {
                GrayQueueSection *section = sgen_section_gray_queue_dequeue (&workers_distribute_gray_queue);
@@ -428,49 +217,6 @@ workers_get_work (WorkerData *data)
        return FALSE;
 }
 
-static void
-workers_gray_queue_share_redirect (SgenGrayQueue *queue)
-{
-       GrayQueueSection *section;
-       WorkerData *data = queue->alloc_prepare_data;
-
-       if (data->stealable_stack_fill) {
-               /*
-                * There are still objects in the stealable stack, so
-                * wake up any workers that might be sleeping
-                */
-               workers_signal_enqueue_work_if_necessary (workers_num);
-               return;
-       }
-
-       /* The stealable stack is empty, so fill it. */
-       mono_mutex_lock (&data->stealable_stack_mutex);
-
-       while (data->stealable_stack_fill < STEALABLE_STACK_SIZE &&
-                       (section = sgen_gray_object_dequeue_section (queue))) {
-               int num = MIN (section->size, STEALABLE_STACK_SIZE - data->stealable_stack_fill);
-
-               memcpy (data->stealable_stack + data->stealable_stack_fill,
-                               section->entries + section->size - num,
-                               sizeof (GrayQueueEntry) * num);
-
-               section->size -= num;
-               data->stealable_stack_fill += num;
-
-               if (section->size)
-                       sgen_gray_object_enqueue_section (queue, section);
-               else
-                       sgen_gray_object_free_queue_section (section);
-       }
-
-       if (sgen_gray_object_queue_is_empty (queue))
-               workers_steal (data, data, FALSE);
-
-       mono_mutex_unlock (&data->stealable_stack_mutex);
-
-       workers_signal_enqueue_work_if_necessary (workers_num);
-}
-
 static void
 concurrent_enqueue_check (char *obj)
 {
@@ -482,73 +228,66 @@ concurrent_enqueue_check (char *obj)
 static void
 init_private_gray_queue (WorkerData *data)
 {
-       sgen_gray_object_queue_init_with_alloc_prepare (&data->private_gray_queue,
-                       sgen_get_major_collector ()->is_concurrent ? concurrent_enqueue_check : NULL,
-                       workers_gray_queue_share_redirect, data);
+       sgen_gray_object_queue_init (&data->private_gray_queue,
+                       sgen_get_major_collector ()->is_concurrent ? concurrent_enqueue_check : NULL);
 }
 
-static mono_native_thread_return_t
-workers_thread_func (void *data_untyped)
+static void
+thread_pool_init_func (void *data_untyped)
 {
        WorkerData *data = data_untyped;
        SgenMajorCollector *major = sgen_get_major_collector ();
 
        mono_thread_info_register_small_id ();
 
+       if (!major->is_concurrent)
+               return;
+
        if (major->init_worker_thread)
                major->init_worker_thread (data->major_collector_data);
 
        init_private_gray_queue (data);
+}
 
-       for (;;) {
-               gboolean did_work = FALSE;
-
-               SGEN_ASSERT (0, sgen_get_current_collection_generation () != GENERATION_NURSERY, "Why are we doing work while there's a nursery collection happening?");
-
-               while (workers_state.data.state == STATE_WORKING && workers_dequeue_and_do_job (data)) {
-                       did_work = TRUE;
-                       /* FIXME: maybe distribute the gray queue here? */
-               }
-
-               if (!sgen_gray_object_queue_is_empty (&data->private_gray_queue) || workers_get_work (data)) {
-                       SgenObjectOperations *ops = sgen_concurrent_collection_in_progress ()
-                               ? &major->major_concurrent_ops
-                               : &major->major_ops;
-                       ScanCopyContext ctx = { ops->scan_object, NULL, &data->private_gray_queue };
+static gboolean
+marker_idle_func (void *data_untyped)
+{
+       WorkerData *data = data_untyped;
+       SgenMajorCollector *major = sgen_get_major_collector ();
 
-                       g_assert (!sgen_gray_object_queue_is_empty (&data->private_gray_queue));
+       if (workers_state.data.state != STATE_WORKING)
+               return FALSE;
 
-                       while (!sgen_drain_gray_stack (32, ctx)) {
-                               if (workers_state.data.state == STATE_NURSERY_COLLECTION)
-                                       workers_wait ();
+       SGEN_ASSERT (0, sgen_get_current_collection_generation () != GENERATION_NURSERY, "Why are we doing work while there's a nursery collection happening?");
 
-                               workers_gray_queue_share_redirect (&data->private_gray_queue);
-                       }
-                       g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
+       if (!sgen_gray_object_queue_is_empty (&data->private_gray_queue) || workers_get_work (data)) {
+               SgenObjectOperations *ops = sgen_concurrent_collection_in_progress ()
+                       ? &major->major_concurrent_ops
+                       : &major->major_ops;
+               ScanCopyContext ctx = { ops->scan_object, NULL, &data->private_gray_queue };
 
-                       init_private_gray_queue (data);
+               SGEN_ASSERT (0, !sgen_gray_object_queue_is_empty (&data->private_gray_queue), "How is our gray queue empty if we just got work?");
 
-                       did_work = TRUE;
-               }
+               sgen_drain_gray_stack (32, ctx);
 
-               if (!did_work)
-                       workers_wait ();
+               return TRUE;
        }
 
-       /* dummy return to make compilers happy */
-       return NULL;
+       worker_finish ();
+
+       return FALSE;
 }
 
 static void
-init_distribute_gray_queue (gboolean locked)
+init_distribute_gray_queue (void)
 {
        if (workers_distribute_gray_queue_inited) {
                g_assert (sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue));
-               g_assert (!workers_distribute_gray_queue.locked == !locked);
+               g_assert (workers_distribute_gray_queue.locked);
                return;
        }
 
-       sgen_section_gray_queue_init (&workers_distribute_gray_queue, locked,
+       sgen_section_gray_queue_init (&workers_distribute_gray_queue, TRUE,
                        sgen_get_major_collector ()->is_concurrent ? concurrent_enqueue_check : NULL);
        workers_distribute_gray_queue_inited = TRUE;
 }
@@ -556,19 +295,21 @@ init_distribute_gray_queue (gboolean locked)
 void
 sgen_workers_init_distribute_gray_queue (void)
 {
-       if (!collection_needs_workers ())
-               return;
-
-       init_distribute_gray_queue (sgen_get_major_collector ()->is_concurrent);
+       SGEN_ASSERT (0, sgen_get_major_collector ()->is_concurrent && collection_needs_workers (),
+                       "Why should we init the distribute gray queue if we don't need it?");
+       init_distribute_gray_queue ();
 }
 
 void
 sgen_workers_init (int num_workers)
 {
        int i;
+       void *workers_data_ptrs [num_workers];
 
-       if (!sgen_get_major_collector ()->is_concurrent)
+       if (!sgen_get_major_collector ()->is_concurrent) {
+               sgen_thread_pool_init (num_workers, thread_pool_init_func, NULL, NULL);
                return;
+       }
 
        //g_print ("initing %d workers\n", num_workers);
 
@@ -577,10 +318,7 @@ sgen_workers_init (int num_workers)
        workers_data = sgen_alloc_internal_dynamic (sizeof (WorkerData) * num_workers, INTERNAL_MEM_WORKER_DATA, TRUE);
        memset (workers_data, 0, sizeof (WorkerData) * num_workers);
 
-       MONO_SEM_INIT (&workers_waiting_sem, 0);
-       MONO_SEM_INIT (&workers_done_sem, 0);
-
-       init_distribute_gray_queue (sgen_get_major_collector ()->is_concurrent);
+       init_distribute_gray_queue ();
 
        if (sgen_get_major_collector ()->alloc_worker_data)
                workers_gc_thread_major_collector_data = sgen_get_major_collector ()->alloc_worker_data ();
@@ -588,104 +326,43 @@ sgen_workers_init (int num_workers)
        for (i = 0; i < workers_num; ++i) {
                workers_data [i].index = i;
 
-               /* private gray queue is inited by the thread itself */
-               mono_mutex_init (&workers_data [i].stealable_stack_mutex);
-               workers_data [i].stealable_stack_fill = 0;
-
                if (sgen_get_major_collector ()->alloc_worker_data)
                        workers_data [i].major_collector_data = sgen_get_major_collector ()->alloc_worker_data ();
-       }
-
-       LOCK_INIT (workers_job_queue_mutex);
-
-       sgen_register_fixed_internal_mem_type (INTERNAL_MEM_JOB_QUEUE_ENTRY, sizeof (JobQueueEntry));
-
-       mono_counters_register ("Stolen from self lock", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_stolen_from_self_lock);
-       mono_counters_register ("Stolen from self no lock", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_stolen_from_self_no_lock);
-       mono_counters_register ("Stolen from others", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_stolen_from_others);
-       mono_counters_register ("# workers waited", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_num_waited);
-}
 
-/* only the GC thread is allowed to start and join workers */
+               workers_data_ptrs [i] = &workers_data [i];
+       }
 
-static void
-workers_start_worker (int index)
-{
-       g_assert (index >= 0 && index < workers_num);
+       sgen_thread_pool_init (num_workers, thread_pool_init_func, marker_idle_func, workers_data_ptrs);
 
-       g_assert (!workers_data [index].thread);
-       mono_native_thread_create (&workers_data [index].thread, workers_thread_func, &workers_data [index]);
+       mono_counters_register ("# workers finished", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_num_finished);
 }
 
 void
 sgen_workers_start_all_workers (void)
 {
-       State old_state, new_state;
-       int i;
-       gboolean result;
-
        if (!collection_needs_workers ())
                return;
 
        if (sgen_get_major_collector ()->init_worker_thread)
                sgen_get_major_collector ()->init_worker_thread (workers_gc_thread_major_collector_data);
 
-       old_state = new_state = workers_state;
-       assert_not_working (old_state);
-
-       g_assert (workers_job_queue_num_entries == 0);
-       workers_num_jobs_enqueued = 0;
-       workers_num_jobs_finished = 0;
-
-       if (workers_started) {
-               workers_signal_enqueue_work (workers_num, FALSE);
-               return;
-       }
-
-       new_state.data.state = STATE_WORKING;
-       new_state.data.num_awake = workers_num;
-       result = set_state (old_state, new_state);
-       SGEN_ASSERT (0, result, "Nobody else should have modified the state - workers have not been started yet");
-
-       for (i = 0; i < workers_num; ++i)
-               workers_start_worker (i);
-
-       workers_started = TRUE;
-}
-
-gboolean
-sgen_workers_have_started (void)
-{
-       return workers_started;
+       workers_signal_enqueue_work (FALSE);
 }
 
 void
 sgen_workers_join (void)
 {
-       State old_state;
        int i;
 
        if (!collection_needs_workers ())
                return;
 
-       for (;;) {
-               old_state = workers_state;
-               SGEN_ASSERT (0, old_state.data.state != STATE_NURSERY_COLLECTION, "Can't be in nursery collection when joining");
-
-               if (old_state.data.state == STATE_WORKING) {
-                       State new_state = old_state;
-
-                       SGEN_ASSERT (0, !old_state.data.post_done, "Why is post_done already set?");
-                       new_state.data.post_done = 1;
-                       if (!set_state (old_state, new_state))
-                               continue;
-
-                       MONO_SEM_WAIT (&workers_done_sem);
-
-                       old_state = workers_state;
-               }
+       sgen_thread_pool_wait_for_all_jobs ();
 
-               assert_not_working (old_state);
+       for (;;) {
+               SGEN_ASSERT (0, workers_state.data.state != STATE_NURSERY_COLLECTION, "Can't be in nursery collection when joining");
+               sgen_thread_pool_idle_wait ();
+               assert_not_working (workers_state);
 
                /*
                 * Checking whether there is still work left and, if not, going to sleep,
@@ -693,10 +370,10 @@ sgen_workers_join (void)
                 * workers.  Therefore there's a race condition where work can be added
                 * after they've checked for work, and before they've gone to sleep.
                 */
-               if (!workers_job_queue_num_entries && sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue))
+               if (sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue))
                        break;
 
-               workers_signal_enqueue_work (workers_num, FALSE);
+               workers_signal_enqueue_work (FALSE);
        }
 
        /* At this point all the workers have stopped. */
@@ -706,12 +383,9 @@ sgen_workers_join (void)
                        sgen_get_major_collector ()->reset_worker_data (workers_data [i].major_collector_data);
        }
 
-       g_assert (workers_job_queue_num_entries == 0);
        g_assert (sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue));
-       for (i = 0; i < workers_num; ++i) {
-               g_assert (!workers_data [i].stealable_stack_fill);
+       for (i = 0; i < workers_num; ++i)
                g_assert (sgen_gray_object_queue_is_empty (&workers_data [i].private_gray_queue));
-       }
 }
 
 gboolean
@@ -723,23 +397,14 @@ sgen_workers_all_done (void)
 gboolean
 sgen_workers_are_working (void)
 {
-       State state = workers_state;
-       return state.data.num_awake > 0 || state.data.num_posted > 0;
+       return workers_state.data.state == STATE_WORKING;
 }
 
-gboolean
-sgen_is_worker_thread (MonoNativeThreadId thread)
+void
+sgen_workers_wait (void)
 {
-       int i;
-
-       if (sgen_get_major_collector ()->is_worker_thread && sgen_get_major_collector ()->is_worker_thread (thread))
-               return TRUE;
-
-       for (i = 0; i < workers_num; ++i) {
-               if (workers_data [i].thread == thread)
-                       return TRUE;
-       }
-       return FALSE;
+       sgen_thread_pool_idle_wait ();
+       SGEN_ASSERT (0, sgen_workers_all_done (), "Why are the workers not done after we wait for them?");
 }
 
 SgenSectionGrayQueue*