mono/utils/lock-free-array-queue.c

   1 /*
   2  * lock-free-array-queue.c: A lock-free somewhat-queue that doesn't
   3  * require hazard pointers.
   4  *
   5  * (C) Copyright 2011 Xamarin Inc.
   6  * Licensed under the MIT license. See LICENSE file in the project root for full license information.
   7  */
   8
   9 /*
  10  * The queue is a linked list of arrays (chunks).  Chunks are never
  11  * removed from the list, only added to the end, in a lock-free manner.
  12  *
  13  * Adding or removing an entry in the queue is only possible at the
  14  * end.  To do so, the thread first has to increment or decrement
  15  * q->num_used_entries.  The entry thus added or removed now "belongs"
  16  * to that thread.  It first CASes the state to BUSY, writes/reads the
  17  * entry data, and then sets the state to USED or FREE.
  18  */
  19
  20 #include <string.h>
  21
  22 #include <mono/utils/atomic.h>
  23 #include <mono/utils/mono-membar.h>
  24 #ifdef SGEN_WITHOUT_MONO
  25 #include <mono/sgen/sgen-gc.h>
  26 #include <mono/sgen/sgen-client.h>
  27 #else
  28 #include <mono/utils/mono-mmap.h>
  29 #endif
  30
  31 #include <mono/utils/lock-free-array-queue.h>
  32
  33 struct _MonoLockFreeArrayChunk {
  34         MonoLockFreeArrayChunk *next;
  35         gint32 num_entries;
  36         char entries [MONO_ZERO_LEN_ARRAY];
  37 };
  38
  39 typedef MonoLockFreeArrayChunk Chunk;
  40
  41 #define CHUNK_NTH(arr,chunk,index)      ((chunk)->entries + (index) * (arr)->entry_size)
  42
  43 static Chunk*
  44 alloc_chunk (MonoLockFreeArray *arr)
  45 {
  46         int size = mono_pagesize ();
  47         int num_entries = (size - (sizeof (Chunk) - arr->entry_size * MONO_ZERO_LEN_ARRAY)) / arr->entry_size;
  48         Chunk *chunk = (Chunk *) mono_valloc (NULL, size, MONO_MMAP_READ | MONO_MMAP_WRITE);
  49         g_assert (chunk);
  50         chunk->num_entries = num_entries;
  51         return chunk;
  52 }
  53
  54 static void
  55 free_chunk (Chunk *chunk)
  56 {
  57         mono_vfree (chunk, mono_pagesize ());
  58 }
  59
  60 gpointer
  61 mono_lock_free_array_nth (MonoLockFreeArray *arr, int index)
  62 {
  63         Chunk *chunk;
  64
  65         g_assert (index >= 0);
  66
  67         if (!arr->chunk_list) {
  68                 chunk = alloc_chunk (arr);
  69                 mono_memory_write_barrier ();
  70                 if (InterlockedCompareExchangePointer ((volatile gpointer *)&arr->chunk_list, chunk, NULL) != NULL)
  71                         free_chunk (chunk);
  72         }
  73
  74         chunk = arr->chunk_list;
  75         g_assert (chunk);
  76
  77         while (index >= chunk->num_entries) {
  78                 Chunk *next = chunk->next;
  79                 if (!next) {
  80                         next = alloc_chunk (arr);
  81                         mono_memory_write_barrier ();
  82                         if (InterlockedCompareExchangePointer ((volatile gpointer *) &chunk->next, next, NULL) != NULL) {
  83                                 free_chunk (next);
  84                                 next = chunk->next;
  85                                 g_assert (next);
  86                         }
  87                 }
  88                 index -= chunk->num_entries;
  89                 chunk = next;
  90         }
  91
  92         return CHUNK_NTH (arr, chunk, index);
  93 }
  94
  95 gpointer
  96 mono_lock_free_array_iterate (MonoLockFreeArray *arr, MonoLockFreeArrayIterateFunc func, gpointer user_data)
  97 {
  98         Chunk *chunk;
  99         for (chunk = arr->chunk_list; chunk; chunk = chunk->next) {
 100                 int i;
 101                 for (i = 0; i < chunk->num_entries; ++i) {
 102                         gpointer result = func (i, CHUNK_NTH (arr, chunk, i), user_data);
 103                         if (result)
 104                                 return result;
 105                 }
 106         }
 107         return NULL;
 108 }
 109
 110 void
 111 mono_lock_free_array_cleanup (MonoLockFreeArray *arr)
 112 {
 113         Chunk *chunk;
 114
 115         chunk = arr->chunk_list;
 116         arr->chunk_list = NULL;
 117         while (chunk) {
 118                 Chunk *next = chunk->next;
 119                 free_chunk (chunk);
 120                 chunk = next;
 121         }
 122 }
 123
 124 enum {
 125         STATE_FREE,
 126         STATE_USED,
 127         STATE_BUSY
 128 };
 129
 130 typedef struct {
 131         gint32 state;
 132         gpointer data [MONO_ZERO_LEN_ARRAY];
 133 } Entry;
 134
 135 typedef MonoLockFreeArrayQueue Queue;
 136
 137 /* The queue's entry size, calculated from the array's. */
 138 #define ENTRY_SIZE(q)   ((q)->array.entry_size - sizeof (gpointer))
 139
 140 void
 141 mono_lock_free_array_queue_push (MonoLockFreeArrayQueue *q, gpointer entry_data_ptr)
 142 {
 143         int index, num_used;
 144         Entry *entry;
 145
 146         do {
 147                 index = InterlockedIncrement (&q->num_used_entries) - 1;
 148                 entry = (Entry *) mono_lock_free_array_nth (&q->array, index);
 149         } while (InterlockedCompareExchange (&entry->state, STATE_BUSY, STATE_FREE) != STATE_FREE);
 150
 151         mono_memory_write_barrier ();
 152
 153         memcpy (entry->data, entry_data_ptr, ENTRY_SIZE (q));
 154
 155         mono_memory_write_barrier ();
 156
 157         entry->state = STATE_USED;
 158
 159         mono_memory_barrier ();
 160
 161         do {
 162                 num_used = q->num_used_entries;
 163                 if (num_used > index)
 164                         break;
 165         } while (InterlockedCompareExchange (&q->num_used_entries, index + 1, num_used) != num_used);
 166
 167         mono_memory_write_barrier ();
 168 }
 169
 170 gboolean
 171 mono_lock_free_array_queue_pop (MonoLockFreeArrayQueue *q, gpointer entry_data_ptr)
 172 {
 173         int index;
 174         Entry *entry;
 175
 176         do {
 177                 do {
 178                         index = q->num_used_entries;
 179                         if (index == 0)
 180                                 return FALSE;
 181                 } while (InterlockedCompareExchange (&q->num_used_entries, index - 1, index) != index);
 182
 183                 entry = (Entry *) mono_lock_free_array_nth (&q->array, index - 1);
 184         } while (InterlockedCompareExchange (&entry->state, STATE_BUSY, STATE_USED) != STATE_USED);
 185
 186         /* Reading the item must happen before CASing the state. */
 187         mono_memory_barrier ();
 188
 189         memcpy (entry_data_ptr, entry->data, ENTRY_SIZE (q));
 190
 191         mono_memory_barrier ();
 192
 193         entry->state = STATE_FREE;
 194
 195         mono_memory_write_barrier ();
 196
 197         return TRUE;
 198 }
 199
 200 void
 201 mono_lock_free_array_queue_cleanup (MonoLockFreeArrayQueue *q)
 202 {
 203         mono_lock_free_array_cleanup (&q->array);
 204         q->num_used_entries = 0;
 205 }