bags.c

Go to the documentation of this file.

// Shared variables
block_t * globalHeadBlock[NR_THREADS];

// Thread-local storage
block_t * threadBlock, stealBlock, stealPrev;
bool foundAdd;
int threadHead, stealHead, stealIndex;
int threadID; // Unique number between 0 ... NR_THREADS

struct blockp_t {block_t * p; bool mark2 :1; bool mark1 :1;};
struct block_t : public node_t {
  void * nodes[BLOCK_SIZE];
  long notifyAdd[NR_THREADS/WORD_SIZE];
  blockp_t next;
};
void Mark1Block(block_t *block) {
  for(;;) {
    blockp_t next = block->next;
    if(next.p == NULL || next.mark1 || CAS(&block->next,next,{next.p,next.mark2,true})) break;
  }
}
block_t * NewBlock() {
  block_t * block = NewNode(sizeof(block_t));
  block->next = NULL;
  NotifyAll(block);
  for(int i=0;i<BLOCK_SIZE;i++) block->nodes[i]=NULL;
  return block;
}
void NotifyAll(block_t *block) {
  for(int i=0;i<NR_THREADS/WORD_SIZE;i++)
    block->notifyAdd[i] = 0;
}
void NotifyStart(block_t *block, int Id) {
  do {
    long old = block->notifyAdd[Id/WORD_SIZE];
  } while(!CAS(&block->notifyAdd[Id/WORD_SIZE],old,old|(1<<(Id%WORD_SIZE))));
}

bool NotifyCheck(block_t *block, int Id) {
  return (block->notifyAdd[Id/WORD_SIZE]&(1<<(Id%WORD_SIZE)))==0;
}

void InitBag() {
  for(int i=0;i<NR_THREADS;i++) globalHeadBlock[i]=NULL;
}
void InitThread() {
  threadBlock = globalHeadBlock[threadID];
  threadHead = BLOCK_SIZE;
  stealIndex = 0;
  stealBlock = NULL; stealPrev = NULL;
  stealHead = BLOCK_SIZE;
}

void Add(void *item) {
  int head = threadHead;
  block_t * block = threadBlock;
  for(;;) {
    if(head==BLOCK_SIZE) {
      block_t *oldblock = block;
      block = NewBlock();
      block->next=oldblock;
      globalHeadBlock[threadID]=block;
      threadBlock = block;
      head = 0;
    }
    else if(block->nodes[head]==NULL) {
      NotifyAll(block);
      block->nodes[head]=item;
      threadHead = head+1;
      return;
    }
    else head++;
  }
}

void * TryRemoveAny() {
  int head = threadHead-1;
  block_t * block = threadBlock;
  int round = 0;
  for(;;) {
    if(block == NULL || (head<0 && block->next.p ==NULL)) {
      do {
    int i=0;
    do {
      void *result = TryStealBlock(round);
      if(result!=NULL) return result;
      if(foundAdd) {
        round=0; i=0;
      }
      else if(stealBlock==NULL) i++;
    } while(i<NR_THREADS);
      } while(++round<=NR_THREADS);
      return NULL;
    }
    if(head<0) {
      Mark1Block(block);
      for(;;) {
    blockp_t next=DeRefLink(&block->next);
    if(next.mark2) Mark1Block(next.p);
    if(next.mark1) {
      if(next.p) NotifyAll(next.p);
      if(CAS(&globalHeadBlock[threadId],block,next.p)) {
        block->next = {NULL,false,true};
        DeleteNode(block); ReScan(next);
        block=next.p;
      }
      else block=DeRefLink(&globalHeadBlock[threadId]);
    }
    else break;
      }
      threadBlock = block;
      threadHead = BLOCK_SIZE;
      head = BLOCK_SIZE-1;
    }
    else {
      void *data = block->nodes[head];
      if(data==NULL) head--;
      else if(CAS(&block->nodes[head],data,NULL)) {
    threadHead = head;
    return data;
      }
    }
  }
}

/*
gc api:
node_t * NewNode(int size);
void DeleteNode(node_t *node);
node_t * DeRefLink(node_t **link);
void ReleaseRef(node_t *node);
void ReScan(node_t *node);
*/

void * TryStealBlock(int round) {
  int head = stealHead;
  block_t * block = stealBlock;
  foundAdd = false;
  if(block==NULL) {
    block = DeRefLink(&globalHeadBlock[stealIndex]);
    stealBlock = block;
    stealHead = head = 0;
  }
  if(head==BLOCK_SIZE) {
    stealBlock = block = NextStealBlock(block);
    head = 0;
  }
  if(block==NULL) {
    stealIndex = (stealIndex+1)%NR_THREADS;
    stealHead = 0;
    stealBlock = NULL; stealPrev = NULL;
    return NULL;
  }
  if(round==1) NotifyStart(block,threadId);
  else if(round>1 && NotifyCheck(block,threadId))foundAdd = true;
  for(;;) {
    if(head==BLOCK_SIZE) {
      stealHead = head;
      return NULL;
    }
    else {
      void *data = block->nodes[head];
      if(data==NULL) head++;
      else if(CAS(&block->nodes[head],data,NULL)) {
    stealHead = head;
    return data;
      }
    }
  }
}

block_t * NextStealBlock(block_t *block) {
  blockp_t next;
  for(;;) {
    if(block==NULL) {
      block = DeRefLink(&globalHeadBlock[stealIndex]);
      break;
    }
    next = DeRefLink(&block->next);
    if(next.mark2) Mark1Block(next.p);
    if(stealPrev == NULL || next.p == NULL) {
      if(next.mark1) {
    if(next.p) NotifyAll(next.p);
    if(CAS(&globalHeadBlock[stealIndex],block,next.p)) {
      block->next = {NULL,false,true};
      DeleteNode(block); ReScan(next);
    }
    else {
      stealPrev = NULL;
      block = DeRefLink(&globalHeadBlock[stealIndex]);
      continue;
    }
      }
      else stealPrev = block;
    }
    else {
      if(next.mark1) {
    blockp_t prevnext = {block,stealPrev->next.mark2,false};
    if(CAS(&stealPrev->next,prevnext,next.p)){
      block->next = {NULL,false,true};
      DeleteNode(block); ReScan(next);
    }
    else {
      stealPrev = NULL;
      block = DeRefLink(&globalHeadBlock[stealIndex]);
      continue;
    }
      }
      else if(block==stealBlock) {
    if(CAS(&stealPrev->next,block,{block,true,false})) {
      Mark1Block(block);
      continue;
    }
    else {
      stealPrev = NULL;
      block = DeRefLink(&globalHeadBlock[stealIndex]);
      continue;
    }
      }
      else stealPrev = block;
    }
    if(block == stealBlock || next.p == stealBlock) {
      block=next.p;
      break;
    }
    block=next.p;
  }
  return block;
}