1 /* -----------------------------------------------------------------------------
3 * (c) The GHC Team, 2009
5 * Work-stealing Deque data structure
7 * The implementation uses Double-Ended Queues with lock-free access
8 * (thereby often called "deque") as described in
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
14 * Author: Jost Berthold MSRC 07-09/2008
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.
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
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.
33 * Both popWSDeque and stealWSDeque also return NULL when the queue is empty.
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.
39 * ---------------------------------------------------------------------------*/
44 #include "SMP.h" // for cas
46 #if defined(THREADED_RTS)
48 #define CASTOP(addr,old,new) ((old) == cas(((StgPtr)addr),(old),(new)))
50 /* -----------------------------------------------------------------------------
52 * -------------------------------------------------------------------------- */
54 /* internal helpers ... */
61 /* StgWord is unsigned anyway, only catch 0 */
63 barf("DeQue,roundUp2: invalid size 0 requested");
65 /* at least 1 bit set, shift up to its place */
67 rounded = rounded << 1;
68 } while (0 != (val = val>>1));
78 realsize = roundUp2(size); /* to compute modulo as a bitwise & */
80 q = (WSDeque*) stgMallocBytes(sizeof(WSDeque), /* admin fields */
82 q->elements = stgMallocBytes(realsize * sizeof(StgClosurePtr), /* dataspace */
83 "newWSDeque:data space");
86 q->topBound=0; /* read by writer, updated each time top is read */
88 q->size = realsize; /* power of 2 */
89 q->moduloSize = realsize - 1; /* n % size == n & moduloSize */
91 ASSERT_WSDEQUE_INVARIANTS(q);
95 /* -----------------------------------------------------------------------------
97 * -------------------------------------------------------------------------- */
100 freeWSDeque (WSDeque *q)
102 stgFree(q->elements);
106 /* -----------------------------------------------------------------------------
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
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
117 * -------------------------------------------------------------------------- */
120 popWSDeque (WSDeque *q)
122 /* also a bit tricky, has to avoid concurrent steal() calls by
123 accessing top with cas, when there is only one element left */
129 ASSERT_WSDEQUE_INVARIANTS(q);
132 /* "decrement b as a test, see what happens" */
134 pos = (q->elements) + (b & (q->moduloSize));
135 t = q->top; /* using topBound would give an *upper* bound, we
136 need a lower bound. We use the real top here, but
137 can update the topBound value */
139 currSize = (long)b - (long)t;
140 if (currSize < 0) { /* was empty before decrementing b, set b
141 consistently and abort */
146 if (currSize > 0) { /* no danger, still elements in buffer after b-- */
147 // debugBelch("popWSDeque: t=%ld b=%ld = %ld\n", t, b, removed);
150 /* otherwise, has someone meanwhile stolen the same (last) element?
151 Check and increment top value to know */
152 if ( !(CASTOP(&(q->top),t,t+1)) ) {
153 removed = NULL; /* no success, but continue adjusting bottom */
155 q->bottom = t+1; /* anyway, empty now. Adjust bottom consistently. */
156 q->topBound = t+1; /* ...and cached top value as well */
158 ASSERT_WSDEQUE_INVARIANTS(q);
159 ASSERT(q->bottom >= q->top);
161 // debugBelch("popWSDeque: t=%ld b=%ld = %ld\n", t, b, removed);
166 /* -----------------------------------------------------------------------------
168 * -------------------------------------------------------------------------- */
171 stealWSDeque_ (WSDeque *q)
179 // Can't do this on someone else's spark pool:
180 // ASSERT_WSDEQUE_INVARIANTS(q);
185 // NB. b and t are unsigned; we need a signed value for the test
186 // below, because it is possible that t > b during a
187 // concurrent popWSQueue() operation.
188 if ((long)b - (long)t <= 0 ) {
189 return NULL; /* already looks empty, abort */
192 /* now access array, see pushBottom() */
193 arraybase = q->elements;
195 pos = arraybase + (t & sz);
198 /* now decide whether we have won */
199 if ( !(CASTOP(&(q->top),t,t+1)) ) {
200 /* lost the race, someon else has changed top in the meantime */
202 } /* else: OK, top has been incremented by the cas call */
204 // debugBelch("stealWSDeque_: t=%d b=%d\n", t, b);
206 // Can't do this on someone else's spark pool:
207 // ASSERT_WSDEQUE_INVARIANTS(q);
213 stealWSDeque (WSDeque *q)
218 stolen = stealWSDeque_(q);
219 } while (stolen == NULL && !looksEmptyWSDeque(q));
224 /* -----------------------------------------------------------------------------
226 * -------------------------------------------------------------------------- */
230 /* enqueue an element. Should always succeed by resizing the array
231 (not implemented yet, silently fails in that case). */
233 pushWSDeque (WSDeque* q, void * elem)
237 StgWord sz = q->moduloSize;
238 StgWord b = q->bottom;
240 ASSERT_WSDEQUE_INVARIANTS(q);
242 /* we try to avoid reading q->top (accessed by all) and use
243 q->topBound (accessed only by writer) instead.
244 This is why we do not just call empty(q) here.
247 if ( (StgInt)b - (StgInt)t >= (StgInt)sz ) {
248 /* NB. 1. sz == q->size - 1, thus ">="
249 2. signed comparison, it is possible that t > b
251 /* could be full, check the real top value in this case */
254 if (b - t >= sz) { /* really no space left :-( */
255 /* reallocate the array, copying the values. Concurrent steal()s
256 will in the meantime use the old one and modify only top.
257 This means: we cannot safely free the old space! Can keep it
258 on a free list internally here...
260 Potential bug in combination with steal(): if array is
261 replaced, it is unclear which one concurrent steal operations
262 use. Must read the array base address in advance in steal().
264 #if defined(DISCARD_NEW)
265 ASSERT_WSDEQUE_INVARIANTS(q);
266 return rtsFalse; // we didn't push anything
268 /* could make room by incrementing the top position here. In
269 * this case, should use CASTOP. If this fails, someone else has
270 * removed something, and new room will be available.
272 ASSERT_WSDEQUE_INVARIANTS(q);
276 pos = (q->elements) + (b & sz);
280 ASSERT_WSDEQUE_INVARIANTS(q);