Robust/src/Runtime/DSTM/interface/queue.c

   1 #include "queue.h"
   2
   3 volatile int headoffset, tailoffset;
   4 char * memory;
   5 pthread_mutex_t qlock;
   6 pthread_mutexattr_t qlockattr;
   7 pthread_cond_t qcond;
   8
   9 #define QSIZE 2048 //2 KB
  10
  11 #ifdef LOGEVENTS
  12 extern char bigarray[16*1024*1024];
  13 extern int bigindex;
  14 #define LOGEVENT(x) { \
  15     int tmp=bigindex++;                         \
  16     bigarray[tmp]=x;                            \
  17 }
  18 #else
  19 #define LOGEVENT(x)
  20 #endif
  21
  22 void queueInit(void) {
  23   /* Intitialize primary queue */
  24   headoffset=0;
  25   tailoffset=0;
  26   memory=malloc(QSIZE+sizeof(int)); //leave space for -1
  27   pthread_mutexattr_init(&qlockattr);
  28   pthread_mutexattr_settype(&qlockattr, PTHREAD_MUTEX_RECURSIVE_NP);
  29   pthread_mutex_init(&qlock, &qlockattr);
  30   pthread_cond_init(&qcond, NULL);
  31 }
  32
  33 void * getmemory(int size) {
  34   int tmpoffset=headoffset+size+sizeof(int);
  35   if (tmpoffset>QSIZE) {
  36     //Wait for tail to go past end
  37     tmpoffset=size+sizeof(int);
  38     if (headoffset<tailoffset) {
  39       pthread_cond_signal(&qcond); //wake the other thread up
  40       return NULL;
  41     }
  42     //Wait for tail to go past new start
  43     if (tailoffset<=tmpoffset) {
  44       pthread_cond_signal(&qcond); //wake the other thread up
  45       return NULL;
  46     }
  47     *((int *)(memory+headoffset))=-1; //safe because we left space
  48     *((int*)memory)=size+sizeof(int);
  49     return memory+sizeof(int);
  50   } else {
  51     if (headoffset<tailoffset&&tailoffset<=tmpoffset) {
  52       pthread_cond_signal(&qcond); //wake the other thread up
  53       return NULL;
  54     }
  55     *((int*)(memory+headoffset))=size+sizeof(int);
  56     return memory+headoffset+sizeof(int);
  57   }
  58 }
  59
  60 void movehead(int size) {
  61   int tmpoffset=headoffset+size+sizeof(int);
  62   if (tmpoffset>QSIZE) {
  63     headoffset=size+sizeof(int);
  64   } else
  65     headoffset=tmpoffset;
  66   pthread_cond_signal(&qcond); //wake the other thread up
  67 }
  68
  69 void * gettail() {
  70   while(tailoffset==headoffset) {
  71     //Sleep
  72     LOGEVENT('T');
  73     pthread_mutex_lock(&qlock);
  74     if (tailoffset==headoffset)
  75       pthread_cond_wait(&qcond, &qlock);
  76     pthread_mutex_unlock(&qlock);
  77     LOGEVENT('W');
  78   }
  79   if (*((int *)(memory+tailoffset))==-1) {
  80     tailoffset=0; //do loop
  81   }
  82
  83   return memory+tailoffset+sizeof(int);
  84 }
  85
  86
  87 int numavailable() {
  88   int tmp=tailoffset;
  89   int available=0;
  90   if (*((int *)(memory+tmp))==-1) {
  91     tmp=0;
  92   }
  93   while(tmp!=headoffset) {
  94     available++;
  95     tmp=tmp+*((int *)(memory+tmp));
  96     if (tmp>QSIZE|| (*((int *)(memory+tmp))==-1)) {
  97       break;
  98     }
  99   }
 100   return available;
 101 }
 102
 103 void incmulttail(int num) {
 104   int i;
 105   for(i=0; i<num; i++) {
 106     int tmpoffset=tailoffset+*((int *)(memory+tailoffset));
 107     if (tmpoffset>QSIZE)
 108       tailoffset=0;
 109     else
 110       tailoffset=tmpoffset;
 111   }
 112 }
 113
 114 void resetqueue() {
 115   headoffset=0;
 116   tailoffset=0;
 117 }
 118
 119 void inctail() {
 120   int tmpoffset=tailoffset+*((int *)(memory+tailoffset));
 121   if (tmpoffset>QSIZE)
 122     tailoffset=0;
 123   else
 124     tailoffset=tmpoffset;
 125 }
 126
 127 void predealloc() {
 128   free(memory);
 129 }
 130