Generality and Speed in Dual Containers

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.


Generic dual container

  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
  }
  

Extensions to generic dual container for nonblocking follow-up

  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;
  }
  

Peekable Treiber stack

  
  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
  }
  

Single polarity dual queue (SPDQ)

  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;
        }
      }
    }
  }
  

Extensions to SPDQ for nonblocking follow-up

  
  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
  } 
  

Multi polarity dual queue (MPDQ)

  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;
        }
      }
    }
  }
  

Extensions to MPDQ for nonblocking follow-up

  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
  


Last Change: 16 August 2016