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 popBottom and steal also return NULL when the queue is empty.
  34  *
  35  * ---------------------------------------------------------------------------*/
  36
  37 #include "Rts.h"
  38 #include "RtsUtils.h"
  39 #include "WSDeque.h"
  40 #include "SMP.h" // for cas
  41
  42 #if defined(THREADED_RTS)
  43
  44 #define CASTOP(addr,old,new) ((old) == cas(((StgPtr)addr),(old),(new)))
  45
  46 /* -----------------------------------------------------------------------------
  47  * newWSDeque
  48  * -------------------------------------------------------------------------- */
  49
  50 /* internal helpers ... */
  51
  52 static StgWord
  53 roundUp2(StgWord val)
  54 {
  55     StgWord rounded = 1;
  56
  57     /* StgWord is unsigned anyway, only catch 0 */
  58     if (val == 0) {
  59         barf("DeQue,roundUp2: invalid size 0 requested");
  60     }
  61     /* at least 1 bit set, shift up to its place */
  62     do {
  63         rounded = rounded << 1;
  64     } while (0 != (val = val>>1));
  65     return rounded;
  66 }
  67
  68 WSDeque *
  69 newWSDeque (nat size)
  70 {
  71     StgWord realsize;
  72     WSDeque *q;
  73
  74     realsize = roundUp2(size); /* to compute modulo as a bitwise & */
  75
  76     q = (WSDeque*) stgMallocBytes(sizeof(WSDeque),   /* admin fields */
  77                                   "newWSDeque");
  78     q->elements = stgMallocBytes(realsize * sizeof(StgClosurePtr), /* dataspace */
  79                                  "newWSDeque:data space");
  80     q->top=0;
  81     q->bottom=0;
  82     q->topBound=0; /* read by writer, updated each time top is read */
  83
  84     q->size = realsize;  /* power of 2 */
  85     q->moduloSize = realsize - 1; /* n % size == n & moduloSize  */
  86
  87     ASSERT_WSDEQUE_INVARIANTS(q);
  88     return q;
  89 }
  90
  91 /* -----------------------------------------------------------------------------
  92  * freeWSDeque
  93  * -------------------------------------------------------------------------- */
  94
  95 void
  96 freeWSDeque (WSDeque *q)
  97 {
  98     stgFree(q->elements);
  99     stgFree(q);
 100 }
 101
 102 /* -----------------------------------------------------------------------------
 103  *
 104  * popWSDeque: remove an element from the write end of the queue.
 105  * Returns the removed spark, and NULL if a race is lost or the pool
 106  * empty.
 107  *
 108  * If only one spark is left in the pool, we synchronise with
 109  * concurrently stealing threads by using cas to modify the top field.
 110  * This routine should NEVER be called by a task which does not own
 111  * this deque.
 112  *
 113  * -------------------------------------------------------------------------- */
 114
 115 void *
 116 popWSDeque (WSDeque *q)
 117 {
 118     /* also a bit tricky, has to avoid concurrent steal() calls by
 119        accessing top with cas, when there is only one element left */
 120     StgWord t, b;
 121     void ** pos;
 122     long  currSize;
 123     void * removed;
 124
 125     ASSERT_WSDEQUE_INVARIANTS(q);
 126
 127     b = q->bottom;
 128     /* "decrement b as a test, see what happens" */
 129     q->bottom = --b;
 130     pos = (q->elements) + (b & (q->moduloSize));
 131     t = q->top; /* using topBound would give an *upper* bound, we
 132                    need a lower bound. We use the real top here, but
 133                    can update the topBound value */
 134     q->topBound = t;
 135     currSize = b - t;
 136     if (currSize < 0) { /* was empty before decrementing b, set b
 137                            consistently and abort */
 138         q->bottom = t;
 139         return NULL;
 140     }
 141     removed = *pos;
 142     if (currSize > 0) { /* no danger, still elements in buffer after b-- */
 143         return removed;
 144     }
 145     /* otherwise, has someone meanwhile stolen the same (last) element?
 146        Check and increment top value to know  */
 147     if ( !(CASTOP(&(q->top),t,t+1)) ) {
 148         removed = NULL; /* no success, but continue adjusting bottom */
 149     }
 150     q->bottom = t+1; /* anyway, empty now. Adjust bottom consistently. */
 151     q->topBound = t+1; /* ...and cached top value as well */
 152
 153     ASSERT_WSDEQUE_INVARIANTS(q);
 154
 155     return removed;
 156 }
 157
 158 /* -----------------------------------------------------------------------------
 159  * stealWSDeque
 160  * -------------------------------------------------------------------------- */
 161
 162 void *
 163 stealWSDeque_ (WSDeque *q)
 164 {
 165     void ** pos;
 166     void ** arraybase;
 167     StgWord sz;
 168     void * stolen;
 169     StgWord b,t;
 170
 171 // Can't do this on someone else's spark pool:
 172 // ASSERT_WSDEQUE_INVARIANTS(q);
 173
 174     b = q->bottom;
 175     t = q->top;
 176
 177     // NB. b and t are unsigned; we need a signed value for the test
 178     // below.
 179     if ((long)b - (long)t <= 0 ) {
 180         return NULL; /* already looks empty, abort */
 181   }
 182
 183     /* now access array, see pushBottom() */
 184     arraybase = q->elements;
 185     sz = q->moduloSize;
 186     pos = arraybase + (t & sz);
 187     stolen = *pos;
 188
 189     /* now decide whether we have won */
 190     if ( !(CASTOP(&(q->top),t,t+1)) ) {
 191         /* lost the race, someon else has changed top in the meantime */
 192         return NULL;
 193     }  /* else: OK, top has been incremented by the cas call */
 194
 195 // Can't do this on someone else's spark pool:
 196 // ASSERT_WSDEQUE_INVARIANTS(q);
 197
 198     return stolen;
 199 }
 200
 201 void *
 202 stealWSDeque (WSDeque *q)
 203 {
 204     void *stolen;
 205
 206     do {
 207         stolen = stealWSDeque_(q);
 208     } while (stolen == NULL && !looksEmptyWSDeque(q));
 209
 210     return stolen;
 211 }
 212
 213
 214 #define DISCARD_NEW
 215
 216 /* enqueue an element. Should always succeed by resizing the array
 217    (not implemented yet, silently fails in that case). */
 218 rtsBool
 219 pushWSDeque (WSDeque* q, void * elem)
 220 {
 221     StgWord t;
 222     void ** pos;
 223     StgWord sz = q->moduloSize;
 224     StgWord b = q->bottom;
 225
 226     ASSERT_WSDEQUE_INVARIANTS(q);
 227
 228     /* we try to avoid reading q->top (accessed by all) and use
 229        q->topBound (accessed only by writer) instead.
 230        This is why we do not just call empty(q) here.
 231     */
 232     t = q->topBound;
 233     if ( (StgInt)b - (StgInt)t >= (StgInt)sz ) {
 234         /* NB. 1. sz == q->size - 1, thus ">="
 235            2. signed comparison, it is possible that t > b
 236         */
 237         /* could be full, check the real top value in this case */
 238         t = q->top;
 239         q->topBound = t;
 240         if (b - t >= sz) { /* really no space left :-( */
 241             /* reallocate the array, copying the values. Concurrent steal()s
 242                will in the meantime use the old one and modify only top.
 243                This means: we cannot safely free the old space! Can keep it
 244                on a free list internally here...
 245
 246                Potential bug in combination with steal(): if array is
 247                replaced, it is unclear which one concurrent steal operations
 248                use. Must read the array base address in advance in steal().
 249             */
 250 #if defined(DISCARD_NEW)
 251             ASSERT_WSDEQUE_INVARIANTS(q);
 252             return rtsFalse; // we didn't push anything
 253 #else
 254             /* could make room by incrementing the top position here.  In
 255              * this case, should use CASTOP. If this fails, someone else has
 256              * removed something, and new room will be available.
 257              */
 258             ASSERT_WSDEQUE_INVARIANTS(q);
 259 #endif
 260         }
 261     }
 262     pos = (q->elements) + (b & sz);
 263     *pos = elem;
 264     (q->bottom)++;
 265
 266     ASSERT_WSDEQUE_INVARIANTS(q);
 267     return rtsTrue;
 268 }
 269
 270 #endif