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 "queue.h"
  10 #include <pthread.h>
  11 #include <sys/types.h>
  12 #include <sys/socket.h>
  13 #include <netdb.h>
  14 #include <netinet/in.h>
  15 #include <sys/types.h>
  16 #include <unistd.h>
  17 #include <errno.h>
  18 #include <time.h>
  19 #include <string.h>
  20 #include <pthread.h>
  21
  22 #define LISTEN_PORT 2156
  23 #define RECEIVE_BUFFER_SIZE 2048
  24 #define NUM_THREADS 10
  25 #define PREFETCH_CACHE_SIZE 1048576 //1MB
  26
  27 /* Global Variables */
  28 extern int classsize[];
  29 extern primarypfq_t pqueue; // shared prefetch queue
  30 extern mcpileq_t mcqueue;  //Shared queue containing prefetch requests sorted by remote machineids
  31 objstr_t *prefetchcache; //Global Prefetch cache
  32 extern prehashtable_t pflookup; //Global Prefetch cache's lookup table
  33 pthread_t wthreads[NUM_THREADS]; //Worker threads for working on the prefetch queue
  34 pthread_t tPrefetch;
  35 extern objstr_t *mainobjstore;
  36
  37 plistnode_t *createPiles(transrecord_t *);
  38 inline int arrayLength(int *array) {
  39         int i;
  40         for(i=0 ;array[i] != -1; i++)
  41                 ;
  42         return i;
  43 }
  44 inline int findmax(int *array, int arraylength) {
  45         int max, i;
  46         max = array[0];
  47         for(i = 0; i < arraylength; i++){
  48                 if(array[i] > max) {
  49                         max = array[i];
  50                 }
  51         }
  52         return max;
  53 }
  54 /* This function is a prefetch call generated by the compiler that
  55  * populates the shared primary prefetch queue*/
  56 void prefetch(int ntuples, unsigned int *oids, short *endoffsets, short *arrayfields) {
  57         int qnodesize;
  58         int len = 0;
  59
  60         /* Allocate for the queue node*/
  61         char *node;
  62         qnodesize = sizeof(prefetchqelem_t) + sizeof(int) + ntuples * (sizeof(short) + sizeof(unsigned int)) + endoffsets[ntuples - 1] * sizeof(short);
  63         if((node = calloc(1, qnodesize)) == NULL) {
  64                 printf("Calloc Error %s, %d\n", __FILE__, __LINE__);
  65                 return;
  66         }
  67         /* Set queue node values */
  68         len = sizeof(prefetchqelem_t);
  69         memcpy(node + len, &ntuples, sizeof(int));
  70         len += sizeof(int);
  71         memcpy(node + len, oids, ntuples*sizeof(unsigned int));
  72         len += ntuples * sizeof(unsigned int);
  73         memcpy(node + len, endoffsets, ntuples*sizeof(short));
  74         len += ntuples * sizeof(short);
  75         memcpy(node + len, arrayfields, endoffsets[ntuples-1]*sizeof(short));
  76         /* Lock and insert into primary prefetch queue */
  77         pthread_mutex_lock(&pqueue.qlock);
  78         pre_enqueue((prefetchqelem_t *)node);
  79         pthread_cond_signal(&pqueue.qcond);
  80         pthread_mutex_unlock(&pqueue.qlock);
  81 }
  82
  83
  84 /* This function starts up the transaction runtime. */
  85 int dstmStartup(const char * option) {
  86   pthread_t thread_Listen;
  87   pthread_attr_t attr;
  88   int master=strcmp(option, "master")==0;
  89
  90   dstmInit();
  91   transInit();
  92
  93   if (master) {
  94     pthread_attr_init(&attr);
  95     pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
  96     pthread_create(&thread_Listen, &attr, dstmListen, NULL);
  97     return 1;
  98   } else {
  99     dstmListen();
 100     return 0;
 101   }
 102
 103 }
 104
 105
 106 /* This function initiates the prefetch thread
 107  * A queue is shared between the main thread of execution
 108  * and the prefetch thread to process the prefetch call
 109  * Call from compiler populates the shared queue with prefetch requests while prefetch thread
 110  * processes the prefetch requests */
 111 void transInit() {
 112         int t, rc;
 113         //Create and initialize prefetch cache structure
 114         prefetchcache = objstrCreate(PREFETCH_CACHE_SIZE);
 115         //Create prefetch cache lookup table
 116         if(prehashCreate(HASH_SIZE, LOADFACTOR))
 117                 return; //Failure
 118         //Initialize primary shared queue
 119         queueInit();
 120         //Initialize machine pile w/prefetch oids and offsets shared queue
 121         mcpileqInit();
 122         //Create the primary prefetch thread
 123         pthread_create(&tPrefetch, NULL, transPrefetch, NULL);
 124         //Create and Initialize a pool of threads
 125         /* Threads are active for the entire period runtime is running */
 126         for(t = 0; t< NUM_THREADS; t++) {
 127                 rc = pthread_create(&wthreads[t], NULL, mcqProcess, (void *)t);
 128                 if (rc) {
 129                         printf("Thread create error %s, %d\n", __FILE__, __LINE__);
 130                         return;
 131                 }
 132         }
 133 }
 134
 135 /* This function stops the threads spawned */
 136 void transExit() {
 137         int t;
 138         pthread_cancel(tPrefetch);
 139         for(t = 0; t < NUM_THREADS; t++)
 140                 pthread_cancel(wthreads[t]);
 141
 142         return;
 143 }
 144
 145 /* This functions inserts randowm wait delays in the order of msec
 146  * Mostly used when transaction commits retry*/
 147 void randomdelay(void)
 148 {
 149         struct timespec req, rem;
 150         time_t t;
 151
 152         t = time(NULL);
 153         req.tv_sec = 0;
 154         req.tv_nsec = (long)(1000000 + (t%10000000)); //1-11 msec
 155         nanosleep(&req, &rem);
 156         return;
 157 }
 158
 159 /* This function initializes things required in the transaction start*/
 160 transrecord_t *transStart()
 161 {
 162         transrecord_t *tmp = malloc(sizeof(transrecord_t));
 163         tmp->cache = objstrCreate(1048576);
 164         tmp->lookupTable = chashCreate(HASH_SIZE, LOADFACTOR);
 165 #ifdef COMPILER
 166         tmp->revertlist=NULL;
 167 #endif
 168         return tmp;
 169 }
 170
 171 /* This function finds the location of the objects involved in a transaction
 172  * and returns the pointer to the object if found in a remote location */
 173 objheader_t *transRead(transrecord_t *record, unsigned int oid) {
 174         printf("Inside transaction read call\n");
 175         unsigned int machinenumber;
 176         objheader_t *tmp, *objheader;
 177         void *objcopy;
 178         int size, rc, found = 0;
 179         void *buf;
 180         struct timespec ts;
 181         struct timeval tp;
 182
 183         rc = gettimeofday(&tp, NULL);
 184
 185         /* Convert from timeval to timespec */
 186         ts.tv_nsec = tp.tv_usec * 1000;
 187
 188         /* Search local transaction cache */
 189         if((objheader = (objheader_t *)chashSearch(record->lookupTable, oid)) != NULL){
 190                 printf("Inside transaction cache \n");
 191                 return(objheader);
 192         } else if ((objheader = (objheader_t *) mhashSearch(oid)) != NULL) {
 193                 /* Look up in machine lookup table  and copy  into cache*/
 194                 printf("Inside mainobject store \n");
 195                 tmp = mhashSearch(oid);
 196                 size = sizeof(objheader_t)+classsize[TYPE(tmp)];
 197                 objcopy = objstrAlloc(record->cache, size);
 198                 memcpy(objcopy, (void *)objheader, size);
 199                 /* Insert into cache's lookup table */
 200                 chashInsert(record->lookupTable, OID(objheader), objcopy);
 201                 return(objcopy);
 202         } else if((tmp = (objheader_t *) prehashSearch(oid)) != NULL) { /* Look up in prefetch cache */
 203                 printf("Inside prefetch cache \n");
 204                 found = 1;
 205                 size = sizeof(objheader_t)+classsize[TYPE(tmp)];
 206                 objcopy = objstrAlloc(record->cache, size);
 207                 memcpy(objcopy, (void *)tmp, size);
 208                 /* Insert into cache's lookup table */
 209                 chashInsert(record->lookupTable, OID(tmp), objcopy);
 210                 return(objcopy);
 211         } else { /* If not found anywhere, then block until object appears in prefetch cache */
 212                 printf("Inside remote machine\n");
 213                 pthread_mutex_lock(&pflookup.lock);
 214                 while(!found) {
 215                         rc = pthread_cond_timedwait(&pflookup.cond, &pflookup.lock, &ts);
 216                         if(rc == ETIMEDOUT) {
 217                                 printf("Wait timed out\n");
 218                                 /* Check Prefetch cache again */
 219                                 if((tmp = (objheader_t *) prehashSearch(oid)) != NULL) { /* Look up in prefetch cache */
 220                                         found = 1;
 221                                         size = sizeof(objheader_t)+classsize[TYPE(tmp)];
 222                                         objcopy = objstrAlloc(record->cache, size);
 223                                         memcpy(objcopy, (void *)tmp, size);
 224                                         /* Insert into cache's lookup table */
 225                                         chashInsert(record->lookupTable, OID(tmp), objcopy);
 226                                         return(objcopy);
 227                                 } else {
 228                                         pthread_mutex_unlock(&pflookup.lock);
 229                                         break;
 230                                 }
 231                                 pthread_mutex_unlock(&pflookup.lock);
 232                         }
 233                 }
 234                 /* Get the object from the remote location */
 235                 machinenumber = lhashSearch(oid);
 236                 objcopy = getRemoteObj(record, machinenumber, oid);
 237                 if(objcopy == NULL) {
 238                         //If object is not found in Remote location
 239                         //printf("Object oid = %d not found in Machine %d\n", oid, machinenumber);
 240                         return NULL;
 241                 }
 242                 else {
 243                         //printf("Object oid = %d found in Machine %d\n", oid, machinenumber);
 244                         return(objcopy);
 245                 }
 246         }
 247 }
 248
 249 /* This function creates objects in the transaction record */
 250 objheader_t *transCreateObj(transrecord_t *record, unsigned short type)
 251 {
 252         objheader_t *tmp = (objheader_t *) objstrAlloc(record->cache, (sizeof(objheader_t) + classsize[type]));
 253         OID(tmp) = getNewOID();
 254         TYPE(tmp) = type;
 255         tmp->version = 1;
 256         tmp->rcount = 0; //? not sure how to handle this yet
 257         STATUS(tmp) = NEW;
 258         chashInsert(record->lookupTable, OID(tmp), tmp);
 259         return tmp;
 260 }
 261
 262 /* This function creates machine piles based on all machines involved in a
 263  * transaction commit request */
 264 plistnode_t *createPiles(transrecord_t *record) {
 265         int i = 0;
 266         unsigned int size;/* Represents number of bins in the chash table */
 267         chashlistnode_t *curr, *ptr, *next;
 268         plistnode_t *pile = NULL;
 269         unsigned int machinenum;
 270         void *localmachinenum;
 271         objheader_t *headeraddr;
 272
 273         ptr = record->lookupTable->table;
 274         size = record->lookupTable->size;
 275
 276         for(i = 0; i < size ; i++) {
 277                 curr = &ptr[i];
 278                 /* Inner loop to traverse the linked list of the cache lookupTable */
 279                 while(curr != NULL) {
 280                         //if the first bin in hash table is empty
 281                         if(curr->key == 0) {
 282                                 break;
 283                         }
 284                         next = curr->next;
 285                         //Get machine location for object id
 286
 287                         if ((machinenum = lhashSearch(curr->key)) == 0) {
 288                                 printf("Error: No such machine %s, %d\n", __FILE__, __LINE__);
 289                                 return NULL;
 290                         }
 291
 292                         if ((headeraddr = chashSearch(record->lookupTable, curr->key)) == NULL) {
 293                                 printf("Error: No such oid %s, %d\n", __FILE__, __LINE__);
 294                                 return NULL;
 295                         }
 296                         //Make machine groups
 297                         if ((pile = pInsert(pile, headeraddr, machinenum, record->lookupTable->numelements)) == NULL) {
 298                                 printf("pInsert error %s, %d\n", __FILE__, __LINE__);
 299                                 return NULL;
 300                         }
 301
 302                         /* Check if local or not */
 303                         if((localmachinenum = mhashSearch(curr->key)) != NULL) {
 304                                 /* Set the pile->local flag*/
 305                                 pile->local = 1; //True i.e. local
 306                         }
 307
 308                         curr = next;
 309                 }
 310         }
 311         return pile;
 312 }
 313
 314 /* This function initiates the transaction commit process
 315  * Spawns threads for each of the new connections with Participants
 316  * and creates new piles by calling the createPiles(),
 317  * Fills the piles with necesaary information and
 318  * Sends a transrequest() to each pile*/
 319 int transCommit(transrecord_t *record) {
 320         unsigned int tot_bytes_mod, *listmid;
 321         plistnode_t *pile, *pile_ptr;
 322         int i, rc, val;
 323         int pilecount = 0, offset, threadnum = 0, trecvcount = 0, tmachcount = 0;
 324         char buffer[RECEIVE_BUFFER_SIZE],control;
 325         char transid[TID_LEN];
 326         trans_req_data_t *tosend;
 327         trans_commit_data_t transinfo;
 328         static int newtid = 0;
 329         char treplyctrl = 0, treplyretry = 0; /* keeps track of the common response that needs to be sent */
 330         char localstat = 0;
 331
 332
 333         /* Look through all the objects in the transaction record and make piles
 334          * for each machine involved in the transaction*/
 335         pile_ptr = pile = createPiles(record);
 336
 337         /* Create the packet to be sent in TRANS_REQUEST */
 338
 339         /* Count the number of participants */
 340         pilecount = pCount(pile);
 341
 342         /* Create a list of machine ids(Participants) involved in transaction   */
 343         if((listmid = calloc(pilecount, sizeof(unsigned int))) == NULL) {
 344                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 345                 return 1;
 346         }
 347         pListMid(pile, listmid);
 348
 349
 350         /* Initialize thread variables,
 351          * Spawn a thread for each Participant involved in a transaction */
 352         pthread_t thread[pilecount];
 353         pthread_attr_t attr;
 354         pthread_cond_t tcond;
 355         pthread_mutex_t tlock;
 356         pthread_mutex_t tlshrd;
 357
 358         thread_data_array_t *thread_data_array;
 359         thread_data_array = (thread_data_array_t *) malloc(sizeof(thread_data_array_t)*pilecount);
 360         local_thread_data_array_t *ltdata;
 361         if((ltdata = calloc(1, sizeof(local_thread_data_array_t))) == NULL) {
 362                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 363                 return 1;
 364         }
 365
 366         thread_response_t rcvd_control_msg[pilecount];  /* Shared thread array that keeps track of responses of participants */
 367
 368         /* Initialize and set thread detach attribute */
 369         pthread_attr_init(&attr);
 370         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
 371         pthread_mutex_init(&tlock, NULL);
 372         pthread_cond_init(&tcond, NULL);
 373
 374         /* Process each machine pile */
 375         while(pile != NULL) {
 376                 //Create transaction id
 377                 newtid++;
 378                 if ((tosend = calloc(1, sizeof(trans_req_data_t))) == NULL) {
 379                         printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 380                         return 1;
 381                 }
 382                 tosend->f.control = TRANS_REQUEST;
 383                 sprintf(tosend->f.trans_id, "%x_%d", pile->mid, newtid);
 384                 tosend->f.mcount = pilecount;
 385                 tosend->f.numread = pile->numread;
 386                 tosend->f.nummod = pile->nummod;
 387                 tosend->f.sum_bytes = pile->sum_bytes;
 388                 tosend->listmid = listmid;
 389                 tosend->objread = pile->objread;
 390                 tosend->oidmod = pile->oidmod;
 391                 thread_data_array[threadnum].thread_id = threadnum;
 392                 thread_data_array[threadnum].mid = pile->mid;
 393                 thread_data_array[threadnum].pilecount = pilecount;
 394                 thread_data_array[threadnum].buffer = tosend;
 395                 thread_data_array[threadnum].recvmsg = rcvd_control_msg;
 396                 thread_data_array[threadnum].threshold = &tcond;
 397                 thread_data_array[threadnum].lock = &tlock;
 398                 thread_data_array[threadnum].count = &trecvcount;
 399                 thread_data_array[threadnum].replyctrl = &treplyctrl;
 400                 thread_data_array[threadnum].replyretry = &treplyretry;
 401                 thread_data_array[threadnum].rec = record;
 402                 /* If local do not create any extra connection */
 403                 if(pile->local != 1) { /* Not local */
 404                         rc = pthread_create(&thread[threadnum], NULL, transRequest, (void *) &thread_data_array[threadnum]);
 405                         if (rc) {
 406                                 perror("Error in pthread create\n");
 407                                 return 1;
 408                         }
 409                 } else { /*Local*/
 410                         /*Unset the pile->local flag*/
 411                         pile->local = 0;
 412                         /*Set flag to identify that Local machine is involved*/
 413                         ltdata->tdata = &thread_data_array[threadnum];
 414                         ltdata->transinfo = &transinfo;
 415                         val = pthread_create(&thread[threadnum], NULL, handleLocalReq, (void *) ltdata);
 416                         if (val) {
 417                                 perror("Error in pthread create\n");
 418                                 return 1;
 419                         }
 420                 }
 421                 threadnum++;
 422                 pile = pile->next;
 423         }
 424
 425         /* Free attribute and wait for the other threads */
 426         pthread_attr_destroy(&attr);
 427         for (i = 0 ;i < pilecount ; i++) {
 428                 rc = pthread_join(thread[i], NULL);
 429                 if (rc)
 430                 {
 431                         printf("ERROR return code from pthread_join() is %d\n", rc);
 432                         return 1;
 433                 }
 434                 free(thread_data_array[i].buffer);
 435         }
 436
 437         /* Free resources */
 438         pthread_cond_destroy(&tcond);
 439         pthread_mutex_destroy(&tlock);
 440         free(listmid);
 441         pDelete(pile_ptr);
 442         free(thread_data_array);
 443         free(ltdata);
 444
 445         /* Retry trans commit procedure if not sucessful in the first try */
 446         if(treplyretry == 1) {
 447                 /* wait a random amount of time */
 448                 randomdelay();
 449                 /* Retry the commiting transaction again */
 450                 transCommit(record);
 451         }
 452
 453         return 0;
 454 }
 455
 456 /* This function sends information involved in the transaction request and
 457  * accepts a response from particpants.
 458  * It calls decideresponse() to decide on what control message
 459  * to send next and sends the message using sendResponse()*/
 460 void *transRequest(void *threadarg) {
 461         int sd, i, n;
 462         struct sockaddr_in serv_addr;
 463         struct hostent *server;
 464         thread_data_array_t *tdata;
 465         objheader_t *headeraddr;
 466         char buffer[RECEIVE_BUFFER_SIZE], control, recvcontrol;
 467         char machineip[16], retval;
 468
 469         tdata = (thread_data_array_t *) threadarg;
 470
 471         /* Send Trans Request */
 472         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 473                 perror("Error in socket for TRANS_REQUEST\n");
 474                 return NULL;
 475         }
 476         bzero((char*) &serv_addr, sizeof(serv_addr));
 477         serv_addr.sin_family = AF_INET;
 478         serv_addr.sin_port = htons(LISTEN_PORT);
 479         midtoIP(tdata->mid,machineip);
 480         machineip[15] = '\0';
 481         serv_addr.sin_addr.s_addr = inet_addr(machineip);
 482         /* Open Connection */
 483         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
 484                 perror("Error in connect for TRANS_REQUEST\n");
 485                 return NULL;
 486         }
 487
 488         printf("DEBUG-> trans.c Sending TRANS_REQUEST to mid %s\n", machineip);
 489         /* Send bytes of data with TRANS_REQUEST control message */
 490         if (send(sd, &(tdata->buffer->f), sizeof(fixed_data_t),MSG_NOSIGNAL) < sizeof(fixed_data_t)) {
 491                 perror("Error sending fixed bytes for thread\n");
 492                 return NULL;
 493         }
 494         /* Send list of machines involved in the transaction */
 495         {
 496                 int size=sizeof(unsigned int)*tdata->pilecount;
 497                 if (send(sd, tdata->buffer->listmid, size, MSG_NOSIGNAL) < size) {
 498                         perror("Error sending list of machines for thread\n");
 499                         return NULL;
 500                 }
 501         }
 502         /* Send oids and version number tuples for objects that are read */
 503         {
 504                 int size=(sizeof(unsigned int)+sizeof(short))*tdata->buffer->f.numread;
 505                 if (send(sd, tdata->buffer->objread, size, MSG_NOSIGNAL) < size) {
 506                         perror("Error sending tuples for thread\n");
 507                         return NULL;
 508                 }
 509         }
 510         /* Send objects that are modified */
 511         for(i = 0; i < tdata->buffer->f.nummod ; i++) {
 512                 int size;
 513                 headeraddr = chashSearch(tdata->rec->lookupTable, tdata->buffer->oidmod[i]);
 514                 size=sizeof(objheader_t)+classsize[TYPE(headeraddr)];
 515                 if (send(sd, headeraddr, size, MSG_NOSIGNAL)  < size) {
 516                         perror("Error sending obj modified for thread\n");
 517                         return NULL;
 518                 }
 519         }
 520
 521         /* Read control message from Participant */
 522         if((n = read(sd, &control, sizeof(char))) <= 0) {
 523                 perror("Error in reading control message from Participant\n");
 524                 return NULL;
 525         }
 526         recvcontrol = control;
 527
 528         /* Update common data structure and increment count */
 529         tdata->recvmsg[tdata->thread_id].rcv_status = recvcontrol;
 530
 531         /* Lock and update count */
 532         //Thread sleeps until all messages from pariticipants are received by coordinator
 533         pthread_mutex_lock(tdata->lock);
 534
 535         (*(tdata->count))++; /* keeps track of no of messages received by the coordinator */
 536
 537         /* Wake up the threads and invoke decideResponse (once) */
 538         if(*(tdata->count) == tdata->pilecount) {
 539                 if (decideResponse(tdata) != 0) {
 540                         printf("decideResponse returned error %s,%d\n", __FILE__, __LINE__);
 541                         pthread_mutex_unlock(tdata->lock);
 542                         close(sd);
 543                         return NULL;
 544                 }
 545                 pthread_cond_broadcast(tdata->threshold);
 546         } else {
 547                 pthread_cond_wait(tdata->threshold, tdata->lock);
 548         }
 549         pthread_mutex_unlock(tdata->lock);
 550
 551         /* Send the final response such as TRANS_COMMIT or TRANS_ABORT t
 552          * to all participants in their respective socket */
 553         if (sendResponse(tdata, sd) == 0) {
 554                 printf("sendResponse returned error %s,%d\n", __FILE__, __LINE__);
 555                 pthread_mutex_unlock(tdata->lock);
 556                 close(sd);
 557                 return NULL;
 558         }
 559
 560         /* Close connection */
 561         close(sd);
 562         pthread_exit(NULL);
 563 }
 564
 565 /* This function decides the reponse that needs to be sent to
 566  * all Participant machines involved in the transaction commit */
 567 int decideResponse(thread_data_array_t *tdata) {
 568         char control;
 569         int i, transagree = 0, transdisagree = 0, transsoftabort = 0; /* Counters to formulate decision of what
 570                                                                          message to send */
 571
 572         //Check common data structure
 573         for (i = 0 ; i < tdata->pilecount ; i++) {
 574                 /*Switch on response from Participant */
 575                 control = tdata->recvmsg[i].rcv_status; /* tdata: keeps track of all participant responses
 576                                                            written onto the shared array */
 577                 switch(control) {
 578                         case TRANS_DISAGREE:
 579                                 printf("DEBUG-> trans.c Recv TRANS_DISAGREE\n");
 580                                 transdisagree++;
 581                                 break;
 582
 583                         case TRANS_AGREE:
 584                                 printf("DEBUG-> trans.c Recv TRANS_AGREE\n");
 585                                 transagree++;
 586                                 break;
 587
 588                         case TRANS_SOFT_ABORT:
 589                                 printf("DEBUG-> trans.c Recv TRANS_SOFT_ABORT\n");
 590                                 transsoftabort++;
 591                                 break;
 592                         default:
 593                                 printf("Participant sent unknown message in %s, %d\n", __FILE__, __LINE__);
 594                                 return -1;
 595                 }
 596         }
 597
 598         /* Decide what control message to send to Participant */
 599         if(transdisagree > 0) {
 600                 /* Send Abort */
 601                 *(tdata->replyctrl) = TRANS_ABORT;
 602                 printf("DEBUG-> trans.c Sending TRANS_ABORT\n");
 603                 /* Free resources */
 604                 objstrDelete(tdata->rec->cache);
 605                 chashDelete(tdata->rec->lookupTable);
 606                 free(tdata->rec);
 607         } else if(transagree == tdata->pilecount){
 608                 /* Send Commit */
 609                 *(tdata->replyctrl) = TRANS_COMMIT;
 610                 printf("DEBUG-> trans.c Sending TRANS_COMMIT\n");
 611                 /* Free resources */
 612                 objstrDelete(tdata->rec->cache);
 613                 chashDelete(tdata->rec->lookupTable);
 614                 free(tdata->rec);
 615         } else if(transsoftabort > 0 && transdisagree == 0) {
 616                 /* Send Abort in soft abort case followed by retry commiting transaction again*/
 617                 *(tdata->replyctrl) = TRANS_ABORT;
 618                 *(tdata->replyretry) = 1;
 619                 printf("DEBUG-> trans.c Sending TRANS_ABORT\n");
 620         } else {
 621                 printf("DEBUG -> %s, %d: Error: undecided response\n", __FILE__, __LINE__);
 622                 return -1;
 623         }
 624
 625         return 0;
 626 }
 627 /* This function sends the final response to remote machines per thread in their respective socket id */
 628 char sendResponse(thread_data_array_t *tdata, int sd) {
 629         int n, N, sum, oidcount = 0;
 630         char *ptr, retval = 0;
 631         unsigned int *oidnotfound;
 632
 633         /* If the decided response is due to a soft abort and missing objects at the Participant's side */
 634         if(tdata->recvmsg[tdata->thread_id].rcv_status == TRANS_SOFT_ABORT) {
 635                 /* Read list of objects missing */
 636                 if((read(sd, &oidcount, sizeof(int)) != 0) && (oidcount != 0)) {
 637                         N = oidcount * sizeof(unsigned int);
 638                         if((oidnotfound = calloc(oidcount, sizeof(unsigned int))) == NULL) {
 639                                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 640                         }
 641                         ptr = (char *) oidnotfound;
 642                         do {
 643                                 n = read(sd, ptr+sum, N-sum);
 644                                 sum += n;
 645                         } while(sum < N && n !=0);
 646                 }
 647                 retval =  TRANS_SOFT_ABORT;
 648         }
 649         /* If the decided response is TRANS_ABORT */
 650         if(*(tdata->replyctrl) == TRANS_ABORT) {
 651                 retval = TRANS_ABORT;
 652         } else if(*(tdata->replyctrl) == TRANS_COMMIT) { /* If the decided response is TRANS_COMMIT */
 653                 retval = TRANS_COMMIT;
 654         }
 655         /* Send response to the Participant */
 656         if (send(sd, tdata->replyctrl, sizeof(char),MSG_NOSIGNAL) < sizeof(char)) {
 657                 perror("Error sending ctrl message for participant\n");
 658         }
 659
 660         return retval;
 661 }
 662
 663 /* This function opens a connection, places an object read request to the
 664  * remote machine, reads the control message and object if available  and
 665  * copies the object and its header to the local cache.
 666  * TODO replace mnum and midtoIP() with MACHINE_IP address later */
 667
 668 void *getRemoteObj(transrecord_t *record, unsigned int mnum, unsigned int oid) {
 669         int sd, size, val;
 670         struct sockaddr_in serv_addr;
 671         struct hostent *server;
 672         char control;
 673         char machineip[16];
 674         objheader_t *h;
 675         void *objcopy;
 676
 677         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 678                 perror("Error in socket\n");
 679                 return NULL;
 680         }
 681         bzero((char*) &serv_addr, sizeof(serv_addr));
 682         serv_addr.sin_family = AF_INET;
 683         serv_addr.sin_port = htons(LISTEN_PORT);
 684         //serv_addr.sin_addr.s_addr = inet_addr(MACHINE_IP);
 685         midtoIP(mnum,machineip);
 686         machineip[15] = '\0';
 687         serv_addr.sin_addr.s_addr = inet_addr(machineip);
 688         /* Open connection */
 689         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
 690                 perror("Error in connect\n");
 691                 return NULL;
 692         }
 693         char readrequest[sizeof(char)+sizeof(unsigned int)];
 694         readrequest[0] = READ_REQUEST;
 695         *((unsigned int *)(&readrequest[1])) = oid;
 696         if (send(sd, &readrequest, sizeof(readrequest), MSG_NOSIGNAL) < sizeof(readrequest)) {
 697                 perror("Error sending message\n");
 698                 return NULL;
 699         }
 700
 701 #ifdef DEBUG1
 702         printf("DEBUG -> ready to rcv ...\n");
 703 #endif
 704         /* Read response from the Participant */
 705         if((val = read(sd, &control, sizeof(char))) <= 0) {
 706                 perror("No control response for getRemoteObj sent\n");
 707                 return NULL;
 708         }
 709         switch(control) {
 710                 case OBJECT_NOT_FOUND:
 711                         printf("DEBUG -> Control OBJECT_NOT_FOUND received\n");
 712                         return NULL;
 713                 case OBJECT_FOUND:
 714                         /* Read object if found into local cache */
 715                         if((val = read(sd, &size, sizeof(int))) <= 0) {
 716                                 perror("No size is read from the participant\n");
 717                                 return NULL;
 718                         }
 719                         objcopy = objstrAlloc(record->cache, size);
 720                         if((val = read(sd, objcopy, size)) <= 0) {
 721                                 perror("No objects are read from the remote participant\n");
 722                                 return NULL;
 723                         }
 724                         /* Insert into cache's lookup table */
 725                         chashInsert(record->lookupTable, oid, objcopy);
 726                         break;
 727                 default:
 728                         printf("Error in recv request from participant on a READ_REQUEST %s, %d\n",__FILE__, __LINE__);
 729                         return NULL;
 730         }
 731         /* Close connection */
 732         close(sd);
 733         return objcopy;
 734 }
 735
 736 /*This function handles the local trans requests involved in a transaction commiting process
 737  * makes a decision if the local machine sends AGREE or DISAGREE or SOFT_ABORT
 738  * Activates the other nonlocal threads that are waiting for the decision and the
 739  * based on common decision by all groups involved in the transaction it
 740  * either commits or aborts the transaction.
 741  * It also frees the calloced memory resources
 742  */
 743
 744 void *handleLocalReq(void *threadarg) {
 745         int val, i = 0;
 746         short version;
 747         char control = 0, *ptr;
 748         unsigned int oid;
 749         unsigned int *oidnotfound = NULL, *oidlocked = NULL, *oidmod = NULL;
 750         void *mobj, *modptr;
 751         objheader_t *headptr;
 752         local_thread_data_array_t *localtdata;
 753
 754         localtdata = (local_thread_data_array_t *) threadarg;
 755
 756         /* Counters and arrays to formulate decision on control message to be sent */
 757         oidnotfound = (unsigned int *) calloc((localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod), sizeof(unsigned int));
 758         oidlocked = (unsigned int *) calloc((localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod), sizeof(unsigned int));
 759         oidmod = (unsigned int *) calloc(localtdata->tdata->buffer->f.nummod, sizeof(unsigned int));
 760         int objnotfound = 0, objlocked = 0, objmod =0, v_nomatch = 0, v_matchlock = 0, v_matchnolock = 0;
 761         int objmodnotfound = 0, nummodfound = 0;
 762
 763         /* modptr points to the beginning of the object store
 764          * created at the Pariticipant */
 765         if ((modptr = objstrAlloc(mainobjstore, localtdata->tdata->buffer->f.sum_bytes)) == NULL) {
 766                 printf("objstrAlloc error for modified objects %s, %d\n", __FILE__, __LINE__);
 767                 return NULL;
 768         }
 769
 770         ptr = modptr;
 771
 772         /* Process each oid in the machine pile/ group per thread */
 773         for (i = 0; i < localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod; i++) {
 774                 if (i < localtdata->tdata->buffer->f.numread) {//Objs only read and not modified
 775                         int incr = sizeof(unsigned int) + sizeof(short);// Offset that points to next position in the objread array
 776                         incr *= i;
 777                         oid = *((unsigned int *)(localtdata->tdata->buffer->objread + incr));
 778                         incr += sizeof(unsigned int);
 779                         version = *((short *)(localtdata->tdata->buffer->objread + incr));
 780                 } else {//Objs modified
 781                         headptr = (objheader_t *) ptr;
 782                         oid = OID(headptr);
 783                         oidmod[objmod] = oid;//Array containing modified oids
 784                         objmod++;
 785                         version = headptr->version;
 786                         ptr += sizeof(objheader_t) + classsize[TYPE(headptr)];
 787                 }
 788
 789                 /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
 790
 791                 /* Save the oids not found and number of oids not found for later use */
 792                 if ((mobj = mhashSearch(oid)) == NULL) {/* Obj not found */
 793                         /* Save the oids not found and number of oids not found for later use */
 794
 795                         oidnotfound[objnotfound] = OID(((objheader_t *)mobj));
 796                         objnotfound++;
 797                 } else { /* If Obj found in machine (i.e. has not moved) */
 798                         /* Check if Obj is locked by any previous transaction */
 799                         if (STATUS(((objheader_t *)mobj)) & LOCK) {
 800                                 if (version == ((objheader_t *)mobj)->version) {      /* If not locked then match versions */
 801                                         v_matchlock++;
 802                                 } else {/* If versions don't match ...HARD ABORT */
 803                                         v_nomatch++;
 804                                         /* Send TRANS_DISAGREE to Coordinator */
 805                                         localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
 806                                         printf("DEBUG -> Sending TRANS_DISAGREE\n");
 807                                         //return tdata->recvmsg[tdata->thread_id].rcv_status;
 808                                 }
 809                         } else {/* If Obj is not locked then lock object */
 810                                 STATUS(((objheader_t *)mobj)) |= LOCK;
 811                                 //TODO Remove this for Testing
 812                                 randomdelay();
 813
 814                                 /* Save all object oids that are locked on this machine during this transaction request call */
 815                                 oidlocked[objlocked] = OID(((objheader_t *)mobj));
 816                                 objlocked++;
 817                                 if (version == ((objheader_t *)mobj)->version) { /* Check if versions match */
 818                                         v_matchnolock++;
 819                                 } else { /* If versions don't match ...HARD ABORT */
 820                                         v_nomatch++;
 821                                         /* Send TRANS_DISAGREE to Coordinator */
 822                                         localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
 823                                         printf("DEBUG -> Sending TRANS_DISAGREE\n");
 824                                         //      return tdata->recvmsg[tdata->thread_id].rcv_status;
 825                                 }
 826                         }
 827                 }
 828         }
 829
 830         /*Decide the response to be sent to the Coordinator( the local machine in this case)*/
 831
 832         /* Condition to send TRANS_AGREE */
 833         if(v_matchnolock == localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod) {
 834                 localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_AGREE;
 835                 printf("DEBUG -> Sending TRANS_AGREE\n");
 836         }
 837         /* Condition to send TRANS_SOFT_ABORT */
 838         if((v_matchlock > 0 && v_nomatch == 0) || (objnotfound > 0 && v_nomatch == 0)) {
 839                 localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_SOFT_ABORT;
 840                 printf("DEBUG -> Sending TRANS_SOFT_ABORT\n");
 841                 //TODO  currently the only soft abort case that is supported is when object locked by previous
 842                 //transaction => v_matchlock > 0
 843                 //The other case for SOFT ABORT i.e. when object is not found but versions match is not supported
 844                 /* Send number of oids not found and the missing oids if objects are missing in the machine */
 845                 /* TODO Remember to store the oidnotfound for later use
 846                    if(objnotfound != 0) {
 847                    int size = sizeof(unsigned int)* objnotfound;
 848                    }
 849                    */
 850         }
 851
 852         /* Fill out the trans_commit_data_t data structure. This is required for a trans commit process
 853          * if Participant receives a TRANS_COMMIT */
 854         localtdata->transinfo->objmod = oidmod;
 855         localtdata->transinfo->objlocked = oidlocked;
 856         localtdata->transinfo->objnotfound = oidnotfound;
 857         localtdata->transinfo->modptr = modptr;
 858         localtdata->transinfo->nummod = localtdata->tdata->buffer->f.nummod;
 859         localtdata->transinfo->numlocked = objlocked;
 860         localtdata->transinfo->numnotfound = objnotfound;
 861
 862         /*Set flag to show that common data structure for this individual thread has been written to */
 863         //*(tdata->localstatus) |= LM_UPDATED;
 864
 865         /* Lock and update count */
 866         //Thread sleeps until all messages from pariticipants are received by coordinator
 867         pthread_mutex_lock(localtdata->tdata->lock);
 868         (*(localtdata->tdata->count))++; /* keeps track of no of messages received by the coordinator */
 869
 870         /* Wake up the threads and invoke decideResponse (once) */
 871         if(*(localtdata->tdata->count) == localtdata->tdata->pilecount) {
 872                 if (decideResponse(localtdata->tdata) != 0) {
 873                         printf("decideResponse returned error %s,%d\n", __FILE__, __LINE__);
 874                         pthread_mutex_unlock(localtdata->tdata->lock);
 875                         return NULL;
 876                 }
 877                 pthread_cond_broadcast(localtdata->tdata->threshold);
 878         } else {
 879                 pthread_cond_wait(localtdata->tdata->threshold, localtdata->tdata->lock);
 880         }
 881         pthread_mutex_unlock(localtdata->tdata->lock);
 882
 883         /*Based on DecideResponse(), Either COMMIT or ABORT the operation*/
 884         if(*(localtdata->tdata->replyctrl) == TRANS_ABORT){
 885                 if(transAbortProcess(modptr,oidlocked, localtdata->transinfo->numlocked, localtdata->transinfo->nummod, localtdata->tdata->buffer->f.numread) != 0) {
 886                         printf("Error in transAbortProcess() %s,%d\n", __FILE__, __LINE__);
 887                         return NULL;
 888                 }
 889         }else if(*(localtdata->tdata->replyctrl) == TRANS_COMMIT){
 890                 if(transComProcess(localtdata->transinfo) != 0) {
 891                         printf("Error in transComProcess() %s,%d\n", __FILE__, __LINE__);
 892                         return NULL;
 893                 }
 894         }
 895
 896         /* Free memory */
 897         printf("DEBUG -> Freeing...\n");
 898         fflush(stdout);
 899         if (localtdata->transinfo->objmod != NULL) {
 900                 free(localtdata->transinfo->objmod);
 901                 localtdata->transinfo->objmod = NULL;
 902         }
 903         if (localtdata->transinfo->objlocked != NULL) {
 904                 free(localtdata->transinfo->objlocked);
 905                 localtdata->transinfo->objlocked = NULL;
 906         }
 907         if (localtdata->transinfo->objnotfound != NULL) {
 908                 free(localtdata->transinfo->objnotfound);
 909                 localtdata->transinfo->objnotfound = NULL;
 910         }
 911
 912         pthread_exit(NULL);
 913 }
 914 /* This function completes the ABORT process if the transaction is aborting
 915 */
 916 int transAbortProcess(void *modptr, unsigned int *objlocked, int numlocked, int nummod, int numread) {
 917         char *ptr;
 918         int i;
 919         objheader_t *tmp_header;
 920         void *header;
 921
 922         printf("DEBUG -> Recv TRANS_ABORT\n");
 923         /* Set all ref counts as 1 and do garbage collection */
 924         ptr = modptr;
 925         for(i = 0; i< nummod; i++) {
 926                 tmp_header = (objheader_t *)ptr;
 927                 tmp_header->rcount = 1;
 928                 ptr += sizeof(objheader_t) + classsize[TYPE(tmp_header)];
 929         }
 930         /* Unlock objects that was locked due to this transaction */
 931         for(i = 0; i< numlocked; i++) {
 932                 header = mhashSearch(objlocked[i]);// find the header address
 933                 STATUS(((objheader_t *)header)) &= ~(LOCK);
 934         }
 935
 936         /* Send ack to Coordinator */
 937         printf("DEBUG-> TRANS_SUCCESSFUL\n");
 938
 939         /*Free the pointer */
 940         ptr = NULL;
 941         return 0;
 942 }
 943
 944 /*This function completes the COMMIT process is the transaction is commiting
 945 */
 946 int transComProcess(trans_commit_data_t *transinfo) {
 947         objheader_t *header;
 948         int i = 0, offset = 0;
 949         char control;
 950
 951         printf("DEBUG -> Recv TRANS_COMMIT\n");
 952         /* Process each modified object saved in the mainobject store */
 953         for(i=0; i<transinfo->nummod; i++) {
 954                 if((header = (objheader_t *) mhashSearch(transinfo->objmod[i])) == NULL) {
 955                         printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
 956                 }
 957                 /* Change reference count of older address and free space in objstr ?? */
 958                 header->rcount = 1; //TODO Not sure what would be the val
 959
 960                 /* Change ptr address in mhash table */
 961                 mhashRemove(transinfo->objmod[i]);
 962                 mhashInsert(transinfo->objmod[i], (transinfo->modptr + offset));
 963                 offset += sizeof(objheader_t) + classsize[TYPE(header)];
 964
 965                 /* Update object version number */
 966                 header = (objheader_t *) mhashSearch(transinfo->objmod[i]);
 967                 header->version += 1;
 968         }
 969
 970         /* Unlock locked objects */
 971         for(i=0; i<transinfo->numlocked; i++) {
 972                 header = (objheader_t *) mhashSearch(transinfo->objlocked[i]);
 973                 STATUS(header) &= ~(LOCK);
 974         }
 975
 976         //TODO Update location lookup table
 977
 978         /* Send ack to Coordinator */
 979         printf("DEBUG-> TRANS_SUCESSFUL\n");
 980         return 0;
 981 }
 982
 983 /* This function checks if the prefetch oids are same and have same offsets
 984  * for case x.a.b and y.a.b where x and y have same oid's
 985  * or if a.b.c is a subset of x.b.c.d*/
 986 /* check for case where the generated request a.y.z or x.y.z.g then
 987  * prefetch needs to be generated for x.y.z.g  if oid of a and x are same*/
 988 void checkPrefetchTuples(prefetchqelem_t *node) {
 989         int i,j, count,k, sindex, index;
 990         char *ptr, *tmp;
 991         int ntuples, slength;
 992         unsigned int *oid;
 993         short *endoffsets, *arryfields;
 994
 995         /* Check for the case x.y.z and a.b.c are same oids */
 996         ptr = (char *) node;
 997         ntuples = *(GET_NTUPLES(ptr));
 998         oid = GET_PTR_OID(ptr);
 999         endoffsets = GET_PTR_EOFF(ptr, ntuples);
1000         arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1001         /* Find offset length for each tuple */
1002         int numoffset[ntuples];
1003         numoffset[0] = endoffsets[0];
1004         for(i = 1; i<ntuples; i++) {
1005                 numoffset[i] = endoffsets[i] - endoffsets[i-1];
1006         }
1007         /* Check for redundant tuples by comparing oids of each tuple */
1008         for(i = 0; i < ntuples; i++) {
1009                 if(oid[i] == -1)
1010                         continue;
1011                 for(j = i+1 ; j < ntuples; j++) {
1012                         if(oid[j] == -1)
1013                                 continue;
1014                         /*If oids of tuples match */
1015                         if (oid[i] == oid[j]) {
1016                                 /* Find the smallest offset length of two tuples*/
1017                                 if(numoffset[i] >  numoffset[j]){
1018                                         slength = numoffset[j];
1019                                         sindex = j;
1020                                 }
1021                                 else {
1022                                         slength = numoffset[i];
1023                                         sindex = i;
1024                                 }
1025
1026                                 /* Compare the offset values based on the current indices
1027                                  * break if they do not match
1028                                  * if all offset values match then pick the largest tuple*/
1029
1030                                 if(i == 0) {
1031                                         k = 0;
1032                                         index = endoffsets[j -1];
1033                                         for(count = 0; count < slength; count ++) {
1034                                                 if (arryfields[k] != arryfields[index]) {
1035                                                         break;
1036                                                 }
1037                                                 index++;
1038                                                 k++;
1039                                         }
1040                                 } else {
1041                                         k = endoffsets[i-1];
1042                                         index = endoffsets[j-1];
1043                                         printf("Value of slength = %d\n", slength);
1044                                         for(count = 0; count < slength; count++) {
1045                                                 if(arryfields[k] != arryfields[index]) {
1046                                                         break;
1047                                                 }
1048                                                 index++;
1049                                                 k++;
1050                                         }
1051                                 }
1052
1053                                 if(slength == count) {
1054                                         oid[sindex] = -1;
1055                                 }
1056                         }
1057                 }
1058         }
1059 }
1060
1061 void checkPreCache(prefetchqelem_t *node, int *numoffset, int counter, int loopcount, unsigned int objoid, int index, int iter, int oidnfound) {
1062         char *ptr, *tmp;
1063         int ntuples, i, k, flag;
1064         unsigned int * oid;
1065         short *endoffsets, *arryfields;
1066         objheader_t *header;
1067
1068         ptr = (char *) node;
1069         ntuples = *(GET_NTUPLES(ptr));
1070         oid = GET_PTR_OID(ptr);
1071         endoffsets = GET_PTR_EOFF(ptr, ntuples);
1072         arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1073
1074         if(oidnfound == 1) {
1075                 if((header = (objheader_t *) prehashSearch(objoid)) == NULL) {
1076                         return;
1077                 } else { //Found in Prefetch Cache
1078                         //TODO Decide if object is too old, if old remove from cache
1079                         tmp = (char *) header;
1080                         /* Check if any of the offset oid is available in the Prefetch cache */
1081                         for(i = counter; i < loopcount; i++) {
1082                                 objoid = *(tmp + sizeof(objheader_t) + arryfields[counter]);
1083                                 if((header = (objheader_t *)prehashSearch(objoid)) != NULL) {
1084                                         flag = 0;
1085                                 } else {
1086                                         flag = 1;
1087                                         break;
1088                                 }
1089                         }
1090                 }
1091         } else {
1092                 for(i = counter; i<loopcount; i++) {
1093                         if((header = (objheader_t *)prehashSearch(objoid)) != NULL) {
1094                                 tmp = (char *) header;
1095                                 objoid = *(tmp + sizeof(objheader_t) + arryfields[index]);
1096                                 flag = 0;
1097                                 index++;
1098                         } else {
1099                                 flag = 1;
1100                                 break;
1101                         }
1102                 }
1103         }
1104
1105         /* If oid not found locally or in prefetch cache then
1106          * assign the latest oid found as the new oid
1107          * and copy left over offsets into the arrayoffsetfieldarray*/
1108         oid[iter] = objoid;
1109         numoffset[iter] = numoffset[iter] - (i+1);
1110         for(k = 0; k < numoffset[iter] ; k++) {
1111                 arryfields[endoffsets[counter]+k] = arryfields[endoffsets[counter]+k+1];
1112         }
1113
1114         if(flag == 0) {
1115                 oid[iter] = -1;
1116                 numoffset[iter] = 0;
1117         }
1118 }
1119
1120 /* This function makes machine piles to be added into the machine pile queue for each prefetch call */
1121 prefetchpile_t *makePreGroups(prefetchqelem_t *node, int *numoffset) {
1122         char *ptr, *tmp;
1123         int ntuples, slength, i, machinenum;
1124         int maxoffset;
1125         unsigned int *oid;
1126         short *endoffsets, *arryfields, *offset;
1127         prefetchpile_t *head = NULL;
1128
1129         /* Check for the case x.y.z and a.b.c are same oids */
1130         ptr = (char *) node;
1131         ntuples = *(GET_NTUPLES(ptr));
1132         oid = GET_PTR_OID(ptr);
1133         endoffsets = GET_PTR_EOFF(ptr, ntuples);
1134         arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1135
1136         /* Check for redundant tuples by comparing oids of each tuple */
1137         for(i = 0; i < ntuples; i++) {
1138                 if(oid[i] == -1)
1139                         continue;
1140                 /* For each tuple make piles */
1141                 if ((machinenum = lhashSearch(oid[i])) == 0) {
1142                         printf("Error: No such Machine %s, %d\n", __FILE__, __LINE__);
1143                         return NULL;
1144                 }
1145                 /* Insert into machine pile */
1146                 offset = &arryfields[endoffsets[i-1]];
1147                 insertPile(machinenum, oid[i], numoffset[i], offset, head);
1148         }
1149
1150         return head;
1151 }
1152
1153
1154 /* This function checks if the oids within the prefetch tuples are available locally.
1155  * If yes then makes the tuple invalid. If no then rearranges oid and offset values in
1156  * the prefetchqelem_t node to represent a new prefetch tuple */
1157 prefetchpile_t *foundLocal(prefetchqelem_t *node) {
1158         int ntuples,i, j, k, oidnfound = 0, index, flag;
1159         unsigned int *oid;
1160         unsigned int  objoid;
1161         char *ptr, *tmp;
1162         objheader_t *objheader;
1163         short *endoffsets, *arryfields;
1164         prefetchpile_t *head = NULL;
1165
1166         ptr = (char *) node;
1167         ntuples = *(GET_NTUPLES(ptr));
1168         oid = GET_PTR_OID(ptr);
1169         endoffsets = GET_PTR_EOFF(ptr, ntuples);
1170         arryfields = GET_PTR_ARRYFLD(ptr, ntuples);
1171         /* Find offset length for each tuple */
1172         int numoffset[ntuples];//Number of offsets for each tuple
1173         numoffset[0] = endoffsets[0];
1174         for(i = 1; i<ntuples; i++) {
1175                 numoffset[i] = endoffsets[i] - endoffsets[i-1];
1176         }
1177         for(i = 0; i < ntuples; i++) {
1178                 if(oid[i] == -1)
1179                         continue;
1180                 /* If object found locally */
1181                 if((objheader = (objheader_t*) mhashSearch(oid[i])) != NULL) {
1182                         oidnfound = 0;
1183                         tmp = (char *) objheader;
1184                         /* Find the oid of its offset value */
1185                         if(i == 0)
1186                                 index = 0;
1187                         else
1188                                 index = endoffsets[i - 1];
1189                         for(j = 0 ; j < numoffset[i] ; j++) {
1190                                 objoid = *(tmp + sizeof(objheader_t) + arryfields[index]);
1191                                 /*If oid found locally then
1192                                  *assign the latest oid found as the new oid
1193                                  *and copy left over offsets into the arrayoffsetfieldarray*/
1194                                 oid[i] = objoid;
1195                                 numoffset[i] = numoffset[i] - (j+1);
1196                                 for(k = 0; k < numoffset[i]; k++)
1197                                         arryfields[endoffsets[j]+ k] = arryfields[endoffsets[j]+k+1];
1198                                 index++;
1199                                 /*New offset oid not found */
1200                                 if((objheader = (objheader_t*) mhashSearch(objoid)) == NULL) {
1201                                         flag = 1;
1202                                         checkPreCache(node, numoffset, j, numoffset[i], objoid, index, i, oidnfound);
1203                                         break;
1204                                 } else
1205                                         flag = 0;
1206                         }
1207
1208                         /*If all offset oids are found locally,make the prefetch tuple invalid */
1209                         if(flag == 0) {
1210                                 oid[i] = -1;
1211                                 numoffset[i] = 0;
1212                         }
1213                 } else {
1214                         oidnfound = 1;
1215                         /* Look in Prefetch cache */
1216                         checkPreCache(node, numoffset, 0, numoffset[i], oid[i], 0, i, oidnfound);
1217                 }
1218
1219         }
1220         /* Make machine groups */
1221         head = makePreGroups(node, numoffset);
1222         return head;
1223 }
1224
1225 /* This function is called by the thread calling transPrefetch */
1226 void *transPrefetch(void *t) {
1227         //int *offstarray = NULL;
1228         prefetchqelem_t *qnode;
1229         prefetchpile_t *pilehead = NULL;
1230
1231         while(1) {
1232                 /* lock mutex of primary prefetch queue */
1233                 pthread_mutex_lock(&pqueue.qlock);
1234                 /* while primary queue is empty, then wait */
1235                 while((pqueue.front == NULL) && (pqueue.rear == NULL)) {
1236                         pthread_cond_wait(&pqueue.qcond, &pqueue.qlock);
1237                 }
1238
1239                 /* dequeue node to create a machine piles and  finally unlock mutex */
1240                 if((qnode = pre_dequeue()) == NULL) {
1241                         printf("Error: No node returned %s, %d\n", __FILE__, __LINE__);
1242                         return NULL;
1243                 }
1244                 pthread_mutex_unlock(&pqueue.qlock);
1245                 /* Reduce redundant prefetch requests */
1246                 checkPrefetchTuples(qnode);
1247                 /* Check if the tuples are found locally, if yes then reduce them further*/
1248                 /* and group requests by remote machine ids by calling the makePreGroups() */
1249                 pilehead = foundLocal(qnode);
1250
1251                 /* Lock mutex of pool queue */
1252                 pthread_mutex_lock(&mcqueue.qlock);
1253                 /* Update the pool queue with the new remote machine piles generated per prefetch call */
1254                 mcpileenqueue(pilehead);
1255                 /* Broadcast signal on machine pile queue */
1256                 pthread_cond_broadcast(&mcqueue.qcond);
1257                 /* Unlock mutex of  machine pile queue */
1258                 pthread_mutex_unlock(&mcqueue.qlock);
1259                 /* Deallocate the prefetch queue pile node */
1260                 predealloc(qnode);
1261
1262         }
1263 }
1264
1265 /* Each thread in the  pool of threads calls this function to establish connection with
1266  * remote machines, send the prefetch requests and process the reponses from
1267  * the remote machines .
1268  * The thread is active throughout the period of runtime */
1269
1270 void *mcqProcess(void *threadid) {
1271         int tid;
1272         prefetchpile_t *mcpilenode;
1273
1274         tid = (int) threadid;
1275         while(1) {
1276                 /* Lock mutex of mc pile queue */
1277                 pthread_mutex_lock(&mcqueue.qlock);
1278                 /* When mc pile queue is empty, wait */
1279                 while((mcqueue.front == NULL) && (mcqueue.rear == NULL)) {
1280                         pthread_cond_wait(&mcqueue.qcond, &mcqueue.qlock);
1281                 }
1282                 /* Dequeue node to send remote machine connections*/
1283                 if((mcpilenode = mcpiledequeue()) == NULL) {
1284                         printf("Dequeue Error: No node returned %s %d\n", __FILE__, __LINE__);
1285                         return NULL;
1286                 }
1287                 /* Unlock mutex */
1288                 pthread_mutex_unlock(&mcqueue.qlock);
1289
1290                 /*Initiate connection to remote host and send request */
1291                 /* Process Request */
1292                 sendPrefetchReq(mcpilenode, tid);
1293
1294                 /* Deallocate the machine queue pile node */
1295                 mcdealloc(mcpilenode);
1296         }
1297 }
1298
1299 void sendPrefetchReq(prefetchpile_t *mcpilenode, int threadid) {
1300         int sd, i, offset, off, len, endpair, numoffsets, count = 0;
1301         struct sockaddr_in serv_addr;
1302         struct hostent *server;
1303         char machineip[16], control;
1304         objpile_t *tmp;
1305
1306
1307         /* Send Trans Prefetch Request */
1308         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
1309                 perror("Error in socket for TRANS_REQUEST\n");
1310                 return;
1311         }
1312         bzero((char*) &serv_addr, sizeof(serv_addr));
1313         serv_addr.sin_family = AF_INET;
1314         serv_addr.sin_port = htons(LISTEN_PORT);
1315         //serv_addr.sin_addr.s_addr = inet_addr(MACHINE_IP);
1316         midtoIP(mcpilenode->mid ,machineip);
1317         machineip[15] = '\0';
1318         serv_addr.sin_addr.s_addr = inet_addr(machineip);
1319
1320         /* Open Connection */
1321         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
1322                 perror("Error in connect for TRANS_REQUEST\n");
1323                 return;
1324         }
1325
1326         /* Send TRANS_PREFETCH control message */
1327         control = TRANS_PREFETCH;
1328         if(send(sd, &control, sizeof(char), MSG_NOSIGNAL) < sizeof(char)) {
1329                 perror("Error in sending prefetch control\n");
1330                 return;
1331         }
1332
1333         /* Send Oids and offsets in pairs */
1334         tmp = mcpilenode->objpiles;
1335         while(tmp != NULL) {
1336                 off = offset = 0;
1337                 count++;  // Keeps track of the number of oid and offset tuples sent per remote machine
1338                 len = sizeof(int) + sizeof(unsigned int) + ((tmp->numoffset) * sizeof(short));
1339                 char oidnoffset[len];
1340                 memcpy(oidnoffset, &len, sizeof(int));
1341                 off = sizeof(int);
1342                 memcpy(oidnoffset + off, &tmp->oid, sizeof(unsigned int));
1343                 off += sizeof(unsigned int);
1344                 for(i = 0; i < numoffsets; i++) {
1345                         offset = off +  (i * sizeof(short));
1346                         memcpy(oidnoffset + offset, tmp->offset, sizeof(short));
1347                 }
1348                 if (send(sd, &oidnoffset, sizeof(oidnoffset),MSG_NOSIGNAL) < sizeof(oidnoffset)) {
1349                         perror("Error sending fixed bytes for thread\n");
1350                         return;
1351                 }
1352                 tmp = tmp->next;
1353         }
1354
1355         /* Send a special char -1 to represent the end of sending oids + offset pair to remote machine */
1356         endpair = -1;
1357         if (send(sd, &endpair, sizeof(int), MSG_NOSIGNAL) < sizeof(int)) {
1358                 perror("Error sending fixed bytes for thread\n");
1359                 return;
1360         }
1361
1362         /* Get Response from the remote machine */
1363         getPrefetchResponse(count,sd);
1364         close(sd);
1365         return;
1366 }
1367
1368 void getPrefetchResponse(int count, int sd) {
1369         int i = 0, val, n, N, sum, index, objsize;
1370         unsigned int bufsize,oid;
1371         char buffer[RECEIVE_BUFFER_SIZE], control;
1372         char *ptr;
1373         void *modptr, *oldptr;
1374
1375         /* Read  prefetch response from the Remote machine */
1376         if((val = read(sd, &control, sizeof(char))) <= 0) {
1377                 perror("No control response for Prefetch request sent\n");
1378                 return;
1379         }
1380
1381         if(control == TRANS_PREFETCH_RESPONSE) {
1382                 /*For each oid and offset tuple sent as prefetch request to remote machine*/
1383                 while(i < count) {
1384                         /* Clear contents of buffer */
1385                         memset(buffer, 0, RECEIVE_BUFFER_SIZE);
1386                         sum = 0;
1387                         index = 0;
1388                         /* Read the size of buffer to be received */
1389                         if((N = read(sd, buffer, sizeof(unsigned int))) <= 0) {
1390                                 perror("Size of buffer not recv\n");
1391                                 return;
1392                         }
1393                         memcpy(&bufsize, buffer, sizeof(unsigned int));
1394                         ptr = buffer + sizeof(unsigned int);
1395                         /* Keep receiving the buffer containing oid info */
1396                         do {
1397                                 n = recv((int)sd, (void *)ptr+sum, bufsize-sum, 0);
1398                                 sum +=n;
1399                         } while(sum < bufsize && n != 0);
1400                         /* Decode the contents of the buffer */
1401                         index = sizeof(unsigned int);
1402                         while(index < (bufsize - sizeof(unsigned int))) {
1403                                 if(buffer[index] == OBJECT_FOUND) {
1404                                         /* Increment it to get the object */
1405                                         index += sizeof(char);
1406                                         memcpy(&oid, buffer + index, sizeof(unsigned int));
1407                                         index += sizeof(unsigned int);
1408                                         /* Lock the Prefetch Cache look up table*/
1409                                         pthread_mutex_lock(&pflookup.lock);
1410                                         /* For each object found add to Prefetch Cache */
1411                                         memcpy(&objsize, buffer + index, sizeof(int));
1412                                         if ((modptr = objstrAlloc(prefetchcache, objsize)) == NULL) {
1413                                                 printf("objstrAlloc error for copying into prefetch cache %s, %d\n", __FILE__, __LINE__);
1414                                                 return;
1415                                         }
1416                                         memcpy(modptr, buffer+index, objsize);
1417                                         index += sizeof(int);
1418                                         /* Insert the oid and its address into the prefetch hash lookup table */
1419                                         /* Do a version comparison if the oid exists */
1420                                         if((oldptr = prehashSearch(oid)) != NULL) {
1421                                                 /* If older version then update with new object ptr */
1422                                                 if(((objheader_t *)oldptr)->version < ((objheader_t *)modptr)->version) {
1423                                                         prehashRemove(oid);
1424                                                         prehashInsert(oid, modptr);
1425                                                 } else if(((objheader_t *)oldptr)->version == ((objheader_t *)modptr)->version) {
1426                                                         /* Add the new object ptr to hash table */
1427                                                         prehashInsert(oid, modptr);
1428                                                 } else { /* Do nothing */
1429                                                         ;
1430                                                 }
1431                                         } else {/*If doesn't no match found in hashtable, add the object ptr to hash table*/
1432                                                 prehashInsert(oid, modptr);
1433                                         }
1434                                         /* Broadcast signal on prefetch cache condition variable */
1435                                         pthread_cond_broadcast(&pflookup.cond);
1436                                         /* Unlock the Prefetch Cache look up table*/
1437                                         pthread_mutex_unlock(&pflookup.lock);
1438                                 } else if(buffer[index] == OBJECT_NOT_FOUND) {
1439                                         /* Increment it to get the object */
1440                                         /* TODO: For each object not found query DHT for new location and retrieve the object */
1441                                         index += sizeof(char);
1442                                         memcpy(&oid, buffer + index, sizeof(unsigned int));
1443                                         index += sizeof(unsigned int);
1444                                         /* Throw an error */
1445                                         printf("OBJECT NOT FOUND.... THIS SHOULD NOT HAPPEN...TERMINATE PROGRAM\n");
1446                                         exit(-1);
1447                                 } else
1448                                         printf("Error in decoding the index value %s, %d\n",__FILE__, __LINE__);
1449                         }
1450
1451                         i++;
1452                 }
1453         } else
1454                 printf("Error in receving response for prefetch request %s, %d\n",__FILE__, __LINE__);
1455         return;
1456 }