rts/parallel/WSDeque.c

   1 /* -----------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 2009
   4  *
   5  * Work-stealing Deque data structure
   6  *
   7  * The implementation uses Double-Ended Queues with lock-free access
   8  * (thereby often called "deque") as described in
   9  *
  10  * D.Chase and Y.Lev, Dynamic Circular Work-Stealing Deque.
  11  * SPAA'05, July 2005, Las Vegas, USA.
  12  * ACM 1-58113-986-1/05/0007
  13  *
  14  * Author: Jost Berthold MSRC 07-09/2008
  15  *
  16  * The DeQue is held as a circular array with known length. Positions
  17  * of top (read-end) and bottom (write-end) always increase, and the
  18  * array is accessed with indices modulo array-size. While this bears
  19  * the risk of overflow, we assume that (with 64 bit indices), a
  20  * program must run very long to reach that point.
  21  *
  22  * The write end of the queue (position bottom) can only be used with
  23  * mutual exclusion, i.e. by exactly one caller at a time.  At this
  24  * end, new items can be enqueued using pushBottom()/newSpark(), and
  25  * removed using popBottom()/reclaimSpark() (the latter implying a cas
  26  * synchronisation with potential concurrent readers for the case of
  27  * just one element).
  28  *
  29  * Multiple readers can steal from the read end (position top), and
  30  * are synchronised without a lock, based on a cas of the top
  31  * position. One reader wins, the others return NULL for a failure.
  32  *
  33  * Both popWSDeque and stealWSDeque also return NULL when the queue is empty.
  34  *
  35  * Testing: see testsuite/tests/ghc-regress/rts/testwsdeque.c.  If
  36  * there's anything wrong with the deque implementation, this test
  37  * will probably catch it.
  38  *
  39  * ---------------------------------------------------------------------------*/
  40
  41 #include "Rts.h"
  42 #include "RtsUtils.h"
  43 #include "WSDeque.h"
  44 #include "SMP.h" // for cas
  45
  46 #if defined(THREADED_RTS)
  47
  48 #define CASTOP(addr,old,new) ((old) == cas(((StgPtr)addr),(old),(new)))
  49
  50 /* -----------------------------------------------------------------------------
  51  * newWSDeque
  52  * -------------------------------------------------------------------------- */
  53
  54 /* internal helpers ... */
  55
  56 static StgWord
  57 roundUp2(StgWord val)
  58 {
  59     StgWord rounded = 1;
  60
  61     /* StgWord is unsigned anyway, only catch 0 */
  62     if (val == 0) {
  63         barf("DeQue,roundUp2: invalid size 0 requested");
  64     }
  65     /* at least 1 bit set, shift up to its place */
  66     do {
  67         rounded = rounded << 1;
  68     } while (0 != (val = val>>1));
  69     return rounded;
  70 }
  71
  72 WSDeque *
  73 newWSDeque (nat size)
  74 {
  75     StgWord realsize;
  76     WSDeque *q;
  77
  78     realsize = roundUp2(size); /* to compute modulo as a bitwise & */
  79
  80     q = (WSDeque*) stgMallocBytes(sizeof(WSDeque),   /* admin fields */
  81                                   "newWSDeque");
  82     q->elements = stgMallocBytes(realsize * sizeof(StgClosurePtr), /* dataspace */
  83                                  "newWSDeque:data space");
  84     q->top=0;
  85     q->bottom=0;
  86     q->topBound=0; /* read by writer, updated each time top is read */
  87
  88     q->size = realsize;  /* power of 2 */
  89     q->moduloSize = realsize - 1; /* n % size == n & moduloSize  */
  90
  91     ASSERT_WSDEQUE_INVARIANTS(q);
  92     return q;
  93 }
  94
  95 /* -----------------------------------------------------------------------------
  96  * freeWSDeque
  97  * -------------------------------------------------------------------------- */
  98
  99 void
 100 freeWSDeque (WSDeque *q)
 101 {
 102     stgFree(q->elements);
 103     stgFree(q);
 104 }
 105
 106 /* -----------------------------------------------------------------------------
 107  *
 108  * popWSDeque: remove an element from the write end of the queue.
 109  * Returns the removed spark, and NULL if a race is lost or the pool
 110  * empty.
 111  *
 112  * If only one spark is left in the pool, we synchronise with
 113  * concurrently stealing threads by using cas to modify the top field.
 114  * This routine should NEVER be called by a task which does not own
 115  * this deque.
 116  *
 117  * -------------------------------------------------------------------------- */
 118
 119 void *
 120 popWSDeque (WSDeque *q)
 121 {
 122     /* also a bit tricky, has to avoid concurrent steal() calls by
 123        accessing top with cas, when there is only one element left */
 124     StgWord t, b;
 125     void ** pos;
 126     long  currSize;
 127     void * removed;
 128
 129     ASSERT_WSDEQUE_INVARIANTS(q);
 130
 131     b = q->bottom;
 132     /* "decrement b as a test, see what happens" */
 133
 134     b--;
 135     q->bottom = b;
 136
 137     // very important that the following read of q->top does not occur
 138     // before the earlier write to q->bottom.
 139     store_load_barrier();
 140
 141     pos = (q->elements) + (b & (q->moduloSize));
 142     t = q->top; /* using topBound would give an *upper* bound, we
 143                    need a lower bound. We use the real top here, but
 144                    can update the topBound value */
 145     q->topBound = t;
 146     currSize = (long)b - (long)t;
 147     if (currSize < 0) { /* was empty before decrementing b, set b
 148                            consistently and abort */
 149         q->bottom = t;
 150         return NULL;
 151     }
 152     removed = *pos;
 153     if (currSize > 0) { /* no danger, still elements in buffer after b-- */
 154         // debugBelch("popWSDeque: t=%ld b=%ld = %ld\n", t, b, removed);
 155         return removed;
 156     }
 157     /* otherwise, has someone meanwhile stolen the same (last) element?
 158        Check and increment top value to know  */
 159     if ( !(CASTOP(&(q->top),t,t+1)) ) {
 160         removed = NULL; /* no success, but continue adjusting bottom */
 161     }
 162     q->bottom = t+1; /* anyway, empty now. Adjust bottom consistently. */
 163     q->topBound = t+1; /* ...and cached top value as well */
 164
 165     ASSERT_WSDEQUE_INVARIANTS(q);
 166     ASSERT(q->bottom >= q->top);
 167
 168     // debugBelch("popWSDeque: t=%ld b=%ld = %ld\n", t, b, removed);
 169
 170     return removed;
 171 }
 172
 173 /* -----------------------------------------------------------------------------
 174  * stealWSDeque
 175  * -------------------------------------------------------------------------- */
 176
 177 void *
 178 stealWSDeque_ (WSDeque *q)
 179 {
 180     void ** pos;
 181     void ** arraybase;
 182     StgWord sz;
 183     void * stolen;
 184     StgWord b,t;
 185
 186 // Can't do this on someone else's spark pool:
 187 // ASSERT_WSDEQUE_INVARIANTS(q);
 188
 189     // NB. these loads must be ordered, otherwise there is a race
 190     // between steal and pop.
 191     t = q->top;
 192     load_load_barrier();
 193     b = q->bottom;
 194
 195     // NB. b and t are unsigned; we need a signed value for the test
 196     // below, because it is possible that t > b during a
 197     // concurrent popWSQueue() operation.
 198     if ((long)b - (long)t <= 0 ) {
 199         return NULL; /* already looks empty, abort */
 200   }
 201
 202     /* now access array, see pushBottom() */
 203     arraybase = q->elements;
 204     sz = q->moduloSize;
 205     pos = arraybase + (t & sz);
 206     stolen = *pos;
 207
 208     /* now decide whether we have won */
 209     if ( !(CASTOP(&(q->top),t,t+1)) ) {
 210         /* lost the race, someon else has changed top in the meantime */
 211         return NULL;
 212     }  /* else: OK, top has been incremented by the cas call */
 213
 214     // debugBelch("stealWSDeque_: t=%d b=%d\n", t, b);
 215
 216 // Can't do this on someone else's spark pool:
 217 // ASSERT_WSDEQUE_INVARIANTS(q);
 218
 219     return stolen;
 220 }
 221
 222 void *
 223 stealWSDeque (WSDeque *q)
 224 {
 225     void *stolen;
 226
 227     do {
 228         stolen = stealWSDeque_(q);
 229     } while (stolen == NULL && !looksEmptyWSDeque(q));
 230
 231     return stolen;
 232 }
 233
 234 /* -----------------------------------------------------------------------------
 235  * pushWSQueue
 236  * -------------------------------------------------------------------------- */
 237
 238 #define DISCARD_NEW
 239
 240 /* enqueue an element. Should always succeed by resizing the array
 241    (not implemented yet, silently fails in that case). */
 242 rtsBool
 243 pushWSDeque (WSDeque* q, void * elem)
 244 {
 245     StgWord t;
 246     void ** pos;
 247     StgWord sz = q->moduloSize;
 248     StgWord b = q->bottom;
 249
 250     ASSERT_WSDEQUE_INVARIANTS(q);
 251
 252     /* we try to avoid reading q->top (accessed by all) and use
 253        q->topBound (accessed only by writer) instead.
 254        This is why we do not just call empty(q) here.
 255     */
 256     t = q->topBound;
 257     if ( (StgInt)b - (StgInt)t >= (StgInt)sz ) {
 258         /* NB. 1. sz == q->size - 1, thus ">="
 259            2. signed comparison, it is possible that t > b
 260         */
 261         /* could be full, check the real top value in this case */
 262         t = q->top;
 263         q->topBound = t;
 264         if (b - t >= sz) { /* really no space left :-( */
 265             /* reallocate the array, copying the values. Concurrent steal()s
 266                will in the meantime use the old one and modify only top.
 267                This means: we cannot safely free the old space! Can keep it
 268                on a free list internally here...
 269
 270                Potential bug in combination with steal(): if array is
 271                replaced, it is unclear which one concurrent steal operations
 272                use. Must read the array base address in advance in steal().
 273             */
 274 #if defined(DISCARD_NEW)
 275             ASSERT_WSDEQUE_INVARIANTS(q);
 276             return rtsFalse; // we didn't push anything
 277 #else
 278             /* could make room by incrementing the top position here.  In
 279              * this case, should use CASTOP. If this fails, someone else has
 280              * removed something, and new room will be available.
 281              */
 282             ASSERT_WSDEQUE_INVARIANTS(q);
 283 #endif
 284         }
 285     }
 286     pos = (q->elements) + (b & sz);
 287     *pos = elem;
 288     q->bottom = b + 1;
 289
 290     ASSERT_WSDEQUE_INVARIANTS(q);
 291     return rtsTrue;
 292 }
 293
 294 #endif