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 #ifdef SGEN_PARALLEL_MARK
  26
  27 typedef struct _WorkerData WorkerData;
  28 struct _WorkerData {
  29         pthread_t thread;
  30         MonoSemType start_worker_sem;
  31         gboolean is_working;
  32         GrayQueue private_gray_queue; /* only read/written by worker thread */
  33         int shared_buffer_increment;
  34         int shared_buffer_index;
  35 };
  36
  37 static int workers_num;
  38 static WorkerData *workers_data;
  39 static WorkerData workers_gc_thread_data;
  40
  41 static int workers_num_working;
  42
  43 static GrayQueue workers_distribute_gray_queue;
  44 /*
  45  * Must be a power of 2.  It seems that larger values don't help much.
  46  * The main reason to make this larger would be to sustain a bigger
  47  * number of worker threads.
  48  */
  49 #define WORKERS_SHARED_BUFFER_SIZE      16
  50 static GrayQueueSection *workers_shared_buffer [WORKERS_SHARED_BUFFER_SIZE];
  51 static int workers_shared_buffer_used;
  52
  53 static const int workers_primes [] = { 3, 5, 7, 11, 13, 17, 23, 29 };
  54
  55 static MonoSemType workers_done_sem;
  56
  57 static long long stat_shared_buffer_insert_tries;
  58 static long long stat_shared_buffer_insert_full;
  59 static long long stat_shared_buffer_insert_iterations;
  60 static long long stat_shared_buffer_insert_failures;
  61 static long long stat_shared_buffer_remove_tries;
  62 static long long stat_shared_buffer_remove_iterations;
  63 static long long stat_shared_buffer_remove_empty;
  64
  65 static void
  66 workers_gray_queue_share_redirect (GrayQueue *queue)
  67 {
  68         GrayQueueSection *section;
  69         WorkerData *data = queue->alloc_prepare_data;
  70         int increment = data->shared_buffer_increment;
  71
  72         while ((section = gray_object_dequeue_section (queue))) {
  73                 int i, index;
  74
  75                 HEAVY_STAT (++stat_shared_buffer_insert_tries);
  76
  77                 if (workers_shared_buffer_used == WORKERS_SHARED_BUFFER_SIZE) {
  78                         HEAVY_STAT (++stat_shared_buffer_insert_full);
  79                         gray_object_enqueue_section (queue, section);
  80                         return;
  81                 }
  82
  83                 index = data->shared_buffer_index;
  84                 for (i = 0; i < WORKERS_SHARED_BUFFER_SIZE; ++i) {
  85                         GrayQueueSection *old = workers_shared_buffer [index];
  86                         HEAVY_STAT (++stat_shared_buffer_insert_iterations);
  87                         if (!old) {
  88                                 if (SGEN_CAS_PTR ((void**)&workers_shared_buffer [index], section, NULL) == NULL) {
  89                                         SGEN_ATOMIC_ADD (workers_shared_buffer_used, 1);
  90                                         //g_print ("thread %d put section %d\n", data - workers_data, index);
  91                                         break;
  92                                 }
  93                         }
  94                         index = (index + increment) & (WORKERS_SHARED_BUFFER_SIZE - 1);
  95                 }
  96                 data->shared_buffer_index = index;
  97
  98                 if (i == WORKERS_SHARED_BUFFER_SIZE) {
  99                         /* unsuccessful */
 100                         HEAVY_STAT (++stat_shared_buffer_insert_failures);
 101                         gray_object_enqueue_section (queue, section);
 102                         return;
 103                 }
 104         }
 105 }
 106
 107 static gboolean
 108 workers_get_work (WorkerData *data)
 109 {
 110         int i, index;
 111         int increment = data->shared_buffer_increment;
 112
 113         HEAVY_STAT (++stat_shared_buffer_remove_tries);
 114
 115         index = data->shared_buffer_index;
 116         for (i = 0; i < WORKERS_SHARED_BUFFER_SIZE; ++i) {
 117                 GrayQueueSection *section;
 118
 119                 HEAVY_STAT (++stat_shared_buffer_remove_iterations);
 120
 121                 do {
 122                         section = workers_shared_buffer [index];
 123                         if (!section)
 124                                 break;
 125                 } while (SGEN_CAS_PTR ((void**)&workers_shared_buffer [index], NULL, section) != section);
 126
 127                 if (section) {
 128                         SGEN_ATOMIC_ADD (workers_shared_buffer_used, -1);
 129                         gray_object_enqueue_section (&data->private_gray_queue, section);
 130                         data->shared_buffer_index = index;
 131                         //g_print ("thread %d popped section %d\n", data - workers_data, index);
 132                         return TRUE;
 133                 }
 134
 135                 index = (index + increment) & (WORKERS_SHARED_BUFFER_SIZE - 1);
 136         }
 137
 138         HEAVY_STAT (++stat_shared_buffer_remove_empty);
 139
 140         data->shared_buffer_index = index;
 141         return FALSE;
 142 }
 143
 144 /* returns the new value */
 145 static int
 146 workers_change_num_working (int delta)
 147 {
 148         int old, new;
 149         do {
 150                 old = workers_num_working;
 151                 new = old + delta;
 152         } while (InterlockedCompareExchange (&workers_num_working, new, old) != old);
 153         return new;
 154 }
 155
 156 static void*
 157 workers_thread_func (void *data_untyped)
 158 {
 159         WorkerData *data = data_untyped;
 160         SgenInternalAllocator allocator;
 161
 162         memset (&allocator, 0, sizeof (allocator));
 163
 164         gray_object_queue_init_with_alloc_prepare (&data->private_gray_queue, &allocator,
 165                         workers_gray_queue_share_redirect, data);
 166
 167         for (;;) {
 168                 //g_print ("worker waiting for start %d\n", data->start_worker_sem);
 169
 170                 MONO_SEM_WAIT (&data->start_worker_sem);
 171
 172                 //g_print ("worker starting\n");
 173
 174                 for (;;) {
 175                         do {
 176                                 drain_gray_stack (&data->private_gray_queue);
 177                         } while (workers_get_work (data));
 178
 179                         /*
 180                          * FIXME: This might never terminate with
 181                          * multiple threads!
 182                          */
 183
 184                         if (workers_change_num_working (-1) == 0)
 185                                 break;
 186
 187                         /* we weren't the last one working */
 188                         //g_print ("sleeping\n");
 189                         usleep (5000);
 190                         workers_change_num_working (1);
 191                 }
 192
 193                 gray_object_queue_init (&data->private_gray_queue, &allocator);
 194
 195                 MONO_SEM_POST (&workers_done_sem);
 196
 197                 //g_print ("worker done\n");
 198         }
 199 }
 200
 201 static void
 202 workers_distribute_gray_queue_sections (void)
 203 {
 204         workers_gray_queue_share_redirect (&workers_distribute_gray_queue);
 205 }
 206
 207 static void
 208 workers_init (int num_workers)
 209 {
 210         int i;
 211
 212         //g_print ("initing %d workers\n", num_workers);
 213
 214         workers_num = num_workers;
 215         workers_data = mono_sgen_alloc_internal_dynamic (sizeof (WorkerData) * num_workers, INTERNAL_MEM_WORKER_DATA);
 216         MONO_SEM_INIT (&workers_done_sem, 0);
 217         workers_gc_thread_data.shared_buffer_increment = 1;
 218         workers_gc_thread_data.shared_buffer_index = 0;
 219         gray_object_queue_init_with_alloc_prepare (&workers_distribute_gray_queue, mono_sgen_get_unmanaged_allocator (),
 220                         workers_gray_queue_share_redirect, &workers_gc_thread_data);
 221
 222         g_assert (num_workers <= sizeof (workers_primes) / sizeof (workers_primes [0]));
 223         for (i = 0; i < workers_num; ++i) {
 224                 workers_data [i].shared_buffer_increment = workers_primes [i];
 225                 workers_data [i].shared_buffer_index = 0;
 226         }
 227
 228         mono_counters_register ("Shared buffer insert tries", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_shared_buffer_insert_tries);
 229         mono_counters_register ("Shared buffer insert full", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_shared_buffer_insert_full);
 230         mono_counters_register ("Shared buffer insert iterations", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_shared_buffer_insert_iterations);
 231         mono_counters_register ("Shared buffer insert failures", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_shared_buffer_insert_failures);
 232         mono_counters_register ("Shared buffer remove tries", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_shared_buffer_remove_tries);
 233         mono_counters_register ("Shared buffer remove iterations", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_shared_buffer_remove_iterations);
 234         mono_counters_register ("Shared buffer remove empty", MONO_COUNTER_GC | MONO_COUNTER_LONG, &stat_shared_buffer_remove_empty);
 235 }
 236
 237 /* only the GC thread is allowed to start and join workers */
 238
 239 static void
 240 workers_start_worker (int index)
 241 {
 242         g_assert (index >= 0 && index < workers_num);
 243
 244         if (workers_data [index].is_working)
 245                 return;
 246
 247         if (!workers_data [index].thread) {
 248                 //g_print ("initing thread %d\n", index);
 249                 MONO_SEM_INIT (&workers_data [index].start_worker_sem, 0);
 250                 pthread_create (&workers_data [index].thread, NULL, workers_thread_func, &workers_data [index]);
 251         }
 252
 253         workers_data [index].is_working = TRUE;
 254         MONO_SEM_POST (&workers_data [index].start_worker_sem);
 255         //g_print ("posted thread start %d %d\n", index, workers_data [index].start_worker_sem);
 256 }
 257
 258 static void
 259 workers_start_all_workers (int num_additional_workers)
 260 {
 261         int i;
 262
 263         g_assert (workers_num_working == 0);
 264         workers_num_working = workers_num + num_additional_workers;
 265
 266         for (i = 0; i < workers_num; ++i)
 267                 workers_start_worker (i);
 268 }
 269
 270 static void
 271 workers_join (void)
 272 {
 273         int i;
 274
 275         //g_print ("joining\n");
 276         for (i = 0; i < workers_num; ++i) {
 277                 if (workers_data [i].is_working)
 278                         MONO_SEM_WAIT (&workers_done_sem);
 279         }
 280         for (i = 0; i < workers_num; ++i)
 281                 workers_data [i].is_working = FALSE;
 282         //g_print ("joined\n");
 283
 284         g_assert (workers_num_working == 0);
 285         g_assert (workers_shared_buffer_used == 0);
 286
 287         for (i = 0; i < WORKERS_SHARED_BUFFER_SIZE; ++i)
 288                 g_assert (!workers_shared_buffer [i]);
 289 }
 290
 291 gboolean
 292 mono_sgen_is_worker_thread (pthread_t thread)
 293 {
 294         int i;
 295
 296         for (i = 0; i < workers_num; ++i) {
 297                 if (workers_data [i].thread == thread)
 298                         return TRUE;
 299         }
 300         return FALSE;
 301 }
 302
 303 #else
 304
 305 #define workers_distribute_gray_queue gray_queue
 306
 307 static int
 308 workers_change_num_working (int delta)
 309 {
 310         return -1;
 311 }
 312
 313 static void
 314 workers_distribute_gray_queue_sections (void)
 315 {
 316 }
 317
 318 static void
 319 workers_init (int num_workers)
 320 {
 321 }
 322
 323 static void
 324 workers_start_all_workers (int num_additional_workers)
 325 {
 326 }
 327
 328 static void
 329 workers_join (void)
 330 {
 331 }
 332
 333 gboolean
 334 mono_sgen_is_worker_thread (pthread_t thread)
 335 {
 336         return FALSE;
 337 }
 338
 339 #endif