mono/utils/lock-free-queue.c

   1 /*
   2  * lock-free-queue.c: Lock free queue.
   3  *
   4  * (C) Copyright 2011 Novell, Inc
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining
   7  * a copy of this software and associated documentation files (the
   8  * "Software"), to deal in the Software without restriction, including
   9  * without limitation the rights to use, copy, modify, merge, publish,
  10  * distribute, sublicense, and/or sell copies of the Software, and to
  11  * permit persons to whom the Software is furnished to do so, subject to
  12  * the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be
  15  * included in all copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  18  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  19  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  20  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  21  * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  22  * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  23  * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  24  */
  25
  26 /*
  27  * This is an implementation of a lock-free queue, as described in
  28  *
  29  * Simple, Fast, and Practical Non-Blocking and Blocking
  30  *   Concurrent Queue Algorithms
  31  * Maged M. Michael, Michael L. Scott
  32  * 1995
  33  *
  34  * A few slight modifications have been made:
  35  *
  36  * We use hazard pointers to rule out the ABA problem, instead of the
  37  * counter as in the paper.
  38  *
  39  * Memory management of the queue entries is done by the caller, not
  40  * by the queue implementation.  This implies that the dequeue
  41  * function must return the queue entry, not just the data.
  42  *
  43  * Therefore, the dummy entry must never be returned.  We do this by
  44  * re-enqueuing a new dummy entry after we dequeue one and then
  45  * retrying the dequeue.  We need more than one dummy because they
  46  * must be hazardly freed.
  47  */
  48
  49 #include <stdlib.h>
  50 #ifdef HAVE_UNISTD_H
  51 #include <unistd.h>
  52 #endif
  53
  54 #include <mono/utils/mono-membar.h>
  55 #include <mono/utils/hazard-pointer.h>
  56 #include <mono/utils/atomic.h>
  57
  58 #include <mono/utils/lock-free-queue.h>
  59
  60 #define INVALID_NEXT    ((MonoLockFreeQueueNode *volatile)-1)
  61 #define END_MARKER      ((MonoLockFreeQueueNode *volatile)-2)
  62 #define FREE_NEXT       ((MonoLockFreeQueueNode *volatile)-3)
  63
  64 /*
  65  * Initialize a lock-free queue in-place at @q.
  66  */
  67 void
  68 mono_lock_free_queue_init (MonoLockFreeQueue *q)
  69 {
  70         int i;
  71         for (i = 0; i < MONO_LOCK_FREE_QUEUE_NUM_DUMMIES; ++i) {
  72                 q->dummies [i].node.next = (i == 0) ? END_MARKER : FREE_NEXT;
  73                 q->dummies [i].in_use = i == 0 ? 1 : 0;
  74 #ifdef QUEUE_DEBUG
  75                 q->dummies [i].node.in_queue = i == 0 ? TRUE : FALSE;
  76 #endif
  77         }
  78
  79         q->head = q->tail = &q->dummies [0].node;
  80         q->has_dummy = 1;
  81 }
  82
  83 /*
  84  * Initialize @node's state. If @poison is TRUE, @node may not be enqueued to a
  85  * queue - @mono_lock_free_queue_node_unpoison must be called first; otherwise,
  86  * the node can be enqueued right away.
  87  *
  88  * The poisoning feature is mainly intended for ensuring correctness in complex
  89  * lock-free code that uses the queue. For example, in some code that reuses
  90  * nodes, nodes can be poisoned when they're dequeued, and then unpoisoned and
  91  * enqueued in their hazard free callback.
  92  */
  93 void
  94 mono_lock_free_queue_node_init (MonoLockFreeQueueNode *node, gboolean poison)
  95 {
  96         node->next = poison ? INVALID_NEXT : FREE_NEXT;
  97 #ifdef QUEUE_DEBUG
  98         node->in_queue = FALSE;
  99 #endif
 100 }
 101
 102 /*
 103  * Unpoisons @node so that it may be enqueued.
 104  */
 105 void
 106 mono_lock_free_queue_node_unpoison (MonoLockFreeQueueNode *node)
 107 {
 108         g_assert (node->next == INVALID_NEXT);
 109 #ifdef QUEUE_DEBUG
 110         g_assert (!node->in_queue);
 111 #endif
 112         node->next = FREE_NEXT;
 113 }
 114
 115 /*
 116  * Enqueue @node to @q. @node must have been initialized by a prior call to
 117  * @mono_lock_free_queue_node_init, and must not be in a poisoned state.
 118  */
 119 void
 120 mono_lock_free_queue_enqueue (MonoLockFreeQueue *q, MonoLockFreeQueueNode *node)
 121 {
 122         MonoThreadHazardPointers *hp = mono_hazard_pointer_get ();
 123         MonoLockFreeQueueNode *tail;
 124
 125 #ifdef QUEUE_DEBUG
 126         g_assert (!node->in_queue);
 127         node->in_queue = TRUE;
 128         mono_memory_write_barrier ();
 129 #endif
 130
 131         g_assert (node->next == FREE_NEXT);
 132         node->next = END_MARKER;
 133         for (;;) {
 134                 MonoLockFreeQueueNode *next;
 135
 136                 tail = (MonoLockFreeQueueNode *) get_hazardous_pointer ((gpointer volatile*)&q->tail, hp, 0);
 137                 mono_memory_read_barrier ();
 138                 /*
 139                  * We never dereference next so we don't need a
 140                  * hazardous load.
 141                  */
 142                 next = tail->next;
 143                 mono_memory_read_barrier ();
 144
 145                 /* Are tail and next consistent? */
 146                 if (tail == q->tail) {
 147                         g_assert (next != INVALID_NEXT && next != FREE_NEXT);
 148                         g_assert (next != tail);
 149
 150                         if (next == END_MARKER) {
 151                                 /*
 152                                  * Here we require that nodes that
 153                                  * have been dequeued don't have
 154                                  * next==END_MARKER.  If they did, we
 155                                  * might append to a node that isn't
 156                                  * in the queue anymore here.
 157                                  */
 158                                 if (InterlockedCompareExchangePointer ((gpointer volatile*)&tail->next, node, END_MARKER) == END_MARKER)
 159                                         break;
 160                         } else {
 161                                 /* Try to advance tail */
 162                                 InterlockedCompareExchangePointer ((gpointer volatile*)&q->tail, next, tail);
 163                         }
 164                 }
 165
 166                 mono_memory_write_barrier ();
 167                 mono_hazard_pointer_clear (hp, 0);
 168         }
 169
 170         /* Try to advance tail */
 171         InterlockedCompareExchangePointer ((gpointer volatile*)&q->tail, node, tail);
 172
 173         mono_memory_write_barrier ();
 174         mono_hazard_pointer_clear (hp, 0);
 175 }
 176
 177 static void
 178 free_dummy (gpointer _dummy)
 179 {
 180         MonoLockFreeQueueDummy *dummy = (MonoLockFreeQueueDummy *) _dummy;
 181         mono_lock_free_queue_node_unpoison (&dummy->node);
 182         g_assert (dummy->in_use);
 183         mono_memory_write_barrier ();
 184         dummy->in_use = 0;
 185 }
 186
 187 static MonoLockFreeQueueDummy*
 188 get_dummy (MonoLockFreeQueue *q)
 189 {
 190         int i;
 191         for (i = 0; i < MONO_LOCK_FREE_QUEUE_NUM_DUMMIES; ++i) {
 192                 MonoLockFreeQueueDummy *dummy = &q->dummies [i];
 193
 194                 if (dummy->in_use)
 195                         continue;
 196
 197                 if (InterlockedCompareExchange (&dummy->in_use, 1, 0) == 0)
 198                         return dummy;
 199         }
 200         return NULL;
 201 }
 202
 203 static gboolean
 204 is_dummy (MonoLockFreeQueue *q, MonoLockFreeQueueNode *n)
 205 {
 206         return n >= &q->dummies [0].node && n <= &q->dummies [MONO_LOCK_FREE_QUEUE_NUM_DUMMIES-1].node;
 207 }
 208
 209 static gboolean
 210 try_reenqueue_dummy (MonoLockFreeQueue *q)
 211 {
 212         MonoLockFreeQueueDummy *dummy;
 213
 214         if (q->has_dummy)
 215                 return FALSE;
 216
 217         dummy = get_dummy (q);
 218         if (!dummy)
 219                 return FALSE;
 220
 221         if (InterlockedCompareExchange (&q->has_dummy, 1, 0) != 0) {
 222                 dummy->in_use = 0;
 223                 return FALSE;
 224         }
 225
 226         mono_lock_free_queue_enqueue (q, &dummy->node);
 227
 228         return TRUE;
 229 }
 230
 231 /*
 232  * Dequeues a node from @q. Returns NULL if no nodes are available. The returned
 233  * node is hazardous and must be freed with @mono_thread_hazardous_try_free or
 234  * @mono_thread_hazardous_queue_free - it must not be freed directly.
 235  */
 236 MonoLockFreeQueueNode*
 237 mono_lock_free_queue_dequeue (MonoLockFreeQueue *q)
 238 {
 239         MonoThreadHazardPointers *hp = mono_hazard_pointer_get ();
 240         MonoLockFreeQueueNode *head;
 241
 242  retry:
 243         for (;;) {
 244                 MonoLockFreeQueueNode *tail, *next;
 245
 246                 head = (MonoLockFreeQueueNode *) get_hazardous_pointer ((gpointer volatile*)&q->head, hp, 0);
 247                 tail = (MonoLockFreeQueueNode*)q->tail;
 248                 mono_memory_read_barrier ();
 249                 next = head->next;
 250                 mono_memory_read_barrier ();
 251
 252                 /* Are head, tail and next consistent? */
 253                 if (head == q->head) {
 254                         g_assert (next != INVALID_NEXT && next != FREE_NEXT);
 255                         g_assert (next != head);
 256
 257                         /* Is queue empty or tail behind? */
 258                         if (head == tail) {
 259                                 if (next == END_MARKER) {
 260                                         /* Queue is empty */
 261                                         mono_hazard_pointer_clear (hp, 0);
 262
 263                                         /*
 264                                          * We only continue if we
 265                                          * reenqueue the dummy
 266                                          * ourselves, so as not to
 267                                          * wait for threads that might
 268                                          * not actually run.
 269                                          */
 270                                         if (!is_dummy (q, head) && try_reenqueue_dummy (q))
 271                                                 continue;
 272
 273                                         return NULL;
 274                                 }
 275
 276                                 /* Try to advance tail */
 277                                 InterlockedCompareExchangePointer ((gpointer volatile*)&q->tail, next, tail);
 278                         } else {
 279                                 g_assert (next != END_MARKER);
 280                                 /* Try to dequeue head */
 281                                 if (InterlockedCompareExchangePointer ((gpointer volatile*)&q->head, next, head) == head)
 282                                         break;
 283                         }
 284                 }
 285
 286                 mono_memory_write_barrier ();
 287                 mono_hazard_pointer_clear (hp, 0);
 288         }
 289
 290         /*
 291          * The head is dequeued now, so we know it's this thread's
 292          * responsibility to free it - no other thread can.
 293          */
 294         mono_memory_write_barrier ();
 295         mono_hazard_pointer_clear (hp, 0);
 296
 297         g_assert (head->next);
 298         /*
 299          * Setting next here isn't necessary for correctness, but we
 300          * do it to make sure that we catch dereferencing next in a
 301          * node that's not in the queue anymore.
 302          */
 303         head->next = INVALID_NEXT;
 304 #if QUEUE_DEBUG
 305         g_assert (head->in_queue);
 306         head->in_queue = FALSE;
 307         mono_memory_write_barrier ();
 308 #endif
 309
 310         if (is_dummy (q, head)) {
 311                 g_assert (q->has_dummy);
 312                 q->has_dummy = 0;
 313                 mono_memory_write_barrier ();
 314                 mono_thread_hazardous_try_free (head, free_dummy);
 315                 if (try_reenqueue_dummy (q))
 316                         goto retry;
 317                 return NULL;
 318         }
 319
 320         /* The caller must hazardously free the node. */
 321         return head;
 322 }