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