Pseudocode from article of the above name, in TOPC '16, by Joe Izraelevitz and Michael L. Scott. The containers provided here are all dual data structures, which implement a "partial" remove; if the container is empty, the removing thread will block until data becomes available.
tuple Placeholder { enum {INVALID, VALID, ABORTED, SATISFIED}; Object* val = NULL; // 32 bits int state = INVALID; // 2 bits // needed for nonblocking; otherwise ignored: Request* req = NULL; // 30 bits // (target aligned to 4 bytes) Placeholder(Object* o) {val = o; req = NULL;} bool satisfy(Object* v, Request* req) { return CAS(this, <|NULL, VALID, NULL|>, <|v, SATISFIED, req|>); } }; class GDual { container* subcon[2]; bool nonblocking; const DATA = 0, ANTI = 1; Request* activeReq; // for nonblocking GDual(container *dc, *ac) { subcon[DATA] = dc; subcon[ANTI] = ac; activeReq = NULL; } }; Object* GDual:remove() { return remsert(NULL, ANTI); } Object* GDual:insert(Object* val) { return remsert(val, DATA); } Object* GDual:remsert(Object* o, bool polarity) { Placeholder* ph; bool nb = nonblocking && (polarity == DATA); Object* rtn = EMPTY; // allocate placeholder ph = new Placeholder(o); // actual transaction attempt while (rtn == EMPTY) { // begin transaction by emplacing placeholder subcon[polarity]->insert(ph); // do empty check on opposite rtn = doOppositeCheck(ph, polarity, nb); if (rtn != EMPTY) { // satisfied opposite, so finished return rtn; } // empty check failed, so now we try // to validate our placeholder if (CAS(ph, <|o, INVALID, NULL|>, <|o, VALID, NULL|>)) { if (polarity == DATA) return OK; // else spin waiting for data while (ph->state != SATISFIED) {} return ph->val; } // else we couldn't validate our placeholder // which means someone aborted us and // we need to retry ph = new Placeholder(o); } // end while empty } Object* GDual:doOppositeCheck (Placeholder* ph, bool polarity, bool nb) { if (!nb) return oppositeCheck(ph, polarity); else return oppositeCheckNB(ph); } Object* GDual:oppositeCheck (Placeholder* ph, bool polarity) { Placeholder* oph; // loop to remove until empty or // found a valid entry in opposite container while (true) { oph = subcon[!polarity]->remove(); if (oph == EMPTY) return EMPTY; else { // attempt to abort oph Object* oval = oph->val; if (CAS(oph, <|oval, INVALID, NULL|>, <|oval, ABORTED, NULL|>)) { // abort succeeded; continue; } else { // abort failed // placeholder guaranteed to be validated if (polarity == DATA) { *oph = <|ph->val, SATISFIED, NULL|>; return OK; } else return oph->val; } // end abort failed } // end attempt to abort } // end empty check loop }
class Request { Object* contents; placeholder* ph; Key key; }; Object* GDual:oppositeCheckNB (Placeholder* ph) { // this method is called only // by positive threads Placeholder* oph; Request* activeCopy; Request* myReq = new Request(); myReq->contents = ph->contents; Key k; Object* rtn = EMPTY; // loop to peek until empty // or valid entry in opposite queue while (true) { // read active request, help if necessary activeCopy = activeReq; if (activeCopy != NULL) { helpRequestNB(activeCopy); continue; } (k, oph) = subcon[ANTIDATA]->peek(); if (oph == EMPTY) { // opposite is empty // our check is complete return EMPTY; } // set up my request myReq->key = k; myReq->ph = oph; if (CAS(activeReq, NULL, myReq)) { // posted my request as active if (helpRequestNB(myReq) != ABORTED) { // my request was satisfied (by someone) if (oph->req() == ph->req()) { // verify my request ptr was used return OK; } } // my request was either aborted or oph // was previously satisfied by another // peeker, so reallocate and retry myReq = new Request(); myReq->contents = ph->contents; } } // end peeking loop } int GDual:helpRequestNB(Request* req) { int rtn; Placeholder* oph = req->ph; Object* oval = oph->contents; // attempt to abort opposite operation if (CAS(oph, <|NULL, INVALID, NULL|>, <|NULL, ABORTED, NULL|>)) { // abort succeeded rtn = ABORTED; } else if (opp_ph->state == ABORTED) { // someone else aborted the placeholder rtn = ABORTED; } else { // oph must be valid or satisfied oph->satisfy(req->contents, req); rtn = SATISFIED; } // take down posted request CAS(activeReq, req, NULL); // remove placeholder from opposite subcon[ANTIDATA]-> remove_cond(req->key); return rtn; }
struct Node { Object* val; Node* down; Node(Object* o){val=o;down=NULL;} }; class TreiberStack { Node* top = NULL; }; bool TreiberStack:push(Object* obj) { Node* newNode; Node* topCopy; newNode = new Node(obj); while (true) { topCopy = top; // read top pointer newNode->down = topCopy; // swing top; finished if success if (CAS(&top, topCopy, newNode)) { return true; } } } Object* TreiberStack:pop() { Node* topCopy; Node* newTop; while (true) { topCopy = top; // read top pointer // check if empty if (topCopy == NULL) return EMPTY; newTop = topCopy->down; // get new top // swing top; finished if success if (CAS(&top, topCopy, newTop)) { return topCopy->val; } } } (Key, Object*) TreiberStack:peek() { Node* topCopy; Key key; Object* obj; do { topCopy = top; key = (Key)topCopy; if (topCopy != NULL) { obj = topCopy->val; } else obj = NULL; // if empty } while (key != top); return (key,obj); } bool TreiberStack:remove_conditional( Key key) { Node* topCopy; Node* newTop; topCopy = (Node*)key; newTop = topCopy->down; // swing top; finished if success if (CAS(&top, topCopy, newTop)) return true; return false; // key wasn't top anymore }
class SPDQ { SP_CRQ* head, tail; bool nonblocking; }; class SP_CRQ:CRQ { bool sealed; bool polarity; bool seal(); }; class Waiter() { Object* val; Object* spin() { while (val == NULL) {} return val; } bool satisfy(Object* arg) { return CAS(&val, NULL, arg); } }; Object* SPDQ:dequeue() { Waiter* w = new Waiter(); // Waiter contains slot in which // to place satisfying data return denqueue(w, ANTIDATA); } void SPDQ:enqueue(Object* val) { return denqueue(val, DATA); } Object* SPDQ:denqueue (Object* val, bool polarity) { SP_CRQ* h; bool nb = (polarity == DATA_ && nonblocking); while (true) { h = head; // read polarity of queue if (h->polarity == polarity) { v = internal_enqueue(h, val, polarity); if (v != TWISTED) return OK; } else { if (nb) v = internal_dequeue_NB(val); else v = internal_dequeue(val, polarity); if (v != TOO_SLOW) return v; } // if internal operation failed, // head has changed, so retry } } bool SP_CRQ:seal() { int h, t; h = head; <|closed,t|> = tail; if(closed == 1 && h>=t){ sealed = true; return true; } } while (true) { if (sealed) return true; h = head; <|closed,t|> = tail; // check if not empty if (h < t && sealed==false) { return false; // if not, seal failed } // try to close while empty // (if an enqueue occurs, // tail moves, and CAS fails) if (CAS(&tail, t, <|1, h|>)) { // CAS succeeded, so CRQ is // closed and empty sealed = true; return true; } } } Object* SPDQ:internal_enqueue (SP_CRQ* h, Object* val, bool polarity) { SP_CRQ* t, next, newring; while (true) { t = tail; // verify tail is the actual tail if (t->next != NULL) { next = t->next; (void)CAS(&tail, t, next); continue; } // verify correct polarity (detect twisting) if (t->polarity != polarity) { (void)CAS(&head, h, h->next); return TWISTED; } // attempt enqueue on tail if (t->enqueue(val) == OK) { if (polarity == ANTIDATA) return ((Waiter*)val)->spin(); else return OK; } // else the tail is closed newring = new SP_CRQ(polarity); newring->enqueue(val); // append ring if (CAS(&t->next, NULL, newring)) { (void)CAS(&tail, t, newring); if (polarity == ANTIDATA) return ((Waiter*)val)->spin(); else return OK; } } } Object* SPDQ:internal_dequeue (Object* val, bool polarity) { SP_CRQ* h, next, newring; while (true) { h = head; // verify queue polarity didn't change if (h->polarity == polarity) { // head polarity inverted, // so recheck queue state return TOO_SLOW; } // dequeue from head v = h->dequeue(val); if (v != EMPTY) return v; // seal empty SP_CRQ_ so we can remove it else if (!h->seal()) continue; // at this point head SP_CRQ_ is sealed if (h->next != NULL) { // swing the head (void)CAS(&head, h, h->next); } else { // add a new tail and swing head to it newring = new SP_CRQ*(polarity); newring->enqueue(val); // append our new ring to list, // which will cause twisting if (CAS(&h->next, NULL, newring)) { (void)CAS(&tail, h, newring); // swing head to fix twisting (void)CAS(&head, h, h->next); if (polarity == ANTIDATA) return ((Waiter*)val)->spin(); else return v; } } } }
Object* SP_CRQ:internal_dequeue_NB (Object* arg) { <|bool closed, int idx|> h, t; // 1, 31 bits Slot* slot; bool closed, safe; int idx; Object* val; h = FAI(&head); bool paused = false; while (true) { slot = &this->ring[h.idx % R]; <|safe, idx, val|> = *slot; // find the wavefront if (idx > h.idx) { // behind the wavefront; move up h.idx++; continue; } if (idx < h.idx) { // too far ahead; we've lapped h.idx = h.idx - R; continue; } // now we know our index matches the slot // verify we are on the wavefront if (h.idx != 0 && ring[(h.idx - 1) % R].idx == h.idx - 1) { // we aren't on the wavefront, // so wait and check again if (!paused) { usleep(1); paused = true; continue; } else { // we already timed out, so search // backward for the wavefront h.idx--; continue; } } // now we know we're at the wavefront, // so dequeue if (val != NULL) { // slot is nonempty; try to dequeue if (((Waiter*)val)->satisfy(arg)) { (void)CAS(&slot, <|safe, h.idx, val|>, <|safe, h.idx + R, NULL|>); return OK; } else { // someone beat us to the wait structure (void)CAS(&slot, <|safe, h.idx, val|>, <|safe, h.idx + R, NULL|>); h.idx++; // behind wavefront; move up continue; } } else { // if slot is empty, mark for counterpart if (CAS(&slot, <|safe, h.idx, val|>, <|safe, h.idx + R, NULL|>)) { <|closed, t|> = tail; if (t <= h + 1) { fixState(); return EMPTY; } else { h = FAI(&head); continue; } } } // end else val is NULL } // end of main while loop }
tuple MP_Slot_ { bool safe; // 1 bit bool polarity; // 1 bit int idx; // 30 bits Object* val; // 32 bits (int or pointer) // padded to cache line size }; class MP_CRQ_ { // fields on distinct cache lines <|bool closing, int idx|> data_idx; // 1, 31 bits <|bool closing, int idx|> antidata_idx; // 1, 31 bits <|bool closed, int idx|> closed_info; // 1, 31 bits MP_CRQ* next; MP_Slot_ ring[R]; // initially ring[i] = <|1, 1, i, NULL|> forall i }; class MPDQ { // fields on distinct cache lines MP_CRQ* data_ptr; MP_CRQ* antidata_ptr; bool nonblocking; Object* dequeue() { Waiter* w = new Waiter(); return denqueue(w, ANTIDATA); } void enqueue(Object* val) { (void)denqueue(val, DATA); } }; Object* MP_CRQ:denqueue (Object* arg, bool polarity) { <|bool closing, int idx|> p; // 1, 31 bits <|bool closing, int idx|>* my_cntr; <|bool closing, int idx|>* their_cntr; MP_Slot* slot; bool safe; int idx; Object* val; bool slot_polarity; int starvation_counter = 0; int closeIdx = 0; // determine which index to use if (polarity == DATA) { my_cntr = &data_idx; their_cntr = &antidata_idx; } else { my_cntr = &antidata_idx; their_cntr = &data_idx; } // do denqueue while (true) { <|p.closing, p.index|> = FAI(my_cntr); // check for closing if (p.closing == true) { closeIdx = discovered_closing(p.idx, polarity); if (closeIdx <= p.idx) return CLOSED; } slot = &ring[p.idx % R]; while (true) { <|safe, slot_polarity, idx, val|> = *slot; // if slot nonempty if (val != NULL) { // try to dequeue opposite if (idx == p.idx && slot_polarity != polarity) { if (CAS(&slot, <|safe, p.idx, val, slot_polarity|>, <|safe, p.idx+R, NULL, slot_polarity|>)) { if (polarity == ANTIDATA) return val; else { ((Waiter*)val)->satisfy(arg); return NULL; } } else continue; } // failed to dequeue; signal slot unsafe // to prevent corresponding operation else { if (CAS(&slot, <|safe, idx, val, slot_polarity|>, <|0, idx, val, slot_polarity|>)) { break; } else continue; } } // if slot empty, try to enqueue self else { if (safe == 1 || their_cntr->idx <= p.idx) { if (CAS(&slot, <|safe, idx, NULL, slot_polarity|>, <|1, idx, arg, polarity|>)) { return OK; } else continue; } else break; // unsafe, try the next index } } // end inner while loop starvation_counter++; // if fail to make progress, close the ring if ((p.idx-their_cntr->idx >= R) || starvation_counter>STARVATION)) { my_ptr->close(); closeIdx = discovered_closing(p.idx, polarity); if (closeIdx <= p.idx) return CLOSED; } } // end outer while loop } Object* MPDQ:denqueue(Object* arg, bool polarity) { MP_CRQ* m, next, newring; int v; MP_CRQ** my_ptr; bool nb = (polarity == DATA_ && nonblocking); if (polarity == DATA) my_ptr = &data_ptr; else my_ptr = &antidata_ptr; while (true) { m = *my_ptr; // denqueue if (nb) v = m->denqueue_NB(arg); else v = m->denqueue(arg, polarity); // successful denqueue if (v != CLOSED) { if (polarity == ANTIDATA_ && v == OK) return ((Waiter*)val)->spin(); else if (polarity == DATA) return OK; else return v; } // my_ptr is closed, move to next if (m->next != NULL) (void)CAS(my_ptr, m, next); else { // if no next, add it newring = new MP_CRQ(); v = newring->denqueue(arg); if (CAS(&m->next, NULL, newring)) { (void)CAS(my_ptr, m, newring); if (polarity == ANTIDATA) return ((Waiter*)val)->spin(); else return OK; } } } }
Object* MP_CRQ:denqueue_NB(Object* arg) { <|bool closed, int idx|> p; // 1, 31 bits <|bool closed, int idx|>* my_cntr; <|bool closed, int idx|>* their_cntr; MP_Slot* slot; bool closed; bool safe, safe_pr; int idx, idx_pr; Object* val, *val_pr; bool slot_polarity, slot_polarity_pr; MP_Slot* slot_prev; int starvation_counter = 0; int close_idx = 0; bool paused = false; // get heads my_cntr = &data_idx; their_cntr = &antidata_idx; p = FAI(my_cntr); while (true) { // check for closed if (p.closed == 1 || data_idx.closed == 1) { close_idx = discovered_closing(p.idx, DATA); if (close_idx <= p.idx) return CLOSED; } // close queue if we fail to make progress or // if the queue is getting full (to ensure the // closed index is an actual slot) if ((data_idx.idx - anti_data.idx >= (R - 2 * MAX_THREADS) || starvation_counter > STARVATION) && !p.closed) { data_idx.close(); close_idx = discovered_closing( data_idx.idx, DATA); if (close_idx <= p.idx) return CLOSED; } slot = &ring[p.idx % R]; <|safe, slot_polarity, idx, val|> = *slot; starvation_counter++; // find wavefront if (p.idx < idx) { // behind wavefront p.idx++; continue; } if (idx < p.idx) { // lapped ahead of wavefront p.idx = p.idx - R; continue; } // verify at wavefront // and previous operation has finished slot_prev = ring[(p.idx - 1) % R]; <|safe_pr, slot_polarity_pr, idx_pr, val_pr|> = *slot_prev; if (p.idx != 0 && (idx_pr == (idx - 1)) && ((slot_polarity_pr == ANTIDATA && val_pr != NULL) || val_pr == NULL)) { // not ahead; wait and check again if (!paused) { usleep(1); paused = true; continue; } else { // already timed out; search backward p.idx--; continue; } } // on the wavefront, can operate if (val != NULL) { // slot is nonempty; try to dequeue if (idx == p.idx && slot_polarity != DATA) { if (((Waiter*)val)->satisfy(arg)) { (void)CAS(&slot, <|safe, p.idx, val|>, <|safe, p.idx + R, NULL|>); return OK; } else { // someone beat us to // the wait structure (void)CAS(&slot, <|safe, p.idx, val|>, <|safe, p.idx + R, NULL|>); p.idx++; continue; } } else if (slot_polarity == DATA) { // slot has wrong polarity; // we lost race to enqueue p.idx++; continue; } } else { // slot is empty; try to enqueue self if (safe == 1 || antidata_idx.idx <= p.idx) { // enqueue if (CAS(&slot, <|safe, p.idx, NULL|>, <|1, p.idx, arg|>)) return OK; } // else something got in the way - // either counterpart or competition } } // end outer while loop } // end dequeue