mono/metadata/mono-wsq.c

   1 /*
   2  * mono-wsq.c: work-stealing queue
   3  *
   4  * Authors:
   5  *   Gonzalo Paniagua Javier (gonzalo@novell.com)
   6  *
   7  * Copyright (c) 2010 Novell, Inc (http://www.novell.com)
   8  * Copyright 2011 Xamarin, Inc (http://www.xamarin.com)
   9  */
  10
  11 #include <string.h>
  12 #include <mono/metadata/object.h>
  13 #include <mono/metadata/mono-wsq.h>
  14 #include <mono/utils/mono-semaphore.h>
  15 #include <mono/utils/mono-tls.h>
  16 #include <mono/utils/atomic.h>
  17
  18 #define INITIAL_LENGTH  32
  19 #define WSQ_DEBUG(...)
  20 //#define WSQ_DEBUG(...) g_message(__VA_ARGS__)
  21
  22 struct _MonoWSQ {
  23         volatile gint head;
  24         volatile gint tail;
  25         MonoArray *queue;
  26         gint32 mask;
  27         gint32 suspended;
  28         MonoSemType lock;
  29 };
  30
  31 #define NO_KEY ((guint32) -1)
  32 static MonoNativeTlsKey wsq_tlskey;
  33 static gboolean wsq_tlskey_inited = FALSE;
  34
  35 void
  36 mono_wsq_init ()
  37 {
  38         if (wsq_tlskey_inited)
  39                 return;
  40
  41         mono_native_tls_alloc (&wsq_tlskey, NULL);
  42         wsq_tlskey_inited = TRUE;
  43 }
  44
  45 void
  46 mono_wsq_cleanup ()
  47 {
  48         if (!wsq_tlskey_inited)
  49                 return;
  50         mono_native_tls_free (wsq_tlskey);
  51         wsq_tlskey_inited = FALSE;
  52 }
  53
  54 MonoWSQ *
  55 mono_wsq_create ()
  56 {
  57         MonoWSQ *wsq;
  58         MonoDomain *root;
  59
  60         if (!wsq_tlskey_inited)
  61                 return NULL;
  62
  63         wsq = g_new0 (MonoWSQ, 1);
  64         wsq->mask = INITIAL_LENGTH - 1;
  65         wsq->suspended = 0;
  66         MONO_GC_REGISTER_ROOT_SINGLE (wsq->queue);
  67         root = mono_get_root_domain ();
  68         wsq->queue = mono_array_new_cached (root, mono_defaults.object_class, INITIAL_LENGTH);
  69         MONO_SEM_INIT (&wsq->lock, 1);
  70         if (!mono_native_tls_set_value (wsq_tlskey, wsq)) {
  71                 mono_wsq_destroy (wsq);
  72                 wsq = NULL;
  73         }
  74         return wsq;
  75 }
  76
  77 gboolean
  78 mono_wsq_suspend (MonoWSQ *wsq)
  79 {
  80         return InterlockedCompareExchange (&wsq->suspended, 1, 0) == 0;
  81 }
  82
  83 void
  84 mono_wsq_destroy (MonoWSQ *wsq)
  85 {
  86         if (wsq == NULL || wsq->queue == NULL)
  87                 return;
  88
  89         g_assert (mono_wsq_count (wsq) == 0);
  90         MONO_GC_UNREGISTER_ROOT (wsq->queue);
  91         MONO_SEM_DESTROY (&wsq->lock);
  92         memset (wsq, 0, sizeof (MonoWSQ));
  93         if (wsq_tlskey_inited && mono_native_tls_get_value (wsq_tlskey) == wsq)
  94                 mono_native_tls_set_value (wsq_tlskey, NULL);
  95         g_free (wsq);
  96 }
  97
  98 gint
  99 mono_wsq_count (MonoWSQ *wsq)
 100 {
 101         if (!wsq)
 102                 return 0;
 103         return ((wsq->tail - wsq->head) & wsq->mask);
 104 }
 105
 106 gboolean
 107 mono_wsq_local_push (void *obj)
 108 {
 109         int tail;
 110         int head;
 111         int count;
 112         MonoWSQ *wsq;
 113
 114         if (obj == NULL || !wsq_tlskey_inited)
 115                 return FALSE;
 116
 117         wsq = (MonoWSQ *) mono_native_tls_get_value (wsq_tlskey);
 118         if (wsq == NULL) {
 119                 WSQ_DEBUG ("local_push: no wsq\n");
 120                 return FALSE;
 121         }
 122
 123         if (wsq->suspended) {
 124                 WSQ_DEBUG ("local_push: wsq suspended\n");
 125                 return FALSE;
 126         }
 127
 128         tail = wsq->tail;
 129         if (tail < wsq->head + wsq->mask) {
 130                 mono_array_setref (wsq->queue, tail & wsq->mask, (MonoObject *) obj);
 131                 wsq->tail = tail + 1;
 132                 WSQ_DEBUG ("local_push: OK %p %p\n", wsq, obj);
 133                 return TRUE;
 134         }
 135
 136         MONO_SEM_WAIT (&wsq->lock);
 137         head = wsq->head;
 138         count = wsq->tail - wsq->head;
 139         if (count >= wsq->mask) {
 140                 MonoArray *new_array;
 141                 int length;
 142                 int i;
 143
 144                 length = mono_array_length (wsq->queue);
 145                 new_array = mono_array_new_cached (mono_get_root_domain (), mono_defaults.object_class, length * 2);
 146                 for (i = 0; i < length; i++)
 147                         mono_array_setref (new_array, i, mono_array_get (wsq->queue, MonoObject*, (i + head) & wsq->mask));
 148
 149                 mono_gc_bzero_aligned (mono_array_addr (wsq->queue, MonoObject *, 0), sizeof (MonoObject*) * length);
 150                 wsq->queue = new_array;
 151                 wsq->head = 0;
 152                 wsq->tail = tail = count;
 153                 wsq->mask = (wsq->mask << 1) | 1;
 154         }
 155         mono_array_setref (wsq->queue, tail & wsq->mask, obj);
 156         wsq->tail = tail + 1;
 157         MONO_SEM_POST (&wsq->lock);
 158         WSQ_DEBUG ("local_push: LOCK %p  %p\n", wsq, obj);
 159         return TRUE;
 160 }
 161
 162 gboolean
 163 mono_wsq_local_pop (void **ptr)
 164 {
 165         int tail;
 166         gboolean res;
 167         MonoWSQ *wsq;
 168
 169         if (ptr == NULL || !wsq_tlskey_inited)
 170                 return FALSE;
 171
 172         wsq = (MonoWSQ *) mono_native_tls_get_value (wsq_tlskey);
 173         if (wsq == NULL) {
 174                 WSQ_DEBUG ("local_pop: no wsq\n");
 175                 return FALSE;
 176         }
 177
 178         tail = wsq->tail;
 179         if (wsq->head >= tail) {
 180                 WSQ_DEBUG ("local_pop: empty\n");
 181                 return FALSE;
 182         }
 183         tail--;
 184         InterlockedExchange (&wsq->tail, tail);
 185         if (wsq->head <= tail) {
 186                 *ptr = mono_array_get (wsq->queue, void *, tail & wsq->mask);
 187                 mono_array_set (wsq->queue, void *, tail & wsq->mask, NULL);
 188                 WSQ_DEBUG ("local_pop: GOT ONE %p %p\n", wsq, *ptr);
 189                 return TRUE;
 190         }
 191
 192         MONO_SEM_WAIT (&wsq->lock);
 193         if (wsq->head <= tail) {
 194                 *ptr = mono_array_get (wsq->queue, void *, tail & wsq->mask);
 195                 mono_array_set (wsq->queue, void *, tail & wsq->mask, NULL);
 196                 res = TRUE;
 197         } else {
 198                 wsq->tail = tail + 1;
 199                 res = FALSE;
 200         }
 201         MONO_SEM_POST (&wsq->lock);
 202         WSQ_DEBUG ("local_pop: LOCK %d %p %p\n", res, wsq, *ptr);
 203         return res;
 204 }
 205
 206 void
 207 mono_wsq_try_steal (MonoWSQ *wsq, void **ptr, guint32 ms_timeout)
 208 {
 209         if (wsq == NULL || ptr == NULL || *ptr != NULL || !wsq_tlskey_inited)
 210                 return;
 211
 212         if (mono_native_tls_get_value (wsq_tlskey) == wsq)
 213                 return;
 214
 215         if (mono_sem_timedwait (&wsq->lock, ms_timeout, FALSE) == 0) {
 216                 int head;
 217
 218                 head = wsq->head;
 219                 InterlockedExchange (&wsq->head, head + 1);
 220                 if (head < wsq->tail) {
 221                         *ptr = mono_array_get (wsq->queue, void *, head & wsq->mask);
 222                         mono_array_set (wsq->queue, void *, head & wsq->mask, NULL);
 223                         WSQ_DEBUG ("STEAL %p %p\n", wsq, *ptr);
 224                 } else {
 225                         wsq->head = head;
 226                 }
 227                 MONO_SEM_POST (&wsq->lock);
 228         }
 229 }
 230