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