Robust/src/Runtime/mlp_runtime.c

   1 #include "runtime.h"
   2
   3 #include <stdlib.h>
   4 #include <stdio.h>
   5 #include <assert.h>
   6
   7 #include "mem.h"
   8 #include "mlp_runtime.h"
   9 #include "workschedule.h"
  10 #include "methodheaders.h"
  11
  12
  13
  14 /*
  15 __thread struct Queue* seseCallStack;
  16 __thread pthread_once_t mlpOnceObj = PTHREAD_ONCE_INIT;
  17 void mlpInitOncePerThread() {
  18   seseCallStack = createQueue();
  19 }
  20 */
  21 __thread SESEcommon_p seseCaller;
  22
  23 __thread void * baseptr=NULL;
  24
  25 __thread int spaceleft=0;
  26
  27 #define MBLOCK 65536
  28 #define MLOW 1000
  29
  30 void * MLPMALLOC(int size) {
  31   //  printf("%d\n",size);
  32   if (spaceleft<MLOW) {
  33     baseptr=malloc(MBLOCK);
  34     spaceleft=MBLOCK;
  35   }
  36   if (size>spaceleft) {
  37     return malloc(size);
  38   } else {
  39     //rount size up
  40     size=(size+7)&~7;
  41     char *charbase=(char *)baseptr;
  42     baseptr=(void *)(charbase+size);
  43     spaceleft-=size;
  44     return charbase;
  45   }
  46 }
  47
  48 void* mlpAllocSESErecord( int size ) {
  49   void* newrec = MLPMALLOC( size );
  50   if( newrec == 0 ) {
  51     printf( "mlpAllocSESErecord did not obtain memory!\n" );
  52     exit( -1 );
  53   }
  54   return newrec;
  55 }
  56
  57
  58 void mlpFreeSESErecord( void* seseRecord ) {
  59   //  MLPFREE( seseRecord );
  60 }
  61
  62 MemoryQueue** mlpCreateMemoryQueueArray(int numMemoryQueue){
  63   int i;
  64   MemoryQueue** newMemoryQueue=(MemoryQueue**)MLPMALLOC( sizeof( MemoryQueue* ) * numMemoryQueue );
  65   for(i=0; i<numMemoryQueue; i++){
  66     newMemoryQueue[i]=createMemoryQueue();
  67   }
  68   return newMemoryQueue;
  69 }
  70
  71 REntry* mlpCreateREntryArray(){
  72   REntry* newREntryArray=(REntry*)MLPMALLOC(sizeof(REntry)*NUMRENTRY);
  73   return newREntryArray;
  74 }
  75
  76 REntry* mlpCreateFineREntry(int type, void* seseToIssue, void* dynID){
  77   struct ___Object___ * obj=(struct ___Object___*)*((unsigned INTPTR *)dynID);
  78   REntry* newREntry=(REntry*)MLPMALLOC(sizeof(REntry));
  79   newREntry->type=type;
  80   newREntry->seseRec=seseToIssue;
  81   newREntry->pointer=dynID;
  82   if (obj!=NULL)
  83     newREntry->oid=obj->oid;
  84
  85   return newREntry;
  86 }
  87
  88 REntry* mlpCreateREntry(int type, void* seseToIssue){
  89   REntry* newREntry=(REntry*)MLPMALLOC(sizeof(REntry));
  90   newREntry->type=type;
  91   newREntry->seseRec=seseToIssue;
  92   return newREntry;
  93 }
  94
  95 int isParent(REntry *r) {
  96   if (r->type==PARENTREAD || r->type==PARENTWRITE || r->type==PARENTCOARSE) {
  97     return TRUE;
  98   } else {
  99     return FALSE;
 100   }
 101 }
 102
 103 int isParentCoarse(REntry *r){
 104   if (r->type==PARENTCOARSE){
 105     return TRUE;
 106   }else{
 107     return FALSE;
 108   }
 109 }
 110
 111 int isFineRead(REntry *r) {
 112   if (r->type==READ || r->type==PARENTREAD) {
 113     return TRUE;
 114   } else {
 115     return FALSE;
 116   }
 117 }
 118
 119 int isFineWrite(REntry *r) {
 120   if (r->type==WRITE || r->type==PARENTWRITE) {
 121     return TRUE;
 122   } else {
 123     return FALSE;
 124   }
 125 }
 126
 127 int isCoarse(REntry *r){
 128   if(r->type==COARSE || r->type==PARENTCOARSE){
 129     return TRUE;
 130   } else {
 131     return FALSE;
 132   }
 133 }
 134
 135 int isSCC(REntry *r){
 136   if(r->type==SCCITEM){
 137     return TRUE;
 138   } else {
 139     return FALSE;
 140   }
 141 }
 142
 143 int isSingleItem(MemoryQueueItem *qItem){
 144   if(qItem->type==SINGLEITEM){
 145     return TRUE;
 146   } else {
 147     return FALSE;
 148   }
 149 }
 150
 151 int isHashtable(MemoryQueueItem *qItem){
 152   if(qItem->type==HASHTABLE){
 153     return TRUE;
 154   } else {
 155     return FALSE;
 156   }
 157 }
 158
 159 int isVector(MemoryQueueItem *qItem){
 160   if(qItem->type==VECTOR){
 161     return TRUE;
 162   } else {
 163     return FALSE;
 164   }
 165 }
 166
 167 int isReadBinItem(BinItem* b){
 168   if(b->type==READBIN){
 169     return TRUE;
 170   }else{
 171     return FALSE;
 172   }
 173 }
 174
 175 int isWriteBinItem(BinItem* b){
 176   if(b->type==WRITEBIN){
 177     return TRUE;
 178   }else{
 179     return FALSE;
 180   }
 181 }
 182
 183 int generateKey(unsigned int data){
 184   return (data&H_MASK)>> 4;
 185 }
 186
 187 Hashtable* createHashtable(){
 188   int i=0;
 189   Hashtable* newTable=(Hashtable*)MLPMALLOC(sizeof(Hashtable));
 190   newTable->item.type=HASHTABLE;
 191   for(i=0;i<NUMBINS;i++){
 192     newTable->array[i]=(BinElement*)MLPMALLOC(sizeof(BinElement));
 193     newTable->array[i]->head=NULL;
 194     newTable->array[i]->tail=NULL;
 195   }
 196   newTable->unresolvedQueue=NULL;
 197   return newTable;
 198 }
 199
 200 WriteBinItem* createWriteBinItem(){
 201   WriteBinItem* binitem=(WriteBinItem*)MLPMALLOC(sizeof(WriteBinItem));
 202   binitem->item.type=WRITEBIN;
 203   return binitem;
 204 }
 205
 206 ReadBinItem* createReadBinItem(){
 207   ReadBinItem* binitem=(ReadBinItem*)MLPMALLOC(sizeof(ReadBinItem));
 208   binitem->index=0;
 209   binitem->item.type=READBIN;
 210   return binitem;
 211 }
 212
 213 Vector* createVector(){
 214   Vector* vector=(Vector*)MLPMALLOC(sizeof(Vector));
 215   vector->index=0;
 216   vector->item.type=VECTOR;
 217   return vector;
 218 }
 219
 220 SCC* createSCC(){
 221   SCC* scc=(SCC*)MLPMALLOC(sizeof(SCC));
 222   scc->item.type=SINGLEITEM;
 223   return scc;
 224 }
 225
 226 MemoryQueue* createMemoryQueue(){
 227   MemoryQueue* queue = (MemoryQueue*)MLPMALLOC(sizeof(MemoryQueue));
 228   MemoryQueueItem* dummy=(MemoryQueueItem*)MLPMALLOC(sizeof(MemoryQueueItem));
 229   dummy->type=3; // dummy type
 230   dummy->total=0;
 231   dummy->status=READY;
 232   queue->head = dummy;
 233   queue->tail = dummy;
 234   return queue;
 235 }
 236
 237 int ADDRENTRY(MemoryQueue * q, REntry * r) {
 238   if (isFineRead(r) || isFineWrite(r)) {
 239     return ADDTABLE(q, r);
 240   } else if (isCoarse(r)) {
 241     return ADDVECTOR(q, r);
 242   } else if (isSCC(r)) {
 243     return ADDSCC(q, r);
 244   }
 245 }
 246
 247 int ADDTABLE(MemoryQueue *q, REntry *r) {
 248   if(!isHashtable(q->tail)) {
 249     //Fast Case
 250     MemoryQueueItem* tail=q->tail;
 251     if (isParent(r) && tail->total==0 && q->tail==q->head) {
 252       return READY;
 253     }
 254
 255     //Add table
 256     Hashtable* h=createHashtable();
 257     tail->next=(MemoryQueueItem*)h;
 258     //************NEED memory barrier here to ensure compiler does not cache Q.tail.status********
 259     if (BARRIER() && tail->status==READY && tail->total==0 && q->tail==q->head) {
 260       //previous Q item is finished
 261       h->item.status=READY;
 262     }
 263     q->tail=(MemoryQueueItem*)h;
 264     // handle the the queue item case
 265     if(q->head->type==3){
 266       q->head=(MemoryQueueItem*)h;
 267     }
 268   }
 269
 270   //at this point, have table
 271   Hashtable* table=(Hashtable*)q->tail;
 272   r->hashtable=table; // set rentry's hashtable
 273   if((*(r->pointer)==0 || (*(r->pointer)!=0 && BARRIER() && table->unresolvedQueue!=NULL))){
 274     struct Queue* val;
 275     // grab lock on the queue
 276     do {
 277       val=(struct Queue*)0x1;
 278       val=(struct Queue*)LOCKXCHG((unsigned INTPTR*)&(table->unresolvedQueue), (unsigned INTPTR)val);
 279     } while(val==(struct Queue*)0x1);
 280     if(val==NULL){
 281       //queue is null, first case
 282       if(*(r->pointer)!=0){
 283         // check whether pointer is already resolved, or not.
 284         table->unresolvedQueue=NULL; //released lock;
 285         return ADDTABLEITEM(table,r,TRUE);
 286       }
 287       struct Queue* queue=createQueue();
 288       addNewItemBack(queue,r);
 289       atomic_inc(&table->item.total);
 290       table->unresolvedQueue=queue; // expose new queue
 291     }else{
 292       // add unresolved rentry at the end of the queue.
 293       addNewItemBack(val,r);
 294       atomic_inc(&table->item.total);
 295       table->unresolvedQueue=val; // released lock
 296     }
 297     return NOTREADY;
 298   }
 299   BinItem * val;
 300   //int key=generateKey((unsigned int)(unsigned INTPTR)*(r->pointer));
 301   int key=generateKey(r->oid);
 302   do {
 303     val=(BinItem*)0x1;
 304     BinElement* bin=table->array[key];
 305     val=(BinItem*)LOCKXCHG((unsigned INTPTR*)&(bin->head), (unsigned INTPTR)val);//note...talk to me about optimizations here.
 306   } while(val==(BinItem*)0x1);
 307   //at this point have locked bin
 308   if (val==NULL) {
 309     return EMPTYBINCASE(table, table->array[key], r, TRUE);
 310   } else {
 311     if (isFineWrite(r)) {
 312       return WRITEBINCASE(table, r, val, key, TRUE);
 313     } else if (isFineRead(r)) {
 314       return READBINCASE(table, r, val, key, TRUE);
 315     }
 316   }
 317 }
 318
 319 int ADDTABLEITEM(Hashtable* table, REntry* r, int inc){
 320
 321   BinItem * val;
 322   //  int key=generateKey((unsigned int)(unsigned INTPTR)*(r->pointer));
 323   int key=generateKey(r->oid);
 324   do {
 325     val=(BinItem*)0x1;
 326     BinElement* bin=table->array[key];
 327     val=(BinItem*)LOCKXCHG((unsigned INTPTR*)&(bin->head), (unsigned INTPTR)val);
 328   } while(val==(BinItem*)0x1);
 329   //at this point have locked bin
 330   if (val==NULL) {
 331     return EMPTYBINCASE(table, table->array[key], r, inc);
 332   } else {
 333     if (isFineWrite(r)) {
 334       return WRITEBINCASE(table, r, val, key, inc);
 335     } else if (isFineRead(r)) {
 336       return READBINCASE(table, r, val, key, inc);
 337     }
 338   }
 339 }
 340
 341 int EMPTYBINCASE(Hashtable *T, BinElement* be, REntry *r, int inc) {
 342   int retval;
 343   BinItem* b;
 344   if (isFineWrite(r)) {
 345     b=(BinItem*)createWriteBinItem();
 346     ((WriteBinItem*)b)->val=r;//<-only different statement
 347   } else if (isFineRead(r)) {
 348     b=(BinItem*)createReadBinItem();
 349     ReadBinItem* readbin=(ReadBinItem*)b;
 350     readbin->array[readbin->index++]=r;
 351   }
 352   b->total=1;
 353
 354   if (T->item.status==READY) {
 355     //current entry is ready
 356     b->status=READY;
 357     retval=READY;
 358     if (isParent(r)) {
 359       be->head=NULL; // released lock
 360       return retval;
 361     }
 362   } else {
 363     b->status=NOTREADY;
 364     retval=NOTREADY;
 365   }
 366
 367   if(inc){
 368     atomic_inc(&T->item.total);
 369   }
 370   r->hashtable=T;
 371   r->binitem=b;
 372   be->tail=b;
 373   be->head=b;//released lock
 374   return retval;
 375 }
 376
 377 int WRITEBINCASE(Hashtable *T, REntry *r, BinItem *val, int key, int inc) {
 378   //chain of bins exists => tail is valid
 379   //if there is something in front of us, then we are not ready
 380
 381   int retval=NOTREADY;
 382   BinElement* be=T->array[key];
 383
 384   BinItem *bintail=be->tail;
 385
 386   WriteBinItem *b=createWriteBinItem();
 387   b->val=r;
 388   b->item.total=1;
 389
 390   // note: If current table clears all dependencies, then write bin is ready
 391
 392
 393   if(inc){
 394     atomic_inc(&T->item.total);
 395   }
 396
 397   r->hashtable=T;
 398   r->binitem=(BinItem*)b;
 399
 400   be->tail->next=(BinItem*)b;
 401   //need to check if we can go...
 402   BARRIER();
 403   if (T->item.status==READY) {
 404     for(;val!=NULL;val=val->next) {
 405       if (val==((BinItem *)b)) {
 406         //ready to retire
 407         retval=READY;
 408         if (isParent(r)) {
 409           b->item.status=retval;//unsure if really needed at this point..
 410           be->head=NULL; // released lock
 411           return retval;
 412         }
 413         break;
 414       } else if (val->total!=0) {
 415         break;
 416       }
 417     }
 418   }
 419
 420   b->item.status=retval;
 421   be->tail=(BinItem*)b;
 422   be->head=val;
 423   return retval;
 424 }
 425
 426 READBINCASE(Hashtable *T, REntry *r, BinItem *val, int key, int inc) {
 427   BinItem * bintail=T->array[key]->tail;
 428   if (isReadBinItem(bintail)) {
 429     return TAILREADCASE(T, r, val, bintail, key, inc);
 430   } else if (!isReadBinItem(bintail)) {
 431     TAILWRITECASE(T, r, val, bintail, key, inc);
 432     return NOTREADY;
 433   }
 434 }
 435
 436 int TAILREADCASE(Hashtable *T, REntry *r, BinItem *val, BinItem *bintail, int key, int inc) {
 437   ReadBinItem * readbintail=(ReadBinItem*)T->array[key]->tail;
 438   int status, retval;
 439   if (readbintail->item.status==READY) {
 440     status=READY;
 441     retval=READY;
 442     if (isParent(r)) {
 443       T->array[key]->head=val;//released lock
 444       return READY;
 445     }
 446   } else {
 447     status=NOTREADY;
 448     retval=NOTREADY;
 449   }
 450
 451   if (readbintail->index==NUMREAD) { // create new read group
 452     ReadBinItem* rb=createReadBinItem();
 453     rb->array[rb->index++]=r;
 454     rb->item.total=1;//safe only because item could not have started
 455     rb->item.status=status;
 456     T->array[key]->tail->next=(BinItem*)rb;
 457     T->array[key]->tail=(BinItem*)rb;
 458     r->binitem=(BinItem*)rb;
 459   } else { // group into old tail
 460     readbintail->array[readbintail->index++]=r;
 461     atomic_inc(&readbintail->item.total);
 462     r->binitem=(BinItem*)readbintail;
 463     //printf("grouping with %d\n",readbintail->index);
 464   }
 465   if(inc){
 466     atomic_inc(&T->item.total);
 467   }
 468   r->hashtable=T;
 469   T->array[key]->head=val;//released lock
 470   return retval;
 471 }
 472
 473 TAILWRITECASE(Hashtable *T, REntry *r, BinItem *val, BinItem *bintail, int key, int inc) {
 474   //  WriteBinItem* wb=createWriteBinItem();
 475   //wb->val=r;
 476   //wb->item.total=1;//safe because item could not have started
 477   //wb->item.status=NOTREADY;
 478   ReadBinItem* rb=createReadBinItem();
 479   rb->array[rb->index++]=r;
 480   rb->item.total=1;//safe because item could not have started
 481   rb->item.status=NOTREADY;
 482   if(inc){
 483     atomic_inc(&T->item.total);
 484   }
 485   r->hashtable=T;
 486   r->binitem=(BinItem*)rb;
 487   T->array[key]->tail->next=(BinItem*)rb;
 488   T->array[key]->tail=(BinItem*)rb;
 489   T->array[key]->head=val;//released lock
 490 }
 491
 492 ADDVECTOR(MemoryQueue *Q, REntry *r) {
 493   if(!isVector(Q->tail)) {
 494     //Fast Case
 495     if (isParentCoarse(r) && Q->tail->total==0 && Q->tail==Q->head) {
 496       return READY;
 497     }
 498
 499     //added vector
 500     Vector* V=createVector();
 501     Q->tail->next=(MemoryQueueItem*)V;
 502     //************NEED memory barrier here to ensure compiler does not cache Q.tail.status******
 503     if (BARRIER() && Q->tail->status==READY&&Q->tail->total==0) {
 504       //previous Q item is finished
 505       V->item.status=READY;
 506     }
 507     Q->tail=(MemoryQueueItem*)V;
 508     // handle the the queue item case
 509     if(Q->head->type==3){
 510       Q->head=(MemoryQueueItem*)V;
 511     }
 512   }
 513   //at this point, have vector
 514   Vector* V=(Vector*)Q->tail;
 515   if (V->index==NUMITEMS) {
 516     //vector is full
 517     //added vector
 518     V=createVector();
 519     V->item.status=NOTREADY;
 520     Q->tail->next=(MemoryQueueItem*)V;
 521     //***NEED memory barrier here to ensure compiler does not cache Q.tail.status******
 522     if (BARRIER() && Q->tail->status==READY) {
 523       V->item.status=READY;
 524     }
 525     Q->tail=(MemoryQueueItem*)V;
 526   }
 527
 528   atomic_inc(&V->item.total);
 529   //expose entry
 530   int index=V->index;
 531   V->array[index]=r;
 532   //*****NEED memory barrier here to ensure compiler does not reorder writes to V.array and V.index
 533   BARRIER();
 534   V->index++;
 535   //*****NEED memory barrier here to ensure compiler does not cache V.status*********
 536   r->vector=V;
 537   if (BARRIER() && V->item.status==READY) {
 538     void* flag=NULL;
 539     flag=(void*)LOCKXCHG((unsigned INTPTR*)&(V->array[index]), (unsigned INTPTR)flag);
 540     if (flag!=NULL) {
 541       if (isParentCoarse(r)) { //parent's retire immediately
 542         atomic_dec(&V->item.total);
 543       }
 544       return READY;
 545     } else {
 546       return NOTREADY;//<- means that some other dispatcher got this one...so need to do accounting correctly
 547     }
 548   } else {
 549     return NOTREADY;
 550   }
 551 }
 552
 553
 554 //SCC's don't come in parent variety
 555 ADDSCC(MemoryQueue *Q, REntry *r) {
 556   //added SCC
 557   SCC* S=createSCC();
 558   S->item.total=1;
 559   S->val=r;
 560   r->scc=S;
 561   Q->tail->next=(MemoryQueueItem*)S;
 562   //*** NEED BARRIER HERE
 563   if (BARRIER() && Q->tail->status==READY && Q->tail->total==0 && Q->tail==Q->head) {
 564     //previous Q item is finished
 565     S->item.status=READY;
 566     Q->tail=(MemoryQueueItem*)S;
 567     // handle the the queue item case
 568     if(Q->head->type==3){
 569       Q->head=(MemoryQueueItem*)S;
 570     }
 571     void* flag=NULL;
 572     flag=(void*)LOCKXCHG((unsigned INTPTR*)&(S->val), (unsigned INTPTR)flag);
 573     if (flag!=NULL) {
 574       return READY;
 575     } else {
 576       return NOTREADY;//<- means that some other dispatcher got this one...so need to do accounting correctly
 577     }
 578   } else {
 579     Q->tail=(MemoryQueueItem*)S;
 580     return NOTREADY;
 581   }
 582 }
 583
 584
 585 void RETIRERENTRY(MemoryQueue* Q, REntry * r) {
 586   if (isFineWrite(r)||isFineRead(r)) {
 587     RETIREHASHTABLE(Q, r);
 588   } else if (isCoarse(r)) {
 589     RETIREVECTOR(Q, r);
 590   } else if (isSCC(r)) {
 591     RETIRESCC(Q, r);
 592   }
 593 }
 594
 595 RETIRESCC(MemoryQueue *Q, REntry *r) {
 596   SCC* s=r->scc;
 597   s->item.total=0;//don't need atomicdec
 598   RESOLVECHAIN(Q);
 599 }
 600
 601
 602 RETIREHASHTABLE(MemoryQueue *q, REntry *r) {
 603   Hashtable *T=r->hashtable;
 604   BinItem *b=r->binitem;
 605   RETIREBIN(T,r,b);
 606   atomic_dec(&T->item.total);
 607   if (T->item.next!=NULL && T->item.total==0) {
 608     RESOLVECHAIN(q);
 609   }
 610 }
 611
 612 RETIREBIN(Hashtable *T, REntry *r, BinItem *b) {
 613   //  int key=generateKey((unsigned int)(unsigned INTPTR)*(r->pointer));
 614   int key=generateKey(r->oid);
 615   if(isFineRead(r)) {
 616     atomic_dec(&b->total);
 617   }
 618   if (isFineWrite(r) || (isFineRead(r) && b->next!=NULL && b->total==0)) {
 619       // CHECK FIRST IF next is nonnull to guarantee that b.total cannot change
 620     BinItem * val;
 621     do {
 622       val=(BinItem*)0x1;
 623       val=(BinItem*)LOCKXCHG((unsigned INTPTR*)&(T->array[key]->head), (unsigned INTPTR)val);
 624     } while(val==(BinItem*)0x1);
 625     // at this point have locked bin
 626     BinItem *ptr=val;
 627     int haveread=FALSE;
 628     int i;
 629     while (ptr!=NULL) {
 630        if (isReadBinItem(ptr)) {
 631         ReadBinItem* rptr=(ReadBinItem*)ptr;
 632         if (rptr->item.status==NOTREADY) {
 633           for (i=0;i<rptr->index;i++) {
 634             resolveDependencies(rptr->array[i]);
 635             if (isParent(rptr->array[i])) {
 636               //parents go immediately
 637               atomic_dec(&rptr->item.total);
 638               atomic_dec(&T->item.total);
 639             }
 640           }
 641         }
 642         rptr->item.status=READY;
 643         if (rptr->item.next==NULL) {
 644           break;
 645         }
 646         if (rptr->item.total!=0) {
 647           haveread=TRUE;
 648         } else if ((BinItem*)rptr==val) {
 649           val=val->next;
 650         }
 651       } else if(isWriteBinItem(ptr)) {
 652         if (haveread)
 653           break;
 654         if(ptr->status==NOTREADY){
 655           resolveDependencies(((WriteBinItem*)ptr)->val);
 656           ptr->status=READY;
 657           if(isParent(((WriteBinItem*)ptr)->val)){
 658             atomic_dec(&T->item.total);
 659             val=val->next;
 660           }else
 661             break;
 662         }else{ // write bin is already resolved
 663           val=val->next;
 664         }
 665         /*
 666         if(ptr->status==NOTREADY) {
 667           resolveDependencies(((WriteBinItem*)ptr)->val);
 668         }
 669           ptr->status=READY;
 670           if (isParent(((WriteBinItem*)ptr)->val)) {
 671             atomic_dec(&T->item.total);
 672             //val=val->next;
 673             val=ptr->next;
 674           } else
 675             break;
 676         }
 677         */
 678       }
 679       ptr=ptr->next;
 680     }
 681     T->array[key]->head=val; // release lock
 682   }
 683 }
 684
 685
 686 RETIREVECTOR(MemoryQueue *Q, REntry *r) {
 687   Vector* V=r->vector;
 688   atomic_dec(&V->item.total);
 689   if (V->item.next!=NULL && V->item.total==0) { //NOTE: ORDERING CRUCIAL HERE
 690     RESOLVECHAIN(Q);
 691   }
 692 }
 693
 694 RESOLVECHAIN(MemoryQueue *Q) {
 695   while(TRUE) {
 696     MemoryQueueItem* head=Q->head;
 697     if (head->next==NULL||head->total!=0) {
 698       //item is not finished
 699       if (head->status!=READY) {
 700         //need to update status
 701         head->status=READY;
 702         if (isHashtable(head)) {
 703           RESOLVEHASHTABLE(Q, head);
 704         } else if (isVector(head)) {
 705           RESOLVEVECTOR(Q, head);
 706         } else if (isSingleItem(head)) {
 707           RESOLVESCC(head);
 708         }
 709         if (head->next==NULL)
 710           break;
 711         if (head->total!=0)
 712           break;
 713       } else
 714         break;
 715     }
 716     MemoryQueueItem* nextitem=head->next;
 717     CAS((unsigned INTPTR*)&(Q->head), (unsigned INTPTR)head, (unsigned INTPTR)nextitem);
 718     //oldvalue not needed...  if we fail we just repeat
 719   }
 720 }
 721
 722
 723 RESOLVEHASHTABLE(MemoryQueue *Q, Hashtable *T) {
 724   int binidx;
 725   for (binidx=0;binidx<NUMBINS;binidx++) {
 726     BinElement* bin=T->array[binidx];
 727     BinItem* val;
 728     do {
 729       val=(BinItem*)1;
 730       val=(BinItem*)LOCKXCHG((unsigned INTPTR*)&(bin->head), (unsigned INTPTR)val);
 731     } while (val==(BinItem*)1);
 732     //at this point have locked bin
 733     int haveread=FALSE;
 734     BinItem* ptr=val;
 735     if(ptr!=NULL&&ptr->status==NOTREADY) {
 736       do {
 737         if (isWriteBinItem(ptr)) {
 738           if (haveread)
 739             break;
 740           resolveDependencies(((WriteBinItem*)ptr)->val);
 741           ptr->status=READY;
 742           if (isParent(((WriteBinItem*)ptr)->val)) {
 743             atomic_dec(&T->item.total);
 744             val=val->next;
 745           } else
 746             break;
 747         } else if (isReadBinItem(ptr)) {
 748           int i;
 749           ReadBinItem* rptr=(ReadBinItem*)ptr;
 750           for(i=0;i<rptr->index;i++) {
 751             resolveDependencies(rptr->array[i]);
 752             if (isParent(rptr->array[i])) {
 753               atomic_dec(&rptr->item.total);
 754               atomic_dec(&T->item.total);
 755             }
 756           }
 757           if (rptr->item.next==NULL||rptr->item.total!=0) {
 758             haveread=TRUE;
 759           } else if((BinItem*)rptr==val) {
 760             val=val->next;
 761           }
 762           rptr->item.status=READY;
 763         }
 764         ptr=ptr->next;
 765       }while(ptr!=NULL);
 766     }
 767     bin->head=val; // released lock;
 768   }
 769 }
 770
 771 RESOLVEVECTOR(MemoryQueue *q, Vector *V) {
 772   int i;
 773   Vector* tmp=V;
 774   //handle ready cases
 775   while(TRUE) {
 776     //enqueue everything
 777     for (i=0;i<NUMITEMS;i++) {
 778       REntry* val=NULL;
 779       val=(REntry*)LOCKXCHG((unsigned INTPTR*)&(tmp->array[i]), (unsigned INTPTR)val);
 780       if (val!=NULL) {
 781         resolveDependencies(val);
 782         if (isParent(val)) {
 783           atomic_dec(&tmp->item.total);
 784         }
 785       }
 786     }
 787     if (tmp->item.next!=NULL&&isVector(tmp->item.next)) {
 788       tmp=(Vector*)tmp->item.next;
 789     } else {
 790       break;
 791     }
 792   }
 793 }
 794
 795 RESOLVESCC(SCC *S) {
 796   //precondition: SCC's state is READY
 797   void* flag=NULL;
 798   flag=(void*)LOCKXCHG((unsigned INTPTR*)&(S->val), (unsigned INTPTR)flag);
 799   if (flag!=NULL) {
 800     resolveDependencies(flag);
 801   }
 802 }
 803
 804
 805 resolveDependencies(REntry* rentry){
 806   SESEcommon* seseCommon=(SESEcommon*)rentry->seseRec;
 807   if(rentry->type==READ || rentry->type==WRITE || rentry->type==COARSE || rentry->type==SCCITEM){
 808     if( atomic_sub_and_test(1, &(seseCommon->unresolvedDependencies)) ){
 809       workScheduleSubmit(seseCommon);
 810     }
 811   }else if(rentry->type==PARENTREAD || rentry->type==PARENTWRITE ||rentry->type==PARENTCOARSE){
 812      psem_give(&(rentry->parentStallSem));
 813   }
 814 }
 815
 816 void INITIALIZEBUF(MemoryQueue * q){
 817   int i=0;
 818   for(i=0; i<NUMBINS; i++){
 819     q->binbuf[i]=NULL;
 820   }
 821   q->bufcount=0;
 822 }
 823
 824 void ADDRENTRYTOBUF(MemoryQueue * q, REntry * r){
 825   q->buf[q->bufcount]=r;
 826   q->bufcount++;
 827 }
 828
 829 int RESOLVEBUFFORHASHTABLE(MemoryQueue * q, Hashtable* table, SESEcommon *seseCommon){
 830   int i;
 831  // first phase: only consider write rentry
 832   for(i=0; i<q->bufcount;i++){
 833     REntry *r=q->buf[i];
 834     if(r->type==WRITE){
 835       int key=generateKey(r->oid);
 836       if(q->binbuf[key]==NULL){
 837         // for multiple writes, add only the first write that hashes to the same bin
 838         q->binbuf[key]=r;
 839       }else{
 840         q->buf[i]=NULL;
 841       }
 842     }
 843   }
 844   // second phase: enqueue read items if it is eligible
 845   for(i=0; i<q->bufcount;i++){
 846     REntry *r=q->buf[i];
 847     if(r!=NULL && r->type==READ){
 848       int key=generateKey(r->oid);
 849       if(q->binbuf[key]==NULL){
 850         // read item that hashes to the bin which doen't contain any write
 851         seseCommon->rentryArray[seseCommon->rentryIdx++]=r;
 852         if(ADDTABLEITEM(table, r, FALSE)==READY){
 853           resolveDependencies(r);
 854         }
 855       }
 856       q->buf[i]=NULL;
 857     }
 858   }
 859
 860   // then, add only one of write items that hashes to the same bin
 861   for(i=0; i<q->bufcount;i++){
 862     REntry *r=q->buf[i];
 863     if(r!=NULL){
 864       seseCommon->rentryArray[seseCommon->rentryIdx++]=r;
 865       if(ADDTABLEITEM(table, r, FALSE)==READY){
 866         resolveDependencies(r);
 867       }
 868     }
 869   }
 870 }
 871
 872 int RESOLVEBUF(MemoryQueue * q, SESEcommon *seseCommon){
 873   int localCount=0;
 874   int i;
 875   // check if every waiting entry is resolved
 876   // if not, defer every items for hashtable until it is resolved.
 877   int unresolved=FALSE;
 878   for(i=0; i<q->bufcount;i++){
 879      REntry *r=q->buf[i];
 880      if(*(r->pointer)==0){
 881        unresolved=TRUE;
 882      }
 883   }
 884   if(unresolved==TRUE){
 885     for(i=0; i<q->bufcount;i++){
 886       REntry *r=q->buf[i];
 887       r->queue=q;
 888       r->isBufMode=TRUE;
 889       if(ADDRENTRY(q,r)==NOTREADY){
 890         localCount++;
 891       }
 892     }
 893     return localCount;
 894   }
 895
 896   // first phase: only consider write rentry
 897   for(i=0; i<q->bufcount;i++){
 898     REntry *r=q->buf[i];
 899     if(r->type==WRITE){
 900       int key=generateKey(r->oid);
 901       if(q->binbuf[key]==NULL){
 902         // for multiple writes, add only the first write that hashes to the same bin
 903         q->binbuf[key]=r;
 904       }else{
 905         q->buf[i]=NULL;
 906       }
 907     }
 908   }
 909   // second phase: enqueue read items if it is eligible
 910   for(i=0; i<q->bufcount;i++){
 911     REntry *r=q->buf[i];
 912     if(r!=NULL && r->type==READ){
 913       int key=generateKey(r->oid);
 914       if(q->binbuf[key]==NULL){
 915         // read item that hashes to the bin which doen't contain any write
 916         seseCommon->rentryArray[seseCommon->rentryIdx++]=r;
 917         if(ADDRENTRY(q,r)==NOTREADY){
 918           localCount++;
 919         }
 920       }
 921       q->buf[i]=NULL;
 922     }
 923   }
 924
 925   // then, add only one of write items that hashes to the same bin
 926   for(i=0; i<q->bufcount;i++){
 927     REntry *r=q->buf[i];
 928     if(r!=NULL){
 929       seseCommon->rentryArray[seseCommon->rentryIdx++]=r;
 930       if(ADDRENTRY(q,r)==NOTREADY){
 931         localCount++;
 932       }
 933     }
 934   }
 935   return localCount;
 936 }
 937
 938
 939 resolvePointer(REntry* rentry){
 940
 941   Hashtable* table=rentry->hashtable;
 942   MemoryQueue* queue;
 943   if(table==NULL){
 944     //resolved already before related rentry is enqueued to the waiting queue
 945     return;
 946   }
 947   struct Queue* val;
 948   do {
 949     val=(struct Queue*)0x1;
 950     val=(struct Queue*)LOCKXCHG((unsigned INTPTR*)&(table->unresolvedQueue), (unsigned INTPTR)val);
 951   } while(val==(struct Queue*)0x1);
 952   if(val!=NULL && getHead(val)->objectptr==rentry){
 953     // handling pointer is the first item of the queue
 954     // start to resolve until it reaches unresolved pointer or end of queue
 955     INTPTR currentSESE=0;
 956     do{
 957       struct QueueItem* head=getHead(val);
 958       if(head!=NULL){
 959         REntry* rentry=(REntry*)head->objectptr;
 960         if(*(rentry->pointer)==0){
 961           // encounters following unresolved pointer
 962           table->unresolvedQueue=val;//released lock
 963           break;
 964         }
 965         removeItem(val,head);
 966         //now, address is resolved. update OID field.
 967         struct ___Object___ * obj=(struct ___Object___*)((unsigned INTPTR)*rentry->pointer);
 968         rentry->oid=obj->oid;
 969
 970         //check if rentry is buffer mode
 971         if(rentry->isBufMode==TRUE){
 972           if(currentSESE==0){
 973             queue=rentry->queue;
 974             INITIALIZEBUF(queue);
 975             currentSESE=(INTPTR)rentry;
 976             ADDRENTRYTOBUF(queue,rentry);
 977           } else if(currentSESE==(INTPTR)rentry){
 978             ADDRENTRYTOBUF(queue,rentry);
 979           } else if(currentSESE!=(INTPTR)rentry){
 980             RESOLVEBUFFORHASHTABLE(queue,table,(SESEcommon*)rentry->seseRec);
 981             currentSESE=(INTPTR)rentry;
 982             INITIALIZEBUF(queue);
 983             ADDRENTRYTOBUF(rentry->queue,rentry);
 984           }
 985         }else{
 986           if(currentSESE!=0){
 987             //previous SESE has buf mode, need to invoke resolve buffer
 988             RESOLVEBUFFORHASHTABLE(queue,table,(SESEcommon*)rentry->seseRec);
 989             currentSESE=0;
 990           }
 991           //normal mode
 992           if(ADDTABLEITEM(table, rentry, FALSE)==READY){
 993             resolveDependencies(rentry);
 994           }
 995         }
 996       }else{
 997         table->unresolvedQueue=NULL; // set hashtable as normal-mode.
 998         break;
 999       }
1000     }while(TRUE);
1001   }else{
1002     // resolved rentry is not head of queue
1003     table->unresolvedQueue=val;//released lock;
1004   }
1005 }
1006
1007 void rehashMemoryQueue(SESEcommon_p seseParent){
1008 #if 0
1009   // update memory queue
1010   int i,binidx;
1011   for(i=0; i<seseParent->numMemoryQueue; i++){
1012     MemoryQueue *memoryQueue=seseParent->memoryQueueArray[i];
1013     MemoryQueueItem *memoryItem=memoryQueue->head;
1014     MemoryQueueItem *prevItem=NULL;
1015     while(memoryItem!=NULL){
1016       if(memoryItem->type==HASHTABLE){
1017         //do re-hash!
1018         Hashtable* ht=(Hashtable*)memoryItem;
1019         Hashtable* newht=createHashtable();
1020         int binidx;
1021         for(binidx=0; binidx<NUMBINS; binidx++){
1022           BinElement *bin=ht->array[binidx];
1023           BinItem *binItem=bin->head;
1024           //traverse over the list of each bin
1025           while(binItem!=NULL){
1026             if(binItem->type==READBIN){
1027               ReadBinItem* readBinItem=(ReadBinItem*)binItem;
1028               int ridx;
1029               for(ridx=0; ridx<readBinItem->index; ridx++){
1030                 REntry *rentry=readBinItem->array[ridx];
1031                 int newkey=generateKey((unsigned int)(unsigned INTPTR)*(rentry->pointer));
1032                 int status=rentry->binitem->status;
1033                 ADDTABLEITEM(newht,rentry,TRUE);
1034                 rentry->binitem->status=status; // update bin status as before rehash
1035               }
1036             }else{//write bin
1037               REntry *rentry=((WriteBinItem*)binItem)->val;
1038               int newkey=generateKey((unsigned int)(unsigned INTPTR)*(rentry->pointer));
1039               int status=rentry->binitem->status;
1040               ADDTABLEITEM(newht,rentry,TRUE);
1041               int newstatus=rentry->binitem->status;
1042               //printf("[%d]old status=%d new status=%d\n",i,status,newstatus);
1043               rentry->binitem->status=status; // update bin status as before rehash
1044             }
1045             binItem=binItem->next;
1046           }
1047         }
1048         newht->item.status=ht->item.status; // update hashtable status
1049         if(prevItem!=NULL){
1050           prevItem->next=(MemoryQueueItem*)newht;
1051         }else{
1052           if(memoryQueue->head==memoryQueue->tail){
1053             memoryQueue->tail=(MemoryQueueItem*)newht;
1054           }
1055           memoryQueue->head=(MemoryQueueItem*)newht;
1056         }
1057         newht->item.next=ht->item.next;
1058       }
1059       prevItem=memoryItem;
1060       memoryItem=memoryItem->next;
1061     }
1062   }
1063 #endif
1064 }