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

   1 /* Coordinator => Machine that initiates the transaction request call for commiting a transaction
   2  * Participant => Machines that host the objects involved in a transaction commit */
   3
   4 #include <netinet/tcp.h>
   5 #include "dstm.h"
   6 #include "altmlookup.h"
   7 #include "llookup.h"
   8 #include "threadnotify.h"
   9 #include "prefetch.h"
  10 #include <sched.h>
  11 #ifdef COMPILER
  12 #include "thread.h"
  13 #endif
  14 #include "gCollect.h"
  15 #include "readstruct.h"
  16 #include "debugmacro.h"
  17 #ifdef SANDBOX
  18 #include "sandbox.h"
  19 #endif
  20
  21 #define BACKLOG 10 //max pending connections
  22 #define RECEIVE_BUFFER_SIZE 2048
  23
  24 extern int classsize[];
  25 extern int numHostsInSystem;
  26 extern pthread_mutex_t notifymutex;
  27 extern unsigned long long clockoffset;
  28 long long startreq, endreq, diff;
  29
  30 //#define LOGTIMES
  31 #ifdef LOGTIMES
  32 extern char bigarray1[6*1024*1024];
  33 extern unsigned int bigarray2[6*1024*1024];
  34 extern unsigned int bigarray3[6*1024*1024];
  35 extern long long bigarray4[6*1024*1024];
  36 extern int bigarray5[6*1024*1024];
  37 extern int bigindex1;
  38 #define LOGTIME(x,y,z,a,b) { \
  39     int tmp=bigindex1; \
  40     bigarray1[tmp]=x; \
  41     bigarray2[tmp]=y; \
  42     bigarray3[tmp]=z; \
  43     bigarray4[tmp]=a; \
  44     bigarray5[tmp]=b; \
  45     bigindex1++; \
  46 }
  47 #else
  48 #define LOGTIME(x,y,z,a,b)
  49 #endif
  50
  51
  52 long long myrdtsc(void) {
  53   unsigned hi, lo;
  54   __asm__ __volatile__ ("rdtsc" : "=a" (lo), "=d" (hi));
  55   return ( (unsigned long long)lo)|( ((unsigned long long)hi)<<32 );
  56 }
  57
  58 objstr_t *mainobjstore;
  59 pthread_mutex_t mainobjstore_mutex;
  60 pthread_mutex_t lockObjHeader;
  61 pthread_mutexattr_t mainobjstore_mutex_attr; /* Attribute for lock to make it a recursive lock */
  62
  63 sockPoolHashTable_t *transPResponseSocketPool;
  64
  65 /* This function initializes the main objects store and creates the
  66  * global machine and location lookup table */
  67
  68 int dstmInit(void) {
  69   mainobjstore = objstrCreate(DEFAULT_OBJ_STORE_SIZE);
  70   /* Initialize attribute for mutex */
  71   pthread_mutexattr_init(&mainobjstore_mutex_attr);
  72   pthread_mutexattr_settype(&mainobjstore_mutex_attr, PTHREAD_MUTEX_RECURSIVE_NP);
  73   pthread_mutex_init(&mainobjstore_mutex, &mainobjstore_mutex_attr);
  74   pthread_mutex_init(&lockObjHeader,NULL);
  75   if (mhashCreate(MHASH_SIZE, MLOADFACTOR))
  76     return 1;             //failure
  77
  78   if (lhashCreate(HASH_SIZE, LOADFACTOR))
  79     return 1;             //failure
  80
  81   if (notifyhashCreate(N_HASH_SIZE, N_LOADFACTOR))
  82     return 1;             //failure
  83
  84   //Initialize socket pool
  85   if((transPResponseSocketPool = createSockPool(transPResponseSocketPool, DEFAULTSOCKPOOLSIZE)) == NULL) {
  86     printf("Error in creating new socket pool at  %s line %d\n", __FILE__, __LINE__);
  87     return 0;
  88   }
  89
  90   return 0;
  91 }
  92
  93
  94 int startlistening() {
  95   int listenfd;
  96   struct sockaddr_in my_addr;
  97   socklen_t addrlength = sizeof(struct sockaddr);
  98   int setsockflag=1;
  99
 100   listenfd = socket(AF_INET, SOCK_STREAM, 0);
 101   if (listenfd == -1) {
 102     perror("socket");
 103     exit(1);
 104   }
 105
 106   if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &setsockflag, sizeof (setsockflag)) < 0) {
 107     perror("socket");
 108     exit(1);
 109   }
 110 #ifdef MAC
 111   if (setsockopt(listenfd, SOL_SOCKET, SO_NOSIGPIPE, &setsockflag, sizeof (setsockflag)) < 0) {
 112     perror("socket");
 113     exit(1);
 114   }
 115 #endif
 116
 117   my_addr.sin_family = AF_INET;
 118   my_addr.sin_port = htons(LISTEN_PORT);
 119   my_addr.sin_addr.s_addr = INADDR_ANY;
 120   memset(&(my_addr.sin_zero), '\0', 8);
 121
 122   if (bind(listenfd, (struct sockaddr *)&my_addr, addrlength) == -1) {
 123     perror("bind");
 124     exit(1);
 125   }
 126
 127   if (listen(listenfd, BACKLOG) == -1) {
 128     perror("listen");
 129     exit(1);
 130   }
 131   return listenfd;
 132 }
 133
 134 /* This function starts the thread to listen on a socket
 135  * for tranaction calls */
 136 void *dstmListen(void *lfd) {
 137   int listenfd=(int)lfd;
 138   int acceptfd;
 139   struct sockaddr_in client_addr;
 140   socklen_t addrlength = sizeof(struct sockaddr);
 141   pthread_t thread_dstm_accept;
 142
 143   printf("Listening on port %d, fd = %d\n", LISTEN_PORT, listenfd);
 144   while(1) {
 145     int retval;
 146     int flag=1;
 147     acceptfd = accept(listenfd, (struct sockaddr *)&client_addr, &addrlength);
 148     setsockopt(acceptfd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(flag));
 149     do {
 150       retval=pthread_create(&thread_dstm_accept, NULL, dstmAccept, (void *)acceptfd);
 151     } while(retval!=0);
 152     pthread_detach(thread_dstm_accept);
 153   }
 154 }
 155 /* This function accepts a new connection request, decodes the control message in the connection
 156  * and accordingly calls other functions to process new requests */
 157 void *dstmAccept(void *acceptfd) {
 158   int val, retval, size, sum, sockid;
 159   unsigned int oid;
 160   char *buffer;
 161   char control,ctrl;
 162   void *srcObj;
 163   objheader_t *h;
 164   trans_commit_data_t transinfo;
 165   unsigned short objType, *versionarry, version;
 166   unsigned int *oidarry, numoid, mid, threadid;
 167   struct readstruct readbuffer;
 168   readbuffer.head=0;
 169   readbuffer.tail=0;
 170   unsigned int numread=0, nummod=0;
 171 #ifdef SANDBOX
 172   objData_t odata;
 173   char *ptr;
 174 #endif
 175
 176   /* Receive control messages from other machines */
 177   while(1) {
 178     int ret=recv_data_errorcode_buf((int)acceptfd, &readbuffer, &control, sizeof(char));
 179     if (ret==0)
 180       break;
 181     if (ret==-1) {
 182       printf("DEBUG -> RECV Error!.. retrying\n");
 183       break;
 184     }
 185     switch(control) {
 186     case READ_REQUEST:
 187       /* Read oid requested and search if available */
 188       recv_data_buf((int)acceptfd, &readbuffer, &oid, sizeof(unsigned int));
 189       while((srcObj = mhashSearch(oid)) == NULL) {
 190         int ret;
 191         if((ret = sched_yield()) != 0) {
 192           printf("%s(): error no %d in thread yield\n", __func__, errno);
 193         }
 194       }
 195       h = (objheader_t *) srcObj;
 196       /* If object is write locked, just wait */
 197       /* May want to remove at some point */
 198       while((*(volatile int *)STATUSPTR(h))<=0)
 199         sched_yield();
 200       GETSIZE(size, h);
 201       size += sizeof(objheader_t);
 202       sockid = (int) acceptfd;
 203       if (h == NULL) {
 204         ctrl = OBJECT_NOT_FOUND;
 205         send_data(sockid, &ctrl, sizeof(char));
 206       } else {
 207         // Type
 208         char msg[]={OBJECT_FOUND, 0, 0, 0, 0};
 209         *((int *)&msg[1])=size;
 210         send_data(sockid, &msg, sizeof(msg));
 211         send_data(sockid, h, size);
 212       }
 213       break;
 214
 215     case READ_MULT_REQUEST:
 216       break;
 217
 218     case MOVE_REQUEST:
 219       break;
 220
 221     case MOVE_MULT_REQUEST:
 222       break;
 223
 224     case TRANS_REQUEST:
 225       /* Read transaction request */
 226       transinfo.objlocked = NULL;
 227       transinfo.objnotfound = NULL;
 228       transinfo.modptr = NULL;
 229       transinfo.numlocked = 0;
 230       transinfo.numnotfound = 0;
 231       if((val = readClientReq(&transinfo, (int)acceptfd, &readbuffer)) != 0) {
 232         printf("Error: In readClientReq() %s, %d\n", __FILE__, __LINE__);
 233         pthread_exit(NULL);
 234       }
 235       break;
 236
 237     case TRANS_PREFETCH:
 238 #ifdef RANGEPREFETCH
 239       if((val = rangePrefetchReq((int)acceptfd, &readbuffer)) != 0) {
 240         printf("Error: In rangePrefetchReq() %s, %d\n", __FILE__, __LINE__);
 241         break;
 242       }
 243 #else
 244       LOGTIME('X',0,0,myrdtsc(),0);
 245       if((val = prefetchReq((int)acceptfd, &readbuffer)) != 0) {
 246         printf("Error: In prefetchReq() %s, %d\n", __FILE__, __LINE__);
 247         break;
 248       }
 249 #endif
 250       break;
 251
 252     case TRANS_PREFETCH_RESPONSE:
 253 #ifdef RANGEPREFETCH
 254       if((val = getRangePrefetchResponse((int)acceptfd, &readbuffer)) != 0) {
 255         printf("Error: In getRangePrefetchRespose() %s, %d\n", __FILE__, __LINE__);
 256         break;
 257       }
 258 #else
 259       if((val = getPrefetchResponse((int) acceptfd, &readbuffer)) != 0) {
 260         printf("Error: In getPrefetchResponse() %s, %d\n", __FILE__, __LINE__);
 261         break;
 262       }
 263 #endif
 264       break;
 265
 266     case START_REMOTE_THREAD:
 267       recv_data_buf((int)acceptfd, &readbuffer, &oid, sizeof(unsigned int));
 268       objType = getObjType(oid);
 269       startDSMthread(oid, objType);
 270       break;
 271
 272     case THREAD_NOTIFY_REQUEST:
 273       recv_data_buf((int)acceptfd, &readbuffer, &numoid, sizeof(unsigned int));
 274       size = (sizeof(unsigned int) + sizeof(unsigned short)) * numoid + 2 * sizeof(unsigned int);
 275       if((buffer = calloc(1,size)) == NULL) {
 276         printf("%s() Calloc error at %s, %d\n", __func__, __FILE__, __LINE__);
 277         pthread_exit(NULL);
 278       }
 279
 280       recv_data_buf((int)acceptfd, &readbuffer, buffer, size);
 281
 282       oidarry = calloc(numoid, sizeof(unsigned int));
 283       memcpy(oidarry, buffer, sizeof(unsigned int) * numoid);
 284       size = sizeof(unsigned int) * numoid;
 285       versionarry = calloc(numoid, sizeof(unsigned short));
 286       memcpy(versionarry, buffer+size, sizeof(unsigned short) * numoid);
 287       size += sizeof(unsigned short) * numoid;
 288       mid = *((unsigned int *)(buffer+size));
 289       size += sizeof(unsigned int);
 290       threadid = *((unsigned int *)(buffer+size));
 291       processReqNotify(numoid, oidarry, versionarry, mid, threadid);
 292       free(buffer);
 293       break;
 294
 295     case THREAD_NOTIFY_RESPONSE:
 296       size = sizeof(unsigned short) + 2 * sizeof(unsigned int);
 297       if((buffer = calloc(1,size)) == NULL) {
 298         printf("%s() Calloc error at %s, %d\n", __func__, __FILE__, __LINE__);
 299         pthread_exit(NULL);
 300       }
 301
 302       recv_data_buf((int)acceptfd, &readbuffer, buffer, size);
 303
 304       oid = *((unsigned int *)buffer);
 305       size = sizeof(unsigned int);
 306       version = *((unsigned short *)(buffer+size));
 307       size += sizeof(unsigned short);
 308       threadid = *((unsigned int *)(buffer+size));
 309       threadNotify(oid,version,threadid);
 310       free(buffer);
 311       break;
 312
 313 #ifdef SANDBOX
 314     case CHECK_OBJECTS: // check if versions of objects match
 315       size = sizeof(odata) - 1;
 316       ptr = (char*)&odata;
 317       recv_data_buf((int)acceptfd, &readbuffer, ptr+1, size);
 318       numread = odata.numread;
 319       nummod = odata.nummod;
 320       checkObjVersion(&readbuffer, (int) acceptfd, numread, nummod);
 321       break;
 322 #endif
 323
 324     case CLOSE_CONNECTION:
 325       goto closeconnection;
 326
 327     default:
 328       printf("Error: dstmAccept() Unknown opcode %d at %s, %d\n", control, __FILE__, __LINE__);
 329     }
 330   }
 331
 332 closeconnection:
 333   /* Close connection */
 334   if (close((int)acceptfd) == -1)
 335     perror("close");
 336   pthread_exit(NULL);
 337 }
 338
 339 /* This function reads the information available in a transaction request
 340  * and makes a function call to process the request */
 341 int readClientReq(trans_commit_data_t *transinfo, int acceptfd, struct readstruct * readbuffer) {
 342   char *ptr;
 343   void *modptr;
 344   unsigned int *oidmod, oid;
 345   fixed_data_t fixed;
 346   objheader_t *headaddr;
 347   int sum, i, size, n, val;
 348
 349   oidmod = NULL;
 350
 351   /* Read fixed_data_t data structure */
 352   size = sizeof(fixed) - 1;
 353   ptr = (char *)&fixed;
 354   fixed.control = TRANS_REQUEST;
 355   recv_data_buf((int)acceptfd, readbuffer, ptr+1, size);
 356
 357   /* Read list of mids */
 358   int mcount = fixed.mcount;
 359   size = mcount * sizeof(unsigned int);
 360   unsigned int listmid[mcount];
 361   ptr = (char *) listmid;
 362   recv_data_buf((int)acceptfd, readbuffer, ptr, size);
 363
 364   /* Read oid and version tuples for those objects that are not modified in the transaction */
 365   int numread = fixed.numread;
 366   size = numread * (sizeof(unsigned int) + sizeof(unsigned short));
 367   char objread[size];
 368   if(numread != 0) { //If pile contains more than one object to be read,
 369     // keep reading all objects
 370     recv_data_buf((int)acceptfd, readbuffer, objread, size);
 371   }
 372
 373   /* Read modified objects */
 374   if(fixed.nummod != 0) {
 375     if ((modptr = calloc(1, fixed.sum_bytes)) == NULL) {
 376       printf("calloc error for modified objects %s, %d\n", __FILE__, __LINE__);
 377       return 1;
 378     }
 379     size = fixed.sum_bytes;
 380     recv_data_buf((int)acceptfd, readbuffer, modptr, size);
 381   }
 382
 383   /* Create an array of oids for modified objects */
 384   oidmod = (unsigned int *) calloc(fixed.nummod, sizeof(unsigned int));
 385   if (oidmod == NULL) {
 386     printf("calloc error %s, %d\n", __FILE__, __LINE__);
 387     return 1;
 388   }
 389   ptr = (char *) modptr;
 390   for(i = 0; i < fixed.nummod; i++) {
 391     headaddr = (objheader_t *) ptr;
 392     oid = OID(headaddr);
 393     oidmod[i] = oid;
 394     int tmpsize=0;
 395     GETSIZE(tmpsize, headaddr);
 396     ptr += sizeof(objheader_t) + tmpsize;
 397   }
 398
 399   /*Process the information read */
 400   if((val = processClientReq(&fixed, transinfo, listmid, objread, modptr, oidmod, acceptfd, readbuffer)) != 0) {
 401     printf("Error: In processClientReq() %s, %d\n", __FILE__, __LINE__);
 402     /* Free resources */
 403     if(oidmod != NULL) {
 404       free(oidmod);
 405     }
 406     return 1;
 407   }
 408
 409   /* Free resources */
 410   if(oidmod != NULL) {
 411     free(oidmod);
 412   }
 413
 414   return 0;
 415 }
 416
 417 /* This function processes the Coordinator's transaction request using "handleTransReq"
 418  * function and sends a reply to the co-ordinator.
 419  * Following this it also receives a new control message from the co-ordinator and processes this message*/
 420 int processClientReq(fixed_data_t *fixed, trans_commit_data_t *transinfo,
 421                      unsigned int *listmid, char *objread, void *modptr, unsigned int *oidmod, int acceptfd, struct readstruct *readbuffer) {
 422
 423   char control, sendctrl, retval;
 424   objheader_t *tmp_header;
 425   void *header;
 426   int i = 0, val;
 427
 428   /* Send reply to the Coordinator */
 429   if((retval = handleTransReq(fixed, transinfo, listmid, objread, modptr,acceptfd)) == 0 ) {
 430     printf("Error: In handleTransReq() %s, %d\n", __FILE__, __LINE__);
 431     return 1;
 432   }
 433
 434   recv_data_buf((int)acceptfd, readbuffer, &control, sizeof(char));
 435   /* Process the new control message */
 436   switch(control) {
 437   case TRANS_ABORT:
 438     if (fixed->nummod > 0)
 439       free(modptr);
 440     /* Unlock objects that was locked due to this transaction */
 441     int useWriteUnlock = 0; //TODO verify is this piece of unlocking code ever used
 442     for(i = 0; i< transinfo->numlocked; i++) {
 443       if(transinfo->objlocked[i] == -1) {
 444         useWriteUnlock = 1;
 445         continue;
 446       }
 447       if((header = mhashSearch(transinfo->objlocked[i])) == NULL) {
 448         printf("mhashSearch returns NULL at %s, %d\n", __FILE__, __LINE__); // find the header address
 449         return 1;
 450       }
 451       if(useWriteUnlock) {
 452         write_unlock(STATUSPTR(header));
 453       } else {
 454         read_unlock(STATUSPTR(header));
 455       }
 456     }
 457     break;
 458
 459   case TRANS_COMMIT:
 460     /* Invoke the transCommit process() */
 461     if((val = transCommitProcess(modptr, oidmod, transinfo->objlocked, fixed->nummod, transinfo->numlocked, (int)acceptfd)) != 0) {
 462       printf("Error: In transCommitProcess() %s, %d\n", __FILE__, __LINE__);
 463       /* Free memory */
 464       if (transinfo->objlocked != NULL) {
 465         free(transinfo->objlocked);
 466       }
 467       if (transinfo->objnotfound != NULL) {
 468         free(transinfo->objnotfound);
 469       }
 470       return 1;
 471     }
 472     break;
 473
 474   case TRANS_ABORT_BUT_RETRY_COMMIT_WITH_RELOCATING:
 475     break;
 476
 477   default:
 478     printf("Error: No response to TRANS_AGREE OR DISAGREE protocol  control = %d %s, %d\n", control, __FILE__, __LINE__);
 479     //TODO Use fixed.trans_id  TID since Client may have died
 480     break;
 481   }
 482
 483   /* Free memory */
 484   if (transinfo->objlocked != NULL) {
 485     free(transinfo->objlocked);
 486   }
 487   if (transinfo->objnotfound != NULL) {
 488     free(transinfo->objnotfound);
 489   }
 490
 491   return 0;
 492 }
 493
 494 /* This function increments counters while running a voting decision on all objects involved
 495  * in TRANS_REQUEST and If a TRANS_DISAGREE sends the response immediately back to the coordinator */
 496 char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigned int *listmid, char *objread, void *modptr, int acceptfd) {
 497   int val, i = 0, j;
 498   unsigned short version;
 499   char control = 0, *ptr;
 500   unsigned int oid;
 501   unsigned int *oidnotfound, *oidlocked, *oidvernotmatch;
 502   objheader_t *headptr;
 503
 504   /* Counters and arrays to formulate decision on control message to be sent */
 505   oidnotfound = (unsigned int *) calloc(fixed->numread + fixed->nummod, sizeof(unsigned int));
 506   oidlocked = (unsigned int *) calloc(fixed->numread + fixed->nummod + 1, sizeof(unsigned int));
 507   oidvernotmatch = (unsigned int *) calloc(fixed->numread + fixed->nummod, sizeof(unsigned int));
 508   int objnotfound = 0, objlocked = 0, objvernotmatch = 0;
 509   int v_nomatch = 0, v_matchlock = 0, v_matchnolock = 0;
 510   int numBytes = 0;
 511   /* modptr points to the beginning of the object store
 512    * created at the Pariticipant.
 513    * Object store holds the modified objects involved in the transaction request */
 514   ptr = (char *) modptr;
 515
 516   char retval;
 517
 518   /* Process each oid in the machine pile/ group per thread */
 519   for (i = 0; i < fixed->numread + fixed->nummod; i++) {
 520     if (i < fixed->numread) { //Objs only read and not modified
 521       int incr = sizeof(unsigned int) + sizeof(unsigned short); // Offset that points to next position in the objread array
 522       incr *= i;
 523       oid = *((unsigned int *)(objread + incr));
 524
 525       incr += sizeof(unsigned int);
 526       version = *((unsigned short *)(objread + incr));
 527       retval=getCommitCountForObjRead(oidnotfound, oidlocked, oidvernotmatch, &objnotfound, &objlocked, &objvernotmatch,
 528                                       &v_matchnolock, &v_matchlock, &v_nomatch, &numBytes, &control, oid, version);
 529     } else {  //Objs modified
 530       if(i == fixed->numread) {
 531         oidlocked[objlocked++] = -1;
 532       }
 533       int tmpsize;
 534       headptr = (objheader_t *) ptr;
 535       oid = OID(headptr);
 536       version = headptr->version;
 537       GETSIZE(tmpsize, headptr);
 538       ptr += sizeof(objheader_t) + tmpsize;
 539       retval=getCommitCountForObjMod(oidnotfound, oidlocked, oidvernotmatch, &objnotfound,
 540                                      &objlocked, &objvernotmatch, &v_matchnolock, &v_matchlock, &v_nomatch,
 541                                      &numBytes, &control, oid, version);
 542     }
 543     if(retval==TRANS_DISAGREE || retval==TRANS_SOFT_ABORT) {
 544       //unlock objects as soon versions mismatch or locks cannot be acquired)
 545       if (objlocked > 0) {
 546         int useWriteUnlock = 0;
 547         for(j = 0; j < objlocked; j++) {
 548           if(oidlocked[j] == -1) {
 549             useWriteUnlock = 1;
 550             continue;
 551           }
 552           if((headptr = mhashSearch(oidlocked[j])) == NULL) {
 553             printf("mhashSearch returns NULL at %s, %d\n", __FILE__, __LINE__);
 554             return 0;
 555           }
 556           if(useWriteUnlock) {
 557             write_unlock(STATUSPTR(headptr));
 558           } else {
 559             read_unlock(STATUSPTR(headptr));
 560           }
 561         }
 562         if(v_nomatch > 0)
 563           free(oidlocked);
 564       }
 565       objlocked=0;
 566       break;
 567     }
 568   }
 569   //go through rest of the objects for version mismatches
 570   if(retval==TRANS_DISAGREE || retval==TRANS_SOFT_ABORT) {
 571     i++;
 572     procRestObjs(objread, ptr, i, fixed->numread, fixed->nummod, oidnotfound, oidvernotmatch, &objnotfound, &objvernotmatch, &v_nomatch, &numBytes);
 573   }
 574
 575   /* send TRANS_DISAGREE and objs*/
 576   if(v_nomatch > 0) {
 577 #ifdef CACHE
 578     char *objs = calloc(1, numBytes);
 579     int j, offset = 0;
 580     for(j = 0; j<objvernotmatch; j++) {
 581       objheader_t *header = mhashSearch(oidvernotmatch[j]);
 582       int size = 0;
 583       GETSIZE(size, header);
 584       size += sizeof(objheader_t);
 585       memcpy(objs+offset, header, size);
 586       offset += size;
 587     }
 588 #endif
 589     /*
 590        if (objlocked > 0) {
 591        int useWriteUnlock = 0;
 592        for(j = 0; j < objlocked; j++) {
 593         if(oidlocked[j] == -1) {
 594           useWriteUnlock = 1;
 595           continue;
 596         }
 597         if((headptr = mhashSearch(oidlocked[j])) == NULL) {
 598           printf("mhashSearch returns NULL at %s, %d\n", __FILE__, __LINE__);
 599           return 0;
 600         }
 601         if(useWriteUnlock) {
 602           write_unlock(STATUSPTR(headptr));
 603         } else {
 604           read_unlock(STATUSPTR(headptr));
 605         }
 606        }
 607        free(oidlocked);
 608        }
 609      */
 610     control=TRANS_DISAGREE;
 611     send_data(acceptfd, &control, sizeof(char));
 612 #ifdef CACHE
 613     send_data(acceptfd, &numBytes, sizeof(int));
 614     send_data(acceptfd, objs, numBytes);
 615     transinfo->objvernotmatch = oidvernotmatch;
 616     transinfo->numvernotmatch = objvernotmatch;
 617     free(objs);
 618     free(transinfo->objvernotmatch);
 619 #endif
 620     return control;
 621   }
 622
 623   /* Decide what control message to send to Coordinator */
 624   if ((control = decideCtrlMessage(fixed, transinfo, &v_matchnolock, &v_matchlock, &v_nomatch, &objnotfound, &objlocked,
 625                                    modptr, oidnotfound, oidlocked, acceptfd)) == 0) {
 626     printf("Error: In decideCtrlMessage() %s, %d\n", __FILE__, __LINE__);
 627     return 0;
 628   }
 629   return control;
 630 }
 631
 632 /* Update Commit info for objects that are read */
 633 char getCommitCountForObjMod(unsigned int *oidnotfound, unsigned int *oidlocked,
 634                              unsigned int *oidvernotmatch, int *objnotfound, int *objlocked, int *objvernotmatch,
 635                              int *v_matchnolock, int *v_matchlock, int *v_nomatch, int *numBytes,
 636                              char *control, unsigned int oid, unsigned short version) {
 637   void *mobj;
 638   /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
 639
 640   if ((mobj = mhashSearch(oid)) == NULL) {    /* Obj not found */
 641     /* Save the oids not found and number of oids not found for later use */
 642     oidnotfound[*objnotfound] = oid;
 643     (*objnotfound)++;
 644     *control = TRANS_DISAGREE;
 645   } else {     /* If Obj found in machine (i.e. has not moved) */
 646     /* Check if Obj is locked by any previous transaction */
 647     if (write_trylock(STATUSPTR(mobj))) { // Can acquire write lock
 648       if (version == ((objheader_t *)mobj)->version) { /* match versions */
 649         (*v_matchnolock)++;
 650         *control = TRANS_AGREE;
 651       } else { /* If versions don't match ...HARD ABORT */
 652         (*v_nomatch)++;
 653         oidvernotmatch[*objvernotmatch] = oid;
 654         (*objvernotmatch)++;
 655         int size;
 656         GETSIZE(size, mobj);
 657         size += sizeof(objheader_t);
 658         *numBytes += size;
 659         /* Send TRANS_DISAGREE to Coordinator */
 660         *control = TRANS_DISAGREE;
 661       }
 662       //Keep track of oid locked
 663       oidlocked[(*objlocked)++] = OID(((objheader_t *)mobj));
 664     } else {  //we are locked
 665       if (version == ((objheader_t *)mobj)->version) {     /* Check if versions match */
 666         (*v_matchlock)++;
 667         *control=TRANS_SOFT_ABORT;
 668       } else { /* If versions don't match ...HARD ABORT */
 669         (*v_nomatch)++;
 670         oidvernotmatch[*objvernotmatch] = oid;
 671         (*objvernotmatch)++;
 672         int size;
 673         GETSIZE(size, mobj);
 674         size += sizeof(objheader_t);
 675         *numBytes += size;
 676         *control = TRANS_DISAGREE;
 677       }
 678     }
 679   }
 680   return *control;
 681 }
 682
 683 /* Update Commit info for objects that are read */
 684 char getCommitCountForObjRead(unsigned int *oidnotfound, unsigned int *oidlocked, unsigned int *oidvernotmatch,
 685                               int *objnotfound, int *objlocked, int * objvernotmatch, int *v_matchnolock, int *v_matchlock,
 686                               int *v_nomatch, int *numBytes, char *control, unsigned int oid, unsigned short version) {
 687   void *mobj;
 688   /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
 689   if ((mobj = mhashSearch(oid)) == NULL) {    /* Obj not found */
 690     /* Save the oids not found and number of oids not found for later use */
 691     oidnotfound[*objnotfound] = oid;
 692     (*objnotfound)++;
 693     *control = TRANS_DISAGREE;
 694   } else {     /* If Obj found in machine (i.e. has not moved) */
 695     /* Check if Obj is locked by any previous transaction */
 696     if (read_trylock(STATUSPTR(mobj))) { //Can further acquire read locks
 697       if (version == ((objheader_t *)mobj)->version) { /* match versions */
 698         (*v_matchnolock)++;
 699         *control=TRANS_AGREE;
 700       } else { /* If versions don't match ...HARD ABORT */
 701         (*v_nomatch)++;
 702         oidvernotmatch[(*objvernotmatch)++] = oid;
 703         int size;
 704         GETSIZE(size, mobj);
 705         size += sizeof(objheader_t);
 706         *numBytes += size;
 707         /* Send TRANS_DISAGREE to Coordinator */
 708         *control = TRANS_DISAGREE;
 709       }
 710       //Keep track of oid locked
 711       oidlocked[(*objlocked)++] = OID(((objheader_t *)mobj));
 712     } else { /* Some other transaction has aquired a write lock on this object */
 713       if (version == ((objheader_t *)mobj)->version) { /* Check if versions match */
 714         (*v_matchlock)++;
 715         *control=TRANS_SOFT_ABORT;
 716       } else { /* If versions don't match ...HARD ABORT */
 717         (*v_nomatch)++;
 718         oidvernotmatch[*objvernotmatch] = oid;
 719         (*objvernotmatch)++;
 720         int size;
 721         GETSIZE(size, mobj);
 722         size += sizeof(objheader_t);
 723         *numBytes += size;
 724         *control = TRANS_DISAGREE;
 725       }
 726     }
 727   }
 728   return *control;
 729 }
 730
 731 void procRestObjs(char *objread,
 732                   char *objmod,
 733                   int index,
 734                   int numread,
 735                   int nummod,
 736                   unsigned int *oidnotfound,
 737                   unsigned int *oidvernotmatch,
 738                   int *objnotfound,
 739                   int *objvernotmatch,
 740                   int *v_nomatch,
 741                   int *numBytes) {
 742   int i;
 743   unsigned int oid;
 744   unsigned short version;
 745
 746   /* Process each oid in the machine pile/ group per thread */
 747   for (i = index; i < numread+nummod; i++) {
 748     if (i < numread) { //Objs only read and not modified
 749       int incr = sizeof(unsigned int) + sizeof(unsigned short); // Offset that points to next position in the objread array
 750       incr *= i;
 751       oid = *((unsigned int *)(objread + incr));
 752       incr += sizeof(unsigned int);
 753       version = *((unsigned short *)(objread + incr));
 754     } else {  //Objs modified
 755       objheader_t *headptr;
 756       headptr = (objheader_t *) objmod;
 757       oid = OID(headptr);
 758       version = headptr->version;
 759       int tmpsize;
 760       GETSIZE(tmpsize, headptr);
 761       objmod += sizeof(objheader_t) + tmpsize;
 762     }
 763     processVerNoMatch(oidnotfound,
 764                       oidvernotmatch,
 765                       objnotfound,
 766                       objvernotmatch,
 767                       v_nomatch,
 768                       numBytes,
 769                       oid,
 770                       version);
 771   }
 772   return;
 773 }
 774
 775 void processVerNoMatch(unsigned int *oidnotfound,
 776                        unsigned int *oidvernotmatch,
 777                        int *objnotfound,
 778                        int *objvernotmatch,
 779                        int *v_nomatch,
 780                        int *numBytes,
 781                        unsigned int oid,
 782                        unsigned short version) {
 783   void *mobj;
 784   /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
 785
 786   if ((mobj = mhashSearch(oid)) == NULL) {    /* Obj not found */
 787     /* Save the oids not found and number of oids not found for later use */
 788     oidnotfound[*objnotfound] = oid;
 789     (*objnotfound)++;
 790   } else {     /* If Obj found in machine (i.e. has not moved) */
 791     /* Check if Obj is locked by any previous transaction */
 792     //if (!write_trylock(STATUSPTR(mobj))) { // Can acquire write lock
 793     if (version != ((objheader_t *)mobj)->version) { /* match versions */
 794       (*v_nomatch)++;
 795       oidvernotmatch[*objvernotmatch] = oid;
 796       (*objvernotmatch)++;
 797       int size;
 798       GETSIZE(size, mobj);
 799       size += sizeof(objheader_t);
 800       *numBytes += size;
 801     }
 802   }
 803 }
 804
 805 /* This function decides what control message such as TRANS_AGREE, TRANS_DISAGREE or TRANS_SOFT_ABORT
 806  * to send to Coordinator based on the votes of oids involved in the transaction */
 807 char decideCtrlMessage(fixed_data_t *fixed, trans_commit_data_t *transinfo, int *v_matchnolock, int *v_matchlock,
 808                        int *v_nomatch, int *objnotfound, int *objlocked, void *modptr,
 809                        unsigned int *oidnotfound, unsigned int *oidlocked, int acceptfd) {
 810   int val;
 811   char control = 0;
 812
 813   /* Condition to send TRANS_AGREE */
 814   if(*(v_matchnolock) == fixed->numread + fixed->nummod) {
 815     control = TRANS_AGREE;
 816     /* Send control message */
 817     send_data(acceptfd, &control, sizeof(char));
 818   }
 819   /* Condition to send TRANS_SOFT_ABORT */
 820   if((*(v_matchlock) > 0 && *(v_nomatch) == 0) || (*(objnotfound) > 0 && *(v_nomatch) == 0)) {
 821     control = TRANS_SOFT_ABORT;
 822
 823     /* Send control message */
 824     send_data(acceptfd, &control, sizeof(char));
 825
 826     /*  FIXME how to send objs Send number of oids not found and the missing oids if objects are missing in the machine */
 827     if(*(objnotfound) != 0) {
 828       int msg[1];
 829       msg[0] = *(objnotfound);
 830       send_data(acceptfd, &msg, sizeof(int));
 831       int size = sizeof(unsigned int)* *(objnotfound);
 832       send_data(acceptfd, oidnotfound, size);
 833     }
 834   }
 835
 836   /* Fill out the trans_commit_data_t data structure. This is required for a trans commit process
 837    * if Participant receives a TRANS_COMMIT */
 838   transinfo->objlocked = oidlocked;
 839   transinfo->objnotfound = oidnotfound;
 840   transinfo->modptr = modptr;
 841   transinfo->numlocked = *(objlocked);
 842   transinfo->numnotfound = *(objnotfound);
 843   return control;
 844 }
 845
 846 /* This function processes all modified objects involved in a TRANS_COMMIT and updates pointer
 847  * addresses in lookup table and also changes version number
 848  * Sends an ACK back to Coordinator */
 849 int transCommitProcess(void *modptr, unsigned int *oidmod, unsigned int *oidlocked, int nummod, int numlocked, int acceptfd) {
 850   objheader_t *header;
 851   objheader_t *newheader;
 852   int i = 0, offset = 0;
 853   char control;
 854   int tmpsize;
 855
 856   /* Process each modified object saved in the mainobject store */
 857   for(i = 0; i < nummod; i++) {
 858     if((header = (objheader_t *) mhashSearch(oidmod[i])) == NULL) {
 859       printf("Error: mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
 860       return 1;
 861     }
 862     GETSIZE(tmpsize,header);
 863
 864     {
 865       struct ___Object___ *dst=(struct ___Object___*)((char*)header+sizeof(objheader_t));
 866       struct ___Object___ *src=(struct ___Object___*)((char*)modptr+sizeof(objheader_t)+offset);
 867       dst->type=src->type;
 868       dst->___cachedCode___=src->___cachedCode___;
 869       dst->___cachedHash___=src->___cachedHash___;
 870       memcpy(&dst[1], &src[1], tmpsize-sizeof(struct ___Object___));
 871     }
 872
 873     //memory barrier
 874     CFENCE;
 875
 876     header->version += 1;
 877     /* If threads are waiting on this object to be updated, notify them */
 878     if(header->notifylist != NULL) {
 879       notifyAll(&header->notifylist, OID(header), header->version);
 880     }
 881     offset += sizeof(objheader_t) + tmpsize;
 882   }
 883
 884   if (nummod > 0)
 885     free(modptr);
 886
 887   /* Unlock locked objects */
 888   int useWriteUnlock = 0;
 889   for(i = 0; i < numlocked; i++) {
 890     if(oidlocked[i] == -1) {
 891       useWriteUnlock = 1;
 892       continue;
 893     }
 894     if((header = (objheader_t *) mhashSearch(oidlocked[i])) == NULL) {
 895       printf("Error: mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
 896       return 1;
 897     }
 898
 899     if(useWriteUnlock) {
 900       write_unlock(STATUSPTR(header));
 901     } else {
 902       read_unlock(STATUSPTR(header));
 903     }
 904   }
 905   //TODO Update location lookup table
 906   return 0;
 907 }
 908
 909 /* This function recevies the oid and offset tuples from the Coordinator's prefetch call.
 910  * Looks for the objects to be prefetched in the main object store.
 911  * If objects are not found then record those and if objects are found
 912  * then use offset values to prefetch references to other objects */
 913 int prefetchReq(int acceptfd, struct readstruct * readbuffer) {
 914   int i, size, objsize, numoffset = 0, gid=0;
 915   int length;
 916   char *recvbuffer, control;
 917   unsigned int oid, mid=-1;
 918   objheader_t *header;
 919   oidmidpair_t oidmid;
 920   struct writestruct writebuffer;
 921   int sd = -1;
 922
 923   while(1) {
 924     recv_data_buf((int)acceptfd, readbuffer, &numoffset, sizeof(int));
 925     if(numoffset == -1)
 926       break;
 927     recv_data_buf((int)acceptfd, readbuffer, &oidmid, 2*sizeof(unsigned int));
 928     oid = oidmid.oid;
 929     if (mid != oidmid.mid) {
 930       if (mid!=-1) {
 931         forcesend_buf(sd, &writebuffer, NULL, 0);
 932         freeSockWithLock(transPResponseSocketPool, mid, sd);
 933       }
 934       mid=oidmid.mid;
 935       sd = getSockWithLock(transPResponseSocketPool, mid);
 936       writebuffer.offset=0;
 937     }
 938     short offsetarry[numoffset];
 939     recv_data_buf((int)acceptfd, readbuffer, &gid, sizeof(int));
 940     recv_data_buf((int) acceptfd, readbuffer, offsetarry, numoffset*sizeof(short));
 941     LOGTIME('A',oid,0,myrdtsc(),gid);  //after recv the entire prefetch request
 942
 943     /*Process each oid */
 944     if ((header = mhashSearch(oid)) == NULL) { /* Obj not found */
 945       /* Save the oids not found in buffer for later use */
 946       size = sizeof(int)+sizeof(int) + sizeof(char) + sizeof(unsigned int);
 947       char sendbuffer[size+1];
 948       sendbuffer[0]=TRANS_PREFETCH_RESPONSE;
 949       *((int *) (sendbuffer+sizeof(char))) = size;
 950       *((char *)(sendbuffer + sizeof(char)+sizeof(int))) = OBJECT_NOT_FOUND;
 951       *((unsigned int *)(sendbuffer + sizeof(int) + sizeof(char)+sizeof(char))) = oid;
 952       *((int *)(sendbuffer+sizeof(int) + sizeof(char)+sizeof(char)+sizeof(unsigned int))) = gid;
 953       send_buf(sd, &writebuffer, sendbuffer, size+1);
 954       LOGTIME('J',oid, 0,myrdtsc(), gid); //send first oid not found prefetch request
 955     } else { /* Object Found */
 956       int incr = 1;
 957       GETSIZE(objsize, header);
 958       size = sizeof(int)+sizeof(int) + sizeof(char) + sizeof(unsigned int) + sizeof(objheader_t) + objsize;
 959       char sendbuffer[size+1];
 960       sendbuffer[0]=TRANS_PREFETCH_RESPONSE;
 961       *((int *)(sendbuffer + incr)) = size;
 962       incr += sizeof(int);
 963       *((char *)(sendbuffer + incr)) = OBJECT_FOUND;
 964       incr += sizeof(char);
 965       *((unsigned int *)(sendbuffer+incr)) = oid;
 966       incr += sizeof(unsigned int);
 967       *((int *)(sendbuffer+incr)) = gid;
 968       incr += sizeof(int);
 969       memcpy(sendbuffer + incr, header, objsize + sizeof(objheader_t));
 970       send_buf(sd, &writebuffer, sendbuffer, size+1);
 971       LOGOIDTYPE("SRES", oid, TYPE(header), (myrdtsc()-clockoffset));
 972       LOGTIME('C',oid,TYPE(header),myrdtsc(), gid); //send first oid found from prefetch request
 973
 974       /* Calculate the oid corresponding to the offset value */
 975       for(i = 0; i< numoffset; i++) {
 976         /* Check for arrays  */
 977         if(TYPE(header) >= NUMCLASSES) {
 978           int elementsize = classsize[TYPE(header)];
 979           struct ArrayObject *ao = (struct ArrayObject *) (((char *)header) + sizeof(objheader_t));
 980           unsigned short length = ao->___length___;
 981           /* Check if array out of bounds */
 982           if(offsetarry[i]< 0 || offsetarry[i] >= length) {
 983             break;
 984           }
 985           oid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + sizeof(struct ArrayObject) + (elementsize*offsetarry[i])));
 986         } else {
 987           oid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + offsetarry[i]));
 988         }
 989
 990         /* Don't continue if we hit a NULL pointer */
 991         if (oid==0)
 992           break;
 993
 994         LOGTIME('B',oid,0,myrdtsc(),gid); //send next oid found from prefetch request
 995
 996         if((header = mhashSearch(oid)) == NULL) {
 997           size = sizeof(int)+sizeof(int) + sizeof(char) + sizeof(unsigned int);
 998           char sendbuffer[size+1];
 999           sendbuffer[0]=TRANS_PREFETCH_RESPONSE;
1000           *((int *) (sendbuffer+1)) = size;
1001           *((char *)(sendbuffer + sizeof(char)+sizeof(int))) = OBJECT_NOT_FOUND;
1002           *((unsigned int *)(sendbuffer + sizeof(char)+sizeof(int) + sizeof(char))) = oid;
1003           *((int *)(sendbuffer+sizeof(int) + sizeof(char)+sizeof(char)+sizeof(unsigned int))) = gid;
1004
1005           send_buf(sd, &writebuffer, sendbuffer, size+1);
1006           LOGTIME('J',oid, 0,myrdtsc(), gid); //send first oid not found prefetch request
1007           break;
1008         } else { /* Obj Found */
1009           int incr = 1;
1010           GETSIZE(objsize, header);
1011           size = sizeof(int)+sizeof(int) + sizeof(char) + sizeof(unsigned int) + sizeof(objheader_t) + objsize;
1012           char sendbuffer[size+1];
1013           sendbuffer[0]=TRANS_PREFETCH_RESPONSE;
1014           *((int *)(sendbuffer + incr)) = size;
1015           incr += sizeof(int);
1016           *((char *)(sendbuffer + incr)) = OBJECT_FOUND;
1017           incr += sizeof(char);
1018           *((unsigned int *)(sendbuffer+incr)) = oid;
1019           incr += sizeof(unsigned int);
1020           *((int *)(sendbuffer+incr)) = gid;
1021           incr += sizeof(int);
1022           memcpy(sendbuffer + incr, header, objsize + sizeof(objheader_t));
1023           send_buf(sd, &writebuffer, sendbuffer, size+1);
1024           LOGOIDTYPE("SRES", oid, TYPE(header), (myrdtsc()-clockoffset));
1025           LOGTIME('C',oid,TYPE(header),myrdtsc(), gid); //send first oid found from prefetch request
1026         }
1027       } //end of for
1028     }
1029   } //end of while
1030
1031   //Release socket
1032   if (mid!=-1) {
1033     forcesend_buf(sd, &writebuffer, NULL, 0);
1034     freeSockWithLock(transPResponseSocketPool, mid, sd);
1035   }
1036   return 0;
1037 }
1038
1039 void sendPrefetchResponse(int sd, char *control, char *sendbuffer, int *size) {
1040   send_data(sd, control, sizeof(char));
1041   /* Send the buffer with its size */
1042   int length = *(size);
1043   send_data(sd, sendbuffer, length);
1044 }
1045
1046 void processReqNotify(unsigned int numoid, unsigned int *oidarry, unsigned short *versionarry, unsigned int mid, unsigned int threadid) {
1047   objheader_t *header;
1048   unsigned int oid;
1049   unsigned short newversion;
1050   char msg[1+  2 * sizeof(unsigned int) + sizeof(unsigned short)];
1051   int sd;
1052   struct sockaddr_in remoteAddr;
1053   //int bytesSent;
1054   int size;
1055   int i = 0;
1056
1057   while(i < numoid) {
1058     oid = *(oidarry + i);
1059     if((header = (objheader_t *) mhashSearch(oid)) == NULL) {
1060       printf("Error: mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
1061       return;
1062     } else {
1063       /* Check to see if versions are same */
1064 checkversion:
1065       if (write_trylock(STATUSPTR(header))) { // Can acquire write lock
1066         newversion = header->version;
1067         if(newversion == *(versionarry + i)) {
1068           //Add to the notify list
1069           if((header->notifylist = insNode(header->notifylist, threadid, mid)) == NULL) {
1070             printf("Error: Obj notify list points to NULL %s, %d\n", __FILE__, __LINE__);
1071             return;
1072           }
1073           write_unlock(STATUSPTR(header));
1074         } else {
1075           write_unlock(STATUSPTR(header));
1076           if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
1077             perror("processReqNotify():socket()");
1078             return;
1079           }
1080           bzero(&remoteAddr, sizeof(remoteAddr));
1081           remoteAddr.sin_family = AF_INET;
1082           remoteAddr.sin_port = htons(LISTEN_PORT);
1083           remoteAddr.sin_addr.s_addr = htonl(mid);
1084
1085           if (connect(sd, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0) {
1086             printf("Error: processReqNotify():error %d connecting to %s:%d\n", errno,
1087                    inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
1088             close(sd);
1089             return;
1090           } else {
1091
1092             //Send Update notification
1093             msg[0] = THREAD_NOTIFY_RESPONSE;
1094             *((unsigned int *)&msg[1]) = oid;
1095             size = sizeof(unsigned int);
1096             *((unsigned short *)(&msg[1]+size)) = newversion;
1097             size += sizeof(unsigned short);
1098             *((unsigned int *)(&msg[1]+size)) = threadid;
1099             size = 1+ 2*sizeof(unsigned int) + sizeof(unsigned short);
1100             send_data(sd, msg, size);
1101           }
1102           close(sd);
1103         }
1104       } else {
1105         randomdelay();
1106         goto checkversion;
1107       }
1108     }
1109     i++;
1110   }
1111   free(oidarry);
1112   free(versionarry);
1113 }