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

   1 #include "dstm.h"
   2 #include "ip.h"
   3 #include "clookup.h"
   4 #include "machinepile.h"
   5 #include "mlookup.h"
   6 #include "llookup.h"
   7 #include "plookup.h"
   8 #include "prelookup.h"
   9 #include "threadnotify.h"
  10 #include "queue.h"
  11 #include <pthread.h>
  12 #include <sys/types.h>
  13 #include <sys/socket.h>
  14 #include <netdb.h>
  15 #include <netinet/in.h>
  16 #include <sys/types.h>
  17 #include <unistd.h>
  18 #include <errno.h>
  19 #include <time.h>
  20 #include <string.h>
  21
  22 #define LISTEN_PORT 2156
  23 #define RECEIVE_BUFFER_SIZE 2048
  24 #define NUM_THREADS 10
  25 #define PREFETCH_CACHE_SIZE 1048576 //1MB
  26 #define CONFIG_FILENAME "dstm.conf"
  27
  28 /* Global Variables */
  29 extern int classsize[];
  30 extern primarypfq_t pqueue; //Shared prefetch queue
  31 extern mcpileq_t mcqueue;  //Shared queue containing prefetch requests sorted by remote machineids
  32 objstr_t *prefetchcache; //Global Prefetch cache
  33 pthread_mutex_t prefetchcache_mutex;// Mutex to lock Prefetch Cache
  34 pthread_mutexattr_t prefetchcache_mutex_attr; /* Attribute for lock to make it a recursive lock */
  35 extern pthread_mutex_t mainobjstore_mutex;// Mutex to lock main Object store
  36 extern prehashtable_t pflookup; //Global Prefetch cache's lookup table
  37 pthread_t wthreads[NUM_THREADS]; //Worker threads for working on the prefetch queue
  38 pthread_t tPrefetch;            /* Primary Prefetch thread that processes the prefetch queue */
  39 extern objstr_t *mainobjstore;
  40 unsigned int myIpAddr;
  41 unsigned int *hostIpAddrs;
  42 int sizeOfHostArray;
  43 int numHostsInSystem;
  44 int myIndexInHostArray;
  45 unsigned int oidsPerBlock;
  46 unsigned int oidMin;
  47 unsigned int oidMax;
  48 void *mlist[10000];
  49 pthread_mutex_t mlock = PTHREAD_MUTEX_INITIALIZER;
  50
  51 void printhex(unsigned char *, int);
  52 plistnode_t *createPiles(transrecord_t *);
  53
  54 void printhex(unsigned char *ptr, int numBytes)
  55 {
  56         int i;
  57         for (i = 0; i < numBytes; i++)
  58         {
  59                 if (ptr[i] < 16)
  60                         printf("0%x ", ptr[i]);
  61                 else
  62                         printf("%x ", ptr[i]);
  63         }
  64         printf("\n");
  65         return;
  66 }
  67
  68 inline int arrayLength(int *array) {
  69         int i;
  70         for(i=0 ;array[i] != -1; i++)
  71                 ;
  72         return i;
  73 }
  74 inline int findmax(int *array, int arraylength) {
  75         int max, i;
  76         max = array[0];
  77         for(i = 0; i < arraylength; i++){
  78                 if(array[i] > max) {
  79                         max = array[i];
  80                 }
  81         }
  82         return max;
  83 }
  84 /* This function is a prefetch call generated by the compiler that
  85  * populates the shared primary prefetch queue*/
  86 void prefetch(int ntuples, unsigned int *oids, unsigned short *endoffsets, short *arrayfields) {
  87         int qnodesize;
  88         int len = 0;
  89         int i, rc;
  90
  91         //do {
  92         //      rc=pthread_create(&tPrefetch, NULL, transPrefetch, NULL);
  93         //} while(rc!=0);
  94
  95         /* Allocate for the queue node*/
  96         char *node;
  97         if(ntuples > 0) {
  98                 qnodesize = sizeof(prefetchqelem_t) + sizeof(int) + ntuples * (sizeof(unsigned short) + sizeof(unsigned int)) + endoffsets[ntuples - 1] * sizeof(unsigned short);
  99                 if((node = calloc(1, qnodesize)) == NULL) {
 100                         printf("Calloc Error %s, %d\n", __FILE__, __LINE__);
 101                         return;
 102                 }
 103                 /* Set queue node values */
 104                 len = sizeof(prefetchqelem_t);
 105                 memcpy(node + len, &ntuples, sizeof(int));
 106                 len += sizeof(int);
 107                 memcpy(node + len, oids, ntuples*sizeof(unsigned int));
 108                 len += ntuples * sizeof(unsigned int);
 109                 memcpy(node + len, endoffsets, ntuples*sizeof(unsigned short));
 110                 len += ntuples * sizeof(unsigned short);
 111                 memcpy(node + len, arrayfields, endoffsets[ntuples-1]*sizeof(unsigned short));
 112                 /* Lock and insert into primary prefetch queue */
 113                 pthread_mutex_lock(&pqueue.qlock);
 114                 pre_enqueue((prefetchqelem_t *)node);
 115                 pthread_cond_signal(&pqueue.qcond);
 116                 pthread_mutex_unlock(&pqueue.qlock);
 117         }
 118 }
 119
 120 /* This function starts up the transaction runtime. */
 121 int dstmStartup(const char * option) {
 122   pthread_t thread_Listen;
 123   pthread_attr_t attr;
 124   int master=option!=NULL && strcmp(option, "master")==0;
 125
 126         if (processConfigFile() != 0)
 127                 return 0; //TODO: return error value, cause main program to exit
 128
 129         //TODO Remove after testing
 130         //Initializing the global array
 131         int i;
 132         for (i = 0; i < 10000; i++)
 133                 mlist[i] = NULL;
 134         ////////
 135   dstmInit();
 136   transInit();
 137
 138   if (master) {
 139     pthread_attr_init(&attr);
 140     pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 141     pthread_create(&thread_Listen, &attr, dstmListen, NULL);
 142     return 1;
 143   } else {
 144     dstmListen();
 145     return 0;
 146   }
 147
 148 }
 149
 150 //TODO Use this later
 151 void *pCacheAlloc(objstr_t *store, unsigned int size) {
 152         void *tmp;
 153         objstr_t *ptr;
 154         ptr = store;
 155         int success = 0;
 156
 157         while(ptr->next != NULL) {
 158                 /* check if store is empty */
 159                 if(((unsigned int)ptr->top - (unsigned int)ptr - sizeof(objstr_t) + size) <= ptr->size) {
 160                         tmp = ptr->top;
 161                         ptr->top += size;
 162                         success = 1;
 163                         return tmp;
 164                 } else {
 165                         ptr = ptr-> next;
 166                 }
 167         }
 168
 169         if(success == 0) {
 170                 return NULL;
 171         }
 172 }
 173
 174 /* This function initiates the prefetch thread
 175  * A queue is shared between the main thread of execution
 176  * and the prefetch thread to process the prefetch call
 177  * Call from compiler populates the shared queue with prefetch requests while prefetch thread
 178  * processes the prefetch requests */
 179 void transInit() {
 180         int t, rc;
 181         int retval;
 182         //Create and initialize prefetch cache structure
 183         prefetchcache = objstrCreate(PREFETCH_CACHE_SIZE);
 184         //prefetchcache->next = objstrCreate(PREFETCH_CACHE_SIZE);
 185         //prefetchcache->next->next = objstrCreate(PREFETCH_CACHE_SIZE);
 186
 187         /* Initialize attributes for mutex */
 188         pthread_mutexattr_init(&prefetchcache_mutex_attr);
 189         pthread_mutexattr_settype(&prefetchcache_mutex_attr, PTHREAD_MUTEX_RECURSIVE_NP);
 190
 191         pthread_mutex_init(&prefetchcache_mutex, &prefetchcache_mutex_attr);
 192
 193         //Create prefetch cache lookup table
 194         if(prehashCreate(HASH_SIZE, LOADFACTOR))
 195                 return; //Failure
 196         //Initialize primary shared queue
 197         queueInit();
 198         //Initialize machine pile w/prefetch oids and offsets shared queue
 199         mcpileqInit();
 200
 201         //Create the primary prefetch thread
 202         do {
 203           retval=pthread_create(&tPrefetch, NULL, transPrefetch, NULL);
 204         } while(retval!=0);
 205         pthread_detach(tPrefetch);
 206
 207         //Create and Initialize a pool of threads
 208         /* Threads are active for the entire period runtime is running */
 209         for(t = 0; t< NUM_THREADS; t++) {
 210           do {
 211                 rc = pthread_create(&wthreads[t], NULL, mcqProcess, (void *)t);
 212           } while(rc!=0);
 213           pthread_detach(wthreads[t]);
 214         }
 215 }
 216
 217 /* This function stops the threads spawned */
 218 void transExit() {
 219         int t;
 220         pthread_cancel(tPrefetch);
 221         for(t = 0; t < NUM_THREADS; t++)
 222                 pthread_cancel(wthreads[t]);
 223
 224         return;
 225 }
 226
 227 /* This functions inserts randowm wait delays in the order of msec
 228  * Mostly used when transaction commits retry*/
 229 void randomdelay()
 230 {
 231         struct timespec req;
 232         time_t t;
 233
 234         t = time(NULL);
 235         req.tv_sec = 0;
 236         req.tv_nsec = (long)(1000000 + (t%10000000)); //1-11 msec
 237         nanosleep(&req, NULL);
 238         return;
 239 }
 240
 241 /* This function initializes things required in the transaction start*/
 242 transrecord_t *transStart()
 243 {
 244         transrecord_t *tmp = calloc(1, sizeof(transrecord_t));
 245         tmp->cache = objstrCreate(1048576);
 246         tmp->lookupTable = chashCreate(HASH_SIZE, LOADFACTOR);
 247         //TODO Remove after testing
 248         //Filling the global array when transansaction's record lookupTable
 249         //is calloced
 250         pthread_mutex_lock(&mlock);
 251         int ii;
 252         for (ii = 0; ii < 10000; ii++) {
 253                 if (mlist[ii] == NULL) {
 254                         mlist[ii] = (void *)tmp->lookupTable;
 255                         break;
 256                 }
 257         }
 258         if (ii == 10000) { fprintf(stderr, "Error"); }
 259         pthread_mutex_unlock(&mlock);
 260         ////////////
 261 #ifdef COMPILER
 262         tmp->revertlist=NULL;
 263 #endif
 264         return tmp;
 265 }
 266
 267 /* This function finds the location of the objects involved in a transaction
 268  * and returns the pointer to the object if found in a remote location */
 269 objheader_t *transRead(transrecord_t *record, unsigned int oid) {
 270         unsigned int machinenumber;
 271         objheader_t *tmp, *objheader;
 272         objheader_t *objcopy;
 273         int size, rc, found = 0;
 274         void *buf;
 275         struct timespec ts;
 276         struct timeval tp;
 277
 278         if(oid == 0) {
 279                 return NULL;
 280         }
 281
 282         rc = gettimeofday(&tp, NULL);
 283
 284         /* 1ms delay */
 285         tp.tv_usec += 1000;
 286         if (tp.tv_usec >= 1000000)
 287         {
 288                 tp.tv_usec -= 1000000;
 289                 tp.tv_sec += 1;
 290         }
 291         /* Convert from timeval to timespec */
 292         ts.tv_sec = tp.tv_sec;
 293         ts.tv_nsec = tp.tv_usec * 1000;
 294
 295         /* Search local transaction cache */
 296         if((objheader = (objheader_t *)chashSearch(record->lookupTable, oid)) != NULL){
 297 #ifdef COMPILER
 298           return &objheader[1];
 299 #else
 300           return objheader;
 301 #endif
 302         } else if ((objheader = (objheader_t *) mhashSearch(oid)) != NULL) {
 303                 /* Look up in machine lookup table  and copy  into cache*/
 304                 GETSIZE(size, objheader);
 305                 size += sizeof(objheader_t);
 306                 objcopy = (objheader_t *) objstrAlloc(record->cache, size);
 307                 memcpy(objcopy, objheader, size);
 308                 /* Insert into cache's lookup table */
 309                 chashInsert(record->lookupTable, OID(objheader), objcopy);
 310 #ifdef COMPILER
 311                 return &objcopy[1];
 312 #else
 313                 return objcopy;
 314 #endif
 315         } else if((tmp = (objheader_t *) prehashSearch(oid)) != NULL) { /* Look up in prefetch cache */
 316                 GETSIZE(size, tmp);
 317                 size+=sizeof(objheader_t);
 318                 objcopy = (objheader_t *) objstrAlloc(record->cache, size);
 319                 memcpy(objcopy, tmp, size);
 320                 /* Insert into cache's lookup table */
 321                 chashInsert(record->lookupTable, OID(tmp), objcopy);
 322 #ifdef COMPILER
 323                 return &objcopy[1];
 324 #else
 325                 return objcopy;
 326 #endif
 327         } else {
 328                 /*If object not found in prefetch cache then block until object appears in the prefetch cache */
 329                 pthread_mutex_lock(&pflookup.lock);
 330                 while(!found) {
 331                         rc = pthread_cond_timedwait(&pflookup.cond, &pflookup.lock, &ts);
 332                         /* Check Prefetch cache again */
 333                         if((tmp =(objheader_t *) prehashSearch(oid)) != NULL) {
 334                                 found = 1;
 335                                 GETSIZE(size,tmp);
 336                                 size+=sizeof(objheader_t);
 337                                 objcopy = (objheader_t *) objstrAlloc(record->cache, size);
 338                                 memcpy(objcopy, tmp, size);
 339                                 chashInsert(record->lookupTable, OID(tmp), objcopy);
 340                                 pthread_mutex_unlock(&pflookup.lock);
 341 #ifdef COMPILER
 342                                 return &objcopy[1];
 343 #else
 344                                 return objcopy;
 345 #endif
 346                         } else if (rc == ETIMEDOUT) {
 347                                 pthread_mutex_unlock(&pflookup.lock);
 348                                 break;
 349                         }
 350                 }
 351
 352                 /* Get the object from the remote location */
 353                 machinenumber = lhashSearch(oid);
 354                 objcopy = getRemoteObj(record, machinenumber, oid);
 355                 if(objcopy == NULL) {
 356                         printf("Error: Object not found in Remote location %s, %d\n", __FILE__, __LINE__);
 357                         return NULL;
 358                 } else {
 359 #ifdef COMPILER
 360                         return &objcopy[1];
 361 #else
 362                         return objcopy;
 363 #endif
 364                 }
 365         }
 366 }
 367
 368 /* This function creates objects in the transaction record */
 369 objheader_t *transCreateObj(transrecord_t *record, unsigned int size)
 370 {
 371         objheader_t *tmp = (objheader_t *) objstrAlloc(record->cache, (sizeof(objheader_t) + size));
 372         tmp->notifylist = NULL;
 373         OID(tmp) = getNewOID();
 374         tmp->version = 1;
 375         tmp->rcount = 1;
 376         STATUS(tmp) = NEW;
 377         chashInsert(record->lookupTable, OID(tmp), tmp);
 378
 379 #ifdef COMPILER
 380         return &tmp[1]; //want space after object header
 381 #else
 382         return tmp;
 383 #endif
 384 }
 385
 386 /* This function creates machine piles based on all machines involved in a
 387  * transaction commit request */
 388 plistnode_t *createPiles(transrecord_t *record) {
 389         int i = 0;
 390         unsigned int size;/* Represents number of bins in the chash table */
 391         chashlistnode_t *curr, *ptr, *next;
 392         plistnode_t *pile = NULL;
 393         unsigned int machinenum;
 394         void *localmachinenum;
 395         objheader_t *headeraddr;
 396
 397         ptr = record->lookupTable->table;
 398         size = record->lookupTable->size;
 399
 400         for(i = 0; i < size ; i++) {
 401                 curr = &ptr[i];
 402                 /* Inner loop to traverse the linked list of the cache lookupTable */
 403                 while(curr != NULL) {
 404                         //if the first bin in hash table is empty
 405                         if(curr->key == 0) {
 406                                 break;
 407                         }
 408                         next = curr->next;
 409
 410                         if ((headeraddr = chashSearch(record->lookupTable, curr->key)) == NULL) {
 411                                 printf("Error: No such oid %s, %d\n", __FILE__, __LINE__);
 412                                 return NULL;
 413                         }
 414
 415                         //Get machine location for object id (and whether local or not)
 416                         if (STATUS(headeraddr) & NEW || (mhashSearch(curr->key) != NULL)) {
 417                                 machinenum = myIpAddr;
 418                         } else  if ((machinenum = lhashSearch(curr->key)) == 0) {
 419                                 printf("Error: No such machine %s, %d\n", __FILE__, __LINE__);
 420                                 return NULL;
 421                         }
 422
 423                         //Make machine groups
 424                         if ((pile = pInsert(pile, headeraddr, machinenum, record->lookupTable->numelements)) == NULL) {
 425                                 printf("pInsert error %s, %d\n", __FILE__, __LINE__);
 426                                 return NULL;
 427                         }
 428
 429                         curr = next;
 430                 }
 431         }
 432         return pile;
 433 }
 434
 435 /* This function initiates the transaction commit process
 436  * Spawns threads for each of the new connections with Participants
 437  * and creates new piles by calling the createPiles(),
 438  * Sends a transrequest() to each remote machines for objects found remotely
 439  * and calls handleLocalReq() to process objects found locally */
 440 int transCommit(transrecord_t *record) {
 441         unsigned int tot_bytes_mod, *listmid;
 442         plistnode_t *pile, *pile_ptr;
 443         int i, j, rc, val;
 444         int pilecount, offset, threadnum = 0, trecvcount = 0;
 445         char buffer[RECEIVE_BUFFER_SIZE],control;
 446         char transid[TID_LEN];
 447         trans_req_data_t *tosend;
 448         trans_commit_data_t transinfo;
 449         static int newtid = 0;
 450         char treplyctrl = 0, treplyretry = 0; /* keeps track of the common response that needs to be sent */
 451         char localstat = 0;
 452
 453         /* Look through all the objects in the transaction record and make piles
 454          * for each machine involved in the transaction*/
 455         pile_ptr = pile = createPiles(record);
 456
 457         /* Create the packet to be sent in TRANS_REQUEST */
 458
 459         /* Count the number of participants */
 460         pilecount = pCount(pile);
 461
 462         /* Create a list of machine ids(Participants) involved in transaction   */
 463         if((listmid = calloc(pilecount, sizeof(unsigned int))) == NULL) {
 464                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 465                 return 1;
 466         }
 467         pListMid(pile, listmid);
 468
 469
 470         /* Initialize thread variables,
 471          * Spawn a thread for each Participant involved in a transaction */
 472         pthread_t thread[pilecount];
 473         pthread_attr_t attr;
 474         pthread_cond_t tcond;
 475         pthread_mutex_t tlock;
 476         pthread_mutex_t tlshrd;
 477
 478         thread_data_array_t *thread_data_array;
 479         if((thread_data_array = (thread_data_array_t *) calloc(pilecount, sizeof(thread_data_array_t))) == NULL) {
 480                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 481                 pthread_cond_destroy(&tcond);
 482                 pthread_mutex_destroy(&tlock);
 483                 pDelete(pile_ptr);
 484                 free(listmid);
 485                 return 1;
 486         }
 487
 488         local_thread_data_array_t *ltdata;
 489         if((ltdata = calloc(1, sizeof(local_thread_data_array_t))) == NULL) {
 490                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 491                 pthread_cond_destroy(&tcond);
 492                 pthread_mutex_destroy(&tlock);
 493                 pDelete(pile_ptr);
 494                 free(listmid);
 495                 free(thread_data_array);
 496                 return 1;
 497         }
 498
 499         thread_response_t rcvd_control_msg[pilecount];  /* Shared thread array that keeps track of responses of participants */
 500
 501         /* Initialize and set thread detach attribute */
 502         pthread_attr_init(&attr);
 503         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
 504         pthread_mutex_init(&tlock, NULL);
 505         pthread_cond_init(&tcond, NULL);
 506
 507         /* Process each machine pile */
 508         while(pile != NULL) {
 509                 //Create transaction id
 510                 newtid++;
 511                 if ((tosend = calloc(1, sizeof(trans_req_data_t))) == NULL) {
 512                         printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 513                         pthread_cond_destroy(&tcond);
 514                         pthread_mutex_destroy(&tlock);
 515                         pDelete(pile_ptr);
 516                         free(listmid);
 517                         free(thread_data_array);
 518                         free(ltdata);
 519                         return 1;
 520                 }
 521                 tosend->f.control = TRANS_REQUEST;
 522                 sprintf(tosend->f.trans_id, "%x_%d", pile->mid, newtid);
 523                 tosend->f.mcount = pilecount;
 524                 tosend->f.numread = pile->numread;
 525                 tosend->f.nummod = pile->nummod;
 526                 tosend->f.numcreated = pile->numcreated;
 527                 tosend->f.sum_bytes = pile->sum_bytes;
 528                 tosend->listmid = listmid;
 529                 tosend->objread = pile->objread;
 530                 tosend->oidmod = pile->oidmod;
 531                 tosend->oidcreated = pile->oidcreated;
 532                 thread_data_array[threadnum].thread_id = threadnum;
 533                 thread_data_array[threadnum].mid = pile->mid;
 534                 thread_data_array[threadnum].buffer = tosend;
 535                 thread_data_array[threadnum].recvmsg = rcvd_control_msg;
 536                 thread_data_array[threadnum].threshold = &tcond;
 537                 thread_data_array[threadnum].lock = &tlock;
 538                 thread_data_array[threadnum].count = &trecvcount;
 539                 thread_data_array[threadnum].replyctrl = &treplyctrl;
 540                 thread_data_array[threadnum].replyretry = &treplyretry;
 541                 thread_data_array[threadnum].rec = record;
 542                 /* If local do not create any extra connection */
 543                 if(pile->mid != myIpAddr) { /* Not local */
 544                         do {
 545                                 rc = pthread_create(&thread[threadnum], &attr, transRequest, (void *) &thread_data_array[threadnum]);
 546                         } while(rc!=0);
 547                         if(rc) {
 548                                 perror("Error in pthread create\n");
 549                                 pthread_cond_destroy(&tcond);
 550                                 pthread_mutex_destroy(&tlock);
 551                                 pDelete(pile_ptr);
 552                                 free(listmid);
 553                                 for (i = 0; i < threadnum; i++)
 554                                         free(thread_data_array[i].buffer);
 555                                 free(thread_data_array);
 556                                 free(ltdata);
 557                                 return 1;
 558                         }
 559                 } else { /*Local*/
 560                         ltdata->tdata = &thread_data_array[threadnum];
 561                         ltdata->transinfo = &transinfo;
 562                         do {
 563                                 val = pthread_create(&thread[threadnum], &attr, handleLocalReq, (void *) ltdata);
 564                         } while(val!=0);
 565                         if(val) {
 566                                 perror("Error in pthread create\n");
 567                                 pthread_cond_destroy(&tcond);
 568                                 pthread_mutex_destroy(&tlock);
 569                                 pDelete(pile_ptr);
 570                                 free(listmid);
 571                                 for (i = 0; i < threadnum; i++)
 572                                         free(thread_data_array[i].buffer);
 573                                 free(thread_data_array);
 574                                 free(ltdata);
 575                                 return 1;
 576                         }
 577                 }
 578
 579                 threadnum++;
 580                 pile = pile->next;
 581         }
 582
 583         /* Free attribute and wait for the other threads */
 584         pthread_attr_destroy(&attr);
 585
 586         for (i = 0; i < threadnum; i++) {
 587                 rc = pthread_join(thread[i], NULL);
 588                 if(rc)
 589                 {
 590                         printf("Error: return code from pthread_join() is %d\n", rc);
 591                         pthread_cond_destroy(&tcond);
 592                         pthread_mutex_destroy(&tlock);
 593                         pDelete(pile_ptr);
 594                         free(listmid);
 595                         for (j = i; j < threadnum; j++) {
 596                                 free(thread_data_array[j].buffer);
 597                         }
 598                         return 1;
 599                 }
 600                 free(thread_data_array[i].buffer);
 601         }
 602
 603         /* Free resources */
 604         pthread_cond_destroy(&tcond);
 605         pthread_mutex_destroy(&tlock);
 606         free(listmid);
 607         pDelete(pile_ptr);
 608
 609
 610         if(treplyctrl == TRANS_ABORT) {
 611                 /* Free Resources */
 612                 objstrDelete(record->cache);
 613                 //TODO Remove after testing
 614                 pthread_mutex_lock(&mlock);
 615                 int count = 0, jj = 0, ii = 0;
 616                 for (ii = 0; ii < 10000; ii++) {
 617                         if (mlist[ii] == (void *) record->lookupTable) {
 618                                 count++;
 619                                 jj = ii;
 620                         }
 621                 }
 622                 if (count==2 || count == 0) {
 623                         fprintf(stderr, "TRANS_ABORT CASE: Count for same addr:%d\n", count);
 624                 }
 625                 if (count == 1)
 626                         mlist[jj] = 0;
 627                 pthread_mutex_unlock(&mlock);
 628                 ////////////
 629                 chashDelete(record->lookupTable);
 630                 free(record);
 631                 free(thread_data_array);
 632                 free(ltdata);
 633                 return TRANS_ABORT;
 634         } else if(treplyctrl == TRANS_COMMIT) {
 635                 /* Free Resources */
 636                 objstrDelete(record->cache);
 637                 //TODO Remove after Testing
 638                 pthread_mutex_lock(&mlock);
 639                 int count = 0, jj = 0, ii = 0;
 640                 for (ii = 0; ii < 10000; ii++) {
 641                         if (mlist[ii] == (void *) record->lookupTable) {count++; jj = ii;}
 642                 }
 643                 if (count==2 || count == 0) {
 644                         fprintf(stderr, "TRANS_COMMIT CASE: Count for same addr:%d\n", count);
 645                 }
 646                 if (count == 1) mlist[jj] = 0;
 647                 pthread_mutex_unlock(&mlock);
 648                 ///////////////////////
 649                 chashDelete(record->lookupTable);
 650                 free(record);
 651                 free(thread_data_array);
 652                 free(ltdata);
 653                 return 0;
 654         } else {
 655                 //TODO Add other cases
 656                 printf("DEBUG-> THIS SHOULD NOT HAPPEN.....EXIT PROGRAM\n");
 657                 exit(-1);
 658         }
 659
 660         return 0;
 661 }
 662
 663 /* This function sends information involved in the transaction request
 664  * to participants and accepts a response from particpants.
 665  * It calls decideresponse() to decide on what control message
 666  * to send next to participants and sends the message using sendResponse()*/
 667 void *transRequest(void *threadarg) {
 668         int sd, i, n;
 669         struct sockaddr_in serv_addr;
 670         struct hostent *server;
 671         thread_data_array_t *tdata;
 672         objheader_t *headeraddr;
 673         char buffer[RECEIVE_BUFFER_SIZE], control, recvcontrol;
 674         char machineip[16], retval;
 675
 676
 677         tdata = (thread_data_array_t *) threadarg;
 678
 679         /* Send Trans Request */
 680         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 681                 perror("Error in socket for TRANS_REQUEST\n");
 682                 pthread_exit(NULL);
 683         }
 684         bzero((char*) &serv_addr, sizeof(serv_addr));
 685         serv_addr.sin_family = AF_INET;
 686         serv_addr.sin_port = htons(LISTEN_PORT);
 687         midtoIP(tdata->mid,machineip);
 688         machineip[15] = '\0';
 689         serv_addr.sin_addr.s_addr = inet_addr(machineip);
 690         /* Open Connection */
 691         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
 692                 perror("Error in connect for TRANS_REQUEST\n");
 693                 close(sd);
 694                 pthread_exit(NULL);
 695         }
 696
 697         /* Send bytes of data with TRANS_REQUEST control message */
 698         if (send(sd, &(tdata->buffer->f), sizeof(fixed_data_t),MSG_NOSIGNAL) < sizeof(fixed_data_t)) {
 699                 perror("Error sending fixed bytes for thread\n");
 700                 close(sd);
 701                 pthread_exit(NULL);
 702         }
 703         /* Send list of machines involved in the transaction */
 704         {
 705                 int size=sizeof(unsigned int)*tdata->buffer->f.mcount;
 706                 if (send(sd, tdata->buffer->listmid, size, MSG_NOSIGNAL) < size) {
 707                         perror("Error sending list of machines for thread\n");
 708                         close(sd);
 709                         pthread_exit(NULL);
 710                 }
 711         }
 712         /* Send oids and version number tuples for objects that are read */
 713         {
 714                 int size=(sizeof(unsigned int)+sizeof(short))*tdata->buffer->f.numread;
 715                 if (send(sd, tdata->buffer->objread, size, MSG_NOSIGNAL) < size) {
 716                         perror("Error sending tuples for thread\n");
 717                         close(sd);
 718                         pthread_exit(NULL);
 719                 }
 720         }
 721         /* Send objects that are modified */
 722         for(i = 0; i < tdata->buffer->f.nummod ; i++) {
 723                 int size;
 724                 headeraddr = chashSearch(tdata->rec->lookupTable, tdata->buffer->oidmod[i]);
 725                 GETSIZE(size,headeraddr);
 726                 size+=sizeof(objheader_t);
 727                 if (send(sd, headeraddr, size, MSG_NOSIGNAL)  < size) {
 728                         perror("Error sending obj modified for thread\n");
 729                         close(sd);
 730                         pthread_exit(NULL);
 731                 }
 732         }
 733
 734         /* Read control message from Participant */
 735         if((n = read(sd, &control, sizeof(char))) <= 0) {
 736                 perror("Error in reading control message from Participant\n");
 737                 close(sd);
 738                 pthread_exit(NULL);
 739         }
 740
 741         recvcontrol = control;
 742
 743         /* Update common data structure and increment count */
 744         tdata->recvmsg[tdata->thread_id].rcv_status = recvcontrol;
 745
 746         /* Lock and update count */
 747         /* Thread sleeps until all messages from pariticipants are received by coordinator */
 748         pthread_mutex_lock(tdata->lock);
 749
 750         (*(tdata->count))++; /* keeps track of no of messages received by the coordinator */
 751
 752         /* Wake up the threads and invoke decideResponse (once) */
 753         if(*(tdata->count) == tdata->buffer->f.mcount) {
 754                 decideResponse(tdata);
 755                 pthread_cond_broadcast(tdata->threshold);
 756         } else {
 757                 pthread_cond_wait(tdata->threshold, tdata->lock);
 758         }
 759         pthread_mutex_unlock(tdata->lock);
 760
 761         /* Send the final response such as TRANS_COMMIT or TRANS_ABORT t
 762          * to all participants in their respective socket */
 763         if (sendResponse(tdata, sd) == 0) {
 764                 printf("sendResponse returned error %s,%d\n", __FILE__, __LINE__);
 765                 close(sd);
 766                 pthread_exit(NULL);
 767         }
 768
 769         if ((retval = recv((int)sd, &control, sizeof(char), 0))<= 0) {
 770                 printf("Error: In receiving control %s,%d\n", __FILE__, __LINE__);
 771                 close(sd);
 772                 pthread_exit(NULL);
 773         }
 774
 775         if(control == TRANS_UNSUCESSFUL) {
 776                 //printf("DEBUG-> TRANS_ABORTED\n");
 777         } else if(control == TRANS_SUCESSFUL) {
 778                 //printf("DEBUG-> TRANS_SUCCESSFUL\n");
 779         } else {
 780                 //printf("DEBUG-> Error: Incorrect Transaction End Message %d\n", control);
 781         }
 782
 783         /* Close connection */
 784         close(sd);
 785         pthread_exit(NULL);
 786 }
 787
 788 /* This function decides the reponse that needs to be sent to
 789  * all Participant machines after the TRANS_REQUEST protocol */
 790 void decideResponse(thread_data_array_t *tdata) {
 791         char control;
 792         int i, transagree = 0, transdisagree = 0, transsoftabort = 0; /* Counters to formulate decision of what
 793                                                                          message to send */
 794
 795         for (i = 0 ; i < tdata->buffer->f.mcount; i++) {
 796                 control = tdata->recvmsg[i].rcv_status; /* tdata: keeps track of all participant responses
 797                                                            written onto the shared array */
 798                 switch(control) {
 799                         default:
 800                                 printf("Participant sent unknown message in %s, %d\n", __FILE__, __LINE__);
 801                                 /* treat as disagree, pass thru */
 802                         case TRANS_DISAGREE:
 803                                 transdisagree++;
 804                                 break;
 805
 806                         case TRANS_AGREE:
 807                                 transagree++;
 808                                 break;
 809
 810                         case TRANS_SOFT_ABORT:
 811                                 transsoftabort++;
 812                                 break;
 813                 }
 814         }
 815
 816         if(transdisagree > 0) {
 817                 /* Send Abort */
 818                 *(tdata->replyctrl) = TRANS_ABORT;
 819                 *(tdata->replyretry) = 0;
 820                 /* clear objects from prefetch cache */
 821                 for (i = 0; i < tdata->buffer->f.numread; i++)
 822                         prehashRemove(*((unsigned int *)(((char *)tdata->buffer->objread) + (sizeof(unsigned int) + sizeof(unsigned short))*i)));
 823                 for (i = 0; i < tdata->buffer->f.nummod; i++)
 824                         prehashRemove(tdata->buffer->oidmod[i]);
 825         } else if(transagree == tdata->buffer->f.mcount){
 826                 /* Send Commit */
 827                 *(tdata->replyctrl) = TRANS_COMMIT;
 828                 *(tdata->replyretry) = 0;
 829         } else {
 830                 /* Send Abort in soft abort case followed by retry commiting transaction again*/
 831                 *(tdata->replyctrl) = TRANS_ABORT;
 832                 *(tdata->replyretry) = 1;
 833         }
 834
 835         return;
 836 }
 837 /* This function sends the final response to remote machines per thread in their respective socket id
 838  * It returns a char that is only needed to check the correctness of execution of this function inside
 839  * transRequest()*/
 840 char sendResponse(thread_data_array_t *tdata, int sd) {
 841         int n, N, sum, oidcount = 0, control;
 842         char *ptr, retval = 0;
 843         unsigned int *oidnotfound;
 844
 845         /* If the decided response is due to a soft abort and missing objects at the Participant's side */
 846         if(tdata->recvmsg[tdata->thread_id].rcv_status == TRANS_SOFT_ABORT) {
 847                 /* Read list of objects missing */
 848                 if((read(sd, &oidcount, sizeof(int)) != 0) && (oidcount != 0)) {
 849                         N = oidcount * sizeof(unsigned int);
 850                         if((oidnotfound = calloc(oidcount, sizeof(unsigned int))) == NULL) {
 851                                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 852                                 return 0;
 853                         }
 854                         ptr = (char *) oidnotfound;
 855                         do {
 856                                 n = read(sd, ptr+sum, N-sum);
 857                                 sum += n;
 858                         } while(sum < N && n !=0);
 859                 }
 860                 retval =  TRANS_SOFT_ABORT;
 861         }
 862         /* If the decided response is TRANS_ABORT */
 863         if(*(tdata->replyctrl) == TRANS_ABORT) {
 864                 retval = TRANS_ABORT;
 865         } else if(*(tdata->replyctrl) == TRANS_COMMIT) { /* If the decided response is TRANS_COMMIT */
 866                 retval = TRANS_COMMIT;
 867         }
 868
 869         if (send(sd, tdata->replyctrl, sizeof(char),MSG_NOSIGNAL) < sizeof(char)) {
 870                 perror("Error sending ctrl message for participant\n");
 871                 return 0;
 872         }
 873
 874         return retval;
 875 }
 876
 877 /* This function opens a connection, places an object read request to the
 878  * remote machine, reads the control message and object if available  and
 879  * copies the object and its header to the local cache.
 880  * */
 881
 882 void *getRemoteObj(transrecord_t *record, unsigned int mnum, unsigned int oid) {
 883         int sd, size, val;
 884         struct sockaddr_in serv_addr;
 885         struct hostent *server;
 886         char control;
 887         char machineip[16];
 888         objheader_t *h;
 889         void *objcopy;
 890
 891
 892         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 893                 perror("Error in socket\n");
 894                 return NULL;
 895         }
 896
 897         bzero((char*) &serv_addr, sizeof(serv_addr));
 898         serv_addr.sin_family = AF_INET;
 899         serv_addr.sin_port = htons(LISTEN_PORT);
 900         midtoIP(mnum,machineip);
 901         machineip[15] = '\0';
 902         serv_addr.sin_addr.s_addr = inet_addr(machineip);
 903
 904         /* Open connection */
 905         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
 906                 perror("Error in connect\n");
 907                 return NULL;
 908         }
 909         char readrequest[sizeof(char)+sizeof(unsigned int)];
 910         readrequest[0] = READ_REQUEST;
 911         *((unsigned int *)(&readrequest[1])) = oid;
 912         if (send(sd, readrequest, sizeof(readrequest), MSG_NOSIGNAL) < sizeof(readrequest)) {
 913                 perror("Error sending message\n");
 914                 return NULL;
 915         }
 916
 917         /* Read response from the Participant */
 918         if((val = read(sd, &control, sizeof(char))) <= 0) {
 919                 perror("No control response for getRemoteObj sent\n");
 920                 return NULL;
 921         }
 922         switch(control) {
 923                 case OBJECT_NOT_FOUND:
 924                         return NULL;
 925                 case OBJECT_FOUND:
 926                         /* Read object if found into local cache */
 927                         if((val = read(sd, &size, sizeof(int))) <= 0) {
 928                                 perror("No size is read from the participant\n");
 929                                 return NULL;
 930                         }
 931                         objcopy = objstrAlloc(record->cache, size);
 932                         if((val = read(sd, (char *)objcopy, size)) <= 0) {
 933                                 perror("No objects are read from the remote participant\n");
 934                                 return NULL;
 935                         }
 936                         /* Insert into cache's lookup table */
 937                         chashInsert(record->lookupTable, oid, objcopy);
 938                         break;
 939                 default:
 940                         printf("Error in recv request from participant on a READ_REQUEST %s, %d\n",__FILE__, __LINE__);
 941                         return NULL;
 942         }
 943         /* Close connection */
 944         close(sd);
 945         return objcopy;
 946 }
 947
 948 /* This function handles the local objects involved in a transaction commiting process.
 949  * It also makes a decision if this local machine sends AGREE or DISAGREE or SOFT_ABORT to coordinator.
 950  * Note Coordinator = local machine
 951  * It wakes up the other threads from remote participants that are waiting for the coordinator's decision and
 952  * based on common agreement it either commits or aborts the transaction.
 953  * It also frees the memory resources */
 954 void *handleLocalReq(void *threadarg) {
 955         unsigned int *oidnotfound = NULL, *oidlocked = NULL;
 956         local_thread_data_array_t *localtdata;
 957         int objnotfound = 0, objlocked = 0;
 958         int v_nomatch = 0, v_matchlock = 0, v_matchnolock = 0;
 959         int numread, i;
 960         unsigned int oid;
 961         unsigned short version;
 962         void *mobj;
 963         objheader_t *headptr;
 964
 965         localtdata = (local_thread_data_array_t *) threadarg;
 966
 967         /* Counters and arrays to formulate decision on control message to be sent */
 968         oidnotfound = (unsigned int *) calloc((localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod), sizeof(unsigned int));
 969         oidlocked = (unsigned int *) calloc((localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod), sizeof(unsigned int));
 970
 971         numread = localtdata->tdata->buffer->f.numread;
 972         /* Process each oid in the machine pile/ group per thread */
 973         for (i = 0; i < localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod; i++) {
 974                 if (i < localtdata->tdata->buffer->f.numread) {
 975                         int incr = sizeof(unsigned int) + sizeof(short);// Offset that points to next position in the objread array
 976                         incr *= i;
 977                         oid = *((unsigned int *)(((char *)localtdata->tdata->buffer->objread) + incr));
 978                         version = *((short *)(((char *)localtdata->tdata->buffer->objread) + incr + sizeof(unsigned int)));
 979                 } else { // Objects Modified
 980                         int tmpsize;
 981                         headptr = (objheader_t *) chashSearch(localtdata->tdata->rec->lookupTable, localtdata->tdata->buffer->oidmod[i-numread]);
 982                         if (headptr == NULL) {
 983                                 printf("Error: handleLocalReq() returning NULL %s, %d\n", __FILE__, __LINE__);
 984                                 return NULL;
 985                         }
 986                         oid = OID(headptr);
 987                         version = headptr->version;
 988                 }
 989                 /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
 990
 991                 /* Save the oids not found and number of oids not found for later use */
 992                 if ((mobj = mhashSearch(oid)) == NULL) { /* Obj not found */
 993                         /* Save the oids not found and number of oids not found for later use */
 994                         oidnotfound[objnotfound] = oid;
 995                         objnotfound++;
 996                 } else { /* If Obj found in machine (i.e. has not moved) */
 997                         /* Check if Obj is locked by any previous transaction */
 998                         if ((STATUS((objheader_t *)mobj) & LOCK) == LOCK) {
 999                                 if (version == ((objheader_t *)mobj)->version) {      /* If locked then match versions */
1000                                         v_matchlock++;
1001                                 } else {/* If versions don't match ...HARD ABORT */
1002                                         v_nomatch++;
1003                                         /* Send TRANS_DISAGREE to Coordinator */
1004                                         localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
1005                                 }
1006                         } else {/* If Obj is not locked then lock object */
1007                                 STATUS(((objheader_t *)mobj)) |= LOCK;
1008                                 /* Save all object oids that are locked on this machine during this transaction request call */
1009                                 oidlocked[objlocked] = OID(((objheader_t *)mobj));
1010                                 objlocked++;
1011                                 if (version == ((objheader_t *)mobj)->version) { /* Check if versions match */
1012                                         v_matchnolock++;
1013                                 } else { /* If versions don't match ...HARD ABORT */
1014                                         v_nomatch++;
1015                                         /* Send TRANS_DISAGREE to Coordinator */
1016                                         localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
1017                                 }
1018                         }
1019                 }
1020         } // End for
1021         /* Condition to send TRANS_AGREE */
1022         if(v_matchnolock == localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod) {
1023                 localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_AGREE;
1024         }
1025         /* Condition to send TRANS_SOFT_ABORT */
1026         if((v_matchlock > 0 && v_nomatch == 0) || (objnotfound > 0 && v_nomatch == 0)) {
1027                 localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_SOFT_ABORT;
1028         }
1029
1030         /* Fill out the trans_commit_data_t data structure. This is required for a trans commit process
1031          * if Participant receives a TRANS_COMMIT */
1032         localtdata->transinfo->objlocked = oidlocked;
1033         localtdata->transinfo->objnotfound = oidnotfound;
1034         localtdata->transinfo->modptr = NULL;
1035         localtdata->transinfo->numlocked = objlocked;
1036         localtdata->transinfo->numnotfound = objnotfound;
1037         /* Lock and update count */
1038         //Thread sleeps until all messages from pariticipants are received by coordinator
1039         pthread_mutex_lock(localtdata->tdata->lock);
1040         (*(localtdata->tdata->count))++; /* keeps track of no of messages received by the coordinator */
1041
1042         /* Wake up the threads and invoke decideResponse (once) */
1043         if(*(localtdata->tdata->count) == localtdata->tdata->buffer->f.mcount) {
1044                 decideResponse(localtdata->tdata);
1045                 pthread_cond_broadcast(localtdata->tdata->threshold);
1046         } else {
1047                 pthread_cond_wait(localtdata->tdata->threshold, localtdata->tdata->lock);
1048         }
1049         pthread_mutex_unlock(localtdata->tdata->lock);
1050         if(*(localtdata->tdata->replyctrl) == TRANS_ABORT){
1051                 if(transAbortProcess(localtdata) != 0) {
1052                         printf("Error in transAbortProcess() %s,%d\n", __FILE__, __LINE__);
1053                         pthread_exit(NULL);
1054                 }
1055         } else if(*(localtdata->tdata->replyctrl) == TRANS_COMMIT) {
1056                 if(transComProcess(localtdata) != 0) {
1057                         printf("Error in transComProcess() %s,%d\n", __FILE__, __LINE__);
1058                         pthread_exit(NULL);
1059                 }
1060         }
1061         /* Free memory */
1062         if (localtdata->transinfo->objlocked != NULL) {
1063                 free(localtdata->transinfo->objlocked);
1064         }
1065         if (localtdata->transinfo->objnotfound != NULL) {
1066                 free(localtdata->transinfo->objnotfound);
1067         }
1068
1069         pthread_exit(NULL);
1070 }
1071
1072 /* This function completes the ABORT process if the transaction is aborting */
1073 int transAbortProcess(local_thread_data_array_t  *localtdata) {
1074         int i, numlocked;
1075         unsigned int *objlocked;
1076         void *header;
1077
1078         numlocked = localtdata->transinfo->numlocked;
1079         objlocked = localtdata->transinfo->objlocked;
1080
1081         for (i = 0; i < numlocked; i++) {
1082                 if((header = mhashSearch(objlocked[i])) == NULL) {
1083                         printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
1084                         return 1;
1085                 }
1086                 STATUS(((objheader_t *)header)) &= ~(LOCK);
1087         }
1088
1089         return 0;
1090 }
1091
1092 /*This function completes the COMMIT process is the transaction is commiting*/
1093 int transComProcess(local_thread_data_array_t  *localtdata) {
1094         objheader_t *header, *tcptr;
1095         int i, nummod, tmpsize, numcreated, numlocked;
1096         unsigned int *oidmod, *oidcreated, *oidlocked;
1097         void *ptrcreate;
1098
1099         nummod = localtdata->tdata->buffer->f.nummod;
1100         oidmod = localtdata->tdata->buffer->oidmod;
1101         numcreated = localtdata->tdata->buffer->f.numcreated;
1102         oidcreated = localtdata->tdata->buffer->oidcreated;
1103         numlocked = localtdata->transinfo->numlocked;
1104         oidlocked = localtdata->transinfo->objlocked;
1105
1106         for (i = 0; i < nummod; i++) {
1107                 if((header = (objheader_t *) mhashSearch(oidmod[i])) == NULL) {
1108                         printf("Error: transComProcess() mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
1109                         return 1;
1110                 }
1111                 /* Copy from transaction cache -> main object store */
1112                 if ((tcptr = ((objheader_t *) chashSearch(localtdata->tdata->rec->lookupTable, oidmod[i]))) == NULL) {
1113                         printf("Error: transComProcess() chashSearch returned NULL at %s, %d\n", __FILE__, __LINE__);
1114                         return 1;
1115                 }
1116                 GETSIZE(tmpsize, header);
1117                 pthread_mutex_lock(&mainobjstore_mutex);
1118                 memcpy((char*)header+sizeof(objheader_t), (char *)tcptr+ sizeof(objheader_t), tmpsize);
1119                 header->version += 1;
1120                 if(header->notifylist != NULL) {
1121                         notifyAll(&header->notifylist, OID(header), header->version);
1122                 }
1123                 pthread_mutex_unlock(&mainobjstore_mutex);
1124         }
1125         /* If object is newly created inside transaction then commit it */
1126         for (i = 0; i < numcreated; i++) {
1127                 if ((header = ((objheader_t *) chashSearch(localtdata->tdata->rec->lookupTable, oidcreated[i]))) == NULL) {
1128                         printf("Error: transComProcess() chashSearch returned NULL at %s, %d\n", __FILE__, __LINE__);
1129                         return 1;
1130                 }
1131                 GETSIZE(tmpsize, header);
1132                 tmpsize += sizeof(objheader_t);
1133                 pthread_mutex_lock(&mainobjstore_mutex);
1134                 if ((ptrcreate = objstrAlloc(mainobjstore, tmpsize)) == NULL) {
1135                         printf("Error: transComProcess() failed objstrAlloc %s, %d\n", __FILE__, __LINE__);
1136                         pthread_mutex_unlock(&mainobjstore_mutex);
1137                         return 1;
1138                 }
1139                 pthread_mutex_unlock(&mainobjstore_mutex);
1140                 memcpy(ptrcreate, header, tmpsize);
1141                 mhashInsert(oidcreated[i], ptrcreate);
1142                 lhashInsert(oidcreated[i], myIpAddr);
1143         }
1144         /* Unlock locked objects */
1145         for(i = 0; i < numlocked; i++) {
1146                 if((header = (objheader_t *) mhashSearch(oidlocked[i])) == NULL) {
1147                         printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
1148                         return 1;
1149                 }
1150                 STATUS(header) &= ~(LOCK);
1151         }
1152
1153         return 0;
1154 }
1155
1156 /* This function checks if the prefetch oids are same and have same offsets
1157  * for case x.a.b and y.a.b where x and y have same oid's
1158  * or if a.b.c is a subset of x.b.c.d*/
1159 /* check for case where the generated request a.y.z or x.y.z.g then
1160  * prefetch needs to be generated for x.y.z.g  if oid of a and x are same*/
1161 void checkPrefetchTuples(prefetchqelem_t *node) {
1162         int i,j, count,k, sindex, index;
1163         char *ptr, *tmp;
1164         int ntuples, slength;
1165         unsigned int *oid;
1166         short *endoffsets, *arryfields;
1167
1168         /* Check for the case x.y.z and a.b.c are same oids */
1169         ptr = (char *) node;
1170         ntuples = *(GET_NTUPLES(ptr));
1171         oid = GET_PTR_OID(ptr);
1172         endoffsets = GET_PTR_EOFF(ptr, ntuples);
1173         arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1174
1175         /* Find offset length for each tuple */
1176         int numoffset[ntuples];
1177         numoffset[0] = endoffsets[0];
1178         for(i = 1; i<ntuples; i++) {
1179                 numoffset[i] = endoffsets[i] - endoffsets[i-1];
1180         }
1181         /* Check for redundant tuples by comparing oids of each tuple */
1182         for(i = 0; i < ntuples; i++) {
1183                 if(oid[i] == 0)
1184                         continue;
1185                 for(j = i+1 ; j < ntuples; j++) {
1186                         if(oid[j] == 0)
1187                                 continue;
1188                         /*If oids of tuples match */
1189                         if (oid[i] == oid[j]) {
1190                                 /* Find the smallest offset length of two tuples*/
1191                                 if(numoffset[i] >  numoffset[j]){
1192                                         slength = numoffset[j];
1193                                         sindex = j;
1194                                 }
1195                                 else {
1196                                         slength = numoffset[i];
1197                                         sindex = i;
1198                                 }
1199
1200                                 /* Compare the offset values based on the current indices
1201                                  * break if they do not match
1202                                  * if all offset values match then pick the largest tuple*/
1203
1204                                 if(i == 0) {
1205                                         k = 0;
1206                                 } else {
1207                                         k = endoffsets[i-1];
1208                                 }
1209                                 index = endoffsets[j -1];
1210                                 for(count = 0; count < slength; count ++) {
1211                                         if (arryfields[k] != arryfields[index]) {
1212                                                 break;
1213                                         }
1214                                         index++;
1215                                         k++;
1216                                 }
1217                                 if(slength == count) {
1218                                         oid[sindex] = 0;
1219                                 }
1220                         }
1221                 }
1222         }
1223 }
1224 /* This function makes machine piles to be added into the machine pile queue for each prefetch call */
1225 prefetchpile_t *makePreGroups(prefetchqelem_t *node, int *numoffset) {
1226         char *ptr;
1227         int ntuples, i, machinenum, count=0;
1228         unsigned int *oid;
1229         short *endoffsets, *arryfields, *offset;
1230         prefetchpile_t *head = NULL, *tmp = NULL;
1231
1232         /* Check for the case x.y.z and a.b.c are same oids */
1233         ptr = (char *) node;
1234         ntuples = *(GET_NTUPLES(ptr));
1235         oid = GET_PTR_OID(ptr);
1236         endoffsets = GET_PTR_EOFF(ptr, ntuples);
1237         arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1238
1239         if((head = (prefetchpile_t *) calloc(1, sizeof(prefetchpile_t))) == NULL) {
1240                 printf("Calloc error: %s %d\n", __FILE__, __LINE__);
1241                 return NULL;
1242         }
1243
1244         /* Check for redundant tuples by comparing oids of each tuple */
1245         for(i = 0; i < ntuples; i++) {
1246                 if(oid[i] == 0){
1247                         if(head->next != NULL) {
1248                                 if((tmp = (prefetchpile_t *) calloc(1, sizeof(prefetchpile_t))) == NULL) {
1249                                         printf("Calloc error: %s %d\n", __FILE__, __LINE__);
1250                                         return NULL;
1251                                 }
1252                                 tmp->mid = myIpAddr;
1253                                 tmp->next = head;
1254                                 head = tmp;
1255                         } else {
1256                                 head->mid = myIpAddr;
1257                         }
1258                         continue;
1259                 }
1260                 /* For each tuple make piles */
1261                 if ((machinenum = lhashSearch(oid[i])) == 0) {
1262                         printf("Error: No such Machine %s, %d\n", __FILE__, __LINE__);
1263                         return NULL;
1264                 }
1265                 /* Insert into machine pile */
1266                 if(i == 0){
1267                         offset = &arryfields[0];
1268                 } else {
1269                         offset = &arryfields[endoffsets[i-1]];
1270                 }
1271
1272                 if((head = insertPile(machinenum, oid[i], numoffset[i], offset, head)) == NULL){
1273                         printf("Error: Couldn't create a pile %s, %d\n", __FILE__, __LINE__);
1274                         return NULL;
1275                 }
1276         }
1277
1278         return head;
1279 }
1280
1281 prefetchpile_t *foundLocal(prefetchqelem_t *node) {
1282         int ntuples,i, j, k, oidnfound = 0, arryfieldindex,nextarryfieldindex, flag = 0, val;
1283         unsigned int *oid;
1284         unsigned int  objoid;
1285         int isArray = 0;
1286         char *ptr, *tmp;
1287         objheader_t *objheader;
1288         short *endoffsets, *arryfields;
1289         prefetchpile_t *head = NULL;
1290
1291         ptr = (char *) node;
1292         ntuples = *(GET_NTUPLES(ptr));
1293         oid = GET_PTR_OID(ptr);
1294         endoffsets = GET_PTR_EOFF(ptr, ntuples);
1295         arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1296
1297         /* Find offset length for each tuple */
1298         int numoffset[ntuples];//Number of offsets for each tuple
1299         numoffset[0] = endoffsets[0];
1300         for(i = 1; i<ntuples; i++) {
1301                 numoffset[i] = endoffsets[i] - endoffsets[i-1];
1302         }
1303         for(i = 0; i < ntuples; i++) {
1304                 if(oid[i] == 0){
1305                         if(i == 0) {
1306                                 arryfieldindex = 0;
1307                                 nextarryfieldindex =  endoffsets[0];
1308                         }else {
1309                                 arryfieldindex = endoffsets[i-1];
1310                                 nextarryfieldindex =  endoffsets[i];
1311                         }
1312                         numoffset[i] = 0;
1313                         endoffsets[0] = val = numoffset[0];
1314                         for(k = 1; k < ntuples; k++) {
1315                                 val = val + numoffset[k];
1316                                 endoffsets[k] = val;
1317                         }
1318
1319                         for(k = 0; k<endoffsets[ntuples-1]; k++) {
1320                                 arryfields[arryfieldindex+k] = arryfields[nextarryfieldindex+k];
1321                         }
1322                         continue;
1323                 }
1324                 /* If object found locally */
1325                 if((objheader = (objheader_t*) mhashSearch(oid[i])) != NULL) {
1326                         tmp = (char *) objheader;
1327                         int orgnumoffset = numoffset[i];
1328                         if(i == 0) {
1329                                 arryfieldindex = 0;
1330                         }else {
1331                                 arryfieldindex = endoffsets[i-1];
1332                         }
1333
1334                         for(j = 0; j<orgnumoffset; j++) {
1335                                 /* Check for arrays  */
1336                                 if(TYPE(objheader) > NUMCLASSES) {
1337                                         isArray = 1;
1338                                 }
1339                                 if(isArray == 1) {
1340                                         int elementsize = classsize[TYPE(objheader)];
1341                                         objoid = *((unsigned int *)(tmp + sizeof(objheader_t) + sizeof(struct ArrayObject) + (elementsize*arryfields[arryfieldindex])));
1342                                 } else {
1343                                         objoid = *((unsigned int *)(tmp + sizeof(objheader_t) + arryfields[arryfieldindex]));
1344                                 }
1345                                 //Update numoffset array
1346                                 numoffset[i] = numoffset[i] - 1;
1347                                 //Update oid array
1348                                 oid[i] = objoid;
1349                                 //Update endoffset array
1350                                 endoffsets[0] = val = numoffset[0];
1351                                 for(k = 1; k < ntuples; k++) {
1352                                         val = val + numoffset[k];
1353                                         endoffsets[k] = val;
1354                                 }
1355                                 //Update arrayfields array
1356                                 for(k = 0; k < endoffsets[ntuples-1]; k++) {
1357                                         arryfields[arryfieldindex+k] = arryfields[arryfieldindex+k+1];
1358                                 }
1359                                 if((objheader = (objheader_t*) mhashSearch(oid[i])) == NULL) {
1360                                         flag = 1;
1361                                         checkPreCache(node, numoffset, oid[i], i);
1362                                         break;
1363                                 }
1364                                 tmp = (char *) objheader;
1365                                 isArray = 0;
1366                         }
1367                         /*If all offset oids are found locally,make the prefetch tuple invalid */
1368                         if(flag == 0) {
1369                                 oid[i] = 0;
1370                         }
1371                 } else {
1372                         /* Look in Prefetch cache */
1373                         checkPreCache(node, numoffset, oid[i],i);
1374                 }
1375         }
1376
1377         /* Make machine groups */
1378         if((head = makePreGroups(node, numoffset)) == NULL) {
1379                 printf("Error in makePreGroups() %s %d\n", __FILE__, __LINE__);
1380                 return NULL;
1381         }
1382
1383         return head;
1384 }
1385
1386 void checkPreCache(prefetchqelem_t *node, int *numoffset, unsigned int objoid, int index) {
1387         char *ptr, *tmp;
1388         int ntuples, i, k, flag=0, isArray =0, arryfieldindex, val;
1389         unsigned int * oid;
1390         short *endoffsets, *arryfields;
1391         objheader_t *header;
1392
1393         ptr = (char *) node;
1394         ntuples = *(GET_NTUPLES(ptr));
1395         oid = GET_PTR_OID(ptr);
1396         endoffsets = GET_PTR_EOFF(ptr, ntuples);
1397         arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1398
1399         if((header = (objheader_t *) prehashSearch(objoid)) == NULL) {
1400                 return;
1401         } else { //Found in Prefetch Cache
1402                 //TODO Decide if object is too old, if old remove from cache
1403                 tmp = (char *) header;
1404                 int loopcount = numoffset[index];
1405                 if(index == 0)
1406                         arryfieldindex = 0;
1407                 else
1408                         arryfieldindex = endoffsets[(index - 1)];
1409                 // Check if any of the offset oid is available in the Prefetch cache
1410                 for(i = 0; i < loopcount; i++) {
1411                         /* Check for arrays  */
1412                         if(TYPE(header) > NUMCLASSES) {
1413                                 isArray = 1;
1414                         }
1415                         if(isArray == 1) {
1416                                 int elementsize = classsize[TYPE(header)];
1417                                 objoid = *((unsigned int *)(tmp + sizeof(objheader_t) + sizeof(struct ArrayObject) + (elementsize*arryfields[arryfieldindex])));
1418                         } else {
1419                                 objoid = *((unsigned int *)(tmp + sizeof(objheader_t) + arryfields[arryfieldindex]));
1420                         }
1421                         //Update numoffset array
1422                         numoffset[index] = numoffset[index] - 1;
1423                         //Update oid array
1424                         oid[index] = objoid;
1425                         //Update endoffset array
1426                         endoffsets[0] = val = numoffset[0];
1427                         for(k = 1; k < ntuples; k++) {
1428                                 val = val + numoffset[k];
1429                                 endoffsets[k] = val;
1430                         }
1431                         //Update arrayfields array
1432                         for(k = 0; k < endoffsets[ntuples-1]; k++) {
1433                                 arryfields[arryfieldindex+k] = arryfields[arryfieldindex+k+1];
1434                         }
1435                         if((header = (objheader_t *)prehashSearch(oid[index])) != NULL) {
1436                                 tmp = (char *) header;
1437                                 isArray = 0;
1438                         } else {
1439                                 flag = 1;
1440                                 break;
1441                         }
1442                 }
1443         }
1444         //Found in the prefetch cache
1445         if(flag == 0 && (numoffset[index] == 0)) {
1446                 oid[index] = 0;
1447         }
1448 }
1449
1450
1451
1452 /* This function is called by the thread calling transPrefetch */
1453 void *transPrefetch(void *t) {
1454         prefetchqelem_t *qnode;
1455         prefetchpile_t *pilehead = NULL;
1456         prefetchpile_t *ptr = NULL, *piletail = NULL;
1457
1458         while(1) {
1459                 /* lock mutex of primary prefetch queue */
1460                 pthread_mutex_lock(&pqueue.qlock);
1461                 /* while primary queue is empty, then wait */
1462                 while((pqueue.front == NULL) && (pqueue.rear == NULL)) {
1463                         pthread_cond_wait(&pqueue.qcond, &pqueue.qlock);
1464                 }
1465
1466                 /* dequeue node to create a machine piles and  finally unlock mutex */
1467                 if((qnode = pre_dequeue()) == NULL) {
1468                         printf("Error: No node returned %s, %d\n", __FILE__, __LINE__);
1469                         pthread_mutex_unlock(&pqueue.qlock);
1470                         pthread_exit(NULL);
1471                 }
1472                 pthread_mutex_unlock(&pqueue.qlock);
1473
1474                 /* Reduce redundant prefetch requests */
1475                 checkPrefetchTuples(qnode);
1476                 /* Check if the tuples are found locally, if yes then reduce them further*/
1477                 /* and group requests by remote machine ids by calling the makePreGroups() */
1478                 if((pilehead = foundLocal(qnode)) == NULL) {
1479                         printf("Error: No node created for serving prefetch request %s %d\n", __FILE__, __LINE__);
1480                         pthread_exit(NULL);
1481                 }
1482
1483                 ptr = pilehead;
1484                 while(ptr != NULL) {
1485                         if(ptr->next == NULL) {
1486                                 piletail = ptr;
1487                         }
1488                         ptr = ptr->next;
1489                 }
1490
1491                 /* Lock mutex of pool queue */
1492                 pthread_mutex_lock(&mcqueue.qlock);
1493                 /* Update the pool queue with the new remote machine piles generated per prefetch call */
1494                 mcpileenqueue(pilehead, piletail);
1495                 /* Broadcast signal on machine pile queue */
1496                 pthread_cond_broadcast(&mcqueue.qcond);
1497                 /* Unlock mutex of  machine pile queue */
1498                 pthread_mutex_unlock(&mcqueue.qlock);
1499                 /* Deallocate the prefetch queue pile node */
1500                 predealloc(qnode);
1501         }
1502 }
1503
1504 /* Each thread in the  pool of threads calls this function to establish connection with
1505  * remote machines, send the prefetch requests and process the reponses from
1506  * the remote machines .
1507  * The thread is active throughout the period of runtime */
1508
1509 void *mcqProcess(void *threadid) {
1510         int tid;
1511         prefetchpile_t *mcpilenode;
1512
1513         tid = (int) threadid;
1514         while(1) {
1515                 /* Lock mutex of mc pile queue */
1516                 pthread_mutex_lock(&mcqueue.qlock);
1517                 /* When mc pile queue is empty, wait */
1518                 while((mcqueue.front == NULL) && (mcqueue.rear == NULL)) {
1519                         pthread_cond_wait(&mcqueue.qcond, &mcqueue.qlock);
1520                 }
1521                 /* Dequeue node to send remote machine connections*/
1522                 if((mcpilenode = mcpiledequeue()) == NULL) {
1523                         printf("Dequeue Error: No node returned %s %d\n", __FILE__, __LINE__);
1524                         pthread_mutex_unlock(&mcqueue.qlock);
1525                         pthread_exit(NULL);
1526                 }
1527                 /* Unlock mutex */
1528                 pthread_mutex_unlock(&mcqueue.qlock);
1529
1530                 /*Initiate connection to remote host and send request */
1531                 /* Process Request */
1532                 if(mcpilenode->mid != myIpAddr)
1533                         sendPrefetchReq(mcpilenode);
1534
1535                 /* Deallocate the machine queue pile node */
1536                 mcdealloc(mcpilenode);
1537         }
1538 }
1539
1540 void sendPrefetchReq(prefetchpile_t *mcpilenode) {
1541         int sd, i, off, len, endpair, count = 0;
1542         struct sockaddr_in serv_addr;
1543         struct hostent *server;
1544         char machineip[16], control;
1545         objpile_t *tmp;
1546
1547         /* Send Trans Prefetch Request */
1548         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
1549                 perror("Error in socket for SEND_PREFETCH_REQUEST\n");
1550                 return;
1551         }
1552
1553         bzero((char*) &serv_addr, sizeof(serv_addr));
1554         serv_addr.sin_family = AF_INET;
1555         serv_addr.sin_port = htons(LISTEN_PORT);
1556         midtoIP(mcpilenode->mid ,machineip);
1557         machineip[15] = '\0';
1558         serv_addr.sin_addr.s_addr = inet_addr(machineip);
1559
1560         /* Open Connection */
1561         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
1562                 perror("Error in connect for SEND_PREFETCH_REQUEST\n");
1563                 close(sd);
1564                 return;
1565         }
1566
1567         /* Send TRANS_PREFETCH control message */
1568         control = TRANS_PREFETCH;
1569         if(send(sd, &control, sizeof(char), MSG_NOSIGNAL) < sizeof(char)) {
1570                 perror("Error in sending prefetch control\n");
1571                 close(sd);
1572                 return;
1573         }
1574
1575         /* Send Oids and offsets in pairs */
1576         tmp = mcpilenode->objpiles;
1577         while(tmp != NULL) {
1578                 off = 0;
1579                 count++;  /* Keeps track of the number of oid and offset tuples sent per remote machine */
1580                 len = sizeof(int) + sizeof(unsigned int) + ((tmp->numoffset) * sizeof(unsigned short));
1581                 char oidnoffset[len];
1582                 bzero(oidnoffset, len);
1583                 *((unsigned int*)oidnoffset) = len;
1584                 //memcpy(oidnoffset, &len, sizeof(int));
1585                 off = sizeof(int);
1586                 *((unsigned int *)((char *)oidnoffset + off)) = tmp->oid;
1587                 //memcpy(oidnoffset + off, &tmp->oid, sizeof(unsigned int));
1588                 off += sizeof(unsigned int);
1589                 for(i = 0; i < tmp->numoffset; i++) {
1590                         *((unsigned short*)((char *)oidnoffset + off)) = tmp->offset[i];
1591                         //memcpy(oidnoffset + off, &(tmp->offset[i]), sizeof(short));
1592                         off+=sizeof(unsigned short);
1593                 }
1594
1595                 if (send(sd, oidnoffset, len , MSG_NOSIGNAL) < len) {
1596                         perror("Error sending fixed bytes for thread\n");
1597                         close(sd);
1598                         return;
1599                 }
1600
1601                 tmp = tmp->next;
1602         }
1603
1604         /* Send a special char -1 to represent the end of sending oids + offset pair to remote machine */
1605         endpair = -1;
1606         if (send(sd, &endpair, sizeof(int), MSG_NOSIGNAL) < sizeof(int)) {
1607                 perror("Error sending fixed bytes for thread\n");
1608                 close(sd);
1609                 return;
1610         }
1611
1612         /* Get Response from the remote machine */
1613         getPrefetchResponse(count,sd);
1614         close(sd);
1615         return;
1616 }
1617
1618 void getPrefetchResponse(int count, int sd) {
1619         int i = 0, val, n, N, sum, index, objsize;
1620         unsigned int bufsize,oid;
1621         char buffer[RECEIVE_BUFFER_SIZE], control;
1622         char *ptr;
1623         void *modptr, *oldptr;
1624
1625         /* Read  prefetch response from the Remote machine */
1626         if((val = read(sd, &control, sizeof(char))) <= 0) {
1627                 perror("No control response for Prefetch request sent\n");
1628                 return;
1629         }
1630
1631         if(control == TRANS_PREFETCH_RESPONSE) {
1632                 /*For each oid and offset tuple sent as prefetch request to remote machine*/
1633                 while(i < count) {
1634                         sum = 0;
1635                         index = 0;
1636                         /* Read the size of buffer to be received */
1637                         if((N = read(sd, buffer, sizeof(unsigned int))) <= 0) {
1638                                 perror("Size of buffer not recv\n");
1639                                 return;
1640                         }
1641                         bufsize = *((unsigned int *) buffer);
1642                         ptr = buffer + sizeof(unsigned int);
1643                         /* Keep receiving the buffer containing oid info */
1644                         do {
1645                                 n = recv((int)sd, (void *)ptr+sum, bufsize-sum, 0);
1646                                 sum +=n;
1647                         } while(sum < bufsize && n != 0);
1648
1649                         /* Decode the contents of the buffer */
1650                         index = sizeof(unsigned int);
1651                         while(index < (bufsize - sizeof(unsigned int))) {
1652                                 if(buffer[index] == OBJECT_FOUND) {
1653                                         /* Increment it to get the object */
1654                                         index += sizeof(char);
1655                                         oid = *((unsigned int *)(buffer+index));
1656                                         index += sizeof(unsigned int);
1657                                         /* For each object found add to Prefetch Cache */
1658                                         objsize = *((int *)(buffer+index));
1659                                         index+=sizeof(int);
1660                                         pthread_mutex_lock(&prefetchcache_mutex);
1661                                         if ((modptr = objstrAlloc(prefetchcache, objsize)) == NULL) {
1662                                                 printf("Error: objstrAlloc error for copying into prefetch cache %s, %d\n", __FILE__, __LINE__);
1663                                                 pthread_mutex_unlock(&prefetchcache_mutex);
1664                                                 return;
1665                                         }
1666                                         pthread_mutex_unlock(&prefetchcache_mutex);
1667                                         memcpy(modptr, buffer+index, objsize);
1668                                         index += objsize;
1669                                         /* Insert the oid and its address into the prefetch hash lookup table */
1670                                         /* Do a version comparison if the oid exists */
1671                                         if((oldptr = prehashSearch(oid)) != NULL) {
1672                                                 /* If older version then update with new object ptr */
1673                                                 if(((objheader_t *)oldptr)->version < ((objheader_t *)modptr)->version) {
1674                                                         prehashRemove(oid);
1675                                                         prehashInsert(oid, modptr);
1676                                                 } else if(((objheader_t *)oldptr)->version == ((objheader_t *)modptr)->version) {
1677                                                         /* Add the new object ptr to hash table */
1678                                                         prehashRemove(oid);
1679                                                         prehashInsert(oid, modptr);
1680                                                 } else { /* Do nothing: TODO modptr should be reference counted */
1681                                                         ;
1682                                                 }
1683                                         } else {/*If doesn't no match found in hashtable, add the object ptr to hash table*/
1684                                                 prehashInsert(oid, modptr);
1685                                         }
1686                                         /* Lock the Prefetch Cache look up table*/
1687                                         pthread_mutex_lock(&pflookup.lock);
1688                                         /* Broadcast signal on prefetch cache condition variable */
1689                                         pthread_cond_broadcast(&pflookup.cond);
1690                                         /* Unlock the Prefetch Cache look up table*/
1691                                         pthread_mutex_unlock(&pflookup.lock);
1692                                 } else if(buffer[index] == OBJECT_NOT_FOUND) {
1693                                         /* Increment it to get the object */
1694                                         /* TODO: For each object not found query DHT for new location and retrieve the object */
1695                                         index += sizeof(char);
1696                                         oid = *((unsigned int *)(buffer + index));
1697                                         index += sizeof(unsigned int);
1698                                         /* Throw an error */
1699                                         printf("OBJECT NOT FOUND.... THIS SHOULD NOT HAPPEN...TERMINATE PROGRAM\n");
1700                                         exit(-1);
1701                                 } else {
1702                                         printf("Error in decoding the index value %s, %d\n",__FILE__, __LINE__);
1703                                         return;
1704                                 }
1705                         }
1706
1707                         i++;
1708                 }
1709         } else
1710                 printf("Error in receving response for prefetch request %s, %d\n",__FILE__, __LINE__);
1711         return;
1712 }
1713
1714 unsigned short getObjType(unsigned int oid)
1715 {
1716         objheader_t *objheader;
1717         unsigned short numoffset[] ={0};
1718         short fieldoffset[] ={};
1719
1720         if ((objheader = (objheader_t *) mhashSearch(oid)) == NULL)
1721         {
1722                 if ((objheader = (objheader_t *) prehashSearch(oid)) == NULL)
1723                 {
1724                         prefetch(1, &oid, numoffset, fieldoffset);
1725                         pthread_mutex_lock(&pflookup.lock);
1726                         while ((objheader = (objheader_t *) prehashSearch(oid)) == NULL)
1727                         {
1728                                 pthread_cond_wait(&pflookup.cond, &pflookup.lock);
1729                         }
1730                         pthread_mutex_unlock(&pflookup.lock);
1731                 }
1732         }
1733
1734         return TYPE(objheader);
1735 }
1736
1737 int startRemoteThread(unsigned int oid, unsigned int mid)
1738 {
1739         int sock;
1740         struct sockaddr_in remoteAddr;
1741         char msg[1 + sizeof(unsigned int)];
1742         int bytesSent;
1743         int status;
1744
1745         if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0)
1746         {
1747                 perror("startRemoteThread():socket()");
1748                 return -1;
1749         }
1750
1751         bzero(&remoteAddr, sizeof(remoteAddr));
1752         remoteAddr.sin_family = AF_INET;
1753         remoteAddr.sin_port = htons(LISTEN_PORT);
1754         remoteAddr.sin_addr.s_addr = htonl(mid);
1755
1756         if (connect(sock, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0)
1757         {
1758                 printf("startRemoteThread():error %d connecting to %s:%d\n", errno,
1759                         inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
1760                 status = -1;
1761         }
1762         else
1763         {
1764                 msg[0] = START_REMOTE_THREAD;
1765                 memcpy(&msg[1], &oid, sizeof(unsigned int));
1766
1767                 bytesSent = send(sock, msg, 1 + sizeof(unsigned int), 0);
1768                 if (bytesSent < 0)
1769                 {
1770                         perror("startRemoteThread():send()");
1771                         status = -1;
1772                 }
1773                 else if (bytesSent != 1 + sizeof(unsigned int))
1774                 {
1775                         printf("startRemoteThread(): error, sent %d bytes\n", bytesSent);
1776                         status = -1;
1777                 }
1778                 else
1779                 {
1780                         status = 0;
1781                 }
1782         }
1783
1784         close(sock);
1785         return status;
1786 }
1787
1788 //TODO: when reusing oids, make sure they are not already in use!
1789 unsigned int getNewOID(void) {
1790         static unsigned int id = 0xFFFFFFFF;
1791
1792         id += 2;
1793         if (id > oidMax || id < oidMin)
1794         {
1795                 id = (oidMin | 1);
1796         }
1797         return id;
1798 }
1799
1800 int processConfigFile()
1801 {
1802         FILE *configFile;
1803         const int maxLineLength = 200;
1804         char lineBuffer[maxLineLength];
1805         char *token;
1806         const char *delimiters = " \t\n";
1807         char *commentBegin;
1808         in_addr_t tmpAddr;
1809
1810         configFile = fopen(CONFIG_FILENAME, "r");
1811         if (configFile == NULL)
1812         {
1813                 printf("error opening %s:\n", CONFIG_FILENAME);
1814                 perror("");
1815                 return -1;
1816         }
1817
1818         numHostsInSystem = 0;
1819         sizeOfHostArray = 8;
1820         hostIpAddrs = calloc(sizeOfHostArray, sizeof(unsigned int));
1821
1822         while(fgets(lineBuffer, maxLineLength, configFile) != NULL)
1823         {
1824                 commentBegin = strchr(lineBuffer, '#');
1825                 if (commentBegin != NULL)
1826                         *commentBegin = '\0';
1827                 token = strtok(lineBuffer, delimiters);
1828                 while (token != NULL)
1829                 {
1830                         tmpAddr = inet_addr(token);
1831                         if ((int)tmpAddr == -1)
1832                         {
1833                                 printf("error in %s: bad token:%s\n", CONFIG_FILENAME, token);
1834                                 fclose(configFile);
1835                                 return -1;
1836                         }
1837                         else
1838                                 addHost(htonl(tmpAddr));
1839                         token = strtok(NULL, delimiters);
1840                 }
1841         }
1842
1843         fclose(configFile);
1844
1845         if (numHostsInSystem < 1)
1846         {
1847                 printf("error in %s: no IP Adresses found\n", CONFIG_FILENAME);
1848                 return -1;
1849         }
1850 #ifdef MAC
1851         myIpAddr = getMyIpAddr("en1");
1852 #else
1853         myIpAddr = getMyIpAddr("eth0");
1854 #endif
1855         myIndexInHostArray = findHost(myIpAddr);
1856         if (myIndexInHostArray == -1)
1857         {
1858                 printf("error in %s: IP Address of eth0 not found\n", CONFIG_FILENAME);
1859                 return -1;
1860         }
1861         oidsPerBlock = (0xFFFFFFFF / numHostsInSystem) + 1;
1862         oidMin = oidsPerBlock * myIndexInHostArray;
1863         if (myIndexInHostArray == numHostsInSystem - 1)
1864                 oidMax = 0xFFFFFFFF;
1865         else
1866                 oidMax = oidsPerBlock * (myIndexInHostArray + 1) - 1;
1867
1868         return 0;
1869 }
1870
1871 void addHost(unsigned int hostIp)
1872 {
1873         unsigned int *tmpArray;
1874
1875         if (findHost(hostIp) != -1)
1876                 return;
1877
1878         if (numHostsInSystem == sizeOfHostArray)
1879         {
1880                 tmpArray = calloc(sizeOfHostArray * 2, sizeof(unsigned int));
1881                 memcpy(tmpArray, hostIpAddrs, sizeof(unsigned int) * numHostsInSystem);
1882                 free(hostIpAddrs);
1883                 hostIpAddrs = tmpArray;
1884         }
1885
1886         hostIpAddrs[numHostsInSystem++] = hostIp;
1887
1888         return;
1889 }
1890
1891 int findHost(unsigned int hostIp)
1892 {
1893         int i;
1894         for (i = 0; i < numHostsInSystem; i++)
1895                 if (hostIpAddrs[i] == hostIp)
1896                         return i;
1897
1898         //not found
1899         return -1;
1900 }
1901
1902 /* This function sends notification request per thread waiting on object(s) whose version
1903  * changes */
1904 int reqNotify(unsigned int *oidarry, unsigned short *versionarry, unsigned int numoid) {
1905         int sock,i;
1906         objheader_t *objheader;
1907         struct sockaddr_in remoteAddr;
1908         char msg[1 + numoid * (sizeof(short) + sizeof(unsigned int)) +  3 * sizeof(unsigned int)];
1909         char *ptr;
1910         int bytesSent;
1911         int status, size;
1912         unsigned short version;
1913         unsigned int oid,mid;
1914         static unsigned int threadid = 0;
1915         pthread_mutex_t threadnotify = PTHREAD_MUTEX_INITIALIZER; //Lock and condition var for threadjoin and notification
1916         pthread_cond_t threadcond = PTHREAD_COND_INITIALIZER;
1917         notifydata_t *ndata;
1918
1919         //FIXME currently all oids belong to one machine
1920         oid = oidarry[0];
1921         mid = lhashSearch(oid);
1922
1923         if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0){
1924                 perror("reqNotify():socket()");
1925                 return -1;
1926         }
1927
1928         bzero(&remoteAddr, sizeof(remoteAddr));
1929         remoteAddr.sin_family = AF_INET;
1930         remoteAddr.sin_port = htons(LISTEN_PORT);
1931         remoteAddr.sin_addr.s_addr = htonl(mid);
1932
1933         /* Generate unique threadid */
1934         threadid++;
1935
1936         /* Save threadid, numoid, oidarray, versionarray, pthread_cond_variable for later processing */
1937         if((ndata = calloc(1, sizeof(notifydata_t))) == NULL) {
1938                 printf("Calloc Error %s, %d\n", __FILE__, __LINE__);
1939                 return -1;
1940         }
1941         ndata->numoid = numoid;
1942         ndata->threadid = threadid;
1943         ndata->oidarry = oidarry;
1944         ndata->versionarry = versionarry;
1945         ndata->threadcond = threadcond;
1946         ndata->threadnotify = threadnotify;
1947         if((status = notifyhashInsert(threadid, ndata)) != 0) {
1948                 printf("reqNotify(): Insert into notify hash table not successful %s, %d\n", __FILE__, __LINE__);
1949                 free(ndata);
1950                 return -1;
1951         }
1952
1953         /* Send  number of oids, oidarry, version array, machine id and threadid */
1954         if (connect(sock, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0){
1955                 printf("reqNotify():error %d connecting to %s:%d\n", errno,
1956                                 inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
1957                 free(ndata);
1958                 return -1;
1959         } else {
1960                 msg[0] = THREAD_NOTIFY_REQUEST;
1961                 *((unsigned int *)(&msg[1])) = numoid;
1962                 /* Send array of oids  */
1963                 size = sizeof(unsigned int);
1964                 {
1965                         i = 0;
1966                         while(i < numoid) {
1967                                 oid = oidarry[i];
1968                                 *((unsigned int *)(&msg[1] + size)) = oid;
1969                                 size += sizeof(unsigned int);
1970                                 i++;
1971                         }
1972                 }
1973
1974                 /* Send array of version  */
1975                 {
1976                         i = 0;
1977                         while(i < numoid) {
1978                                 version = versionarry[i];
1979                                 *((unsigned short *)(&msg[1] + size)) = version;
1980                                 size += sizeof(unsigned short);
1981                                 i++;
1982                         }
1983                 }
1984
1985                 *((unsigned int *)(&msg[1] + size)) = myIpAddr;
1986                 size += sizeof(unsigned int);
1987                 *((unsigned int *)(&msg[1] + size)) = threadid;
1988
1989                 pthread_mutex_lock(&(ndata->threadnotify));
1990                 bytesSent = send(sock, msg, 1 + numoid * (sizeof(unsigned int) + sizeof(unsigned short)) + 3 * sizeof(unsigned int) , 0);
1991                 if (bytesSent < 0){
1992                         perror("reqNotify():send()");
1993                         status = -1;
1994                 } else if (bytesSent != 1 + numoid*(sizeof(unsigned int) + sizeof(unsigned short)) + 3 * sizeof(unsigned int)){
1995                         printf("reNotify(): error, sent %d bytes %s, %d\n", bytesSent, __FILE__, __LINE__);
1996                         status = -1;
1997                 } else {
1998                         status = 0;
1999                 }
2000
2001                 pthread_cond_wait(&(ndata->threadcond), &(ndata->threadnotify));
2002                 pthread_mutex_unlock(&(ndata->threadnotify));
2003         }
2004
2005         pthread_cond_destroy(&threadcond);
2006         pthread_mutex_destroy(&threadnotify);
2007         free(ndata);
2008         close(sock);
2009         return status;
2010 }
2011
2012 void threadNotify(unsigned int oid, unsigned short version, unsigned int tid) {
2013         notifydata_t *ndata;
2014         int i, objIsFound = 0, index;
2015         void *ptr;
2016
2017         //Look up the tid and call the corresponding pthread_cond_signal
2018         if((ndata = notifyhashSearch(tid)) == NULL) {
2019                 printf("threadnotify(): No such threadid is present %s, %d\n", __FILE__, __LINE__);
2020                 return;
2021         } else  {
2022                 for(i = 0; i < ndata->numoid; i++) {
2023                         if(ndata->oidarry[i] == oid){
2024                                 objIsFound = 1;
2025                                 index = i;
2026                         }
2027                 }
2028                 if(objIsFound == 0){
2029                         printf("threadNotify(): Oid not found %s, %d\n", __FILE__, __LINE__);
2030                         return;
2031                 } else {
2032                         if(version <= ndata->versionarry[index]){
2033                                 printf("threadNotify(): New version %d has not changed since last version %s, %d\n", version, __FILE__, __LINE__);
2034                                 return;
2035                         } else {
2036                                 /* Clear from prefetch cache and free thread related data structure */
2037                                 if((ptr = prehashSearch(oid)) != NULL) {
2038                                         prehashRemove(oid);
2039                                 }
2040                                 pthread_cond_signal(&(ndata->threadcond));
2041                         }
2042                 }
2043         }
2044         return;
2045 }
2046
2047 int notifyAll(threadlist_t **head, unsigned int oid, unsigned int version) {
2048         threadlist_t *ptr;
2049         unsigned int mid;
2050         struct sockaddr_in remoteAddr;
2051         char msg[1 + sizeof(unsigned short) + 2*sizeof(unsigned int)];
2052         int sock, status, size, bytesSent;
2053
2054         while(*head != NULL) {
2055                 ptr = *head;
2056                 mid = ptr->mid;
2057                 //create a socket connection to that machine
2058                 if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0){
2059                         perror("notifyAll():socket()");
2060                         return -1;
2061                 }
2062
2063                 bzero(&remoteAddr, sizeof(remoteAddr));
2064                 remoteAddr.sin_family = AF_INET;
2065                 remoteAddr.sin_port = htons(LISTEN_PORT);
2066                 remoteAddr.sin_addr.s_addr = htonl(mid);
2067                 //send Thread Notify response and threadid to that machine
2068                 if (connect(sock, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0){
2069                         printf("notifyAll():error %d connecting to %s:%d\n", errno,
2070                                         inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
2071                         status = -1;
2072                 } else {
2073                         bzero(msg, (1+sizeof(unsigned short) + 2*sizeof(unsigned int)));
2074                         msg[0] = THREAD_NOTIFY_RESPONSE;
2075                         *((unsigned int *)&msg[1]) = oid;
2076                         size = sizeof(unsigned int);
2077                         *((unsigned short *)(&msg[1]+ size)) = version;
2078                         size+= sizeof(unsigned short);
2079                         *((unsigned int *)(&msg[1]+ size)) = ptr->threadid;
2080
2081                         bytesSent = send(sock, msg, (1 + 2*sizeof(unsigned int) + sizeof(unsigned short)), 0);
2082                         if (bytesSent < 0){
2083                                 perror("notifyAll():send()");
2084                                 status = -1;
2085                         } else if (bytesSent != 1 + 2*sizeof(unsigned int) + sizeof(unsigned short)){
2086                                 printf("notifyAll(): error, sent %d bytes %s, %d\n",
2087                                                 bytesSent, __FILE__, __LINE__);
2088                                 status = -1;
2089                         } else {
2090                                 status = 0;
2091                         }
2092                 }
2093                 //close socket
2094                 close(sock);
2095                 // Update head
2096                 *head = ptr->next;
2097                 free(ptr);
2098         }
2099         return status;
2100 }
2101
2102 void transAbort(transrecord_t *trans) {
2103         objstrDelete(trans->cache);
2104         chashDelete(trans->lookupTable);
2105         free(trans);
2106 }