Merge pull request #3806 from BrzVlad/feature-parallel-gc-final
[mono.git] / mono / sgen / 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  * Licensed under the MIT license. See LICENSE file in the project root for full license information.
9  */
10
11 #include "config.h"
12 #ifdef HAVE_SGEN_GC
13
14 #include <string.h>
15
16 #include "mono/sgen/sgen-gc.h"
17 #include "mono/sgen/sgen-workers.h"
18 #include "mono/sgen/sgen-thread-pool.h"
19 #include "mono/utils/mono-membar.h"
20 #include "mono/sgen/sgen-client.h"
21
22 static int workers_num;
23 static int active_workers_num;
24 static volatile gboolean forced_stop;
25 static WorkerData *workers_data;
26 static SgenWorkerCallback worker_init_cb;
27
28 /*
29  * When using multiple workers, we need to have the last worker
30  * enqueue the preclean jobs (if there are any). This lock ensures
31  * that when the last worker takes it, all the other workers have
32  * gracefully finished, so it can restart them.
33  */
34 static mono_mutex_t finished_lock;
35 static volatile gboolean workers_finished;
36 static int worker_awakenings;
37
38 static SgenSectionGrayQueue workers_distribute_gray_queue;
39 static gboolean workers_distribute_gray_queue_inited;
40
41 /*
42  * Allowed transitions:
43  *
44  * | from \ to          | NOT WORKING | WORKING | WORK ENQUEUED |
45  * |--------------------+-------------+---------+---------------+
46  * | NOT WORKING        | -           | -       | main / worker |
47  * | WORKING            | worker      | -       | main / worker |
48  * | WORK ENQUEUED      | -           | worker  | -             |
49  *
50  * The WORK ENQUEUED state guarantees that the worker thread will inspect the queue again at
51  * least once.  Only after looking at the queue will it go back to WORKING, and then,
52  * eventually, to NOT WORKING.  After enqueuing work the main thread transitions the state
53  * to WORK ENQUEUED.  Signalling the worker thread to wake up is only necessary if the old
54  * state was NOT WORKING.
55  */
56
57 enum {
58         STATE_NOT_WORKING,
59         STATE_WORKING,
60         STATE_WORK_ENQUEUED
61 };
62
63 typedef gint32 State;
64
65 static SgenObjectOperations * volatile idle_func_object_ops;
66 static SgenObjectOperations *idle_func_object_ops_par, *idle_func_object_ops_nopar;
67 /*
68  * finished_callback is called only when the workers finish work normally (when they
69  * are not forced to finish). The callback is used to enqueue preclean jobs.
70  */
71 static volatile SgenWorkersFinishCallback finish_callback;
72
73 static guint64 stat_workers_num_finished;
74
75 static gboolean
76 set_state (WorkerData *data, State old_state, State new_state)
77 {
78         SGEN_ASSERT (0, old_state != new_state, "Why are we transitioning to the same state?");
79         if (new_state == STATE_NOT_WORKING)
80                 SGEN_ASSERT (0, old_state == STATE_WORKING, "We can only transition to NOT WORKING from WORKING");
81         else if (new_state == STATE_WORKING)
82                 SGEN_ASSERT (0, old_state == STATE_WORK_ENQUEUED, "We can only transition to WORKING from WORK ENQUEUED");
83         if (new_state == STATE_NOT_WORKING || new_state == STATE_WORKING)
84                 SGEN_ASSERT (6, sgen_thread_pool_is_thread_pool_thread (mono_native_thread_id_get ()), "Only the worker thread is allowed to transition to NOT_WORKING or WORKING");
85
86         return InterlockedCompareExchange (&data->state, new_state, old_state) == old_state;
87 }
88
89 static gboolean
90 state_is_working_or_enqueued (State state)
91 {
92         return state == STATE_WORKING || state == STATE_WORK_ENQUEUED;
93 }
94
95 static void
96 sgen_workers_ensure_awake (void)
97 {
98         int i;
99         gboolean need_signal = FALSE;
100
101         /*
102          * All workers are awaken, make sure we reset the parallel context.
103          * We call this function only when starting the workers so nobody is running,
104          * or when the last worker is enqueuing preclean work. In both cases we can't
105          * have a worker working using a nopar context, which means it is safe.
106          */
107         idle_func_object_ops = (active_workers_num > 1) ? idle_func_object_ops_par : idle_func_object_ops_nopar;
108         workers_finished = FALSE;
109
110         for (i = 0; i < active_workers_num; i++) {
111                 State old_state;
112                 gboolean did_set_state;
113
114                 do {
115                         old_state = workers_data [i].state;
116
117                         if (old_state == STATE_WORK_ENQUEUED)
118                                 break;
119
120                         did_set_state = set_state (&workers_data [i], old_state, STATE_WORK_ENQUEUED);
121                 } while (!did_set_state);
122
123                 if (!state_is_working_or_enqueued (old_state))
124                         need_signal = TRUE;
125         }
126
127         if (need_signal)
128                 sgen_thread_pool_idle_signal ();
129 }
130
131 static void
132 worker_try_finish (WorkerData *data)
133 {
134         State old_state;
135         int i, working = 0;
136
137         ++stat_workers_num_finished;
138
139         mono_os_mutex_lock (&finished_lock);
140
141         for (i = 0; i < active_workers_num; i++) {
142                 if (state_is_working_or_enqueued (workers_data [i].state))
143                         working++;
144         }
145
146         if (working == 1) {
147                 SgenWorkersFinishCallback callback = finish_callback;
148                 SGEN_ASSERT (0, idle_func_object_ops == idle_func_object_ops_nopar, "Why are we finishing with parallel context");
149                 /* We are the last one left. Enqueue preclean job if we have one and awake everybody */
150                 SGEN_ASSERT (0, data->state != STATE_NOT_WORKING, "How did we get from doing idle work to NOT WORKING without setting it ourselves?");
151                 if (callback) {
152                         finish_callback = NULL;
153                         callback ();
154                         worker_awakenings = 0;
155                         /* Make sure each worker has a chance of seeing the enqueued jobs */
156                         sgen_workers_ensure_awake ();
157                         SGEN_ASSERT (0, data->state == STATE_WORK_ENQUEUED, "Why did we fail to set our own state to ENQUEUED");
158                         goto work_available;
159                 }
160         }
161
162         do {
163                 old_state = data->state;
164
165                 SGEN_ASSERT (0, old_state != STATE_NOT_WORKING, "How did we get from doing idle work to NOT WORKING without setting it ourselves?");
166                 if (old_state == STATE_WORK_ENQUEUED)
167                         goto work_available;
168                 SGEN_ASSERT (0, old_state == STATE_WORKING, "What other possibility is there?");
169         } while (!set_state (data, old_state, STATE_NOT_WORKING));
170
171         /*
172          * If we are second to last to finish, we set the scan context to the non-parallel
173          * version so we can speed up the last worker. This helps us maintain same level
174          * of performance as non-parallel mode even if we fail to distribute work properly.
175          */
176         if (working == 2)
177                 idle_func_object_ops = idle_func_object_ops_nopar;
178
179         workers_finished = TRUE;
180         mono_os_mutex_unlock (&finished_lock);
181
182         binary_protocol_worker_finish (sgen_timestamp (), forced_stop);
183
184         sgen_gray_object_queue_trim_free_list (&data->private_gray_queue);
185         return;
186
187 work_available:
188         mono_os_mutex_unlock (&finished_lock);
189 }
190
191 void
192 sgen_workers_enqueue_job (SgenThreadPoolJob *job, gboolean enqueue)
193 {
194         if (!enqueue) {
195                 job->func (NULL, job);
196                 sgen_thread_pool_job_free (job);
197                 return;
198         }
199
200         sgen_thread_pool_job_enqueue (job);
201 }
202
203 static gboolean
204 workers_get_work (WorkerData *data)
205 {
206         SgenMajorCollector *major;
207
208         g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
209
210         /* If we're concurrent, steal from the workers distribute gray queue. */
211         major = sgen_get_major_collector ();
212         if (major->is_concurrent) {
213                 GrayQueueSection *section = sgen_section_gray_queue_dequeue (&workers_distribute_gray_queue);
214                 if (section) {
215                         sgen_gray_object_enqueue_section (&data->private_gray_queue, section, major->is_parallel);
216                         return TRUE;
217                 }
218         }
219
220         /* Nobody to steal from */
221         g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
222         return FALSE;
223 }
224
225 static gboolean
226 workers_steal_work (WorkerData *data)
227 {
228         SgenMajorCollector *major = sgen_get_major_collector ();
229         GrayQueueSection *section = NULL;
230         int i, current_worker;
231
232         if (!major->is_parallel)
233                 return FALSE;
234
235         /* If we're parallel, steal from other workers' private gray queues  */
236         g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
237
238         current_worker = (int) (data - workers_data);
239
240         for (i = 1; i < active_workers_num && !section; i++) {
241                 int steal_worker = (current_worker + i) % active_workers_num;
242                 if (state_is_working_or_enqueued (workers_data [steal_worker].state))
243                         section = sgen_gray_object_steal_section (&workers_data [steal_worker].private_gray_queue);
244         }
245
246         if (section) {
247                 sgen_gray_object_enqueue_section (&data->private_gray_queue, section, TRUE);
248                 return TRUE;
249         }
250
251         /* Nobody to steal from */
252         g_assert (sgen_gray_object_queue_is_empty (&data->private_gray_queue));
253         return FALSE;
254 }
255
256 static void
257 concurrent_enqueue_check (GCObject *obj)
258 {
259         g_assert (sgen_concurrent_collection_in_progress ());
260         g_assert (!sgen_ptr_in_nursery (obj));
261         g_assert (SGEN_LOAD_VTABLE (obj));
262 }
263
264 static void
265 init_private_gray_queue (WorkerData *data)
266 {
267         sgen_gray_object_queue_init (&data->private_gray_queue,
268                         sgen_get_major_collector ()->is_concurrent ? concurrent_enqueue_check : NULL,
269                         FALSE);
270 }
271
272 static void
273 thread_pool_init_func (void *data_untyped)
274 {
275         WorkerData *data = (WorkerData *)data_untyped;
276         SgenMajorCollector *major = sgen_get_major_collector ();
277
278         sgen_client_thread_register_worker ();
279
280         if (!major->is_concurrent)
281                 return;
282
283         init_private_gray_queue (data);
284
285         if (worker_init_cb)
286                 worker_init_cb (data);
287 }
288
289 static gboolean
290 continue_idle_func (void *data_untyped)
291 {
292         if (data_untyped) {
293                 WorkerData *data = (WorkerData *)data_untyped;
294                 return state_is_working_or_enqueued (data->state);
295         } else {
296                 /* Return if any of the threads is working */
297                 return !sgen_workers_all_done ();
298         }
299 }
300
301 static gboolean
302 should_work_func (void *data_untyped)
303 {
304         WorkerData *data = (WorkerData*)data_untyped;
305         int current_worker = (int) (data - workers_data);
306
307         return current_worker < active_workers_num;
308 }
309
310 static void
311 marker_idle_func (void *data_untyped)
312 {
313         WorkerData *data = (WorkerData *)data_untyped;
314
315         SGEN_ASSERT (0, continue_idle_func (data_untyped), "Why are we called when we're not supposed to work?");
316         SGEN_ASSERT (0, sgen_concurrent_collection_in_progress (), "The worker should only mark in concurrent collections.");
317
318         if (data->state == STATE_WORK_ENQUEUED) {
319                 set_state (data, STATE_WORK_ENQUEUED, STATE_WORKING);
320                 SGEN_ASSERT (0, data->state != STATE_NOT_WORKING, "How did we get from WORK ENQUEUED to NOT WORKING?");
321         }
322
323         if (!forced_stop && (!sgen_gray_object_queue_is_empty (&data->private_gray_queue) || workers_get_work (data) || workers_steal_work (data))) {
324                 ScanCopyContext ctx = CONTEXT_FROM_OBJECT_OPERATIONS (idle_func_object_ops, &data->private_gray_queue);
325
326                 SGEN_ASSERT (0, !sgen_gray_object_queue_is_empty (&data->private_gray_queue), "How is our gray queue empty if we just got work?");
327
328                 sgen_drain_gray_stack (ctx);
329
330                 if (data->private_gray_queue.num_sections > 16 && workers_finished && worker_awakenings < active_workers_num) {
331                         /* We bound the number of worker awakenings just to be sure */
332                         worker_awakenings++;
333                         sgen_workers_ensure_awake ();
334                 }
335         } else {
336                 worker_try_finish (data);
337         }
338 }
339
340 static void
341 init_distribute_gray_queue (void)
342 {
343         if (workers_distribute_gray_queue_inited) {
344                 g_assert (sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue));
345                 g_assert (workers_distribute_gray_queue.locked);
346                 return;
347         }
348
349         sgen_section_gray_queue_init (&workers_distribute_gray_queue, TRUE,
350                         sgen_get_major_collector ()->is_concurrent ? concurrent_enqueue_check : NULL);
351         workers_distribute_gray_queue_inited = TRUE;
352 }
353
354 void
355 sgen_workers_init_distribute_gray_queue (void)
356 {
357         SGEN_ASSERT (0, sgen_get_major_collector ()->is_concurrent,
358                         "Why should we init the distribute gray queue if we don't need it?");
359         init_distribute_gray_queue ();
360 }
361
362 void
363 sgen_workers_init (int num_workers, SgenWorkerCallback callback)
364 {
365         int i;
366         void **workers_data_ptrs = (void **)alloca(num_workers * sizeof(void *));
367
368         if (!sgen_get_major_collector ()->is_concurrent) {
369                 sgen_thread_pool_init (num_workers, thread_pool_init_func, NULL, NULL, NULL, NULL);
370                 return;
371         }
372
373         mono_os_mutex_init (&finished_lock);
374         //g_print ("initing %d workers\n", num_workers);
375
376         workers_num = num_workers;
377         active_workers_num = num_workers;
378
379         workers_data = (WorkerData *)sgen_alloc_internal_dynamic (sizeof (WorkerData) * num_workers, INTERNAL_MEM_WORKER_DATA, TRUE);
380         memset (workers_data, 0, sizeof (WorkerData) * num_workers);
381
382         init_distribute_gray_queue ();
383
384         for (i = 0; i < num_workers; ++i)
385                 workers_data_ptrs [i] = (void *) &workers_data [i];
386
387         worker_init_cb = callback;
388
389         sgen_thread_pool_init (num_workers, thread_pool_init_func, marker_idle_func, continue_idle_func, should_work_func, workers_data_ptrs);
390
391         mono_counters_register ("# workers finished", MONO_COUNTER_GC | MONO_COUNTER_ULONG, &stat_workers_num_finished);
392 }
393
394 void
395 sgen_workers_stop_all_workers (void)
396 {
397         finish_callback = NULL;
398         mono_memory_write_barrier ();
399         forced_stop = TRUE;
400
401         sgen_thread_pool_wait_for_all_jobs ();
402         sgen_thread_pool_idle_wait ();
403         SGEN_ASSERT (0, sgen_workers_all_done (), "Can only signal enqueue work when in no work state");
404 }
405
406 void
407 sgen_workers_set_num_active_workers (int num_workers)
408 {
409         if (num_workers) {
410                 SGEN_ASSERT (0, active_workers_num <= workers_num, "We can't start more workers than we initialized");
411                 active_workers_num = num_workers;
412         } else {
413                 active_workers_num = workers_num;
414         }
415 }
416
417 void
418 sgen_workers_start_all_workers (SgenObjectOperations *object_ops_nopar, SgenObjectOperations *object_ops_par, SgenWorkersFinishCallback callback)
419 {
420         idle_func_object_ops_par = object_ops_par;
421         idle_func_object_ops_nopar = object_ops_nopar;
422         forced_stop = FALSE;
423         finish_callback = callback;
424         worker_awakenings = 0;
425         mono_memory_write_barrier ();
426
427         sgen_workers_ensure_awake ();
428 }
429
430 void
431 sgen_workers_join (void)
432 {
433         int i;
434
435         sgen_thread_pool_wait_for_all_jobs ();
436         sgen_thread_pool_idle_wait ();
437         SGEN_ASSERT (0, sgen_workers_all_done (), "Can only signal enqueue work when in no work state");
438
439         /* At this point all the workers have stopped. */
440
441         SGEN_ASSERT (0, sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue), "Why is there still work left to do?");
442         for (i = 0; i < active_workers_num; ++i)
443                 SGEN_ASSERT (0, sgen_gray_object_queue_is_empty (&workers_data [i].private_gray_queue), "Why is there still work left to do?");
444 }
445
446 /*
447  * Can only be called if the workers are stopped.
448  * If we're stopped, there are also no pending jobs.
449  */
450 gboolean
451 sgen_workers_have_idle_work (void)
452 {
453         int i;
454
455         SGEN_ASSERT (0, forced_stop && sgen_workers_all_done (), "Checking for idle work should only happen if the workers are stopped.");
456
457         if (!sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue))
458                 return TRUE;
459
460         for (i = 0; i < active_workers_num; ++i) {
461                 if (!sgen_gray_object_queue_is_empty (&workers_data [i].private_gray_queue))
462                         return TRUE;
463         }
464
465         return FALSE;
466 }
467
468 gboolean
469 sgen_workers_all_done (void)
470 {
471         int i;
472
473         for (i = 0; i < active_workers_num; i++) {
474                 if (state_is_working_or_enqueued (workers_data [i].state))
475                         return FALSE;
476         }
477         return TRUE;
478 }
479
480 /* Must only be used for debugging */
481 gboolean
482 sgen_workers_are_working (void)
483 {
484         return !sgen_workers_all_done ();
485 }
486
487 void
488 sgen_workers_assert_gray_queue_is_empty (void)
489 {
490         SGEN_ASSERT (0, sgen_section_gray_queue_is_empty (&workers_distribute_gray_queue), "Why is the workers gray queue not empty?");
491 }
492
493 void
494 sgen_workers_take_from_queue (SgenGrayQueue *queue)
495 {
496         sgen_gray_object_spread (queue, sgen_workers_get_job_split_count ());
497
498         for (;;) {
499                 GrayQueueSection *section = sgen_gray_object_dequeue_section (queue);
500                 if (!section)
501                         break;
502                 sgen_section_gray_queue_enqueue (&workers_distribute_gray_queue, section);
503         }
504
505         SGEN_ASSERT (0, !sgen_workers_are_working (), "We should fully populate the distribute gray queue before we start the workers");
506 }
507
508 SgenObjectOperations*
509 sgen_workers_get_idle_func_object_ops (void)
510 {
511         return (idle_func_object_ops_par) ? idle_func_object_ops_par : idle_func_object_ops_nopar;
512 }
513
514 /*
515  * If we have a single worker, splitting into multiple jobs makes no sense. With
516  * more than one worker, we split into a larger number of jobs so that, in case
517  * the work load is uneven, a worker that finished quickly can take up more jobs
518  * than another one.
519  */
520 int
521 sgen_workers_get_job_split_count (void)
522 {
523         return (active_workers_num > 1) ? active_workers_num * 4 : 1;
524 }
525
526 void
527 sgen_workers_foreach (SgenWorkerCallback callback)
528 {
529         int i;
530
531         for (i = 0; i < workers_num; i++)
532                 callback (&workers_data [i]);
533 }
534
535 #endif