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