Robust/src/Runtime/DSTM/interface_recovery/trans.c

   1 #include "machinepile.h"
   2 #include "mlookup.h"
   3 #include "llookup.h"
   4 #include "plookup.h"
   5 #include "prelookup.h"
   6 #include "threadnotify.h"
   7 #include "queue.h"
   8 #include "addUdpEnhance.h"
   9 #include "addPrefetchEnhance.h"
  10 #include "gCollect.h"
  11 #include "dsmlock.h"
  12 #include "prefetch.h"
  13 #ifdef COMPILER
  14 #include "thread.h"
  15 #endif
  16 #ifdef ABORTREADERS
  17 #include "abortreaders.h"
  18 #endif
  19 #include "trans.h"
  20
  21 #ifdef RECOVERY
  22 #include <unistd.h>
  23 #include <signal.h>
  24 #include <sys/select.h>
  25 #define WAIT_TIME 3
  26 #endif
  27
  28 #define NUM_THREADS 1
  29 #define CONFIG_FILENAME "dstm.conf"
  30
  31 /* Thread transaction variables */
  32
  33 __thread objstr_t *t_cache;
  34 __thread struct ___Object___ *revertlist;
  35 #ifdef ABORTREADERS
  36 __thread int t_abort;
  37 __thread jmp_buf aborttrans;
  38 #endif
  39
  40 /* Global Variables */
  41 extern int classsize[];
  42 pfcstats_t *evalPrefetch;
  43 extern int numprefetchsites; //Global variable containing number of prefetch sites
  44 extern pthread_mutex_t mainobjstore_mutex; // Mutex to lock main Object store
  45 pthread_mutex_t prefetchcache_mutex; // Mutex to lock Prefetch Cache
  46 pthread_mutexattr_t prefetchcache_mutex_attr; /* Attribute for lock to make it a recursive lock */
  47 extern prehashtable_t pflookup; //Global Prefetch cache's lookup table
  48 pthread_t wthreads[NUM_THREADS]; //Worker threads for working on the prefetch queue
  49 pthread_t tPrefetch;            /* Primary Prefetch thread that processes the prefetch queue */
  50 extern objstr_t *mainobjstore;
  51 unsigned int myIpAddr;
  52 unsigned int *hostIpAddrs;
  53 int sizeOfHostArray;
  54 int numHostsInSystem;
  55 int myIndexInHostArray;
  56 unsigned int oidsPerBlock;
  57 unsigned int oidMin;
  58 unsigned int oidMax;
  59
  60 sockPoolHashTable_t *transReadSockPool;
  61 sockPoolHashTable_t *transPrefetchSockPool;
  62 sockPoolHashTable_t *transRequestSockPool;
  63 pthread_mutex_t notifymutex;
  64 pthread_mutex_t atomicObjLock;
  65
  66 /***********************************
  67  * Global Variables for statistics
  68  **********************************/
  69 int numTransCommit = 0;
  70 int numTransAbort = 0;
  71 int nchashSearch = 0;
  72 int nmhashSearch = 0;
  73 int nprehashSearch = 0;
  74 int nRemoteSend = 0;
  75 int nSoftAbort = 0;
  76 int bytesSent = 0;
  77 int bytesRecv = 0;
  78 int totalObjSize = 0;
  79
  80 /***********************************
  81  * Global variables for Duplication
  82  ***********************************/
  83 int *liveHosts;
  84 int liveHostsValid;
  85 int numLiveHostsInSystem;
  86 int flipBit;                                                            // Used to distribute requests between primary and backup evenly
  87 unsigned int *locateObjHosts;
  88 __thread int timeoutFlag;
  89 extern int leaderFixing;
  90 extern pthread_mutex_t leaderFixing_mutex;
  91 extern pthread_mutex_t liveHosts_mutex;
  92
  93 unsigned int liveTransactions[25];
  94 unsigned int transIDMax;
  95 unsigned int transIDMin;
  96 unsigned int transIDIndex;
  97 #ifdef DEBUG
  98 char ip[16];
  99 #endif
 100
 101 /******************************
 102  * Global variables for Paxos
 103  ******************************/
 104 int n_a;
 105 unsigned int v_a;
 106 int n_h;
 107 int my_n;
 108 unsigned int leader;
 109 unsigned int origleader;
 110 unsigned int temp_v_a;
 111 int paxosRound;
 112
 113 void printhex(unsigned char *, int);
 114 plistnode_t *createPiles();
 115 plistnode_t *sortPiles(plistnode_t *pileptr);
 116
 117 /*******************************
 118 * Send and Recv function calls
 119 *******************************/
 120 void send_data(int fd, void *buf, int buflen) {
 121 #ifdef DEBUG
 122 //      printf("%s-> Start; fd:%d, buflen:%d\n", __func__, fd, buflen);
 123 #endif
 124         char *buffer = (char *)(buf);
 125   int size = buflen;
 126   int numbytes;
 127   while (size > 0) {
 128                 numbytes = send(fd, buffer, size, 0);
 129                 bytesSent = bytesSent + numbytes;
 130 #ifdef RECOVERY
 131 #ifdef DEBUG
 132 //              printf("%s-> numbytes: %d\n", __func__, numbytes);
 133 #endif
 134                 if(errno == ECONNRESET) {       // EINT/EPIPE??; Connection reset, possible disconnected machine
 135 #ifdef DEBUG
 136                         printf("%s-> errno = ECONNRESET; connection reset\n", __func__);
 137                         printf("***SETTING TIMEOUTFLAG***\n");
 138 #endif
 139                         errno = 0;
 140                         timeoutFlag = 1;
 141                         return;
 142                 }
 143                 else if(errno == EAGAIN || errno == EWOULDBLOCK) {
 144 #ifdef DEBUG
 145                         printf("%s-> errno = EAGAIN|EWOULDBLOCK; socket timeout\n", __func__);
 146                         printf("***SETTING TIMEOUTFLAG***\n");
 147 #endif
 148                         errno = 0;
 149                         timeoutFlag = 1;
 150                         return;
 151                 }
 152                 else if(numbytes == -1) {
 153 #ifdef DEBUG
 154                         printf("%s-> numbytes = -1; socket timeout\n", __func__);
 155                         printf("***SETTING TIMEOUTFLAG***\n");
 156 #endif
 157                         timeoutFlag = 1;
 158                         return;
 159                 }
 160 #else
 161                 if (numbytes == -1) {
 162                         perror("send");
 163                         exit(0);
 164                 }
 165 #endif
 166                 buffer += numbytes;
 167                 size -= numbytes;
 168         }
 169 #ifdef DEBUG
 170 //      printf("%s-> Exiting\n", __func__);
 171 #endif
 172 }
 173
 174 void recv_data(int fd, void *buf, int buflen) {
 175 #ifdef DEBUG
 176 //      printf("%s-> Start; fd:%d, buflen:%d\n", __func__, fd, buflen);
 177 #endif
 178         char *buffer = (char *)(buf);
 179         int size = buflen;
 180   int numbytes;
 181         while (size > 0) {
 182         numbytes = recv(fd, buffer, size, 0);
 183     bytesRecv = bytesRecv + numbytes;
 184 #ifdef RECOVERY
 185 #ifdef DEBUG
 186 //              printf("%s-> numbytes: %d\n", __func__, numbytes);
 187 #endif
 188                 if(errno == ECONNRESET) {
 189 #ifdef DEBUG
 190                         printf("%s-> errno = ECONNRESET; connection reset\n", __func__);
 191                         printf("***SETTING TIMEOUTFLAG***\n");
 192 #endif
 193                         errno = 0;
 194                         timeoutFlag = 1;
 195                         return;
 196                 }
 197                 else if(errno == EAGAIN || errno == EWOULDBLOCK) {
 198 #ifdef DEBUG
 199                         printf("%s-> errno = EAGAIN|EWOULDBLOCK; socket timeout\n", __func__);
 200                         printf("***SETTING TIMEOUTFLAG***\n");
 201 #endif
 202                         errno = 0;
 203                         timeoutFlag = 1;
 204                         return;
 205                 }
 206                 else if(numbytes == -1) {
 207 #ifdef DEBUG
 208                         printf("%s-> numbytes = -1; socket timeout\n", __func__);
 209                         printf("***SETTING TIMEOUTFLAG***\n");
 210 #endif
 211                         timeoutFlag = 1;
 212                         return;
 213                 }
 214 #else
 215                 if (numbytes == -1) {
 216       perror("recv");
 217       exit(0);
 218     }
 219 #endif
 220     buffer += numbytes;
 221                 size -= numbytes;
 222         }
 223 #ifdef DEBUG
 224 //      printf("%s-> Exiting\n", __func__);
 225 #endif
 226 }
 227
 228 void recv_data_block(int fd, void *buf, int buflen) {
 229 #ifdef DEBUG
 230         printf("%s-> Start; fd:%d, buflen:%d\n", __func__, fd, buflen);
 231 #endif
 232         char *buffer = (char *)(buf);
 233         int size = buflen;
 234         int numbytes;
 235         while (size > 0) {
 236                 numbytes = recv(fd, buffer, size, 0);
 237 #ifdef DEBUG
 238                 printf("%s-> numbytes: %d\n", __func__, numbytes);
 239 #endif
 240                 if(errno == EAGAIN || errno == EWOULDBLOCK) {
 241                         errno = 0;
 242                 }
 243                 if(numbytes != -1) {
 244                         bytesRecv = bytesRecv + numbytes;
 245                         buffer += numbytes;
 246                         size -= numbytes;
 247                 }
 248         }
 249 #ifdef DEBUG
 250         printf("%s-> Exiting\n", __func__);
 251 #endif
 252 }
 253
 254 int recv_data_errorcode(int fd, void *buf, int buflen) {
 255 #ifdef DEBUG
 256         printf("%s-> Start; fd:%d, buflen:%d\n", __func__, fd, buflen);
 257 #endif
 258   char *buffer = (char *)(buf);
 259   int size = buflen;
 260         int numbytes;
 261         while (size > 0) {
 262                 numbytes = recv(fd, buffer, size, 0);
 263 #ifdef DEBUG
 264                 printf("%s-> numbytes: %d\n", __func__, numbytes);
 265 #endif
 266                 if (numbytes==0)
 267                         return 0;
 268                 else if (numbytes == -1) {
 269                         perror("recv_data_errorcode");
 270                         return -1;
 271                 }
 272                 buffer += numbytes;
 273                 size -= numbytes;
 274         }
 275 #ifdef DEBUG
 276         printf("%s-> Exiting\n", __func__);
 277 #endif
 278         return 1;
 279 }
 280
 281 void printhex(unsigned char *ptr, int numBytes) {
 282   int i;
 283   for (i = 0; i < numBytes; i++) {
 284     if (ptr[i] < 16)
 285       printf("0%x ", ptr[i]);
 286     else
 287       printf("%x ", ptr[i]);
 288   }
 289   printf("\n");
 290   return;
 291 }
 292
 293 inline int arrayLength(int *array) {
 294   int i;
 295   for(i=0 ; array[i] != -1; i++)
 296     ;
 297   return i;
 298 }
 299
 300 inline int findmax(int *array, int arraylength) {
 301   int max, i;
 302   max = array[0];
 303   for(i = 0; i < arraylength; i++) {
 304     if(array[i] > max) {
 305       max = array[i];
 306     }
 307   }
 308   return max;
 309 }
 310
 311 char* midtoIPString(unsigned int mid){
 312                 midtoIP(mid, ip);
 313                 return ip;
 314 }
 315 /* This function is a prefetch call generated by the compiler that
 316  * populates the shared primary prefetch queue*/
 317 void prefetch(int siteid, int ntuples, unsigned int *oids, unsigned short *endoffsets, short *arrayfields) {
 318   /* Allocate for the queue node*/
 319   int qnodesize = 2*sizeof(int) + ntuples * (sizeof(unsigned short) + sizeof(unsigned int)) + endoffsets[ntuples - 1] * sizeof(short);
 320   int len;
 321   char * node= getmemory(qnodesize);
 322   int top=endoffsets[ntuples-1];
 323
 324   if (node==NULL)
 325     return;
 326   /* Set queue node values */
 327
 328   /* TODO: Remove this after testing */
 329   evalPrefetch[siteid].callcount++;
 330
 331   *((int *)(node))=siteid;
 332   *((int *)(node + sizeof(int))) = ntuples;
 333   len = 2*sizeof(int);
 334   memcpy(node+len, oids, ntuples*sizeof(unsigned int));
 335   memcpy(node+len+ntuples*sizeof(unsigned int), endoffsets, ntuples*sizeof(unsigned short));
 336   memcpy(node+len+ntuples*(sizeof(unsigned int)+sizeof(short)), arrayfields, top*sizeof(short));
 337
 338   /* Lock and insert into primary prefetch queue */
 339   movehead(qnodesize);
 340 }
 341
 342 /* This function starts up the transaction runtime. */
 343 int dstmStartup(const char * option) {
 344   pthread_t thread_Listen, udp_thread_Listen;
 345   pthread_attr_t attr;
 346   int master=option!=NULL && strcmp(option, "master")==0;
 347   int fd;
 348   int udpfd;
 349
 350   if (processConfigFile() != 0)
 351     return 0; //TODO: return error value, cause main program to exit
 352 #ifdef COMPILER
 353   if (!master)
 354     threadcount--;
 355 #endif
 356
 357 #ifdef TRANSSTATS
 358   printf("Trans stats is on\n");
 359   fflush(stdout);
 360 #endif
 361 #ifdef ABORTREADERS
 362   initreaderlist();
 363 #endif
 364
 365   //Initialize socket pool
 366   transReadSockPool = createSockPool(transReadSockPool, DEFAULTSOCKPOOLSIZE);
 367   transPrefetchSockPool = createSockPool(transPrefetchSockPool, DEFAULTSOCKPOOLSIZE);
 368   transRequestSockPool = createSockPool(transRequestSockPool, DEFAULTSOCKPOOLSIZE);
 369
 370   dstmInit();
 371   transInit();
 372
 373   fd=startlistening();
 374   pthread_attr_init(&attr);
 375   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 376 #ifdef CACHE
 377   udpfd = udpInit();
 378   pthread_create(&udp_thread_Listen, &attr, udpListenBroadcast, (void*)udpfd);
 379 #endif
 380   if (master) {
 381                 pthread_create(&thread_Listen, &attr, dstmListen, (void*)fd);
 382 #ifdef RECOVERY
 383                 updateLiveHosts();
 384                 setLocateObjHosts();
 385                 updateLiveHostsCommit();
 386                 leader = paxos();
 387                 if(!allHostsLive()) {
 388                         printf("Not all hosts live. Exiting.\n");
 389                         exit(-1);
 390                 }
 391 #endif
 392                 return 1;
 393         } else {
 394                 dstmListen((void *)fd);
 395                 return 0;
 396   }
 397 }
 398
 399 //TODO Use this later
 400 void *pCacheAlloc(objstr_t *store, unsigned int size) {
 401   void *tmp;
 402   objstr_t *ptr;
 403   ptr = store;
 404   int success = 0;
 405
 406   while(ptr->next != NULL) {
 407     /* check if store is empty */
 408     if(((unsigned int)ptr->top - (unsigned int)ptr - sizeof(objstr_t) + size) <= ptr->size) {
 409       tmp = ptr->top;
 410       ptr->top += size;
 411       success = 1;
 412       return tmp;
 413     } else {
 414       ptr = ptr->next;
 415     }
 416   }
 417
 418   if(success == 0) {
 419     return NULL;
 420   }
 421 }
 422
 423 /* This function initiates the prefetch thread A queue is shared
 424  * between the main thread of execution and the prefetch thread to
 425  * process the prefetch call Call from compiler populates the shared
 426  * queue with prefetch requests while prefetch thread processes the
 427  * prefetch requests */
 428
 429 void transInit() {
 430   //Create and initialize prefetch cache structure
 431 #ifdef CACHE
 432   initializePCache();
 433   if((evalPrefetch = initPrefetchStats()) == NULL) {
 434     printf("%s() Error allocating memory at %s, %d\n", __func__, __FILE__, __LINE__);
 435     exit(0);
 436   }
 437 #endif
 438
 439   /* Initialize attributes for mutex */
 440   pthread_mutexattr_init(&prefetchcache_mutex_attr);
 441   pthread_mutexattr_settype(&prefetchcache_mutex_attr, PTHREAD_MUTEX_RECURSIVE_NP);
 442
 443   pthread_mutex_init(&prefetchcache_mutex, &prefetchcache_mutex_attr);
 444   pthread_mutex_init(&notifymutex, NULL);
 445   pthread_mutex_init(&atomicObjLock, NULL);
 446 #ifdef CACHE
 447   //Create prefetch cache lookup table
 448   if(prehashCreate(PHASH_SIZE, PLOADFACTOR)) {
 449     printf("ERROR\n");
 450     return; //Failure
 451   }
 452
 453   //Initialize primary shared queue
 454   queueInit();
 455   //Initialize machine pile w/prefetch oids and offsets shared queue
 456   mcpileqInit();
 457
 458   //Create the primary prefetch thread
 459   int retval;
 460 #ifdef RANGEPREFETCH
 461   do {
 462     retval=pthread_create(&tPrefetch, NULL, transPrefetchNew, NULL);
 463   } while(retval!=0);
 464 #else
 465   do {
 466     retval=pthread_create(&tPrefetch, NULL, transPrefetch, NULL);
 467   } while(retval!=0);
 468 #endif
 469   pthread_detach(tPrefetch);
 470 #endif
 471 }
 472
 473 /* This function stops the threads spawned */
 474 void transExit() {
 475 #ifdef CACHE
 476   int t;
 477   pthread_cancel(tPrefetch);
 478   for(t = 0; t < NUM_THREADS; t++)
 479     pthread_cancel(wthreads[t]);
 480 #endif
 481
 482   return;
 483 }
 484
 485 /* This functions inserts randowm wait delays in the order of msec
 486  * Mostly used when transaction commits retry*/
 487 void randomdelay() {
 488   struct timespec req;
 489   time_t t;
 490
 491   t = time(NULL);
 492   req.tv_sec = 0;
 493   req.tv_nsec = (long)(1000 + (t%10000)); //1-11 microsec
 494   nanosleep(&req, NULL);
 495   return;
 496 }
 497
 498 /* This function initializes things required in the transaction start*/
 499 void transStart() {
 500   t_cache = objstrCreate(1048576);
 501   t_chashCreate(CHASH_SIZE, CLOADFACTOR);
 502   revertlist=NULL;
 503 #ifdef ABORTREADERS
 504   t_abort=0;
 505 #endif
 506 }
 507
 508 // Search for an address for a given oid
 509 /*#define INLINE    inline __attribute__((always_inline))
 510
 511 INLINE void * chashSearchI(chashtable_t *table, unsigned int key) {
 512   //REMOVE HASH FUNCTION CALL TO MAKE SURE IT IS INLINED HERE
 513   chashlistnode_t *node = &table->table[(key & table->mask)>>1];
 514
 515   do {
 516     if(node->key == key) {
 517       return node->val;
 518     }
 519     node = node->next;
 520   } while(node != NULL);
 521
 522   return NULL;
 523   }*/
 524
 525
 526
 527
 528 /* This function finds the location of the objects involved in a transaction
 529  * and returns the pointer to the object if found in a remote location */
 530 __attribute__((pure)) objheader_t *transRead(unsigned int oid) {
 531   unsigned int machinenumber;
 532   objheader_t *tmp, *objheader;
 533   objheader_t *objcopy;
 534   int size;
 535   void *buf;
 536   chashlistnode_t *node;
 537
 538         if(oid == 0) {
 539     return NULL;
 540   }
 541
 542   node= &c_table[(oid & c_mask)>>1];
 543   do {
 544     if(node->key == oid) {
 545 #ifdef TRANSSTATS
 546     nchashSearch++;
 547 #endif
 548 #ifdef COMPILER
 549     return &((objheader_t*)node->val)[1];
 550 #else
 551     return node->val;
 552 #endif
 553     }
 554     node = node->next;
 555   } while(node != NULL);
 556
 557
 558   /*
 559   if((objheader = chashSearchI(record->lookupTable, oid)) != NULL) {
 560 #ifdef TRANSSTATS
 561     nchashSearch++;
 562 #endif
 563 #ifdef COMPILER
 564     return &objheader[1];
 565 #else
 566     return objheader;
 567 #endif
 568   } else
 569   */
 570
 571 #ifdef ABORTREADERS
 572   if (t_abort) {
 573     //abort this transaction
 574     //printf("ABORTING\n");
 575     removetransactionhash();
 576     objstrDelete(t_cache);
 577     t_chashDelete();
 578     _longjmp(aborttrans,1);
 579   } else
 580     addtransaction(oid);
 581 #endif
 582
 583   if ((objheader = (objheader_t *) mhashSearch(oid)) != NULL) {
 584 #ifdef TRANSSTATS
 585     nmhashSearch++;
 586 #endif
 587     /* Look up in machine lookup table  and copy  into cache*/
 588     GETSIZE(size, objheader);
 589     size += sizeof(objheader_t);
 590     objcopy = (objheader_t *) objstrAlloc(&t_cache, size);
 591     memcpy(objcopy, objheader, size);
 592     /* Insert into cache's lookup table */
 593     STATUS(objcopy)=0;
 594     t_chashInsert(OID(objheader), objcopy);
 595 #ifdef COMPILER
 596     return &objcopy[1];
 597 #else
 598     return objcopy;
 599 #endif
 600   } else {
 601 #ifdef CACHE
 602     if((tmp = (objheader_t *) prehashSearch(oid)) != NULL) {
 603 #ifdef TRANSSTATS
 604       nprehashSearch++;
 605 #endif
 606       /* Look up in prefetch cache */
 607       GETSIZE(size, tmp);
 608       size+=sizeof(objheader_t);
 609       objcopy = (objheader_t *) objstrAlloc(&t_cache, size);
 610       memcpy(objcopy, tmp, size);
 611       /* Insert into cache's lookup table */
 612       t_chashInsert(OID(tmp), objcopy);
 613 #ifdef COMPILER
 614       return &objcopy[1];
 615 #else
 616       return objcopy;
 617 #endif
 618     }
 619 #endif
 620     /* Get the object from the remote location */
 621     if((machinenumber = lhashSearch(oid)) == 0) {
 622       printf("Error: %s() No machine found for oid =% %s,%dx\n",__func__, machinenumber, __FILE__, __LINE__);
 623       return NULL;
 624     }
 625     objcopy = getRemoteObj(machinenumber, oid);
 626
 627     if(objcopy == NULL) {
 628       printf("Error: Object not found in Remote location %s, %d\n", __FILE__, __LINE__);
 629       return NULL;
 630     } else {
 631 #ifdef TRANSSTATS
 632       nRemoteSend++;
 633 #endif
 634 #ifdef COMPILER
 635       return &objcopy[1];
 636 #else
 637       return objcopy;
 638 #endif
 639     }
 640   }
 641 }
 642
 643
 644 /* This function finds the location of the objects involved in a transaction
 645  * and returns the pointer to the object if found in a remote location */
 646 __attribute__((pure)) objheader_t *transRead2(unsigned int oid) {
 647   unsigned int machinenumber;
 648   objheader_t *tmp, *objheader;
 649   objheader_t *objcopy;
 650   int size;
 651
 652 #ifdef DEBUG
 653         printf("%s-> Start, oid:%u\n", __func__, oid);
 654 #endif
 655
 656 #ifdef ABORTREADERS
 657   if (t_abort) {
 658     //abort this transaction
 659     //printf("ABORTING\n");
 660     removetransactionhash();
 661     objstrDelete(t_cache);
 662     t_chashDelete();
 663     _longjmp(aborttrans,1);
 664   } else
 665     addtransaction(oid);
 666 #endif
 667
 668   if ((objheader = (objheader_t *) mhashSearch(oid)) != NULL) {
 669 #ifdef DEBUG
 670                 printf("%s-> Grab from this machine\n", __func__);
 671 #endif
 672 #ifdef TRANSSTATS
 673     nmhashSearch++;
 674 #endif
 675     /* Look up in machine lookup table  and copy  into cache*/
 676     GETSIZE(size, objheader);
 677     size += sizeof(objheader_t);
 678     objcopy = (objheader_t *) objstrAlloc(&t_cache, size);
 679     memcpy(objcopy, objheader, size);
 680     /* Insert into cache's lookup table */
 681     STATUS(objcopy)=0;
 682     t_chashInsert(OID(objheader), objcopy);
 683 #ifdef COMPILER
 684     return &objcopy[1];
 685 #else
 686     return objcopy;
 687 #endif
 688   } else {
 689 #ifdef CACHE
 690     , TYPE(header)if((tmp = (objheader_t *) prehashSearch(oid)) != NULL) {
 691 #ifdef TRANSSTATS
 692       nprehashSearch++;
 693 #endif
 694       /* Look up in prefetch cache */
 695       GETSIZE(size, tmp);
 696       size+=sizeof(objheader_t);
 697       objcopy = (objheader_t *) objstrAlloc(&t_cache, size);
 698       memcpy(objcopy, tmp, size);
 699       /* Insert into cache's lookup table */
 700       t_chashInsert(OID(tmp), objcopy);
 701 #ifdef COMPILER
 702       return &objcopy[1];
 703 #else
 704                         return objcopy;
 705 #endif
 706                 }
 707 #endif
 708                 /* Get the object from the remote location */
 709 #ifdef DEBUG
 710                         printf("%s-> Grab from remote machine\n", __func__);
 711 #endif
 712 #ifdef RECOVERY
 713                 //while(!liveHostsValid) {
 714                 //}
 715                 /*if(!liveHostsValid){
 716                         sleep(WAIT_TIME);
 717                 }*/
 718                 unsigned int mindex = findHost(lhashSearch(oid));
 719                 machinenumber = locateObjHosts[2*mindex+flipBit];
 720                 flipBit ^= 1;
 721                 printf("mindex:%d, oid:%d, machinenumber:%s\n", mindex, oid, midtoIPString(machinenumber));
 722 #else
 723                 if((machinenumber = lhashSearch(oid)) == 0) {
 724                         printf("Error: %s() No machine found for oid =% %s,%dx\n",__func__, machinenumber, __FILE__, __LINE__);
 725                         return NULL;
 726                 }
 727 #endif
 728                 objcopy = getRemoteObj(machinenumber, oid);
 729
 730                 if(objcopy == NULL) {
 731                         printf("Error: Object not found in Remote location %s, %d\n", __FILE__, __LINE__);
 732                         return NULL;
 733                 } else {
 734 #ifdef TRANSSTATS
 735                         nRemoteSend++;
 736 #endif
 737 #ifdef COMPILER
 738                         return &objcopy[1];
 739 #else
 740                         return objcopy;
 741 #endif
 742                 }
 743   }
 744 }
 745
 746 /* This function creates objects in the transaction record */
 747 objheader_t *transCreateObj(unsigned int size) {
 748   objheader_t *tmp = (objheader_t *) objstrAlloc(&t_cache, (sizeof(objheader_t) + size));
 749   OID(tmp) = getNewOID();
 750   tmp->version = 1;
 751   tmp->rcount = 1;
 752         tmp->isBackup = 0;
 753   STATUS(tmp) = NEW;
 754   t_chashInsert(OID(tmp), tmp);
 755
 756 #ifdef COMPILER
 757   return &tmp[1]; //want space after object header
 758 #else
 759   return tmp;
 760 #endif
 761 }
 762
 763
 764 #if 1
 765 /* This function creates machine piles based on all machines involved in a
 766  * transaction commit request */
 767 plistnode_t *createPiles() {
 768   int i;
 769         unsigned int oid;
 770   plistnode_t *pile = NULL;
 771   unsigned int machinenum;
 772         unsigned int destMachine[2];
 773   objheader_t *headeraddr;
 774   chashlistnode_t * ptr = c_table;
 775   /* Represents number of bins in the chash table */
 776   unsigned int size = c_size;
 777
 778         for(i = 0; i < size ; i++) {
 779                 chashlistnode_t * curr = &ptr[i];
 780                 /* Inner loop to traverse the linked list of the cache lookupTable */
 781                 while(curr != NULL) {
 782                         //if the first bin in hash table is empty
 783                         if(curr->key == 0)
 784                                 break;
 785                         headeraddr=(objheader_t *) curr->val;
 786
 787 #if RECOVERY
 788                         oid = OID(headeraddr);
 789 #ifdef DEBUG
 790                         printf("%s-> oid:%u, version:%d, status:%d, type:%d\n", __func__, OID(headeraddr), headeraddr->version, STATUS(headeraddr), TYPE(headeraddr));
 791
 792                         if (STATUS(headeraddr) & NEW) {  // new/local object
 793                                 printf("%s-> new/local object\n", __func__);
 794                         }
 795                         else if ((mhashSearch(curr->key) != NULL)) {    //local/nonnew
 796                                 if(STATUS(headeraddr) & DIRTY) {        // modified
 797                                         printf("%s-> old/local/mod object\n", __func__);
 798                                 }
 799                                 else {  //read
 800                                         printf("%s-> old/local/read object\n", __func__);
 801                                 }
 802                         } else if ((machinenum = lhashSearch(curr->key)) != 0) { // remote/nonnew object
 803                                 if(STATUS(headeraddr) & DIRTY) {                //modified
 804                                         printf("%s-> remote/local/mod object\n", __func__);
 805                                 }
 806                                 else {  //read
 807                                         printf("%s-> remote/local/read object\n", __func__);
 808                                 }
 809                         } else {
 810                                 printf("Error: No such machine %s, %d\n", __FILE__, __LINE__);
 811                                 return NULL;
 812                         }
 813                         unsigned int pmid = getPrimaryMachine(lhashSearch(oid));
 814                         unsigned int bmid = getBackupMachine(lhashSearch(oid));
 815                         printf("%s-> Primary Machine: [%s], ", __func__, midtoIPString(pmid));
 816                         printf("Backup Machine: [%s]\n", midtoIPString(bmid));
 817 #endif
 818                         int makedirty = 0;
 819                         if(STATUS(headeraddr) & DIRTY || STATUS(headeraddr) & NEW) {
 820                                 makedirty = 1;
 821                         }
 822                         pile = pInsert(pile, headeraddr, getPrimaryMachine(lhashSearch(oid)), c_numelements);
 823 //problem here
 824                         if(makedirty) {
 825                                 STATUS(headeraddr) = DIRTY;
 826                         }
 827                         pile = pInsert(pile, headeraddr, getBackupMachine(lhashSearch(oid)), c_numelements);
 828 #else
 829                         // Get machine location for object id (and whether local or not)
 830                         if (STATUS(headeraddr) & NEW || (mhashSearch(curr->key) != NULL)) {
 831                                 machinenum = myIpAddr;
 832                         } else if ((machinenum = lhashSearch(curr->key)) == 0) {
 833                                         printf("Error: No such machine %s, %d\n", __FILE__, __LINE__);
 834                                         return NULL;
 835                                 }
 836
 837                         //Make machine groups
 838                         pile = pInsert(pile, headeraddr, machinenum, c_numelements);
 839 #endif
 840                         curr = curr->next;
 841                 }
 842         }
 843         return pile;
 844 }
 845 #else
 846 /* This function creates machine piles based on all machines involved in a
 847  * transaction commit request */
 848 plistnode_t *createPiles() {
 849   int i;
 850   plistnode_t *pile = NULL;
 851   unsigned int machinenum;
 852         unsigned int destMachine[2];
 853   objheader_t *headeraddr;
 854   struct chashentry * ptr = c_table;
 855   /* Represents number of bins in the chash table */
 856   unsigned int size = c_size;
 857
 858   for(i = 0; i < size ; i++) {
 859     struct chashentry * curr = & ptr[i];
 860     /* Inner loop to traverse the linked list of the cache lookupTable */
 861     // if the first bin in hash table is empty
 862     if(curr->key == 0)
 863       continue;
 864     headeraddr=(objheader_t *) curr->ptr;
 865
 866     //Get machine location for object id (and whether local or not)
 867     if (STATUS(headeraddr) & NEW || (mhashSearch(curr->key) != NULL)) {
 868       machinenum = myIpAddr;
 869     } else if ((machinenum = lhashSearch(curr->key)) == 0) {
 870       printf("Error: No such machine %s, %d\n", __FILE__, __LINE__);
 871       return NULL;
 872     }
 873
 874     //Make machine groups
 875     pile = pInsert(pile, headeraddr, machinenum, c_numelements);
 876   }
 877   return pile;
 878 }
 879 #endif
 880
 881 /* This function initiates the transaction commit process
 882  * Spawns threads for each of the new connections with Participants
 883  * and creates new piles by calling the createPiles(),
 884  * Sends a transrequest() to each remote machines for objects found remotely
 885  * and calls handleLocalReq() to process objects found locally */
 886 int transCommit() {
 887   unsigned int tot_bytes_mod, *listmid;
 888   plistnode_t *pile, *pile_ptr;
 889   int trecvcount;
 890   char treplyretry; /* keeps track of the common response that needs to be sent */
 891   int firsttime=1;
 892   trans_commit_data_t transinfo; /* keeps track of objs locked during transaction */
 893   char finalResponse;
 894
 895         int tmpTransIndex = (transIDIndex++)%25;
 896         liveTransactions[tmpTransIndex] = getNewTransID();
 897
 898 #ifdef DEBUG
 899         printf("%s-> Start, transID:%d\n", __func__, liveTransactions[tmpTransIndex]);
 900 #endif
 901
 902 #ifdef ABORTREADERS
 903   if (t_abort) {
 904     //abort this transaction
 905     /* Debug
 906      * printf("ABORTING TRANSACTION AT COMMIT\n");
 907      */
 908     removetransactionhash();
 909     objstrDelete(t_cache);
 910     t_chashDelete();
 911 #ifdef DEBUG
 912           printf("%s-> End, line:%d\n\n", __func__, __LINE__);
 913 #endif
 914 #ifdef RECOVERY
 915           liveTransactions[tmpTransIndex] = 0;
 916 #endif
 917     return 1;
 918   }
 919 #endif
 920
 921
 922   do {
 923     trecvcount = 0;
 924     treplyretry = 0;
 925
 926     /* Look through all the objects in the transaction record and make piles
 927      * for each machine involved in the transaction*/
 928     if (firsttime) {
 929       pile_ptr = pile = createPiles();
 930       pile_ptr = pile = sortPiles(pile);
 931     } else {
 932       pile = pile_ptr;
 933     }
 934     firsttime = 0;
 935     /* Create the packet to be sent in TRANS_REQUEST */
 936
 937     /* Count the number of participants */
 938     int pilecount;
 939     pilecount = pCount(pile);
 940
 941     /* Create a list of machine ids(Participants) involved in transaction   */
 942     listmid = calloc(pilecount, sizeof(unsigned int));
 943     pListMid(pile, listmid);
 944
 945     /* Create a socket and getReplyCtrl array, initialize */
 946     int socklist[pilecount];
 947     int loopcount;
 948     for(loopcount = 0 ; loopcount < pilecount; loopcount++)
 949       socklist[loopcount] = 0;
 950     char getReplyCtrl[pilecount];
 951     for(loopcount = 0 ; loopcount < pilecount; loopcount++)
 952       getReplyCtrl[loopcount] = 0;
 953
 954     /* Process each machine pile */
 955     int sockindex = 0;
 956                 int localReqsock = -1;
 957     trans_req_data_t *tosend;
 958     tosend = calloc(pilecount, sizeof(trans_req_data_t));
 959     while(pile != NULL) {
 960 #ifdef DEBUG
 961                         printf("%s-> New pile:[%s],", __func__, midtoIPString(pile->mid));
 962                         printf(" myIp:[%s]\n", midtoIPString(myIpAddr));
 963 #endif
 964       tosend[sockindex].f.control = TRANS_REQUEST;
 965                         tosend[sockindex].f.mcount = pilecount;
 966                         tosend[sockindex].f.numread = pile->numread;
 967                         tosend[sockindex].f.nummod = pile->nummod;
 968                         tosend[sockindex].f.numcreated = pile->numcreated;
 969 #ifdef DEBUG
 970                         printf("%s-> numread:%d, nummod:%d, numcreated:%d\n", __func__, pile->numread, pile->nummod, pile->numcreated);
 971 #endif
 972                         tosend[sockindex].f.sum_bytes = pile->sum_bytes;
 973                         tosend[sockindex].listmid = listmid;
 974                         tosend[sockindex].objread = pile->objread;
 975                         tosend[sockindex].oidmod = pile->oidmod;
 976                         tosend[sockindex].oidcreated = pile->oidcreated;
 977                         int sd = 0;
 978                         if(pile->mid != myIpAddr) {
 979 #ifdef RECOVERY
 980                                 if((sd = getSockWithLock(transRequestSockPool, pile->mid)) < 0) {
 981 #else
 982                                 if((sd = getSock2WithLock(transRequestSockPool, pile->mid)) < 0) {
 983 #endif
 984                                         printf("\ntransRequest(): socket create error\n");
 985                                         free(listmid);
 986                                         free(tosend);
 987 #ifdef DEBUG
 988                                         printf("%s-> End, line:%d\n\n", __func__, __LINE__);
 989 #endif
 990 #ifdef RECOVERY
 991                                         liveTransactions[tmpTransIndex] = 0;
 992 #endif
 993                                         return 1;
 994                                 }
 995                                 socklist[sockindex] = sd;
 996                                 /* Send bytes of data with TRANS_REQUEST control message */
 997                                 send_data(sd, &(tosend[sockindex].f), sizeof(fixed_data_t));
 998                                 /*if(timeoutFlag) {
 999                                                 printf("send_data: remote machine dead, line:%d\n", __LINE__);
1000                                                 timeoutFlag = 0;
1001                                                 exit(1);
1002                                         }*/
1003                                 /* Send list of machines involved in the transaction */
1004                                 {
1005                                         int size=sizeof(unsigned int)*(tosend[sockindex].f.mcount);
1006                                         send_data(sd, tosend[sockindex].listmid, size);
1007                                 }
1008
1009                                 /* Send oids and version number tuples for objects that are read */
1010                                 {
1011                                         int size=(sizeof(unsigned int)+sizeof(unsigned short))*(tosend[sockindex].f.numread);
1012                                         send_data(sd, tosend[sockindex].objread, size);
1013                                 }
1014
1015                                 /* Send objects that are modified */
1016                                 void *modptr;
1017                                 if((modptr = calloc(1, tosend[sockindex].f.sum_bytes)) == NULL) {
1018                                         printf("Calloc error for modified objects %s, %d\n", __FILE__, __LINE__);
1019                                         free(listmid);
1020                                         free(tosend);
1021 #ifdef DEBUG
1022                                         printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1023 #endif
1024 #ifdef RECOVERY
1025                                         liveTransactions[tmpTransIndex] = 0;
1026 #endif
1027                                         return 1;
1028                                 }
1029                                 int offset = 0;
1030                                 int i;
1031                                 for(i = 0; i < tosend[sockindex].f.nummod ; i++) {
1032                                         int size;
1033                                         objheader_t *headeraddr;
1034                                         if((headeraddr = t_chashSearch(tosend[sockindex].oidmod[i])) == NULL) {
1035                                                 printf("%s() Error: No such oid %s, %d\n", __func__, __FILE__, __LINE__);
1036                                                 free(modptr);
1037                                                 free(listmid);
1038                                                 free(tosend);
1039 #ifdef DEBUG
1040                                                 printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1041 #endif
1042 #ifdef RECOVERY
1043                                                 liveTransactions[tmpTransIndex] = 0;
1044 #endif
1045                                                 return 1;
1046                                         }
1047                                         GETSIZE(size,headeraddr);
1048                                         size+=sizeof(objheader_t);
1049                                         memcpy(modptr+offset, headeraddr, size);
1050                                         offset+=size;
1051                                 }
1052                                 send_data(sd, modptr, tosend[sockindex].f.sum_bytes);
1053                                 free(modptr);
1054                         } else { //handle request locally
1055                                 localReqsock = sockindex;
1056                                 handleLocalReq(&tosend[sockindex], &transinfo, &getReplyCtrl[sockindex]);
1057                         }
1058                         sockindex++;
1059                         pile = pile->next;
1060                 } //end of pile processing
1061
1062                 /* Recv Ctrl msgs from all machines */
1063 #ifdef DEBUG
1064                 printf("%s-> Finished sending transaction read/mod objects\n",__func__);
1065 #endif
1066                 int i;
1067                 for(i = 0; i < pilecount; i++) {
1068                         printf("i:%d\n", i);
1069                         if(i == localReqsock)
1070                                 continue;
1071                         int sd = socklist[i];
1072                         if(sd != 0) {
1073                                 char control;
1074                                 recv_data(sd, &control, sizeof(char));
1075                                 /*if(timeoutFlag) {
1076                                         printf("recv_data: remote machine dead, timeoutFlag:%d, timeoutFlag:%d, line:%d\n", timeoutFlag, timeoutFlag, __LINE__);
1077                                         timeoutFlag = 0;
1078                                         exit(1);
1079                                 }*/
1080                                 //Update common data structure with new ctrl msg
1081                                 getReplyCtrl[i] = control;
1082                                 /* Recv Objects if participant sends TRANS_DISAGREE */
1083                                 //printf("getReplyCtrl[%d] = %d\n", i, (int)getReplyCtrl[i]);
1084 #ifdef CACHE
1085                                 if(control == TRANS_DISAGREE) {
1086                                         int length;
1087                                         recv_data(sd, &length, sizeof(int));
1088                                         void *newAddr;
1089                                         pthread_mutex_lock(&prefetchcache_mutex);
1090                                         if ((newAddr = prefetchobjstrAlloc((unsigned int)length)) == NULL) {
1091                                                 printf("Error: %s() objstrAlloc error for copying into prefetch cache %s, %d\n", __func__, __FILE__, __LINE__);
1092                                                 free(tosend);
1093                                                 free(listmid);
1094                                                 pthread_mutex_unlock(&prefetchcache_mutex);
1095 #ifdef DEBUG
1096                                                 printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1097 #endif
1098 #ifdef RECOVERY
1099                                                 liveTransactions[tmpTransIndex] = 0;
1100 #endif
1101                                                 return 1;
1102                                         }
1103                                         pthread_mutex_unlock(&prefetchcache_mutex);
1104                                         recv_data(sd, newAddr, length);
1105                                         int offset = 0;
1106                                         while(length != 0) {
1107                                                 unsigned int oidToPrefetch;
1108                                                 objheader_t * header;
1109                                                 header = (objheader_t *)(((char *)newAddr) + offset);
1110                                                 oidToPrefetch = OID(header);
1111                                                 STATUS(header)=0;
1112                                                 int size = 0;
1113                                                 GETSIZE(size, header);
1114                                                 size += sizeof(objheader_t);
1115                                                 //make an entry in prefetch hash table
1116                                                 void *oldptr;
1117                                                 if((oldptr = prehashSearch(oidToPrefetch)) != NULL) {
1118                                                         prehashRemove(oidToPrefetch);
1119                                                         prehashInsert(oidToPrefetch, header);
1120                                                 } else {
1121                                                         prehashInsert(oidToPrefetch, header);
1122                                                 }
1123                                                 length = length - size;
1124                                                 offset += size;
1125                                         }
1126                                 } //end of receiving objs
1127 #endif
1128                         }
1129                 }
1130 #ifdef DEBUG
1131                 printf("%s-> Decide final response now\n", __func__);
1132 #endif
1133                 /* Decide the final response */
1134                 if((finalResponse = decideResponse(getReplyCtrl, &treplyretry, pilecount)) == 0) {
1135                         printf("Error: %s() in updating prefetch cache %s, %d\n", __func__, __FILE__, __LINE__);
1136                         free(tosend);
1137                         free(listmid);
1138 #ifdef DEBUG
1139                         printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1140 #endif
1141 #ifdef RECOVERY
1142                         liveTransactions[tmpTransIndex] = 0;
1143 #endif
1144                         return 1;
1145                 }
1146 #ifdef DEBUG
1147     printf("%s-> Final Response: %d\n", __func__, (int)finalResponse);
1148 #endif
1149                 /* Send responses to all machines */
1150                 for(i = 0; i < pilecount; i++) {
1151                         int sd = socklist[i];
1152                         if(sd != 0) {
1153 #ifdef CACHE
1154                                 if(finalResponse == TRANS_COMMIT) {
1155                                         int retval;
1156                                         /* Update prefetch cache */
1157                                         if((retval = updatePrefetchCache(&(tosend[i]))) != 0) {
1158                                                 printf("Error: %s() in updating prefetch cache %s, %d\n", __func__, __FILE__, __LINE__);
1159                                                 free(tosend);
1160                                                 free(listmid);
1161 #ifdef DEBUG
1162                                                 printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1163 #endif
1164 #ifdef RECOVERY
1165                                                 liveTransactions[tmpTransIndex] = 0;
1166 #endif
1167                                                 return 1;
1168                                         }
1169
1170
1171                                         /* Invalidate objects in other machine cache */
1172                                         if(tosend[i].f.nummod > 0) {
1173                                                 if((retval = invalidateObj(&(tosend[i]))) != 0) {
1174                                                         printf("Error: %s() in invalidating Objects %s, %d\n", __func__, __FILE__, __LINE__);
1175                                                         free(tosend);
1176                                                         free(listmid);
1177 #ifdef DEBUG
1178                                                         printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1179 #endif
1180 #ifdef RECOVERY
1181                                                         liveTransactions[tmpTransIndex] = 0;
1182 #endif
1183                                                         return 1;
1184                                                 }
1185                                         }
1186 #ifdef ABORTREADERS
1187                                         removetransaction(tosend[i].oidmod,tosend[i].f.nummod);
1188                                         removethisreadtransaction(tosend[i].objread, tosend[i].f.numread);
1189 #endif
1190                                 }
1191 #ifdef ABORTREADERS
1192                                 else if (!treplyretry) {
1193                                         removethistransaction(tosend[i].oidmod,tosend[i].f.nummod);
1194                                         removethisreadtransaction(tosend[i].objread,tosend[i].f.numread);
1195                                 }
1196 #endif
1197 #endif
1198                                 send_data(sd, &finalResponse, sizeof(char));
1199                         } else {
1200                                 /* Complete local processing */
1201                                 doLocalProcess(finalResponse, &(tosend[i]), &transinfo);
1202 #ifdef ABORTREADERS
1203                                 if(finalResponse == TRANS_COMMIT) {
1204                                         removetransaction(tosend[i].oidmod,tosend[i].f.nummod);
1205                                         removethisreadtransaction(tosend[i].objread,tosend[i].f.numread);
1206                                 } else if (!treplyretry) {
1207                                         removethistransaction(tosend[i].oidmod,tosend[i].f.nummod);
1208                                         removethisreadtransaction(tosend[i].objread,tosend[i].f.numread);
1209                                 }
1210 #endif
1211                         }
1212                 }
1213
1214 #ifdef RECOVERY
1215 #ifdef DEBUG
1216                 printf("%s-> Free sockets\n", __func__);
1217 #endif
1218                 for(i = 0; i < pilecount; i++) {
1219                         if(socklist[i] != 0) {
1220                           freeSockWithLock(transRequestSockPool, listmid[i], socklist[i]);
1221                         }
1222                 }
1223 #endif
1224                         /* Free resources */
1225     free(tosend);
1226     free(listmid);
1227     if (!treplyretry)
1228       pDelete(pile_ptr);
1229     /* wait a random amount of time before retrying to commit transaction*/
1230     if(treplyretry) {
1231       randomdelay();
1232 #ifdef TRANSSTATS
1233                         nSoftAbort++;
1234 #endif
1235                 }
1236                 /* Retry trans commit procedure during soft_abort case */
1237         } while (treplyretry);
1238
1239         if(finalResponse == TRANS_ABORT) {
1240                 //printf("Aborting trans\n");
1241 #ifdef TRANSSTATS
1242                 numTransAbort++;
1243 #endif
1244     /* Free Resources */
1245     objstrDelete(t_cache);
1246     t_chashDelete();
1247 #ifdef DEBUG
1248           printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1249 #endif
1250 #ifdef RECOVERY
1251           liveTransactions[tmpTransIndex] = 0;
1252 #endif
1253     return TRANS_ABORT;
1254   } else if(finalResponse == TRANS_COMMIT) {
1255 #ifdef TRANSSTATS
1256                 numTransCommit++;
1257 #endif
1258     /* Free Resources */
1259     objstrDelete(t_cache);
1260     t_chashDelete();
1261 #ifdef DEBUG
1262                                         printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1263 #endif
1264 #ifdef RECOVERY
1265                                         liveTransactions[tmpTransIndex] = 0;
1266 #endif
1267     return 0;
1268   } else {
1269     //TODO Add other cases
1270     printf("Error: in %s() THIS SHOULD NOT HAPPEN.....EXIT PROGRAM\n", __func__);
1271 #ifdef DEBUG
1272         printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1273 #endif
1274 #ifdef RECOVERY
1275         liveTransactions[tmpTransIndex] = 0;
1276 #endif
1277     exit(-1);
1278   }
1279 #ifdef DEBUG
1280         printf("%s-> End, line:%d\n\n", __func__, __LINE__);
1281 #endif
1282 #ifdef RECOVERY
1283         liveTransactions[tmpTransIndex] = 0;
1284 #endif
1285   return 0;
1286 }
1287
1288 /* This function handles the local objects involved in a transaction
1289  * commiting process.  It also makes a decision if this local machine
1290  * sends AGREE or DISAGREE or SOFT_ABORT to coordinator */
1291 void handleLocalReq(trans_req_data_t *tdata, trans_commit_data_t *transinfo, char *getReplyCtrl) {
1292   unsigned int *oidnotfound = NULL, *oidlocked = NULL;
1293   int numoidnotfound = 0, numoidlocked = 0;
1294   int v_nomatch = 0, v_matchlock = 0, v_matchnolock = 0;
1295   int numread, i;
1296   unsigned int oid;
1297   unsigned short version;
1298
1299   /* Counters and arrays to formulate decision on control message to be sent */
1300   oidnotfound = (unsigned int *) calloc((tdata->f.numread + tdata->f.nummod), sizeof(unsigned int));
1301         oidlocked = (unsigned int *) calloc((tdata->f.numread + tdata->f.nummod +1), sizeof(unsigned int)); // calloc additional 1 byte for
1302         //setting a divider between read and write locks
1303         numread = tdata->f.numread;
1304         /* Process each oid in the machine pile/ group per thread */
1305         for (i = 0; i < tdata->f.numread + tdata->f.nummod; i++) {
1306                 if (i < tdata->f.numread) {
1307                         int incr = sizeof(unsigned int) + sizeof(unsigned short); // Offset that points to next position in the objread array
1308                         incr *= i;
1309                         oid = *((unsigned int *)(((char *)tdata->objread) + incr));
1310                         version = *((unsigned short *)(((char *)tdata->objread) + incr + sizeof(unsigned int)));
1311                         commitCountForObjRead(getReplyCtrl, oidnotfound, oidlocked, &numoidnotfound, &numoidlocked, &v_nomatch, &v_matchlock, &v_matchnolock, oid, version);
1312                 } else { // Objects Modified
1313                         if(i == tdata->f.numread) {
1314                                 oidlocked[numoidlocked++] = -1;
1315                         }
1316                         int tmpsize;
1317                         objheader_t *headptr;
1318                         headptr = (objheader_t *) t_chashSearch(tdata->oidmod[i-numread]);
1319                         if (headptr == NULL) {
1320                                 printf("Error: handleLocalReq() returning NULL, no such oid %s, %d\n", __FILE__, __LINE__);
1321                                 return;
1322                         }
1323                         oid = OID(headptr);
1324                         version = headptr->version;
1325                         commitCountForObjMod(getReplyCtrl, oidnotfound, oidlocked, &numoidnotfound, &numoidlocked, &v_nomatch, &v_matchlock, &v_matchnolock, oid, version);
1326                 }
1327   }
1328
1329         /* Fill out the trans_commit_data_t data structure. This is required for a trans commit process
1330          * if Participant receives a TRANS_COMMIT */
1331         transinfo->objlocked = oidlocked;
1332         transinfo->objnotfound = oidnotfound;
1333         transinfo->modptr = NULL;
1334         transinfo->numlocked = numoidlocked;
1335         transinfo->numnotfound = numoidnotfound;
1336
1337   /* Condition to send TRANS_AGREE */
1338   if(v_matchnolock == tdata->f.numread + tdata->f.nummod) {
1339     *getReplyCtrl = TRANS_AGREE;
1340   }
1341   /* Condition to send TRANS_SOFT_ABORT */
1342   if((v_matchlock > 0 && v_nomatch == 0) || (numoidnotfound > 0 && v_nomatch == 0)) {
1343     *getReplyCtrl = TRANS_SOFT_ABORT;
1344   }
1345 }
1346
1347 void doLocalProcess(char finalResponse, trans_req_data_t *tdata, trans_commit_data_t *transinfo) {
1348   if(finalResponse == TRANS_ABORT) {
1349     if(transAbortProcess(transinfo) != 0) {
1350       printf("Error in transAbortProcess() %s,%d\n", __FILE__, __LINE__);
1351       fflush(stdout);
1352       return;
1353     }
1354   } else if(finalResponse == TRANS_COMMIT) {
1355 #ifdef CACHE
1356     /* Invalidate objects in other machine cache */
1357     if(tdata->f.nummod > 0) {
1358       int retval;
1359       if((retval = invalidateObj(tdata)) != 0) {
1360         printf("Error: %s() in invalidating Objects %s, %d\n", __func__, __FILE__, __LINE__);
1361         return;
1362       }
1363     }
1364 #endif
1365     if(transComProcess(tdata, transinfo) != 0) {
1366       printf("Error in transComProcess() %s,%d\n", __FILE__, __LINE__);
1367       fflush(stdout);
1368       return;
1369     }
1370   } else {
1371     printf("ERROR...No Decision\n");
1372   }
1373
1374   /* Free memory */
1375   if (transinfo->objlocked != NULL) {
1376     free(transinfo->objlocked);
1377   }
1378   if (transinfo->objnotfound != NULL) {
1379     free(transinfo->objnotfound);
1380   }
1381 }
1382
1383 /* This function decides the reponse that needs to be sent to
1384  * all Participant machines after the TRANS_REQUEST protocol */
1385 char decideResponse(char *getReplyCtrl, char *treplyretry, int pilecount) {
1386   int i, transagree = 0, transdisagree = 0, transsoftabort = 0; /* Counters to formulate decision of what
1387                                                                    message to send */
1388   for (i = 0 ; i < pilecount; i++) {
1389     char control;
1390     control = getReplyCtrl[i];
1391     switch(control) {
1392     default:
1393 #ifdef DEBUG
1394       printf("%s-> Participant sent unknown message, i:%d, Control: %d\n", __func__, i, (int)control);
1395 #endif
1396
1397       /* treat as disagree, pass thru */
1398     case TRANS_DISAGREE:
1399       transdisagree++;
1400 #ifdef DEBUG
1401       printf("%s-> Participant sent TRANS_DISAGREE, i:%d, Control: %d\n", __func__, i, (int)control);
1402 #endif
1403       break;
1404
1405     case TRANS_AGREE:
1406       transagree++;
1407 #ifdef DEBUG
1408       printf("%s-> Participant sent TRANS_AGREE, i:%d, Control: %d\n", __func__, i, (int)control);
1409 #endif
1410       break;
1411
1412     case TRANS_SOFT_ABORT:
1413       transsoftabort++;
1414 #ifdef DEBUG
1415       printf("%s-> Participant sent TRANS_SOFT_ABORT, i:%d, Control: %d\n", __func__, i, (int)control);
1416 #endif
1417       break;
1418     }
1419   }
1420
1421   if(transdisagree > 0) {
1422     /* Send Abort */
1423     *treplyretry = 0;
1424     return TRANS_ABORT;
1425 #ifdef CACHE
1426     /* clear objects from prefetch cache */
1427     cleanPCache();
1428 #endif
1429   } else if(transagree == pilecount) {
1430     /* Send Commit */
1431     *treplyretry = 0;
1432     return TRANS_COMMIT;
1433   } else {
1434     /* Send Abort in soft abort case followed by retry commiting transaction again*/
1435     *treplyretry = 1;
1436     return TRANS_ABORT;
1437   }
1438   return 0;
1439 }
1440
1441 /* This function opens a connection, places an object read request to
1442  * the remote machine, reads the control message and object if
1443  * available and copies the object and its header to the local
1444  * cache. */
1445
1446 void *getRemoteObj(unsigned int mnum, unsigned int oid) {
1447   int size, val;
1448   struct sockaddr_in serv_addr;
1449   char machineip[16];
1450   char control = 0;
1451   objheader_t *h;
1452   void *objcopy = NULL;
1453
1454   int sd = getSock2(transReadSockPool, mnum);
1455   char readrequest[sizeof(char)+sizeof(unsigned int)];
1456   readrequest[0] = READ_REQUEST;
1457   *((unsigned int *)(&readrequest[1])) = oid;
1458   send_data(sd, readrequest, sizeof(readrequest));
1459
1460   /* Read response from the Participant */
1461   recv_data(sd, &control, sizeof(char));
1462
1463   if (control==OBJECT_NOT_FOUND) {
1464     objcopy = NULL;
1465   } else if(control==OBJECT_FOUND) {
1466     /* Read object if found into local cache */
1467     recv_data(sd, &size, sizeof(int));
1468  objcopy = objstrAlloc(&t_cache, size);
1469     recv_data(sd, objcopy, size);
1470     STATUS(objcopy)=0;
1471     /* Insert into cache's lookup table */
1472     t_chashInsert(oid, objcopy);
1473 #ifdef TRANSSTATS
1474     totalObjSize += size;
1475 #endif
1476         }
1477
1478 #ifdef RECOVERY
1479         if( detectMachineFailure(mnum) ) { //check for timeouts
1480                 printf("looking for oid:%d\n", oid);
1481                 restoreDuplicationState(mnum);  // suspect machine failure, restore state
1482
1483                 objheader_t *temp;
1484                 temp = transRead2(oid);         // retry transRead
1485 #ifdef COMPILER
1486                 temp -= 1;              // return object w/ objheader
1487 #endif
1488                 return (void *)temp;
1489         }
1490 #endif
1491         return objcopy;
1492 }
1493
1494 int detectMachineFailure(unsigned int mid) {
1495         if(timeoutFlag == 1) {
1496 #ifdef DEBUG
1497                 printf("%s-> Suspect machine failure: [%s]\n", __func__, midtoIPString(mid));
1498 #endif
1499                 timeoutFlag = 0;
1500                 return 1;
1501         }
1502         else
1503                 return 0;
1504 }
1505
1506 void restoreDuplicationState(unsigned int deadHost) {
1507         int sd;
1508         char ctrl;
1509
1510         if(!liveHosts[findHost(deadHost)]) {
1511                 sleep(WAIT_TIME);
1512                 return;
1513         }
1514         if(deadHost == leader)
1515                 paxos();
1516
1517 #ifdef DEBUG
1518         printf("%s-> leader?:%s, me?:%d\n", __func__, midtoIPString(leader), (myIpAddr == leader));
1519 #endif
1520
1521         if(leader == myIpAddr) {
1522                 pthread_mutex_lock(&leaderFixing_mutex);
1523                 if(!leaderFixing) {
1524                         leaderFixing = 1;
1525                         pthread_mutex_unlock(&leaderFixing_mutex);
1526                         //fixit
1527                         updateLiveHosts();
1528
1529                         if(!liveHosts[findHost(deadHost)]) {    //confirmed dead
1530                                 duplicateLostObjects(deadHost);
1531                         }
1532                         if(updateLiveHostsCommit() != 0) {
1533                                 printf("error updateLiveHostsCommit()\n");
1534                                 exit(1);
1535                         }
1536                 pthread_mutex_lock(&leaderFixing_mutex);
1537                         leaderFixing = 0;
1538                         pthread_mutex_unlock(&leaderFixing_mutex);
1539                 }
1540                 else {
1541                         pthread_mutex_unlock(&leaderFixing_mutex);
1542                         sleep(WAIT_TIME);
1543                         //while(leaderFixing);
1544                         return;
1545                 }
1546         }
1547         else {
1548                 if((sd = getSock2WithLock(transRequestSockPool, leader)) < 0) {
1549                         printf("restoreDuplicationState(): socket create error\n");
1550                         exit(-1);
1551                 }
1552                 ctrl = REMOTE_RESTORE_DUPLICATED_STATE;
1553                 send_data(sd, &ctrl, sizeof(char));
1554                 send_data(sd, &deadHost, sizeof(unsigned int));
1555                 recv_data(sd, &ctrl, sizeof(char));
1556           sleep(WAIT_TIME);
1557                 return;
1558         }
1559 }
1560
1561 /*  Commit info for objects modified */
1562 void commitCountForObjMod(char *getReplyCtrl, unsigned int *oidnotfound, unsigned int *oidlocked, int *numoidnotfound,
1563                           int *numoidlocked, int *v_nomatch, int *v_matchlock, int *v_matchnolock, unsigned int oid, unsigned short version) {
1564   void *mobj;
1565   /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
1566   /* Save the oids not found and number of oids not found for later use */
1567   if ((mobj = mhashSearch(oid)) == NULL) { /* Obj not found */
1568     /* Save the oids not found and number of oids not found for later use */
1569     oidnotfound[*numoidnotfound] = oid;
1570     (*numoidnotfound)++;
1571   } else { /* If Obj found in machine (i.e. has not moved) */
1572     /* Check if Obj is locked by any previous transaction */
1573     if (write_trylock(STATUSPTR(mobj))) { // Can acquire write lock
1574       if (version == ((objheader_t *)mobj)->version) {      /* match versions */
1575         (*v_matchnolock)++;
1576         //Keep track of what is locked
1577         oidlocked[(*numoidlocked)++] = OID(((objheader_t *)mobj));
1578       } else { /* If versions don't match ...HARD ABORT */
1579         (*v_nomatch)++;
1580         /* Send TRANS_DISAGREE to Coordinator */
1581         *getReplyCtrl = TRANS_DISAGREE;
1582
1583         //Keep track of what is locked
1584         oidlocked[(*numoidlocked)++] = OID(((objheader_t *)mobj));
1585         //printf("%s() oid = %d, type = %d\t", __func__, OID(mobj), TYPE((objheader_t *)mobj));
1586         return;
1587       }
1588     } else { //A lock is acquired some place else
1589                       if (version == ((objheader_t *)mobj)->version) { /* Check if versions match */
1590         (*v_matchlock)++;
1591       } else { /* If versions don't match ...HARD ABORT */
1592         (*v_nomatch)++;
1593         /* Send TRANS_DISAGREE to Coordinator */
1594         *getReplyCtrl = TRANS_DISAGREE;
1595         //printf("%s() oid = %d, type = %d\t", __func__, OID(mobj), TYPE((objheader_t *)mobj));
1596         return;
1597       }
1598     }
1599   }
1600 }
1601
1602 /*  Commit info for objects modified */
1603 void commitCountForObjRead(char *getReplyCtrl, unsigned int *oidnotfound, unsigned int *oidlocked, int *numoidnotfound,
1604                            int *numoidlocked, int *v_nomatch, int *v_matchlock, int *v_matchnolock, unsigned int oid, unsigned short version) {
1605   void *mobj;
1606   /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
1607   /* Save the oids not found and number of oids not found for later use */
1608   if ((mobj = mhashSearch(oid)) == NULL) { /* Obj not found */
1609                 /* Save the oids not found and number of oids not found for later use */
1610                 oidnotfound[*numoidnotfound] = oid;
1611                 (*numoidnotfound)++;
1612         } else { /* If Obj found in machine (i.e. has not moved) */
1613                 /* Check if Obj is locked by any previous transaction */
1614                 if (read_trylock(STATUSPTR(mobj))) { // Can further acquire read locks
1615                         if (version == ((objheader_t *)mobj)->version) {      /* If locked then match versions */
1616                                 (*v_matchnolock)++;
1617                                 //Keep track of what is locked
1618                                 oidlocked[(*numoidlocked)++] = OID(((objheader_t *)mobj));
1619                         } else { /* If versions don't match ...HARD ABORT */
1620                                 (*v_nomatch)++;
1621                                 /* Send TRANS_DISAGREE to Coordinator */
1622                                 *getReplyCtrl = TRANS_DISAGREE;
1623                                 //Keep track of what is locked
1624                                 oidlocked[(*numoidlocked)++] = OID(((objheader_t *)mobj));
1625                                 //printf("%s() oid = %d, type = %d\t", __func__, OID(mobj), TYPE((objheader_t *)mobj));
1626                                 return;
1627                         }
1628                 } else { //Has reached max number of readers or some other transaction
1629                         //has acquired a lock on this object
1630                         if (version == ((objheader_t *)mobj)->version) { /* Check if versions match */
1631                                 (*v_matchlock)++;
1632                         } else { /* If versions don't match ...HARD ABORT */
1633                                 (*v_nomatch)++;
1634                                 /* Send TRANS_DISAGREE to Coordinator */
1635                                 *getReplyCtrl = TRANS_DISAGREE;
1636                                 //printf("%s() oid = %d, type = %d\t", __func__, OID(mobj), TYPE((objheader_t *)mobj));
1637                                 return;
1638                         }
1639     }
1640   }
1641 }
1642
1643 /* This function completes the ABORT process if the transaction is aborting */
1644 int transAbortProcess(trans_commit_data_t *transinfo) {
1645   int i, numlocked;
1646   unsigned int *objlocked;
1647   void *header;
1648
1649   numlocked = transinfo->numlocked;
1650   objlocked = transinfo->objlocked;
1651
1652   int useWriteUnlock = 0;
1653   for (i = 0; i < numlocked; i++) {
1654     if(objlocked[i] == -1) {
1655       useWriteUnlock = 1;
1656       continue;
1657     }
1658     if((header = mhashSearch(objlocked[i])) == NULL) {
1659       printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
1660       return 1;
1661     }
1662     if(!useWriteUnlock) {
1663       read_unlock(STATUSPTR(header));
1664     } else {
1665       write_unlock(STATUSPTR(header));
1666     }
1667   }
1668
1669   return 0;
1670 }
1671
1672 /*This function completes the COMMIT process if the transaction is commiting*/
1673 int transComProcess(trans_req_data_t *tdata, trans_commit_data_t *transinfo) {
1674   objheader_t *header, *tcptr;
1675   int i, nummod, tmpsize, numcreated, numlocked;
1676   unsigned int *oidmod, *oidcreated, *oidlocked;
1677   void *ptrcreate;
1678 #ifdef DEBUG
1679         printf("%s-> Entering transComProcess, trans.c\n", __func__);
1680 #endif
1681
1682   nummod = tdata->f.nummod;
1683   oidmod = tdata->oidmod;
1684   numcreated = tdata->f.numcreated;
1685   oidcreated = tdata->oidcreated;
1686   numlocked = transinfo->numlocked;
1687   oidlocked = transinfo->objlocked;
1688
1689
1690 #ifdef DEBUG
1691         printf("%s-> nummod: %d, numcreated: %d, numlocked: %d\n", __func__, nummod, numcreated, numlocked);
1692 #endif
1693
1694   for (i = 0; i < nummod; i++) {
1695     if((header = (objheader_t *) mhashSearch(oidmod[i])) == NULL) {
1696       printf("Error: transComProcess() mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
1697       return 1;
1698     }
1699     /* Copy from transaction cache -> main object store */
1700     if ((tcptr = ((objheader_t *) t_chashSearch(oidmod[i]))) == NULL) {
1701       printf("Error: transComProcess() chashSearch returned NULL at %s, %d\n", __FILE__, __LINE__);
1702       return 1;
1703     }
1704     GETSIZE(tmpsize, header);
1705     char *tmptcptr = (char *) tcptr;
1706     {
1707       struct ___Object___ *dst=(struct ___Object___*)((char*)header+sizeof(objheader_t));
1708       struct ___Object___ *src=(struct ___Object___*)((char*)tmptcptr+sizeof(objheader_t));
1709       dst->___cachedCode___=src->___cachedCode___;
1710       dst->___cachedHash___=src->___cachedHash___;
1711
1712       memcpy(&dst[1], &src[1], tmpsize-sizeof(struct ___Object___));
1713     }
1714
1715     header->version += 1;
1716     //printf("oid: %u, new header version: %d\n", oidmod[i], header->version);
1717     if(header->notifylist != NULL) {
1718       notifyAll(&header->notifylist, OID(header), header->version);
1719     }
1720   }
1721   /* If object is newly created inside transaction then commit it */
1722   for (i = 0; i < numcreated; i++) {
1723     if ((header = ((objheader_t *) t_chashSearch(oidcreated[i]))) == NULL) {
1724       printf("Error: transComProcess() chashSearch returned NULL for oid = %x at %s, %d\n", oidcreated[i], __FILE__, __LINE__);
1725       return 1;
1726     }
1727     header->version += 1;
1728     //printf("oid: %u, new header version: %d\n", oidcreated[i], header->version);
1729     GETSIZE(tmpsize, header);
1730     tmpsize += sizeof(objheader_t);
1731     pthread_mutex_lock(&mainobjstore_mutex);
1732     if ((ptrcreate = objstrAlloc(&mainobjstore, tmpsize)) == NULL) {
1733       printf("Error: transComProcess() failed objstrAlloc %s, %d\n", __FILE__, __LINE__);
1734       pthread_mutex_unlock(&mainobjstore_mutex);
1735       return 1;
1736     }
1737     pthread_mutex_unlock(&mainobjstore_mutex);
1738     /* Initialize read and write locks */
1739     initdsmlocks(STATUSPTR(header));
1740     memcpy(ptrcreate, header, tmpsize);
1741     mhashInsert(oidcreated[i], ptrcreate);
1742     lhashInsert(oidcreated[i], myIpAddr);
1743   }
1744   /* Unlock locked objects */
1745   int useWriteUnlock = 0;
1746   for(i = 0; i < numlocked; i++) {
1747     if(oidlocked[i] == -1) {
1748       useWriteUnlock = 1;
1749       continue;
1750     }
1751     if((header = (objheader_t *) mhashSearch(oidlocked[i])) == NULL) {
1752       printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
1753       return 1;
1754     }
1755     if(!useWriteUnlock) {
1756       read_unlock(STATUSPTR(header));
1757     } else {
1758       write_unlock(STATUSPTR(header));
1759     }
1760   }
1761   return 0;
1762 }
1763
1764 prefetchpile_t *foundLocal(char *ptr) {
1765         int siteid = *(GET_SITEID(ptr));
1766         int ntuples = *(GET_NTUPLES(ptr));
1767         unsigned int * oidarray = GET_PTR_OID(ptr);
1768         unsigned short * endoffsets = GET_PTR_EOFF(ptr, ntuples);
1769         short * arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1770         prefetchpile_t * head=NULL;
1771         int numLocal = 0;
1772
1773         int i;
1774         for(i=0; i<ntuples; i++) {
1775                 unsigned short baseindex=(i==0) ? 0 : endoffsets[i-1];
1776                 unsigned short endindex=endoffsets[i];
1777                 unsigned int oid=oidarray[i];
1778                 int newbase;
1779                 int machinenum;
1780                 if (oid==0)
1781                         continue;
1782                 //Look up fields locally
1783                 for(newbase=baseindex; newbase<endindex; newbase++) {
1784                         if (!lookupObject(&oid, arryfields[newbase]))
1785                                 break;
1786                         //Ended in a null pointer...
1787                         if (oid==0)
1788                                 goto tuple;
1789                 }
1790                 //Entire prefetch is local
1791                 if (newbase==endindex&&checkoid(oid)) {
1792                         numLocal++;
1793                         goto tuple;
1794                 }
1795                 //Add to remote requests
1796                 machinenum=lhashSearch(oid);
1797                 insertPile(machinenum, oid, endindex-newbase, &arryfields[newbase], &head);
1798 tuple:
1799                 ;
1800         }
1801
1802         /* handle dynamic prefetching */
1803         handleDynPrefetching(numLocal, ntuples, siteid);
1804         return head;
1805 }
1806
1807 int checkoid(unsigned int oid) {
1808         objheader_t *header;
1809         if ((header=mhashSearch(oid))!=NULL) {
1810                 //Found on machine
1811                 return 1;
1812         } else if ((header=prehashSearch(oid))!=NULL) {
1813                 //Found in cache
1814                 return 1;
1815         } else {
1816                 return 0;
1817         }
1818 }
1819
1820 int lookupObject(unsigned int * oid, short offset) {
1821   objheader_t *header;
1822   if ((header=mhashSearch(*oid))!=NULL) {
1823     //Found on machine
1824     ;
1825   } else if ((header=prehashSearch(*oid))!=NULL) {
1826     //Found in cache
1827     ;
1828   } else {
1829     return 0;
1830   }
1831
1832   if(TYPE(header) >= NUMCLASSES) {
1833     int elementsize = classsize[TYPE(header)];
1834     struct ArrayObject *ao = (struct ArrayObject *) (((char *)header) + sizeof(objheader_t));
1835     int length = ao->___length___;
1836     /* Check if array out of bounds */
1837     if(offset < 0 || offset >= length) {
1838       //if yes treat the object as found
1839       (*oid)=0;
1840       return 1;
1841     }
1842     (*oid) = *((unsigned int *)(((char *)ao) + sizeof(struct ArrayObject) + (elementsize*offset)));
1843     return 1;
1844   } else {
1845     (*oid) = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + offset));
1846     return 1;
1847   }
1848 }
1849
1850
1851 /* This function is called by the thread calling transPrefetch */
1852 void *transPrefetch(void *t) {
1853   while(1) {
1854     /* read from prefetch queue */
1855     void *node=gettail();
1856     /* Check if the tuples are found locally, if yes then reduce them further*/
1857     /* and group requests by remote machine ids by calling the makePreGroups() */
1858     prefetchpile_t *pilehead = foundLocal(node);
1859
1860     if (pilehead!=NULL) {
1861       // Get sock from shared pool
1862
1863       /* Send  Prefetch Request */
1864       prefetchpile_t *ptr = pilehead;
1865       while(ptr != NULL) {
1866         int sd = getSock2(transPrefetchSockPool, ptr->mid);
1867         sendPrefetchReq(ptr, sd);
1868         ptr = ptr->next;
1869       }
1870
1871       /* Release socket */
1872       //        freeSock(transPrefetchSockPool, pilehead->mid, sd);
1873
1874       /* Deallocated pilehead */
1875       mcdealloc(pilehead);
1876     }
1877     // Deallocate the prefetch queue pile node
1878     inctail();
1879   }
1880 }
1881
1882 void sendPrefetchReqnew(prefetchpile_t *mcpilenode, int sd) {
1883   objpile_t *tmp;
1884
1885   int size=sizeof(char)+sizeof(int);
1886   for(tmp=mcpilenode->objpiles; tmp!=NULL; tmp=tmp->next) {
1887     size += sizeof(int) + sizeof(unsigned int) + sizeof(unsigned int) + ((tmp->numoffset) * sizeof(short));
1888   }
1889
1890   char buft[size];
1891   char *buf=buft;
1892   *buf=TRANS_PREFETCH;
1893   buf+=sizeof(char);
1894
1895   for(tmp=mcpilenode->objpiles; tmp!=NULL; tmp=tmp->next) {
1896     int len = sizeof(int) + sizeof(unsigned int) + sizeof(unsigned int) + ((tmp->numoffset) * sizeof(short));
1897     *((int*)buf)=len;
1898     buf+=sizeof(int);
1899     *((unsigned int *)buf)=tmp->oid;
1900     buf+=sizeof(unsigned int);
1901     *((unsigned int *)(buf)) = myIpAddr;
1902     buf+=sizeof(unsigned int);
1903     memcpy(buf, tmp->offset, tmp->numoffset*sizeof(short));
1904     buf+=tmp->numoffset*sizeof(short);
1905   }
1906   *((int *)buf)=-1;
1907   send_data(sd, buft, size);
1908   return;
1909 }
1910
1911 void sendPrefetchReq(prefetchpile_t *mcpilenode, int sd) {
1912   int len, endpair;
1913   char control;
1914   objpile_t *tmp;
1915
1916   /* Send TRANS_PREFETCH control message */
1917   control = TRANS_PREFETCH;
1918   send_data(sd, &control, sizeof(char));
1919
1920   /* Send Oids and offsets in pairs */
1921   tmp = mcpilenode->objpiles;
1922   while(tmp != NULL) {
1923     len = sizeof(int) + sizeof(unsigned int) + sizeof(unsigned int) + ((tmp->numoffset) * sizeof(short));
1924     char oidnoffset[len];
1925     char *buf=oidnoffset;
1926     *((int*)buf) = tmp->numoffset;
1927     buf+=sizeof(int);
1928     *((unsigned int *)buf) = tmp->oid;
1929     buf+=sizeof(unsigned int);
1930     *((unsigned int *)buf) = myIpAddr;
1931     buf += sizeof(unsigned int);
1932     memcpy(buf, tmp->offset, (tmp->numoffset)*sizeof(short));
1933     send_data(sd, oidnoffset, len);
1934     tmp = tmp->next;
1935   }
1936
1937   /* Send a special char -1 to represent the end of sending oids + offset pair to remote machine */
1938   endpair = -1;
1939   send_data(sd, &endpair, sizeof(int));
1940
1941   return;
1942 }
1943
1944 int getPrefetchResponse(int sd) {
1945   int length = 0, size = 0;
1946   char control;
1947   unsigned int oid;
1948   void *modptr, *oldptr;
1949
1950   recv_data((int)sd, &length, sizeof(int));
1951   size = length - sizeof(int);
1952   char recvbuffer[size];
1953
1954   recv_data((int)sd, recvbuffer, size);
1955   control = *((char *) recvbuffer);
1956   if(control == OBJECT_FOUND) {
1957     oid = *((unsigned int *)(recvbuffer + sizeof(char)));
1958     size = size - (sizeof(char) + sizeof(unsigned int));
1959     pthread_mutex_lock(&prefetchcache_mutex);
1960     if ((modptr = prefetchobjstrAlloc(size)) == NULL) {
1961       printf("Error: objstrAlloc error for copying into prefetch cache %s, %d\n", __FILE__, __LINE__);
1962       pthread_mutex_unlock(&prefetchcache_mutex);
1963       return -1;
1964     }
1965     pthread_mutex_unlock(&prefetchcache_mutex);
1966     memcpy(modptr, recvbuffer + sizeof(char) + sizeof(unsigned int), size);
1967     STATUS(modptr)=0;
1968
1969     /* Insert the oid and its address into the prefetch hash lookup table */
1970     /* Do a version comparison if the oid exists */
1971     if((oldptr = prehashSearch(oid)) != NULL) {
1972       /* If older version then update with new object ptr */
1973       if(((objheader_t *)oldptr)->version <= ((objheader_t *)modptr)->version) {
1974         prehashRemove(oid);
1975         prehashInsert(oid, modptr);
1976       }
1977     } else { /* Else add the object ptr to hash table*/
1978       prehashInsert(oid, modptr);
1979     }
1980     /* Lock the Prefetch Cache look up table*/
1981     pthread_mutex_lock(&pflookup.lock);
1982     /* Broadcast signal on prefetch cache condition variable */
1983     pthread_cond_broadcast(&pflookup.cond);
1984     /* Unlock the Prefetch Cache look up table*/
1985     pthread_mutex_unlock(&pflookup.lock);
1986   } else if(control == OBJECT_NOT_FOUND) {
1987     oid = *((unsigned int *)(recvbuffer + sizeof(char)));
1988     /* TODO: For each object not found query DHT for new location and retrieve the object */
1989     /* Throw an error */
1990     //printf("OBJECT %x NOT FOUND.... THIS SHOULD NOT HAPPEN...TERMINATE PROGRAM\n", oid);
1991     //    exit(-1);
1992   } else {
1993     printf("Error: in decoding the control value %d, %s, %d\n",control, __FILE__, __LINE__);
1994   }
1995
1996   return 0;
1997 }
1998
1999 unsigned short getObjType(unsigned int oid) {
2000   objheader_t *objheader;
2001   unsigned short numoffset[] ={0};
2002   short fieldoffset[] ={};
2003
2004   if ((objheader = (objheader_t *) mhashSearch(oid)) == NULL) {
2005 #ifdef CACHE
2006     if ((objheader = (objheader_t *) prehashSearch(oid)) == NULL) {
2007 #endif
2008     unsigned int mid = lhashSearch(oid);
2009     int sd = getSock2(transReadSockPool, mid);
2010     char remotereadrequest[sizeof(char)+sizeof(unsigned int)];
2011     remotereadrequest[0] = READ_REQUEST;
2012     *((unsigned int *)(&remotereadrequest[1])) = oid;
2013     send_data(sd, remotereadrequest, sizeof(remotereadrequest));
2014
2015     /* Read response from the Participant */
2016     char control;
2017     recv_data(sd, &control, sizeof(char));
2018
2019     if (control==OBJECT_NOT_FOUND) {
2020       printf("Error: in %s() THIS SHOULD NOT HAPPEN.....EXIT PROGRAM\n", __func__);
2021       fflush(stdout);
2022       exit(-1);
2023     } else {
2024       /* Read object if found into local cache */
2025       int size;
2026       recv_data(sd, &size, sizeof(int));
2027 #ifdef CACHE
2028       pthread_mutex_lock(&prefetchcache_mutex);
2029       if ((objheader = prefetchobjstrAlloc(size)) == NULL) {
2030         printf("Error: %s() objstrAlloc error for copying into prefetch cache %s, %d\n", __func__, __FILE__, __LINE__);
2031         pthread_exit(NULL);
2032       }
2033       pthread_mutex_unlock(&prefetchcache_mutex);
2034       recv_data(sd, objheader, size);
2035       prehashInsert(oid, objheader);
2036       return TYPE(objheader);
2037 #else
2038       char *buffer;
2039       if((buffer = calloc(1, size)) == NULL) {
2040         printf("%s() Calloc Error %s at line %d\n", __func__, __FILE__, __LINE__);
2041         fflush(stdout);
2042         return 0;
2043       }
2044       recv_data(sd, buffer, size);
2045       objheader = (objheader_t *)buffer;
2046       unsigned short type = TYPE(objheader);
2047       free(buffer);
2048       return type;
2049 #endif
2050     }
2051 #ifdef CACHE
2052   }
2053 #endif
2054   }
2055   return TYPE(objheader);
2056 }
2057
2058 int startRemoteThread(unsigned int oid, unsigned int mid) {
2059   int sock;
2060   struct sockaddr_in remoteAddr;
2061   char msg[1 + sizeof(unsigned int)];
2062   int bytesSent;
2063   int status;
2064
2065   if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
2066     perror("startRemoteThread():socket()");
2067     return -1;
2068   }
2069
2070   bzero(&remoteAddr, sizeof(remoteAddr));
2071   remoteAddr.sin_family = AF_INET;
2072   remoteAddr.sin_port = htons(LISTEN_PORT);
2073   remoteAddr.sin_addr.s_addr = htonl(mid);
2074
2075   if (connect(sock, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0) {
2076     printf("startRemoteThread():error %d connecting to %s:%d\n", errno,
2077            inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
2078     status = -1;
2079   } else
2080   {
2081     msg[0] = START_REMOTE_THREAD;
2082     *((unsigned int *) &msg[1]) = oid;
2083     send_data(sock, msg, 1 + sizeof(unsigned int));
2084   }
2085
2086   close(sock);
2087   return status;
2088 }
2089
2090 //TODO: when reusing oids, make sure they are not already in use!
2091 static unsigned int id = 0xFFFFFFFF;
2092 unsigned int getNewOID(void) {
2093   id += 2;
2094   if (id > oidMax || id < oidMin) {
2095     id = (oidMin | 1);
2096   }
2097   return id;
2098 }
2099
2100 static unsigned int tid = 0xFFFFFFFF;
2101 unsigned int getNewTransID(void) {
2102   tid++;
2103   if (tid > transIDMax || tid < transIDMin) {
2104     tid = (transIDMin | 1);
2105   }
2106   return tid;
2107 }
2108
2109 int processConfigFile() {
2110   FILE *configFile;
2111   const int maxLineLength = 200;
2112   char lineBuffer[maxLineLength];
2113   char *token;
2114   const char *delimiters = " \t\n";
2115   char *commentBegin;
2116   in_addr_t tmpAddr;
2117
2118   configFile = fopen(CONFIG_FILENAME, "r");
2119   if (configFile == NULL) {
2120     printf("error opening %s:\n", CONFIG_FILENAME);
2121     perror("");
2122     return -1;
2123   }
2124
2125   numHostsInSystem = 0;
2126   sizeOfHostArray = 8;
2127   hostIpAddrs = calloc(sizeOfHostArray, sizeof(unsigned int));
2128 #ifdef RECOVERY
2129         liveHosts = calloc(sizeOfHostArray, sizeof(unsigned int));
2130         locateObjHosts = calloc(sizeOfHostArray*2, sizeof(unsigned int));
2131   liveHostsValid = 0;
2132 #endif
2133
2134         while(fgets(lineBuffer, maxLineLength, configFile) != NULL) {
2135                 commentBegin = strchr(lineBuffer, '#');
2136                 if (commentBegin != NULL)
2137                         *commentBegin = '\0';
2138                 token = strtok(lineBuffer, delimiters);
2139                 while (token != NULL) {
2140                         tmpAddr = inet_addr(token);
2141                         if ((int)tmpAddr == -1) {
2142                                 printf("error in %s: bad token:%s\n", CONFIG_FILENAME, token);
2143                                 fclose(configFile);
2144                                 return -1;
2145                         } else
2146                                 addHost(htonl(tmpAddr));
2147                         token = strtok(NULL, delimiters);
2148                 }
2149         }
2150
2151         fclose(configFile);
2152
2153   if (numHostsInSystem < 1) {
2154     printf("error in %s: no IP Adresses found\n", CONFIG_FILENAME);
2155     return -1;
2156   }
2157 #ifdef MAC
2158   myIpAddr = getMyIpAddr("en1");
2159 #else
2160   myIpAddr = getMyIpAddr("eth0");
2161 #endif
2162   myIndexInHostArray = findHost(myIpAddr);
2163 #ifdef RECOVERY
2164         liveHosts[myIndexInHostArray] = 1;
2165         //locateObjHosts[myIndexInHostArray] = myIpAddr;
2166 #endif
2167         if (myIndexInHostArray == -1) {
2168     printf("error in %s: IP Address of eth0 not found\n", CONFIG_FILENAME);
2169     return -1;
2170   }
2171   oidsPerBlock = (0xFFFFFFFF / numHostsInSystem) + 1;
2172   oidMin = oidsPerBlock * myIndexInHostArray;
2173   if (myIndexInHostArray == numHostsInSystem - 1)
2174     oidMax = 0xFFFFFFFF;
2175   else
2176     oidMax = oidsPerBlock * (myIndexInHostArray + 1) - 1;
2177
2178         transIDMin = oidMin;
2179         transIDMax = oidMax;
2180
2181   return 0;
2182 }
2183
2184 unsigned int getDuplicatedPrimaryMachine(unsigned int mid) {
2185         int i;
2186         for(i = 0; i < numHostsInSystem; i++) {
2187                 if(mid == locateObjHosts[(i*2)+1]) {
2188                         return locateObjHosts[i*2];
2189                 }
2190         }
2191         return -1;
2192 }
2193
2194 unsigned int getPrimaryMachine(unsigned int mid) {
2195         unsigned int pmid;
2196         int pmidindex = 2*findHost(mid);
2197
2198         pthread_mutex_lock(&liveHosts_mutex);
2199         pmid = locateObjHosts[pmidindex];
2200         pthread_mutex_unlock(&liveHosts_mutex);
2201         return pmid;
2202 }
2203
2204 unsigned int getBackupMachine(unsigned int mid) {
2205         unsigned int bmid;
2206         int bmidindex = 2*findHost(mid)+1;
2207
2208         pthread_mutex_lock(&liveHosts_mutex);
2209         bmid = locateObjHosts[bmidindex];
2210         pthread_mutex_unlock(&liveHosts_mutex);
2211         return bmid;
2212 }
2213
2214 // updates the leader's liveHostArray and locateObj
2215 void updateLiveHosts() {
2216 #ifdef DEBUG
2217   printf("%s-> Entering updateLiveHosts\n", __func__);
2218 #endif
2219         // update everyone's list
2220         liveHostsValid = 0;
2221   //int *tmpLiveHosts = calloc(sizeOfHostArray, sizeof(unsigned int));
2222                 //foreach in hostipaddrs, ping -> update list of livemachines
2223     //socket connection?
2224
2225                 //liveHosts lock here
2226                 int sd = 0, i, j, tmpNumLiveHosts = 0;
2227                 for(i = 0; i < numHostsInSystem; i++) {
2228                         if(i == myIndexInHostArray)
2229                         {
2230                                 tmpNumLiveHosts++;
2231                                 continue;
2232                         }
2233                         for(j = 0; j < 5; j++) {        // hard define num of retries
2234                                 if((sd = getSock2WithLock(transRequestSockPool, hostIpAddrs[i])) < 0) {
2235                                 printf("updateLiveHosts(): Cannot create socket connection to [%s], attempt %d\n", __func__, midtoIPString(hostIpAddrs[i]), j);
2236                                         usleep(1000);
2237                                         if(j == 4)
2238                                                 liveHosts[i] = 0;
2239                                         continue;
2240                                 }
2241         char liverequest[sizeof(char)];
2242                                 liverequest[0] = RESPOND_LIVE;
2243
2244                                 send_data(sd, &liverequest[0], sizeof(liverequest));
2245                                 char response = 0;
2246                                 recv_data(sd, &response, sizeof(response));
2247
2248                                 //try to send msg
2249                                 //if timeout, dead host
2250                                 printf("YES received %d\n", response);
2251                                 if(response == LIVE) {
2252                                         printf("must enter here\n");
2253                                 liveHosts[i] = 1;
2254                                 tmpNumLiveHosts++;
2255                                 //locateObjHosts[i*2] = hostIpAddrs[i];
2256                                 }
2257                                 else {
2258                                         printf("or here\n");
2259                                         liveHosts[i] = 0;
2260                                         timeoutFlag = 0;
2261                                 }
2262                                 break;
2263
2264                         }
2265                         if(liveHosts[i] == 0)
2266                                 printf("updateLiveHosts(): cannot make connection to machine %s\n", midtoIPString(hostIpAddrs[i]));
2267                 }
2268                 numLiveHostsInSystem = tmpNumLiveHosts;
2269                 printf("numLiveHostsInSystem:%d\n", numLiveHostsInSystem);
2270                 //have updated list of live machines
2271 #ifdef DEBUG
2272   printf("%s-> Exiting updateLiveHosts\n", __func__);
2273         printHostsStatus();
2274 #endif
2275 }
2276
2277 int getNumLiveHostsInSystem() {
2278         int count = 0, i = 0;
2279         for(; i<numHostsInSystem; i++) {
2280                 if(liveHosts[i])
2281                         count++;
2282         }
2283         return count;
2284 }
2285
2286 int updateLiveHostsCommit() {
2287                 int sd = 0, i;
2288
2289                 char updaterequest[sizeof(char)+sizeof(int)*numHostsInSystem+sizeof(unsigned int)*(numHostsInSystem*2)];
2290                 updaterequest[0] = UPDATE_LIVE_HOSTS;
2291
2292                 for(i = 0; i < numHostsInSystem; i++) {
2293                         *((int *)(&updaterequest[i*4+1])) = liveHosts[i];  // clean this up later
2294                 }
2295
2296                 for(i = 0; i < numHostsInSystem*2; i++) {
2297                         *((unsigned int *)(&updaterequest[i*4+(numHostsInSystem*4)+1])) = locateObjHosts[i];    //ditto
2298                 }
2299
2300                 //for each machine send data
2301                 for(i = 1; i < numHostsInSystem; i++) {         // hard define num of retries
2302                                 if(i == myIndexInHostArray)
2303                                         continue;
2304                                 if(liveHosts[i] == 1) {
2305                                         if((sd = getSock2WithLock(transRequestSockPool, hostIpAddrs[i])) < 0) {
2306                                         printf("updateLiveHosts(): socket create error, attempt %d\n", i);
2307                                                 return -1;
2308                                         }
2309                                         send_data(sd, updaterequest, sizeof(updaterequest));
2310                                 }
2311                 }
2312                 liveHostsValid = 1;
2313                 printHostsStatus();
2314                 return 0;
2315 }
2316
2317 /*void updateLocateObjHosts(unsigned int failedmid) {
2318         int failedmidIndex = findHost(failedmid);
2319         int i = 0, validIndex = 0;
2320
2321         for(; i < numHostsInSystem; i++) {
2322                 if(locateObjHosts[(i*2)] == failedmid) {
2323                         while(liveHosts[(i+validIndex)%numHostsInSystem] == 0)
2324                                 validIndex++;
2325                         locateObjHosts[(i*2)] = hostIpAddrs[(i+validIndex)%numHostsInSystem];
2326                         validIndex++;
2327                         while(liveHosts[(i+validIndex)%numHostsInSystem] == 0)
2328                                 validIndex++;
2329                         locateObjHosts[(i*2)+1] = hostIpAddrs[(i+validIndex)%numHostsInSystem];
2330                 }
2331                 else if(locateObjHosts[(i*2)+1] == failedmid) {
2332                         while(liveHosts[(i+validIndex)%numHostsInSystem] == 0)
2333                                 validIndex++;
2334                         locateObjHosts[(i*2)+1] = hostIpAddrs[(i+validIndex)%numHostsInSystem];
2335                         validIndex = 0;
2336                 }
2337         }
2338 }*/
2339
2340 void setLocateObjHosts() {
2341         int i = 0, validIndex = 0;
2342
2343         //check num hosts even valid first
2344
2345         for(;i < numHostsInSystem; i++) {
2346 #ifdef DEBUG
2347     printf("%s-> i:%d\n", __func__, i);
2348 #endif
2349
2350                 while(liveHosts[(i+validIndex)%numHostsInSystem] == 0) {
2351                         validIndex++;
2352                 }
2353                 locateObjHosts[i*2] = hostIpAddrs[(i+validIndex)%numHostsInSystem];
2354 #ifdef DEBUG
2355                 printf("%s-> locateObjHosts[%d]:%s\n", __func__, i*2, midtoIPString(locateObjHosts[(i*2)]));
2356 #endif
2357
2358                 validIndex++;
2359                 while(liveHosts[(i+validIndex)%numHostsInSystem] == 0) {
2360                         validIndex++;
2361                 }
2362 #ifdef DEBUG
2363                 printf("%s-> validIndex:%d, this mid is: [%s]\n", __func__, validIndex, midtoIPString(hostIpAddrs[(i+validIndex)%numHostsInSystem]));
2364 #endif
2365                 locateObjHosts[(i*2)+1] = hostIpAddrs[(i+validIndex)%numHostsInSystem];
2366                 validIndex=0;
2367
2368 #ifdef DEBUG
2369                 printf("%s-> locateObjHosts[%d]:%s\n", __func__, i*2+1, midtoIPString(locateObjHosts[(i*2)+1]));
2370 #endif
2371         }
2372 }
2373
2374 //debug function
2375 void printHostsStatus() {
2376         int i;
2377 #ifdef DEBUG
2378         printf("%s-> *printing live machines and backups*\n", __func__);
2379 #endif
2380         for(i = 0; i < numHostsInSystem; i++) {
2381                 if(liveHosts[i]) {
2382 #ifdef DEBUG
2383                         printf("%s-> [%s]: LIVE\n", __func__, midtoIPString(hostIpAddrs[i]));
2384 #endif
2385                 }
2386                 else {
2387 #ifdef DEBUG
2388                         printf("%s-> [%s]: DEAD\n", __func__, midtoIPString(hostIpAddrs[i]));
2389 #endif
2390                 }
2391 #ifdef DEBUG
2392                         printf("%s-> original:\t[%s]\n", __func__, midtoIPString(locateObjHosts[i*2]));
2393                         printf("%s-> backup:\t[%s]\n", __func__, midtoIPString(locateObjHosts[i*2+1]));
2394 #endif
2395         }
2396 }
2397
2398 int allHostsLive() {
2399         int i;
2400         for(i = 0; i < numHostsInSystem; i++) {
2401                 if(!liveHosts[i])
2402                         return 0;
2403         }
2404         return 1;
2405 }
2406
2407 void duplicateLostObjects(unsigned int mid){
2408
2409 #ifdef DEBUG
2410         printf("%s-> Start, mid: [%s]\n", __func__, midtoIPString(mid));
2411 #endif
2412
2413         //this needs to be changed.
2414         unsigned int backupMid = getBackupMachine(mid);
2415         unsigned int originalMid = getDuplicatedPrimaryMachine(mid);
2416
2417 #ifdef DEBUG
2418         printf("%s-> backupMid: [%s], ", __func__, midtoIPString(backupMid));
2419         printf("originalMid: [%s]\n", midtoIPString(originalMid));
2420 #endif
2421
2422   setLocateObjHosts();
2423         printHostsStatus();
2424
2425         //connect to these machines
2426         //go through their object store copying necessary (in a transaction)
2427         //transRequestSockPool = createSockPool(transRequestSockPool, DEFAULTSOCKPOOLSIZE);
2428         int sd = 0, i, j, tmpNumLiveHosts = 0;
2429
2430         if(originalMid == myIpAddr) {
2431                 originalMid = getPrimaryMachine(mid);
2432                 printf("originalMid: [%s]\n", midtoIPString(originalMid));
2433                 duplicateLocalOriginalObjects(originalMid);
2434         }
2435         else if((sd = getSock2WithLock(transRequestSockPool, originalMid)) < 0) {
2436                 printf("updateLiveHosts(): socket create error, attempt %d\n", j);
2437                 //usleep(1000);
2438         }
2439         else {
2440                 char duperequest;
2441                 duperequest = DUPLICATE_ORIGINAL;
2442                 send_data(sd, &duperequest, sizeof(char));
2443 #ifdef DEBUG
2444           printf("%s-> Sent DUPLICATE_ORIGINAL request\n", __func__);
2445 #endif
2446                 originalMid = getPrimaryMachine(mid);
2447                 printf("originalMid: [%s]\n", midtoIPString(originalMid));
2448                 send_data(sd, &originalMid, sizeof(unsigned int));
2449 #ifdef DEBUG
2450           printf("%s-> Sent originalMid\n", __func__);
2451 #endif
2452                 char response;
2453                 recv_data_block(sd, &response, sizeof(char));
2454                 printf("YES! Received %d\n", response);
2455                 }
2456
2457         if(backupMid == myIpAddr) {
2458                 backupMid = getBackupMachine(mid);
2459                 duplicateLocalBackupObjects(backupMid);
2460         }
2461         else if((sd = getSock2WithLock(transRequestSockPool, backupMid)) < 0) {
2462                 printf("updateLiveHosts(): socket create error, attempt %d\n", j);
2463                 exit(1);
2464         }
2465         else {
2466                 char duperequest;
2467                 duperequest = DUPLICATE_BACKUP;
2468                 send_data(sd, &duperequest, sizeof(char));
2469 #ifdef DEBUG
2470           printf("%s-> Sent DUPLICATE_BACKUP request\n", __func__);
2471 #endif
2472                 backupMid = getBackupMachine(mid);
2473                 send_data(sd, &backupMid, sizeof(unsigned int));
2474 #ifdef DEBUG
2475           printf("%s-> Sent backupMid\n", __func__);
2476 #endif
2477
2478                 char response;
2479                 recv_data_block(sd, &response, sizeof(char));
2480                 printf("YES! Received %d\n", response);
2481         }
2482
2483 #ifdef DEBUG
2484         printf("%s-> End\n", __func__);
2485 #endif
2486 }
2487
2488 void duplicateLocalBackupObjects(unsigned int mid) {
2489         int tempsize, sd;
2490         char *dupeptr, ctrl, response;
2491
2492 #ifdef DEBUG
2493         printf("%s-> Start; backup mid:%s\n", __func__, midtoIPString(mid));
2494 #endif
2495         //copy code from dstmserver here
2496         tempsize = mhashGetDuplicate(&dupeptr, 1);
2497
2498         printf("tempsize:%d, dupeptrfirstvalue:%d\n", tempsize, *((unsigned int *)(dupeptr)));
2499         //send control and dupes after
2500         ctrl = RECEIVE_DUPES;
2501         if((sd = getSockWithLock(transRequestSockPool, mid)) < 0) {
2502                 printf("duplicatelocalbackup: socket create error\n");
2503                 //usleep(1000);
2504         }
2505
2506         printf("sd:%d, tempsize:%d, dupeptrfirstvalue:%d\n", sd, tempsize, *((unsigned int *)(dupeptr)));
2507         send_data(sd, &ctrl, sizeof(char));
2508         send_data(sd, dupeptr, tempsize);
2509         recv_data(sd, &response, sizeof(char));
2510         if(response != DUPLICATION_COMPLETE) {
2511                 //fail message
2512         }
2513
2514         freeSockWithLock(transRequestSockPool, mid, sd);
2515 #ifdef DEBUG
2516         printf("%s-> End\n", __func__);
2517 #endif
2518
2519 }
2520
2521 void duplicateLocalOriginalObjects(unsigned int mid) {
2522         int tempsize, sd;
2523         char *dupeptr, ctrl, response;
2524
2525 #ifdef DEBUG
2526         printf("%s-> Start\n", __func__);
2527 #endif
2528         //copy code fom dstmserver here
2529
2530         tempsize = mhashGetDuplicate(&dupeptr, 0);
2531
2532         //send control and dupes after
2533         ctrl = RECEIVE_DUPES;
2534
2535         if((sd = getSockWithLock(transRequestSockPool, mid)) < 0) {
2536                 printf("DUPLICATE_ORIGINAL: socket create error\n");
2537                 //usleep(1000);
2538         }
2539         printf("sd:%d, tempsize:%d, dupeptrfirstvalue:%d\n", sd, tempsize, *((unsigned int *)(dupeptr)));
2540
2541         send_data(sd, &ctrl, sizeof(char));
2542         send_data(sd, dupeptr, tempsize);
2543
2544         recv_data(sd, &response, sizeof(char));
2545         if(response != DUPLICATION_COMPLETE) {
2546                 //fail message
2547         }
2548         freeSockWithLock(transRequestSockPool, mid, sd);
2549
2550 #ifdef DEBUG
2551         printf("%s-> End\n", __func__);
2552 #endif
2553
2554 }
2555
2556 void addHost(unsigned int hostIp) {
2557   unsigned int *tmpArray;
2558   int *tmpliveHostsArray;
2559         unsigned int *tmplocateObjHostsArray;
2560
2561   if (findHost(hostIp) != -1)
2562     return;
2563
2564   if (numHostsInSystem == sizeOfHostArray) {
2565     tmpArray = calloc(sizeOfHostArray * 2, sizeof(unsigned int));
2566     memcpy(tmpArray, hostIpAddrs, sizeof(unsigned int) * numHostsInSystem);
2567     free(hostIpAddrs);
2568     hostIpAddrs = tmpArray;
2569
2570                 tmpliveHostsArray = calloc(sizeOfHostArray * 2, sizeof(unsigned int));
2571                 memcpy(tmpliveHostsArray, liveHosts, sizeof(unsigned int) * numHostsInSystem);
2572     free(liveHosts);
2573     liveHosts = tmpliveHostsArray;
2574
2575                 tmplocateObjHostsArray = calloc(sizeOfHostArray * 2 * 2, sizeof(unsigned int));
2576                 memcpy(tmplocateObjHostsArray, locateObjHosts, sizeof(unsigned int) * numHostsInSystem);
2577     free(locateObjHosts);
2578     locateObjHosts = tmplocateObjHostsArray;
2579
2580                 sizeOfHostArray *= 2;
2581   }
2582
2583   hostIpAddrs[numHostsInSystem] = hostIp;
2584   liveHosts[numHostsInSystem] = 0;
2585   locateObjHosts[numHostsInSystem*2] = hostIp;
2586
2587         numHostsInSystem++;
2588   return;
2589 }
2590
2591 int findHost(unsigned int hostIp) {
2592   int i;
2593   for (i = 0; i < numHostsInSystem; i++)
2594     if (hostIpAddrs[i] == hostIp)
2595       return i;
2596
2597   //not found
2598   return -1;
2599 }
2600
2601 /* This function sends notification request per thread waiting on object(s) whose version
2602  * changes */
2603 int reqNotify(unsigned int *oidarry, unsigned short *versionarry, unsigned int numoid) {
2604   int sock,i;
2605   objheader_t *objheader;
2606   struct sockaddr_in remoteAddr;
2607   char msg[1 + numoid * (sizeof(unsigned short) + sizeof(unsigned int)) +  3 * sizeof(unsigned int)];
2608   char *ptr;
2609   int bytesSent;
2610   int status, size;
2611   unsigned short version;
2612   unsigned int oid,mid;
2613   static unsigned int threadid = 0;
2614   pthread_mutex_t threadnotify = PTHREAD_MUTEX_INITIALIZER; //Lock and condition var for threadjoin and notification
2615   pthread_cond_t threadcond = PTHREAD_COND_INITIALIZER;
2616   notifydata_t *ndata;
2617
2618   oid = oidarry[0];
2619   if((mid = lhashSearch(oid)) == 0) {
2620     printf("Error: %s() No such machine found for oid =%x\n",__func__, oid);
2621     return;
2622   }
2623
2624   if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
2625     perror("reqNotify():socket()");
2626     return -1;
2627   }
2628
2629   bzero(&remoteAddr, sizeof(remoteAddr));
2630   remoteAddr.sin_family = AF_INET;
2631   remoteAddr.sin_port = htons(LISTEN_PORT);
2632   remoteAddr.sin_addr.s_addr = htonl(mid);
2633
2634   /* Generate unique threadid */
2635   threadid++;
2636
2637   /* Save threadid, numoid, oidarray, versionarray, pthread_cond_variable for later processing */
2638   if((ndata = calloc(1, sizeof(notifydata_t))) == NULL) {
2639     printf("Calloc Error %s, %d\n", __FILE__, __LINE__);
2640     return -1;
2641   }
2642   ndata->numoid = numoid;
2643   ndata->threadid = threadid;
2644   ndata->oidarry = oidarry;
2645   ndata->versionarry = versionarry;
2646   ndata->threadcond = threadcond;
2647   ndata->threadnotify = threadnotify;
2648   if((status = notifyhashInsert(threadid, ndata)) != 0) {
2649     printf("reqNotify(): Insert into notify hash table not successful %s, %d\n", __FILE__, __LINE__);
2650     free(ndata);
2651     return -1;
2652   }
2653
2654   /* Send  number of oids, oidarry, version array, machine id and threadid */
2655   if (connect(sock, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0) {
2656     printf("reqNotify():error %d connecting to %s:%d\n", errno,
2657            inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
2658     free(ndata);
2659     return -1;
2660   } else {
2661     msg[0] = THREAD_NOTIFY_REQUEST;
2662     *((unsigned int *)(&msg[1])) = numoid;
2663     /* Send array of oids  */
2664     size = sizeof(unsigned int);
2665
2666     for(i = 0;i < numoid; i++) {
2667       oid = oidarry[i];
2668       *((unsigned int *)(&msg[1] + size)) = oid;
2669       size += sizeof(unsigned int);
2670     }
2671
2672     /* Send array of version  */
2673     for(i = 0;i < numoid; i++) {
2674       version = versionarry[i];
2675       *((unsigned short *)(&msg[1] + size)) = version;
2676       size += sizeof(unsigned short);
2677     }
2678
2679     *((unsigned int *)(&msg[1] + size)) = myIpAddr; size += sizeof(unsigned int);
2680     *((unsigned int *)(&msg[1] + size)) = threadid;
2681     pthread_mutex_lock(&(ndata->threadnotify));
2682     size = 1 + numoid * (sizeof(unsigned int) + sizeof(unsigned short)) + 3 * sizeof(unsigned int);
2683     send_data(sock, msg, size);
2684     pthread_cond_wait(&(ndata->threadcond), &(ndata->threadnotify));
2685     pthread_mutex_unlock(&(ndata->threadnotify));
2686   }
2687
2688   pthread_cond_destroy(&threadcond);
2689   pthread_mutex_destroy(&threadnotify);
2690   free(ndata);
2691   close(sock);
2692   return status;
2693 }
2694
2695 void threadNotify(unsigned int oid, unsigned short version, unsigned int tid) {
2696   notifydata_t *ndata;
2697   int i, objIsFound = 0, index;
2698   void *ptr;
2699
2700   //Look up the tid and call the corresponding pthread_cond_signal
2701   if((ndata = notifyhashSearch(tid)) == NULL) {
2702     printf("threadnotify(): No such threadid is present %s, %d\n", __FILE__, __LINE__);
2703     return;
2704   } else  {
2705     for(i = 0; i < ndata->numoid; i++) {
2706       if(ndata->oidarry[i] == oid) {
2707         objIsFound = 1;
2708         index = i;
2709       }
2710     }
2711     if(objIsFound == 0) {
2712       printf("threadNotify(): Oid not found %s, %d\n", __FILE__, __LINE__);
2713       return;
2714     } else {
2715       if(version <= ndata->versionarry[index]) {
2716         printf("threadNotify(): New version %d has not changed since last version for oid = %d, %s, %d\n", version, oid, __FILE__, __LINE__);
2717         return;
2718       } else {
2719 #ifdef CACHE
2720         /* Clear from prefetch cache and free thread related data structure */
2721         if((ptr = prehashSearch(oid)) != NULL) {
2722           prehashRemove(oid);
2723         }
2724 #endif
2725         pthread_mutex_lock(&(ndata->threadnotify));
2726         pthread_cond_signal(&(ndata->threadcond));
2727         pthread_mutex_unlock(&(ndata->threadnotify));
2728       }
2729     }
2730   }
2731   return;
2732 }
2733
2734 int notifyAll(threadlist_t **head, unsigned int oid, unsigned int version) {
2735   threadlist_t *ptr;
2736   unsigned int mid;
2737   struct sockaddr_in remoteAddr;
2738   char msg[1 + sizeof(unsigned short) + 2*sizeof(unsigned int)];
2739   int sock, status, size, bytesSent;
2740
2741   while(*head != NULL) {
2742     ptr = *head;
2743     mid = ptr->mid;
2744     //create a socket connection to that machine
2745     if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
2746       perror("notifyAll():socket()");
2747       return -1;
2748     }
2749
2750     bzero(&remoteAddr, sizeof(remoteAddr));
2751     remoteAddr.sin_family = AF_INET;
2752     remoteAddr.sin_port = htons(LISTEN_PORT);
2753     remoteAddr.sin_addr.s_addr = htonl(mid);
2754     //send Thread Notify response and threadid to that machine
2755     if (connect(sock, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0) {
2756       printf("notifyAll():error %d connecting to %s:%d\n", errno,
2757              inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
2758       fflush(stdout);
2759       status = -1;
2760     } else {
2761       bzero(msg, (1+sizeof(unsigned short) + 2*sizeof(unsigned int)));
2762       msg[0] = THREAD_NOTIFY_RESPONSE;
2763       *((unsigned int *)&msg[1]) = oid;
2764       size = sizeof(unsigned int);
2765       *((unsigned short *)(&msg[1]+ size)) = version;
2766       size+= sizeof(unsigned short);
2767       *((unsigned int *)(&msg[1]+ size)) = ptr->threadid;
2768
2769       size = 1 + 2*sizeof(unsigned int) + sizeof(unsigned short);
2770       send_data(sock, msg, size);
2771     }
2772     //close socket
2773     close(sock);
2774     // Update head
2775     *head = ptr->next;
2776     free(ptr);
2777   }
2778   return status;
2779 }
2780
2781 void transAbort() {
2782 #ifdef ABORTREADERS
2783   removetransactionhash();
2784 #endif
2785   objstrDelete(t_cache);
2786   t_chashDelete();
2787 }
2788
2789 /* This function inserts necessary information into
2790  * a machine pile data structure */
2791 plistnode_t *pInsert(plistnode_t *pile, objheader_t *headeraddr, unsigned int mid, int num_objs) {
2792   plistnode_t *ptr, *tmp;
2793   int found = 0, offset = 0;
2794   char ip[16];
2795   tmp = pile;
2796   //Add oid into a machine that is already present in the pile linked list structure
2797   while(tmp != NULL) {
2798     if (tmp->mid == mid) {
2799       int tmpsize;
2800
2801                         if (STATUS(headeraddr) & NEW) {
2802                                 tmp->oidcreated[tmp->numcreated] = OID(headeraddr);
2803                                 tmp->numcreated++;
2804                                 GETSIZE(tmpsize, headeraddr);
2805                                 tmp->sum_bytes += sizeof(objheader_t) + tmpsize;
2806                           /*if(numHostsInSystem > 1) {
2807                                         STATUS(headeraddr) = DIRTY;
2808                                         //printf("Redo pInsert for oid %d, now modified\n", OID(headeraddr));
2809                                         //printf("this machine: %d\n", mid);
2810                                         midtoIP(tmp->mid, ip);
2811                                         pile = pInsert(tmp, headeraddr, locateBackupMachine(headeraddr), num_objs);
2812
2813                                 //      printf("header version: %d\n", headeraddr->version);
2814                                         //printf("Finished Redo pInsert for oid %d, now modified\n", OID(headeraddr));
2815                                 }*/
2816                         } else if (STATUS(headeraddr) & DIRTY) {
2817                                 tmp->oidmod[tmp->nummod] = OID(headeraddr);
2818                                 tmp->nummod++;
2819                                 GETSIZE(tmpsize, headeraddr);
2820                                 tmp->sum_bytes += sizeof(objheader_t) + tmpsize;
2821                                 /*      midtoIP(tmp->mid, ip);
2822                                         printf("pp; Redo? pile->mid: %s, oid: %d, header version: %d\n", ip, OID(headeraddr), headeraddr->version);*/
2823                         } else {
2824                                 offset = (sizeof(unsigned int) + sizeof(short)) * tmp->numread;
2825                                 *((unsigned int *)(((char *)tmp->objread) + offset))=OID(headeraddr);
2826                                 offset += sizeof(unsigned int);
2827                                 *((short *)(((char *)tmp->objread) + offset)) = headeraddr->version;
2828                                 tmp->numread++;
2829       }
2830       found = 1;
2831       break;
2832     }
2833     tmp = tmp->next;
2834   }
2835   //Add oid for any new machine
2836   if (!found) {
2837     int tmpsize;
2838     if((ptr = pCreate(num_objs)) == NULL) {
2839       return NULL;
2840     }
2841     ptr->mid = mid;
2842     if (STATUS(headeraddr) & NEW) {
2843       ptr->oidcreated[ptr->numcreated] = OID(headeraddr);
2844       ptr->numcreated++;
2845       GETSIZE(tmpsize, headeraddr);
2846       ptr->sum_bytes += sizeof(objheader_t) + tmpsize;
2847           /*if(numHostsInSystem > 1) {
2848                                 STATUS(headeraddr) = DIRTY;
2849                                         midtoIP(ptr->mid, ip);
2850
2851                                         printf("np; ptr->mid: %s, oid: %d, header version: %d\n", ip, OID(headeraddr), headeraddr->version);
2852                                         //printf("header version: %d\n", headeraddr->version);
2853                                 pile = pInsert(tmp, headeraddr, locateBackupMachine(headeraddr), num_objs);
2854                                         //printf("header version: %d\n", headeraddr->version);
2855                         }*/
2856           } else if (STATUS(headeraddr) & DIRTY) {
2857       ptr->oidmod[ptr->nummod] = OID(headeraddr);
2858       ptr->nummod++;
2859       GETSIZE(tmpsize, headeraddr);
2860       ptr->sum_bytes += sizeof(objheader_t) + tmpsize;
2861                                 //printf("Redo oid %d?\n", OID(headeraddr));
2862                                 /*      midtoIP(ptr->mid, ip);
2863                                         printf("np; Redo? ptr->mid: %s, oid: %d, header version: %d\n", ip, OID(headeraddr), headeraddr->version);*/
2864     } else {
2865       *((unsigned int *)ptr->objread)=OID(headeraddr);
2866       offset = sizeof(unsigned int);
2867       *((short *)(((char *)ptr->objread) + offset)) = headeraddr->version;
2868       ptr->numread++;
2869     }
2870     ptr->next = pile;
2871     pile = ptr;
2872   }
2873
2874   /* Clear Flags */
2875   STATUS(headeraddr) = 0;
2876
2877   return pile;
2878 }
2879
2880 plistnode_t *sortPiles(plistnode_t *pileptr) {
2881         plistnode_t *head, *ptr, *tail;
2882         head = pileptr;
2883         ptr = pileptr;
2884         /* Get tail pointer */
2885         while(ptr!= NULL) {
2886                 tail = ptr;
2887                 ptr = ptr->next;
2888         }
2889         ptr = pileptr;
2890         plistnode_t *prev = pileptr;
2891         /* Arrange local machine processing at the end of the pile list */
2892         while(ptr != NULL) {
2893                 if(ptr != tail) {
2894                         if(ptr->mid == myIpAddr && (prev != pileptr)) {
2895                                 prev->next = ptr->next;
2896                                 ptr->next = NULL;
2897                                 tail->next = ptr;
2898                                 return pileptr;
2899                         }
2900                         if((ptr->mid == myIpAddr) && (prev == pileptr)) {
2901                                 prev = ptr->next;
2902                                 ptr->next = NULL;
2903                                 tail->next = ptr;
2904                                 return prev;
2905                         }
2906                         prev = ptr;
2907                 }
2908                 ptr = ptr->next;
2909         }
2910         return pileptr;
2911 }
2912
2913 /* Paxo Algorithm:
2914  * Executes when the known leader has failed.
2915  * Guarantees consensus on next leader among all live hosts.  */
2916 int paxos()
2917 {
2918         int origRound = paxosRound;
2919         origleader = leader;
2920         int ret = -1;
2921 #ifdef DEBUG
2922         printf(">> Debug : Starting paxos..\n");
2923 #endif
2924
2925         do {
2926                 ret = paxosPrepare();           // phase 1
2927                 if (ret == 1) {
2928                         ret = paxosAccept();    // phase 2
2929                         if (ret == 1) {
2930                                 paxosLearn();                           // phase 3
2931                                 break;
2932                         }
2933                 }
2934                 // Paxos not successful; wait and retry if new leader is not yet slected
2935                 sleep(WAIT_TIME);
2936                 if(paxosRound != origRound)
2937                         break;
2938         } while (ret == -1);
2939
2940 #ifdef DEBUG
2941         printf("\n>> Debug : Leader : [%s]\n", midtoIPString(leader));
2942 #endif
2943
2944         return ret;
2945 }
2946
2947 int paxosPrepare()
2948 {
2949         char control;
2950         //int origleader = leader;
2951         int remote_n;
2952         int remote_v;
2953         int tmp_n = -1;
2954         int cnt = 0;
2955         int sd;
2956         int i;
2957         temp_v_a = v_a;
2958         my_n = n_h + 1;
2959
2960 #ifdef DEBUG
2961         printf("[Prepare]...\n");
2962 #endif
2963
2964         temp_v_a = myIpAddr;    // if no other value is proposed, make this machine the new leader
2965
2966         for (i = 0; i < numHostsInSystem; ++i) {
2967                 control = PAXOS_PREPARE;
2968                 if(!liveHosts[i])
2969                         continue;
2970
2971                 if ((sd = getSock2WithLock(transRequestSockPool, hostIpAddrs[i])) < 0) {
2972                         printf("paxosPrepare(): socket create error\n");
2973                         continue;
2974                 }
2975 #ifdef DEBUG
2976                 printf("%s-> Send PAXOS_PREPARE to mid [%s] with my_n=%d\n", __func__, midtoIPString(hostIpAddrs[i]), my_n);
2977 #endif
2978                 send_data(sd, &control, sizeof(char));
2979                 send_data(sd, &my_n, sizeof(int));
2980                 recv_data(sd, &control, sizeof(char));
2981                 if ((sd == -1) || (timeoutFlag == 1)) {
2982 #ifdef DEBUG
2983                         printf("%s-> timeout to machine [%s]\n", __func__, midtoIPString(hostIpAddrs[i]));
2984 #endif
2985                         timeoutFlag = 0;
2986                         continue;
2987                 }
2988
2989                 switch (control) {
2990                         case PAXOS_PREPARE_OK:
2991                                 cnt++;
2992                                 recv_data(sd, &remote_n, sizeof(int));
2993                                 recv_data(sd, &remote_v, sizeof(int));
2994 #ifdef DEBUG
2995                                 printf("%s-> Received PAXOS_PREPARE_OK from mindex [%d] with remote_v=%s\n", __func__, i, midtoIPString(remote_v));
2996 #endif
2997                                 if(remote_v != origleader) {
2998                                         if (remote_n > tmp_n) {
2999                                                 tmp_n = remote_n;
3000                                                 temp_v_a = remote_v;
3001                                         }
3002                                 }
3003                                 break;
3004                         case PAXOS_PREPARE_REJECT:
3005                                 break;
3006                 }
3007         }
3008
3009 #ifdef DEBUG
3010         printf("%s-> cnt:%d, numLiveHostsInSystem:%d\n", __func__, cnt, numLiveHostsInSystem);
3011 #endif
3012
3013         if (cnt >= (numLiveHostsInSystem / 2)) {                // majority of OK replies
3014                 return 1;
3015                 }
3016                 else {
3017                         return -1;
3018                 }
3019 }
3020
3021 int paxosAccept()
3022 {
3023         char control;
3024         int i;
3025         int cnt = 0;
3026         int sd;
3027         int remote_v = temp_v_a;
3028
3029 #ifdef DEBUG
3030         printf("[Accept]...\n");
3031 #endif
3032
3033         for (i = 0; i < numHostsInSystem; ++i) {
3034                 control = PAXOS_ACCEPT;
3035                         if(!liveHosts[i])
3036                         continue;
3037
3038         if ((sd = getSock2WithLock(transRequestSockPool, hostIpAddrs[i])) < 0) {
3039                         printf("paxosAccept(): socket create error\n");
3040                         continue;
3041                 }
3042
3043                 send_data(sd, &control, sizeof(char));
3044                 send_data(sd, &my_n, sizeof(int));
3045                 send_data(sd, &remote_v, sizeof(int));
3046
3047                 recv_data(sd, &control, sizeof(char));
3048                 if ((sd == -1) || (timeoutFlag == 1)) {
3049 #ifdef DEBUG
3050                         printf("%s-> timeout to machine [%s]\n", __func__, midtoIPString(hostIpAddrs[i]));
3051 #endif
3052                         timeoutFlag = 0;
3053                         continue;
3054                 }
3055
3056                 switch (control) {
3057                         case PAXOS_ACCEPT_OK:
3058                                 cnt++;
3059                                 break;
3060                         case PAXOS_ACCEPT_REJECT:
3061                                 break;
3062                 }
3063 #ifdef DEBUG
3064                 printf(">> Debug : Accept - n_h [%d], n_a [%d], v_a [%s]\n", n_h, n_a, midtoIPString(v_a));
3065 #endif
3066         }
3067
3068         if (cnt >= (numLiveHostsInSystem / 2)) {
3069                 return 1;
3070         }
3071         else {
3072                 return -1;
3073         }
3074 }
3075
3076 void paxosLearn()
3077 {
3078         char control;
3079         int sd;
3080         int i;
3081
3082 #ifdef DEBUG
3083         printf("[Learn]...\n");
3084 #endif
3085
3086         control = PAXOS_LEARN;
3087         //      transRequestSockPool = createSockPool(transRequestSockPool, DEFAULTSOCKPOOLSIZE);
3088
3089         for (i = 0; i < numHostsInSystem; ++i) {
3090                 if(!liveHosts[i])
3091                         continue;
3092                 if(hostIpAddrs[i] == myIpAddr)
3093                 {
3094                         leader = v_a;
3095                         paxosRound++;
3096 #ifdef DEBUG
3097                         printf("This is my leader!!!: [%s]\n", midtoIPString(leader));
3098 #endif
3099                         continue;
3100                 }
3101                 if ((sd = getSock2WithLock(transRequestSockPool, hostIpAddrs[i])) < 0) {
3102                         continue;
3103                         //                      printf("paxosLearn(): socket create error, attemp\n");
3104                 }
3105
3106                 send_data(sd, &control, sizeof(char));
3107                 send_data(sd, &v_a, sizeof(int));
3108         }
3109         //return v_a;
3110 }