while working on memory pool for task records, fixed bug where mem Q hashtable entrie...
[IRC.git] / Robust / src / Runtime / mlp_runtime.c
index fc0fcef32befe021ff3de531db10b80c07a51071..3e7fe723eff15631a3cda9bffc54632975890dfc 100644 (file)
@@ -7,16 +7,10 @@
 #include "mem.h"
 #include "mlp_runtime.h"
 #include "workschedule.h"
+#include "methodheaders.h"
 
 
-/*
-__thread struct Queue* seseCallStack;
-__thread pthread_once_t mlpOnceObj = PTHREAD_ONCE_INIT;
-void mlpInitOncePerThread() {
-  seseCallStack = createQueue();
-}
-*/
-__thread SESEcommon_p seseCaller;
+__thread SESEcommon* runningSESE;
 
 
 void* mlpAllocSESErecord( int size ) {
@@ -63,7 +57,7 @@ REntry* mlpCreateREntry(int type, void* seseToIssue){
 }
 
 int isParent(REntry *r) {
-  if (r->type==PARENTREAD || r->type==PARENTWRITE) {
+  if (r->type==PARENTREAD || r->type==PARENTWRITE || r->type==PARENTCOARSE) {
     return TRUE;
   } else {
     return FALSE;
@@ -240,7 +234,12 @@ int ADDTABLE(MemoryQueue *q, REntry *r) {
   //at this point, have table
   Hashtable* table=(Hashtable*)q->tail;
   r->hashtable=table; // set rentry's hashtable
-  if((*(r->pointer)==0 || (*(r->pointer)!=0 && BARRIER() && table->unresolvedQueue!=NULL))){
+  if( *(r->pointer)==0 || 
+      ( *(r->pointer)!=0 && 
+        BARRIER() && 
+        table->unresolvedQueue!=NULL
+        )        
+   ){
     struct Queue* val;
     // grab lock on the queue    
     do {  
@@ -267,7 +266,16 @@ int ADDTABLE(MemoryQueue *q, REntry *r) {
     return NOTREADY;
   }
   BinItem * val;
-  int key=generateKey((unsigned int)(unsigned INTPTR)*(r->pointer));
+
+  // leave this--its a helpful test when things are going bonkers
+  //if( OBJPTRPTR_2_OBJOID( r->pointer ) == 0 ) {
+  //  // we started numbering object ID's at 1, if we try to
+  //  // hash a zero oid, something BAD is about to happen!
+  //  printf( "Tried to insert invalid object type=%d into mem Q hashtable!\n",
+  //          OBJPTRPTR_2_OBJTYPE( r->pointer ) );
+  //  exit( -1 );
+  //}
+  int key=generateKey( OBJPTRPTR_2_OBJOID( r->pointer ) );
   do {  
     val=(BinItem*)0x1;       
     BinElement* bin=table->array[key];
@@ -288,7 +296,7 @@ int ADDTABLE(MemoryQueue *q, REntry *r) {
 int ADDTABLEITEM(Hashtable* table, REntry* r, int inc){
  
   BinItem * val;
-  int key=generateKey((unsigned int)(unsigned INTPTR)*(r->pointer));
+  int key=generateKey( OBJPTRPTR_2_OBJOID( r->pointer ) );
   do {  
     val=(BinItem*)0x1;       
     BinElement* bin=table->array[key];
@@ -346,7 +354,7 @@ int WRITEBINCASE(Hashtable *T, REntry *r, BinItem *val, int key, int inc) {
   //chain of bins exists => tail is valid
   //if there is something in front of us, then we are not ready
 
-  int retval;
+  int retval=NOTREADY;
   BinElement* be=T->array[key];
 
   BinItem *bintail=be->tail;
@@ -356,13 +364,7 @@ int WRITEBINCASE(Hashtable *T, REntry *r, BinItem *val, int key, int inc) {
   b->item.total=1;
 
   // note: If current table clears all dependencies, then write bin is ready
-  if (T->item.total==0){
-    retval=READY;    
-  }else{
-    retval=NOTREADY;
-  }
-  b->item.status=retval;
-  //  b->item.status=NOTREADY;
+  
   
   if(inc){
     atomic_inc(&T->item.total);
@@ -372,6 +374,26 @@ int WRITEBINCASE(Hashtable *T, REntry *r, BinItem *val, int key, int inc) {
   r->binitem=(BinItem*)b;
 
   be->tail->next=(BinItem*)b;
+  //need to check if we can go...
+  BARRIER();
+  if (T->item.status==READY) {
+    for(;val!=NULL;val=val->next) {
+      if (val==((BinItem *)b)) {
+       //ready to retire
+       retval=READY;
+       if (isParent(r)) {
+         b->item.status=retval;//unsure if really needed at this point..
+         be->head=NULL; // released lock
+         return retval;
+       }
+       break;
+      } else if (val->total!=0) {
+       break;
+      }
+    }
+  }
+  
+  b->item.status=retval;
   be->tail=(BinItem*)b;
   be->head=val;
   return retval;
@@ -390,7 +412,7 @@ READBINCASE(Hashtable *T, REntry *r, BinItem *val, int key, int inc) {
 int TAILREADCASE(Hashtable *T, REntry *r, BinItem *val, BinItem *bintail, int key, int inc) {
   ReadBinItem * readbintail=(ReadBinItem*)T->array[key]->tail;
   int status, retval;
-  if (readbintail->item.status=READY) { 
+  if (readbintail->item.status==READY) { 
     status=READY;
     retval=READY;
     if (isParent(r)) {
@@ -414,7 +436,6 @@ int TAILREADCASE(Hashtable *T, REntry *r, BinItem *val, BinItem *bintail, int ke
     readbintail->array[readbintail->index++]=r;
     atomic_inc(&readbintail->item.total);
     r->binitem=(BinItem*)readbintail;
-    //printf("grouping with %d\n",readbintail->index);
   }
   if(inc){
     atomic_inc(&T->item.total);
@@ -492,8 +513,9 @@ ADDVECTOR(MemoryQueue *Q, REntry *r) {
     void* flag=NULL;
     flag=(void*)LOCKXCHG((unsigned INTPTR*)&(V->array[index]), (unsigned INTPTR)flag); 
     if (flag!=NULL) {
-      if (isParent(r)) { //parent's retire immediately
+      if (isParentCoarse(r)) { //parent's retire immediately
         atomic_dec(&V->item.total);
+        V->index--;
       }
       return READY;
     } else {
@@ -564,7 +586,7 @@ RETIREHASHTABLE(MemoryQueue *q, REntry *r) {
 }
 
 RETIREBIN(Hashtable *T, REntry *r, BinItem *b) {
-  int key=generateKey((unsigned int)(unsigned INTPTR)*(r->pointer));
+  int key=generateKey( OBJPTRPTR_2_OBJOID( r->pointer ) );
   if(isFineRead(r)) {
     atomic_dec(&b->total);
   }
@@ -578,7 +600,7 @@ RETIREBIN(Hashtable *T, REntry *r, BinItem *b) {
     // at this point have locked bin
     BinItem *ptr=val;
     int haveread=FALSE;
-     int i;
+    int i;
     while (ptr!=NULL) {
        if (isReadBinItem(ptr)) {
        ReadBinItem* rptr=(ReadBinItem*)ptr;
@@ -712,10 +734,9 @@ RESOLVEHASHTABLE(MemoryQueue *Q, Hashtable *T) {
           } else if((BinItem*)rptr==val) {
             val=val->next;
           }
-          rptr->item.status=READY; { 
-         }
-         ptr=ptr->next;
+          rptr->item.status=READY; 
        } 
+       ptr=ptr->next;  
       }while(ptr!=NULL);   
     }
     bin->head=val; // released lock;
@@ -767,9 +788,133 @@ resolveDependencies(REntry* rentry){
   }
 }
 
+void INITIALIZEBUF(MemoryQueue * q){  
+  int i=0;
+  for(i=0; i<NUMBINS; i++){
+    q->binbuf[i]=NULL;
+  }
+  q->bufcount=0;
+}
+
+void ADDRENTRYTOBUF(MemoryQueue * q, REntry * r){
+  q->buf[q->bufcount]=r;
+  q->bufcount++;
+}
+
+int RESOLVEBUFFORHASHTABLE(MemoryQueue * q, Hashtable* table, SESEcommon *seseCommon){  
+  int i;
+ // first phase: only consider write rentry
+  for(i=0; i<q->bufcount;i++){
+    REntry *r=q->buf[i];
+    if(r->type==WRITE){
+      int key=generateKey( OBJPTRPTR_2_OBJOID( r->pointer ) );
+      if(q->binbuf[key]==NULL){
+       // for multiple writes, add only the first write that hashes to the same bin
+       q->binbuf[key]=r;  
+      }else{
+       q->buf[i]=NULL;
+      }
+    }
+  }
+  // second phase: enqueue read items if it is eligible
+  for(i=0; i<q->bufcount;i++){
+    REntry *r=q->buf[i];    
+    if(r!=NULL && r->type==READ){
+      int key=generateKey( OBJPTRPTR_2_OBJOID( r->pointer ) );
+      if(q->binbuf[key]==NULL){
+       // read item that hashes to the bin which doen't contain any write
+       seseCommon->rentryArray[seseCommon->rentryIdx++]=r;
+       if(ADDTABLEITEM(table, r, FALSE)==READY){
+         resolveDependencies(r);
+       }
+      }
+      q->buf[i]=NULL;
+    }
+  }
+  
+  // then, add only one of write items that hashes to the same bin
+  for(i=0; i<q->bufcount;i++){
+    REntry *r=q->buf[i];   
+    if(r!=NULL){
+      seseCommon->rentryArray[seseCommon->rentryIdx++]=r;
+      if(ADDTABLEITEM(table, r, FALSE)==READY){
+       resolveDependencies(r);
+      }      
+    }
+  }
+}
+
+int RESOLVEBUF(MemoryQueue * q, SESEcommon *seseCommon){
+  int localCount=0;
+  int i;
+  // check if every waiting entry is resolved
+  // if not, defer every items for hashtable until it is resolved.
+  int unresolved=FALSE;
+  for(i=0; i<q->bufcount;i++){
+     REntry *r=q->buf[i];
+     if(*(r->pointer)==0){
+       unresolved=TRUE;
+     }
+  }
+  if(unresolved==TRUE){
+    for(i=0; i<q->bufcount;i++){
+      REntry *r=q->buf[i];
+      r->queue=q;
+      r->isBufMode=TRUE;
+      if(ADDRENTRY(q,r)==NOTREADY){
+       localCount++;
+      }
+    }
+    return localCount;
+  }
+
+  // first phase: only consider write rentry
+  for(i=0; i<q->bufcount;i++){
+    REntry *r=q->buf[i];
+    if(r->type==WRITE){
+      int key=generateKey( OBJPTRPTR_2_OBJOID( r->pointer ) );
+      if(q->binbuf[key]==NULL){
+       // for multiple writes, add only the first write that hashes to the same bin
+       q->binbuf[key]=r;  
+      }else{
+       q->buf[i]=NULL;
+      }
+    }
+  }
+  // second phase: enqueue read items if it is eligible
+  for(i=0; i<q->bufcount;i++){
+    REntry *r=q->buf[i];    
+    if(r!=NULL && r->type==READ){
+      int key=generateKey( OBJPTRPTR_2_OBJOID( r->pointer ) );
+      if(q->binbuf[key]==NULL){
+       // read item that hashes to the bin which doen't contain any write
+       seseCommon->rentryArray[seseCommon->rentryIdx++]=r;
+       if(ADDRENTRY(q,r)==NOTREADY){
+         localCount++;
+       }
+      }
+      q->buf[i]=NULL;
+    }
+  }
+  
+  // then, add only one of write items that hashes to the same bin
+  for(i=0; i<q->bufcount;i++){
+    REntry *r=q->buf[i];   
+    if(r!=NULL){
+      seseCommon->rentryArray[seseCommon->rentryIdx++]=r;
+      if(ADDRENTRY(q,r)==NOTREADY){
+       localCount++;
+      }
+    }
+  }
+  return localCount;
+}
+
+
 resolvePointer(REntry* rentry){  
  
   Hashtable* table=rentry->hashtable;
+  MemoryQueue* queue;
   if(table==NULL){
     //resolved already before related rentry is enqueued to the waiting queue
     return;
@@ -782,6 +927,7 @@ resolvePointer(REntry* rentry){
   if(val!=NULL && getHead(val)->objectptr==rentry){
     // handling pointer is the first item of the queue
     // start to resolve until it reaches unresolved pointer or end of queue
+    INTPTR currentSESE=0;
     do{
       struct QueueItem* head=getHead(val);
       if(head!=NULL){
@@ -792,8 +938,34 @@ resolvePointer(REntry* rentry){
          break;
        }
        removeItem(val,head);
-       if(ADDTABLEITEM(table, rentry, FALSE)==READY){
-         resolveDependencies(rentry);
+
+       //now, address is resolved
+       
+       //check if rentry is buffer mode
+       if(rentry->isBufMode==TRUE){
+         if(currentSESE==0){
+           queue=rentry->queue;
+           INITIALIZEBUF(queue);
+           currentSESE=(INTPTR)rentry;
+           ADDRENTRYTOBUF(queue,rentry);
+         } else if(currentSESE==(INTPTR)rentry){
+           ADDRENTRYTOBUF(queue,rentry);
+         } else if(currentSESE!=(INTPTR)rentry){
+           RESOLVEBUFFORHASHTABLE(queue,table,(SESEcommon*)rentry->seseRec);
+           currentSESE=(INTPTR)rentry;
+           INITIALIZEBUF(queue);
+           ADDRENTRYTOBUF(rentry->queue,rentry);
+         }
+       }else{
+         if(currentSESE!=0){ 
+           //previous SESE has buf mode, need to invoke resolve buffer
+           RESOLVEBUFFORHASHTABLE(queue,table,(SESEcommon*)rentry->seseRec);
+           currentSESE=0;
+         }
+         //normal mode
+         if(ADDTABLEITEM(table, rentry, FALSE)==READY){
+           resolveDependencies(rentry);
+         }
        }
       }else{
        table->unresolvedQueue=NULL; // set hashtable as normal-mode.
@@ -805,3 +977,62 @@ resolvePointer(REntry* rentry){
     table->unresolvedQueue=val;//released lock;
   }  
 }
+
+void rehashMemoryQueue(SESEcommon* seseParent){    
+#if 0
+  // update memory queue
+  int i,binidx;
+  for(i=0; i<seseParent->numMemoryQueue; i++){
+    MemoryQueue *memoryQueue=seseParent->memoryQueueArray[i];
+    MemoryQueueItem *memoryItem=memoryQueue->head;
+    MemoryQueueItem *prevItem=NULL;
+    while(memoryItem!=NULL){
+      if(memoryItem->type==HASHTABLE){
+       //do re-hash!
+       Hashtable* ht=(Hashtable*)memoryItem;
+       Hashtable* newht=createHashtable();     
+       int binidx;
+       for(binidx=0; binidx<NUMBINS; binidx++){
+         BinElement *bin=ht->array[binidx];
+         BinItem *binItem=bin->head;
+         //traverse over the list of each bin
+         while(binItem!=NULL){
+           if(binItem->type==READBIN){
+             ReadBinItem* readBinItem=(ReadBinItem*)binItem;
+             int ridx;
+             for(ridx=0; ridx<readBinItem->index; ridx++){
+               REntry *rentry=readBinItem->array[ridx];
+               int newkey=generateKey((unsigned int)(unsigned INTPTR)*(rentry->pointer));      
+               int status=rentry->binitem->status;           
+               ADDTABLEITEM(newht,rentry,TRUE);
+               rentry->binitem->status=status; // update bin status as before rehash
+             }
+           }else{//write bin
+             REntry *rentry=((WriteBinItem*)binItem)->val;
+             int newkey=generateKey((unsigned int)(unsigned INTPTR)*(rentry->pointer));        
+             int status=rentry->binitem->status;             
+             ADDTABLEITEM(newht,rentry,TRUE);                
+             int newstatus=rentry->binitem->status;
+             //printf("[%d]old status=%d new status=%d\n",i,status,newstatus);
+             rentry->binitem->status=status; // update bin status as before rehash
+           }
+           binItem=binItem->next;
+         }
+       }
+       newht->item.status=ht->item.status; // update hashtable status
+       if(prevItem!=NULL){
+         prevItem->next=(MemoryQueueItem*)newht;
+       }else{
+         if(memoryQueue->head==memoryQueue->tail){
+           memoryQueue->tail=(MemoryQueueItem*)newht;
+         }
+         memoryQueue->head=(MemoryQueueItem*)newht;
+       }
+       newht->item.next=ht->item.next; 
+      }
+      prevItem=memoryItem;
+      memoryItem=memoryItem->next;
+    }
+  }
+#endif
+}