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

   1 #include "dstm.h"
   2 #include "ip.h"
   3 #include "clookup.h"
   4 #include "mlookup.h"
   5 #include "llookup.h"
   6 #include "plookup.h"
   7 #include<pthread.h>
   8 #include<sys/types.h>
   9 #include<sys/socket.h>
  10 #include<netdb.h>
  11 #include<netinet/in.h>
  12 #include <sys/types.h>
  13 #include <unistd.h>
  14 #include <time.h>
  15
  16 #define LISTEN_PORT 2156
  17 #define MACHINE_IP "127.0.0.1"
  18 #define RECEIVE_BUFFER_SIZE 2048
  19
  20 extern int classsize[];
  21 objstr_t *mainobjstore;
  22 plistnode_t *createPiles(transrecord_t *);
  23 int checkPrefetchTuples(int **, int *, short);
  24
  25 inline int arrayLength(int *array) {
  26         int i;
  27         for(i=0 ;array[i] != -1; i++)
  28                 ;
  29         return i;
  30 }
  31
  32 /* This functions inserts randowm wait delays in the order of msec */
  33 void randomdelay(void)
  34 {
  35         struct timespec req, rem;
  36         time_t t;
  37
  38         t = time(NULL);
  39         req.tv_sec = 0;
  40         req.tv_nsec = (long)(1000000 + (t%10000000)); //1-11 msec
  41         nanosleep(&req, &rem);
  42         return;
  43 }
  44
  45 transrecord_t *transStart()
  46 {
  47         transrecord_t *tmp = malloc(sizeof(transrecord_t));
  48         tmp->cache = objstrCreate(1048576);
  49         tmp->lookupTable = chashCreate(HASH_SIZE, LOADFACTOR);
  50         return tmp;
  51 }
  52
  53 /* This function finds the location of the objects involved in a transaction
  54  * and returns the pointer to the object if found in a remote location */
  55 objheader_t *transRead(transrecord_t *record, unsigned int oid)
  56 {
  57         unsigned int machinenumber;
  58         objheader_t *tmp, *objheader;
  59         void *objcopy;
  60         int size;
  61         void *buf;
  62                 /* Search local cache */
  63         if((objheader =(objheader_t *)chashSearch(record->lookupTable, oid)) != NULL){
  64                 return(objheader);
  65         } else if ((objheader = (objheader_t *) mhashSearch(oid)) != NULL) {
  66                 /* Look up in machine lookup table  and copy  into cache*/
  67                 tmp = mhashSearch(oid);
  68                 size = sizeof(objheader_t)+classsize[tmp->type];
  69                 objcopy = objstrAlloc(record->cache, size);
  70                 memcpy(objcopy, (void *)tmp, size);
  71                 /* Insert into cache's lookup table */
  72                 chashInsert(record->lookupTable, objheader->oid, objcopy);
  73                 return(objcopy);
  74         } else { /* If not found in machine look up */
  75                 /* Get the object from the remote location */
  76                 machinenumber = lhashSearch(oid);
  77                 objcopy = getRemoteObj(record, machinenumber, oid);
  78                 if(objcopy == NULL) {
  79                         //If object is not found in Remote location
  80                         //printf("Object oid = %d not found in Machine %d\n", oid, machinenumber);
  81                         return NULL;
  82                 }
  83                 else {
  84                         //printf("Object oid = %d found in Machine %d\n", oid, machinenumber);
  85                         return(objcopy);
  86                 }
  87         }
  88 }
  89 /* This function creates objects in the transaction record */
  90 objheader_t *transCreateObj(transrecord_t *record, unsigned short type)
  91 {
  92         objheader_t *tmp = (objheader_t *) objstrAlloc(record->cache, (sizeof(objheader_t) + classsize[type]));
  93         tmp->oid = getNewOID();
  94         tmp->type = type;
  95         tmp->version = 1;
  96         tmp->rcount = 0; //? not sure how to handle this yet
  97         tmp->status = 0;
  98         tmp->status |= NEW;
  99         chashInsert(record->lookupTable, tmp->oid, tmp);
 100         return tmp;
 101 }
 102 /* This function creates machine piles based on all machines involved in a
 103  * transaction commit request */
 104 plistnode_t *createPiles(transrecord_t *record) {
 105         int i = 0;
 106         unsigned int size;/* Represents number of bins in the chash table */
 107         chashlistnode_t *curr, *ptr, *next;
 108         plistnode_t *pile = NULL;
 109         unsigned int machinenum;
 110         void *localmachinenum;
 111         objheader_t *headeraddr;
 112
 113         ptr = record->lookupTable->table;
 114         size = record->lookupTable->size;
 115
 116         for(i = 0; i < size ; i++) {
 117                 curr = &ptr[i];
 118                 /* Inner loop to traverse the linked list of the cache lookupTable */
 119                 while(curr != NULL) {
 120                         //if the first bin in hash table is empty
 121                         if(curr->key == 0) {
 122                                 break;
 123                         }
 124                         next = curr->next;
 125                         //Get machine location for object id
 126
 127                         if ((machinenum = lhashSearch(curr->key)) == 0) {
 128                                printf("Error: No such machine %s, %d\n", __FILE__, __LINE__);
 129                                return NULL;
 130                         }
 131
 132                         if ((headeraddr = chashSearch(record->lookupTable, curr->key)) == NULL) {
 133                                 printf("Error: No such oid %s, %d\n", __FILE__, __LINE__);
 134                                 return NULL;
 135                         }
 136                         //Make machine groups
 137                         if ((pile = pInsert(pile, headeraddr, machinenum, record->lookupTable->numelements)) == NULL) {
 138                                 printf("pInsert error %s, %d\n", __FILE__, __LINE__);
 139                                 return NULL;
 140                         }
 141
 142                         /* Check if local or not */
 143                         if((localmachinenum = mhashSearch(curr->key)) != NULL) {
 144                                 pile->local = 1; //True i.e. local
 145                         }
 146
 147                         curr = next;
 148                 }
 149         }
 150
 151         return pile;
 152 }
 153 /* This function initiates the transaction commit process
 154  * Spawns threads for each of the new connections with Participants
 155  * and creates new piles by calling the createPiles(),
 156  * Fills the piles with necesaary information and
 157  * Sends a transrequest() to each pile*/
 158 int transCommit(transrecord_t *record) {
 159         unsigned int tot_bytes_mod, *listmid;
 160         plistnode_t *pile;
 161         int i, rc, val;
 162         int pilecount = 0, offset, threadnum = 0, trecvcount = 0, tmachcount = 0;
 163         char buffer[RECEIVE_BUFFER_SIZE],control;
 164         char transid[TID_LEN];
 165         trans_req_data_t *tosend;
 166         trans_commit_data_t transinfo;
 167         static int newtid = 0;
 168         char treplyctrl = 0, treplyretry = 0; /* keeps track of the common response that needs to be sent */
 169         char localstat = 0;
 170
 171         /* Look through all the objects in the transaction record and make piles
 172          * for each machine involved in the transaction*/
 173         pile = createPiles(record);
 174
 175         /* Create the packet to be sent in TRANS_REQUEST */
 176
 177         /* Count the number of participants */
 178         pilecount = pCount(pile);
 179
 180         /* Create a list of machine ids(Participants) involved in transaction   */
 181         if((listmid = calloc(pilecount, sizeof(unsigned int))) == NULL) {
 182                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 183                 return 1;
 184         }
 185         pListMid(pile, listmid);
 186
 187
 188         /* Initialize thread variables,
 189          * Spawn a thread for each Participant involved in a transaction */
 190         pthread_t thread[pilecount];
 191         pthread_attr_t attr;
 192         pthread_cond_t tcond;
 193         pthread_mutex_t tlock;
 194         pthread_mutex_t tlshrd;
 195
 196         thread_data_array_t *thread_data_array;
 197         thread_data_array = (thread_data_array_t *) malloc(sizeof(thread_data_array_t)*pilecount);
 198         local_thread_data_array_t *ltdata;
 199         if((ltdata = calloc(1, sizeof(local_thread_data_array_t))) == NULL) {
 200                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 201                 return 1;
 202         }
 203
 204         thread_response_t rcvd_control_msg[pilecount];  /* Shared thread array that keeps track of responses of participants */
 205
 206         /* Initialize and set thread detach attribute */
 207         pthread_attr_init(&attr);
 208         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
 209         pthread_mutex_init(&tlock, NULL);
 210         pthread_cond_init(&tcond, NULL);
 211
 212         /* Process each machine pile */
 213         while(pile != NULL) {
 214                 //Create transaction id
 215                 newtid++;
 216                 //trans_req_data_t *tosend;
 217                 if ((tosend = calloc(1, sizeof(trans_req_data_t))) == NULL) {
 218                         printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 219                         return 1;
 220                 }
 221                 tosend->f.control = TRANS_REQUEST;
 222                 sprintf(tosend->f.trans_id, "%x_%d", pile->mid, newtid);
 223                 tosend->f.mcount = pilecount;
 224                 tosend->f.numread = pile->numread;
 225                 tosend->f.nummod = pile->nummod;
 226                 tosend->f.sum_bytes = pile->sum_bytes;
 227                 tosend->listmid = listmid;
 228                 tosend->objread = pile->objread;
 229                 tosend->oidmod = pile->oidmod;
 230                 thread_data_array[threadnum].thread_id = threadnum;
 231                 thread_data_array[threadnum].mid = pile->mid;
 232                 thread_data_array[threadnum].pilecount = pilecount;
 233                 thread_data_array[threadnum].buffer = tosend;
 234                 thread_data_array[threadnum].recvmsg = rcvd_control_msg;
 235                 thread_data_array[threadnum].threshold = &tcond;
 236                 thread_data_array[threadnum].lock = &tlock;
 237                 thread_data_array[threadnum].count = &trecvcount;
 238                 //thread_data_array[threadnum].localstatus = &localstat;
 239                 thread_data_array[threadnum].replyctrl = &treplyctrl;
 240                 thread_data_array[threadnum].replyretry = &treplyretry;
 241                 thread_data_array[threadnum].rec = record;
 242                 /* If local do not create any extra connection */
 243                 if(pile->local != 1) {
 244                         rc = pthread_create(&thread[threadnum], NULL, transRequest, (void *) &thread_data_array[threadnum]);
 245                         if (rc) {
 246                                 perror("Error in pthread create\n");
 247                                 return 1;
 248                         }
 249                 } else {
 250                         /*Unset the pile->local flag*/
 251                         pile->local = 0;
 252                         /*Set flag to identify that Local machine is involved*/
 253                         ltdata->tdata = &thread_data_array[threadnum];
 254                         ltdata->transinfo = &transinfo;
 255                         val = pthread_create(&thread[threadnum], NULL, handleLocalReq, (void *) ltdata);
 256                         if (val) {
 257                                 perror("Error in pthread create\n");
 258                                 return 1;
 259                         }
 260                 }
 261                 threadnum++;
 262                 pile = pile->next;
 263         }
 264
 265         /* Free attribute and wait for the other threads */
 266         pthread_attr_destroy(&attr);
 267         for (i = 0 ;i < pilecount ; i++) {
 268                 rc = pthread_join(thread[i], NULL);
 269                 if (rc)
 270                 {
 271                         printf("ERROR return code from pthread_join() is %d\n", rc);
 272                         return 1;
 273                 }
 274         }
 275
 276         /* Free resources */
 277         pthread_cond_destroy(&tcond);
 278         pthread_mutex_destroy(&tlock);
 279         free(tosend);
 280         free(listmid);
 281         pDelete(pile);
 282         free(thread_data_array);
 283         free(ltdata);
 284
 285         /* Retry trans commit procedure if not sucessful in the first try */
 286         if(treplyretry == 1) {
 287                 /* wait a random amount of time */
 288                 randomdelay();
 289                 //sleep(1);
 290                 /* Retry the commiting transaction again */
 291                 transCommit(record);
 292         }
 293
 294         return 0;
 295 }
 296
 297 /* This function sends information involved in the transaction request and
 298  * accepts a response from particpants.
 299  * It calls decideresponse() to decide on what control message
 300  * to send next and sends the message using sendResponse()*/
 301 void *transRequest(void *threadarg) {
 302         int sd, i, n;
 303         struct sockaddr_in serv_addr;
 304         struct hostent *server;
 305         thread_data_array_t *tdata;
 306         objheader_t *headeraddr;
 307         char buffer[RECEIVE_BUFFER_SIZE], control, recvcontrol;
 308         char machineip[16], retval;
 309
 310         tdata = (thread_data_array_t *) threadarg;
 311
 312         /* Send Trans Request */
 313         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 314                 perror("Error in socket for TRANS_REQUEST\n");
 315                 return NULL;
 316         }
 317         bzero((char*) &serv_addr, sizeof(serv_addr));
 318         serv_addr.sin_family = AF_INET;
 319         serv_addr.sin_port = htons(LISTEN_PORT);
 320         midtoIP(tdata->mid,machineip);
 321         machineip[15] = '\0';
 322         serv_addr.sin_addr.s_addr = inet_addr(machineip);
 323         /* Open Connection */
 324         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
 325                 perror("Error in connect for TRANS_REQUEST\n");
 326                 return NULL;
 327         }
 328
 329         printf("DEBUG-> trans.c Sending TRANS_REQUEST to mid %s\n", machineip);
 330         /* Send bytes of data with TRANS_REQUEST control message */
 331         if (send(sd, &(tdata->buffer->f), sizeof(fixed_data_t),MSG_NOSIGNAL) < sizeof(fixed_data_t)) {
 332                 perror("Error sending fixed bytes for thread\n");
 333                 return NULL;
 334         }
 335         /* Send list of machines involved in the transaction */
 336         {
 337           int size=sizeof(unsigned int)*tdata->pilecount;
 338           if (send(sd, tdata->buffer->listmid, size, MSG_NOSIGNAL) < size) {
 339             perror("Error sending list of machines for thread\n");
 340             return NULL;
 341           }
 342         }
 343         /* Send oids and version number tuples for objects that are read */
 344         {
 345           int size=(sizeof(unsigned int)+sizeof(short))*tdata->buffer->f.numread;
 346           if (send(sd, tdata->buffer->objread, size, MSG_NOSIGNAL) < size) {
 347             perror("Error sending tuples for thread\n");
 348             return NULL;
 349           }
 350         }
 351         /* Send objects that are modified */
 352         for(i = 0; i < tdata->buffer->f.nummod ; i++) {
 353           int size;
 354           headeraddr = chashSearch(tdata->rec->lookupTable, tdata->buffer->oidmod[i]);
 355           size=sizeof(objheader_t)+classsize[headeraddr->type];
 356           if (send(sd, headeraddr, size, MSG_NOSIGNAL)  < size) {
 357             perror("Error sending obj modified for thread\n");
 358             return NULL;
 359           }
 360         }
 361
 362         /* Read control message from Participant */
 363         if((n = read(sd, &control, sizeof(char))) <= 0) {
 364                 perror("Error in reading control message from Participant\n");
 365                 return NULL;
 366         }
 367         recvcontrol = control;
 368
 369         /* Update common data structure and increment count */
 370         tdata->recvmsg[tdata->thread_id].rcv_status = recvcontrol;
 371
 372         /* Lock and update count */
 373         //Thread sleeps until all messages from pariticipants are received by coordinator
 374         pthread_mutex_lock(tdata->lock);
 375
 376         (*(tdata->count))++; /* keeps track of no of messages received by the coordinator */
 377
 378         /* Wake up the threads and invoke decideResponse (once) */
 379         if(*(tdata->count) == tdata->pilecount) {
 380                 if (decideResponse(tdata) != 0) {
 381                         printf("decideResponse returned error %s,%d\n", __FILE__, __LINE__);
 382                         pthread_mutex_unlock(tdata->lock);
 383                         close(sd);
 384                         return NULL;
 385                 }
 386                 pthread_cond_broadcast(tdata->threshold);
 387         } else {
 388                 pthread_cond_wait(tdata->threshold, tdata->lock);
 389         }
 390         pthread_mutex_unlock(tdata->lock);
 391
 392         /* Send the final response such as TRANS_COMMIT or TRANS_ABORT t
 393          * to all participants in their respective socket */
 394         if (sendResponse(tdata, sd) == 0) {
 395                 printf("sendResponse returned error %s,%d\n", __FILE__, __LINE__);
 396                 pthread_mutex_unlock(tdata->lock);
 397                 close(sd);
 398                 return NULL;
 399         }
 400
 401         /* Close connection */
 402         close(sd);
 403         pthread_exit(NULL);
 404 }
 405
 406 /* This function decides the reponse that needs to be sent to
 407  * all Participant machines involved in the transaction commit */
 408 int decideResponse(thread_data_array_t *tdata) {
 409         char control;
 410         int i, transagree = 0, transdisagree = 0, transsoftabort = 0; /* Counters to formulate decision of what
 411                                                                          message to send */
 412
 413         //Check common data structure
 414         for (i = 0 ; i < tdata->pilecount ; i++) {
 415                 /*Switch on response from Participant */
 416                 control = tdata->recvmsg[i].rcv_status; /* tdata: keeps track of all participant responses
 417                                                            written onto the shared array */
 418                 switch(control) {
 419                         case TRANS_DISAGREE:
 420                                 printf("DEBUG-> trans.c Recv TRANS_DISAGREE\n");
 421                                 transdisagree++;
 422                                 break;
 423
 424                         case TRANS_AGREE:
 425                                 printf("DEBUG-> trans.c Recv TRANS_AGREE\n");
 426                                 transagree++;
 427                                 break;
 428
 429                         case TRANS_SOFT_ABORT:
 430                                 printf("DEBUG-> trans.c Recv TRANS_SOFT_ABORT\n");
 431                                 transsoftabort++;
 432                                 break;
 433                         default:
 434                                 printf("Participant sent unknown message in %s, %d\n", __FILE__, __LINE__);
 435                                 return -1;
 436                 }
 437         }
 438
 439         /* Decide what control message to send to Participant */
 440         if(transdisagree > 0) {
 441                 /* Send Abort */
 442                 *(tdata->replyctrl) = TRANS_ABORT;
 443                 printf("DEBUG-> trans.c Sending TRANS_ABORT\n");
 444                 objstrDelete(tdata->rec->cache);
 445                 chashDelete(tdata->rec->lookupTable);
 446                 free(tdata->rec);
 447         } else if(transagree == tdata->pilecount){
 448                 /* Send Commit */
 449                 *(tdata->replyctrl) = TRANS_COMMIT;
 450                 printf("DEBUG-> trans.c Sending TRANS_COMMIT\n");
 451                 objstrDelete(tdata->rec->cache);
 452                 chashDelete(tdata->rec->lookupTable);
 453                 free(tdata->rec);
 454         } else if(transsoftabort > 0 && transdisagree == 0) {
 455                 /* Send Abort in soft abort case followed by retry commiting transaction again*/
 456                 *(tdata->replyctrl) = TRANS_ABORT;
 457                 *(tdata->replyretry) = 1;
 458                 printf("DEBUG-> trans.c Sending TRANS_ABORT\n");
 459         } else {
 460                 printf("DEBUG -> %s, %d: Error: undecided response\n", __FILE__, __LINE__);
 461                 return -1;
 462         }
 463
 464         return 0;
 465 }
 466 /* This function sends the final response to all threads in their respective socket id */
 467 char sendResponse(thread_data_array_t *tdata, int sd) {
 468         int n, N, sum, oidcount = 0;
 469         char *ptr, retval = 0;
 470         unsigned int *oidnotfound;
 471
 472         /* If the decided response is due to a soft abort and missing objects at the Participant's side */
 473         if(tdata->recvmsg[tdata->thread_id].rcv_status == TRANS_SOFT_ABORT) {
 474                 /* Read list of objects missing */
 475                 if((read(sd, &oidcount, sizeof(int)) != 0) && (oidcount != 0)) {
 476                         N = oidcount * sizeof(unsigned int);
 477                         if((oidnotfound = calloc(oidcount, sizeof(unsigned int))) == NULL) {
 478                                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 479                         }
 480                         ptr = (char *) oidnotfound;
 481                         do {
 482                                 n = read(sd, ptr+sum, N-sum);
 483                                 sum += n;
 484                         } while(sum < N && n !=0);
 485                 }
 486                 retval =  TRANS_SOFT_ABORT;
 487         }
 488         /* If the decided response is TRANS_ABORT */
 489         if(*(tdata->replyctrl) == TRANS_ABORT) {
 490                 retval = TRANS_ABORT;
 491         } else if(*(tdata->replyctrl) == TRANS_COMMIT) { /* If the decided response is TRANS_COMMIT */
 492                 retval = TRANS_COMMIT;
 493         }
 494         /* Send response to the Participant */
 495         if (send(sd, tdata->replyctrl, sizeof(char),MSG_NOSIGNAL) < sizeof(char)) {
 496                 perror("Error sending ctrl message for participant\n");
 497         }
 498
 499         return retval;
 500 }
 501
 502 /* This function opens a connection, places an object read request to the
 503  * remote machine, reads the control message and object if available  and
 504  * copies the object and its header to the local cache.
 505  * TODO replace mnum and midtoIP() with MACHINE_IP address later */
 506
 507 void *getRemoteObj(transrecord_t *record, unsigned int mnum, unsigned int oid) {
 508         int sd, size, val;
 509         struct sockaddr_in serv_addr;
 510         struct hostent *server;
 511         char control;
 512         char machineip[16];
 513         objheader_t *h;
 514         void *objcopy;
 515
 516         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 517                 perror("Error in socket\n");
 518                 return NULL;
 519         }
 520         bzero((char*) &serv_addr, sizeof(serv_addr));
 521         serv_addr.sin_family = AF_INET;
 522         serv_addr.sin_port = htons(LISTEN_PORT);
 523         //serv_addr.sin_addr.s_addr = inet_addr(MACHINE_IP);
 524         midtoIP(mnum,machineip);
 525         machineip[15] = '\0';
 526         serv_addr.sin_addr.s_addr = inet_addr(machineip);
 527         /* Open connection */
 528         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
 529                 perror("Error in connect\n");
 530                 return NULL;
 531         }
 532         char readrequest[sizeof(char)+sizeof(unsigned int)];
 533         readrequest[0] = READ_REQUEST;
 534         *((unsigned int *)(&readrequest[1])) = oid;
 535         if (send(sd, &readrequest, sizeof(readrequest), MSG_NOSIGNAL) < sizeof(readrequest)) {
 536                 perror("Error sending message\n");
 537                 return NULL;
 538         }
 539
 540 #ifdef DEBUG1
 541         printf("DEBUG -> ready to rcv ...\n");
 542 #endif
 543         /* Read response from the Participant */
 544         if((val = read(sd, &control, sizeof(char))) <= 0) {
 545                 perror("No control response for getRemoteObj sent\n");
 546                 return NULL;
 547         }
 548         switch(control) {
 549                 case OBJECT_NOT_FOUND:
 550                         printf("DEBUG -> Control OBJECT_NOT_FOUND received\n");
 551                         return NULL;
 552                 case OBJECT_FOUND:
 553                         /* Read object if found into local cache */
 554                         if((val = read(sd, &size, sizeof(int))) <= 0) {
 555                                 perror("No size is read from the participant\n");
 556                                 return NULL;
 557                         }
 558                         objcopy = objstrAlloc(record->cache, size);
 559                         if((val = read(sd, objcopy, size)) <= 0) {
 560                                 perror("No objects are read from the remote participant\n");
 561                                 return NULL;
 562                         }
 563                         /* Insert into cache's lookup table */
 564                         chashInsert(record->lookupTable, oid, objcopy);
 565                         break;
 566                 default:
 567                         printf("Error in recv request from participant on a READ_REQUEST %s, %d\n",__FILE__, __LINE__);
 568                         return NULL;
 569         }
 570         /* Close connection */
 571         close(sd);
 572         return objcopy;
 573 }
 574
 575 /*This function handles the local trans requests involved in a transaction commiting process
 576  * makes a decision if the local machine sends AGREE or DISAGREE or SOFT_ABORT
 577  * Activates the other nonlocal threads that are waiting for the decision and the
 578  * based on common decision by all groups involved in the transaction it
 579  * either commits or aborts the transaction.
 580  * It also frees the calloced memory resources
 581  */
 582
 583 void *handleLocalReq(void *threadarg) {
 584         int val, i = 0;
 585         short version;
 586         char control = 0, *ptr;
 587         unsigned int oid;
 588         unsigned int *oidnotfound = NULL, *oidlocked = NULL, *oidmod = NULL;
 589         void *mobj, *modptr;
 590         objheader_t *headptr;
 591         local_thread_data_array_t *localtdata;
 592
 593         localtdata = (local_thread_data_array_t *) threadarg;
 594
 595         /* Counters and arrays to formulate decision on control message to be sent */
 596         oidnotfound = (unsigned int *) calloc((localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod), sizeof(unsigned int));
 597         oidlocked = (unsigned int *) calloc((localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod), sizeof(unsigned int));
 598         oidmod = (unsigned int *) calloc(localtdata->tdata->buffer->f.nummod, sizeof(unsigned int));
 599         int objnotfound = 0, objlocked = 0, objmod =0, v_nomatch = 0, v_matchlock = 0, v_matchnolock = 0;
 600         int objmodnotfound = 0, nummodfound = 0;
 601
 602         /* modptr points to the beginning of the object store
 603          * created at the Pariticipant */
 604         if ((modptr = objstrAlloc(mainobjstore, localtdata->tdata->buffer->f.sum_bytes)) == NULL) {
 605                 printf("objstrAlloc error for modified objects %s, %d\n", __FILE__, __LINE__);
 606                 return NULL;
 607         }
 608
 609         ptr = modptr;
 610
 611         /* Process each oid in the machine pile/ group per thread */
 612         for (i = 0; i < localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod; i++) {
 613                 if (i < localtdata->tdata->buffer->f.numread) {//Objs only read and not modified
 614                         int incr = sizeof(unsigned int) + sizeof(short);// Offset that points to next position in the objread array
 615                         incr *= i;
 616                         oid = *((unsigned int *)(localtdata->tdata->buffer->objread + incr));
 617                         incr += sizeof(unsigned int);
 618                         version = *((short *)(localtdata->tdata->buffer->objread + incr));
 619                 } else {//Objs modified
 620                         headptr = (objheader_t *) ptr;
 621                         oid = headptr->oid;
 622                         oidmod[objmod] = oid;//Array containing modified oids
 623                         objmod++;
 624                         version = headptr->version;
 625                         ptr += sizeof(objheader_t) + classsize[headptr->type];
 626                 }
 627
 628                 /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
 629
 630                 /* Save the oids not found and number of oids not found for later use */
 631                 if ((mobj = mhashSearch(oid)) == NULL) {/* Obj not found */
 632                         /* Save the oids not found and number of oids not found for later use */
 633
 634                         oidnotfound[objnotfound] = ((objheader_t *)mobj)->oid;
 635                         objnotfound++;
 636                 } else { /* If Obj found in machine (i.e. has not moved) */
 637                         /* Check if Obj is locked by any previous transaction */
 638                         if ((((objheader_t *)mobj)->status & LOCK) == LOCK) {
 639                                 if (version == ((objheader_t *)mobj)->version) {      /* If not locked then match versions */
 640                                         v_matchlock++;
 641                                 } else {/* If versions don't match ...HARD ABORT */
 642                                         v_nomatch++;
 643                                         /* Send TRANS_DISAGREE to Coordinator */
 644                                         localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
 645                                         printf("DEBUG -> Sending TRANS_DISAGREE\n");
 646                                         //return tdata->recvmsg[tdata->thread_id].rcv_status;
 647                                 }
 648                         } else {/* If Obj is not locked then lock object */
 649                                 ((objheader_t *)mobj)->status |= LOCK;
 650                                 //TODO Remove this for Testing
 651                                 randomdelay();
 652
 653                                 /* Save all object oids that are locked on this machine during this transaction request call */
 654                                 oidlocked[objlocked] = ((objheader_t *)mobj)->oid;
 655                                 objlocked++;
 656                                 if (version == ((objheader_t *)mobj)->version) { /* Check if versions match */
 657                                         v_matchnolock++;
 658                                 } else { /* If versions don't match ...HARD ABORT */
 659                                         v_nomatch++;
 660                                         /* Send TRANS_DISAGREE to Coordinator */
 661                                         localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
 662                                         printf("DEBUG -> Sending TRANS_DISAGREE\n");
 663                                 //      return tdata->recvmsg[tdata->thread_id].rcv_status;
 664                                 }
 665                         }
 666                 }
 667         }
 668
 669         /*Decide the response to be sent to the Coordinator( the local machine in this case)*/
 670
 671         /* Condition to send TRANS_AGREE */
 672         if(v_matchnolock == localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod) {
 673                 localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_AGREE;
 674                 printf("DEBUG -> Sending TRANS_AGREE\n");
 675         }
 676         /* Condition to send TRANS_SOFT_ABORT */
 677         if((v_matchlock > 0 && v_nomatch == 0) || (objnotfound > 0 && v_nomatch == 0)) {
 678                 localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_SOFT_ABORT;
 679                 printf("DEBUG -> Sending TRANS_SOFT_ABORT\n");
 680                 /* Send number of oids not found and the missing oids if objects are missing in the machine */
 681                 /* TODO Remember to store the oidnotfound for later use
 682                 if(objnotfound != 0) {
 683                         int size = sizeof(unsigned int)* objnotfound;
 684                 }
 685                 */
 686         }
 687
 688         /* Fill out the trans_commit_data_t data structure. This is required for a trans commit process
 689          * if Participant receives a TRANS_COMMIT */
 690         localtdata->transinfo->objmod = oidmod;
 691         localtdata->transinfo->objlocked = oidlocked;
 692         localtdata->transinfo->objnotfound = oidnotfound;
 693         localtdata->transinfo->modptr = modptr;
 694         localtdata->transinfo->nummod = localtdata->tdata->buffer->f.nummod;
 695         localtdata->transinfo->numlocked = objlocked;
 696         localtdata->transinfo->numnotfound = objnotfound;
 697
 698         /*Set flag to show that common data structure for this individual thread has been written to */
 699         //*(tdata->localstatus) |= LM_UPDATED;
 700
 701         /* Lock and update count */
 702         //Thread sleeps until all messages from pariticipants are received by coordinator
 703         pthread_mutex_lock(localtdata->tdata->lock);
 704         (*(localtdata->tdata->count))++; /* keeps track of no of messages received by the coordinator */
 705
 706         /* Wake up the threads and invoke decideResponse (once) */
 707         if(*(localtdata->tdata->count) == localtdata->tdata->pilecount) {
 708                 if (decideResponse(localtdata->tdata) != 0) {
 709                         printf("decideResponse returned error %s,%d\n", __FILE__, __LINE__);
 710                         pthread_mutex_unlock(localtdata->tdata->lock);
 711                         return NULL;
 712                 }
 713                 pthread_cond_broadcast(localtdata->tdata->threshold);
 714         } else {
 715                 pthread_cond_wait(localtdata->tdata->threshold, localtdata->tdata->lock);
 716         }
 717         pthread_mutex_unlock(localtdata->tdata->lock);
 718
 719         /*Based on DecideResponse(), Either COMMIT or ABORT the operation*/
 720         if(*(localtdata->tdata->replyctrl) == TRANS_ABORT){
 721                 if(transAbortProcess(modptr,oidlocked, localtdata->transinfo->numlocked, localtdata->transinfo->nummod, localtdata->tdata->buffer->f.numread) != 0) {
 722                         printf("Error in transAbortProcess() %s,%d\n", __FILE__, __LINE__);
 723                         return NULL;
 724                 }
 725         }else if(*(localtdata->tdata->replyctrl) == TRANS_COMMIT){
 726                 if(transComProcess(localtdata->transinfo) != 0) {
 727                         printf("Error in transComProcess() %s,%d\n", __FILE__, __LINE__);
 728                         return NULL;
 729                 }
 730         }
 731
 732         /* Free memory */
 733         printf("DEBUG -> Freeing...\n");
 734         fflush(stdout);
 735         if (localtdata->transinfo->objmod != NULL) {
 736                 free(localtdata->transinfo->objmod);
 737                 localtdata->transinfo->objmod = NULL;
 738         }
 739         if (localtdata->transinfo->objlocked != NULL) {
 740                 free(localtdata->transinfo->objlocked);
 741                 localtdata->transinfo->objlocked = NULL;
 742         }
 743         if (localtdata->transinfo->objnotfound != NULL) {
 744                 free(localtdata->transinfo->objnotfound);
 745                 localtdata->transinfo->objnotfound = NULL;
 746         }
 747
 748         pthread_exit(NULL);
 749 }
 750 /* This function completes the ABORT process if the transaction is aborting
 751  */
 752 int transAbortProcess(void *modptr, unsigned int *objlocked, int numlocked, int nummod, int numread) {
 753         char *ptr;
 754         int i;
 755         objheader_t *tmp_header;
 756         void *header;
 757
 758         printf("DEBUG -> Recv TRANS_ABORT\n");
 759         /* Set all ref counts as 1 and do garbage collection */
 760         ptr = modptr;
 761         for(i = 0; i< nummod; i++) {
 762                 tmp_header = (objheader_t *)ptr;
 763                 tmp_header->rcount = 1;
 764                 ptr += sizeof(objheader_t) + classsize[tmp_header->type];
 765         }
 766         /* Unlock objects that was locked due to this transaction */
 767         for(i = 0; i< numlocked; i++) {
 768                 header = mhashSearch(objlocked[i]);// find the header address
 769                 ((objheader_t *)header)->status &= ~(LOCK);
 770         }
 771         //TODO/* Unset the bit for local objects */
 772
 773         /* Send ack to Coordinator */
 774         printf("DEBUG-> TRANS_SUCCESSFUL\n");
 775
 776         /*Free the pointer */
 777         ptr = NULL;
 778         return 0;
 779 }
 780
 781 /*This function completes the COMMIT process is the transaction is commiting
 782  */
 783  int transComProcess(trans_commit_data_t *transinfo) {
 784          objheader_t *header;
 785          int i = 0, offset = 0;
 786          char control;
 787
 788          printf("DEBUG -> Recv TRANS_COMMIT\n");
 789          /* Process each modified object saved in the mainobject store */
 790          for(i=0; i<transinfo->nummod; i++) {
 791                  if((header = (objheader_t *) mhashSearch(transinfo->objmod[i])) == NULL) {
 792                          printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
 793                  }
 794                  /* Change reference count of older address and free space in objstr ?? */
 795                  header->rcount = 1; //TODO Not sure what would be the val
 796
 797                  /* Change ptr address in mhash table */
 798                  mhashRemove(transinfo->objmod[i]);
 799                  mhashInsert(transinfo->objmod[i], (transinfo->modptr + offset));
 800                  offset += sizeof(objheader_t) + classsize[header->type];
 801
 802                  /* Update object version number */
 803                  header = (objheader_t *) mhashSearch(transinfo->objmod[i]);
 804                  header->version += 1;
 805          }
 806
 807          /* Unlock locked objects */
 808          for(i=0; i<transinfo->numlocked; i++) {
 809                  header = (objheader_t *) mhashSearch(transinfo->objlocked[i]);
 810                  header->status &= ~(LOCK);
 811          }
 812
 813          //TODO Update location lookup table
 814          //TODO/* Unset the bit for local objects */
 815
 816          /* Send ack to Coordinator */
 817          printf("DEBUG-> TRANS_SUCESSFUL\n");
 818          return 0;
 819  }
 820
 821 /*This function makes piles to prefetch records and prefetches the oids from remote machines */
 822 int transPrefetch(int *arrayofoffset[], short numoids){
 823         int i, k;
 824         int arraylength[numoids];
 825         unsigned int machinenumber;
 826         objheader_t *tmp, *objheader;
 827         void *objcopy;
 828         int size;
 829         void *buf;
 830
 831         /* Given tuple find length of tuple*/
 832         for(i = 0; i < numoids ; i++) {
 833                 arraylength[i] = arrayLength(arrayofoffset[i]);
 834         }
 835         /* Check for similar tuples or  other special case tuples that can be combined to a
 836          * prefetch message*/
 837         checkPrefetchTuples(arrayofoffset, arraylength, numoids);
 838
 839
 840         /* Check if part of prefetch request available locally */
 841         for(i = 0; i < numoids ; i++) {
 842                 while(arrayofoffsets[i][k] != -1) {
 843                         if((objheader = (objheader_t *) mhashSearch(arrayofoffsets[i][k])) != NULL) {
 844                                 /* Look up in machine lookup table  and copy  into cache*/
 845                                 tmp = mhashSearch(oid);
 846                                 size = sizeof(objheader_t)+classsize[tmp->type];
 847                                 objcopy = objstrAlloc(record->cache, size);
 848                                 memcpy(objcopy, (void *)tmp, size);
 849                                 /* Insert into cache's lookup table */
 850                                 chashInsert(record->lookupTable, objheader->oid, objcopy);
 851                                 /* Find the offset field*/
 852                                 oid = (int)(tmp + sizeof(objheader_t) + arrayofoffsets[i][k+1]);
 853                         } else  { /* If not found in machine look up */
 854                                 /* Get the object from the remote location */
 855                                 machinenumber = lhashSearch(oid);
 856                                 /* Create  Machine Piles t send prefetch requests use threads*/
 857                                 /* For each Pile in the machine send TRANS_PREFETCH */
 858                         }
 859                         k = k+1;
 860                 }
 861         }
 862
 863 }
 864
 865 int checkPrefetchTuples(int *arrayofoffset[], int *arraylength, short numoids) {
 866         int i, j, k, matchfound = 0, count = 0, slength, length;
 867         int *sarray, *larray;
 868
 869         /* Check if the prefetch oids are same and have same offsets
 870          * for case x.a.b and y.a.b where x and y have same oid's
 871          * or if a.b.c is a subset of x.b.c.d*/
 872         /* check for case where the generated request a.y.z or x.y.z.g then
 873          * prefetch needs to be generated for x.y.z.g */
 874         for(i = 0; i < numoids -1 ; i++) {
 875                 if(arrayofoffset[i][0] == -1)
 876                         continue;
 877                 for( j = 0; j < (numoids - (i+1)); j++) {
 878                         if(arrayofoffset[j][0] == -1)
 879                                 continue;
 880                         /* If the oids of the tuples match */
 881                         if((arrayofoffset[i][0] == arrayofoffset[j][0]) && (i != j )) {
 882                                 /*Find the  smallest length of two arrays, find ptrs to smallest and long array */
 883                                 if(arraylength[i] > arraylength[j]) {
 884                                         slength = arraylength[j];
 885                                         sarray = arrayofoffset[j];
 886                                         larray = arrayofoffset[i];
 887                                 } else {
 888                                         slength = arraylength[i];
 889                                         sarray = arrayofoffset[i];
 890                                         larray = arrayofoffset[j];
 891                                 }
 892                                 /* From first offset until end of tuple compare all offsets of sarray and larray
 893                                  * if not a match then break  */
 894                                 for(k = 1 ; slength -1 ; k++) {
 895                                         if(sarray[k] != larray[k]) {
 896                                                 break;
 897                                         }
 898                                 }
 899                                 /* If number of same offsets encountered is same as
 900                                  * no of offsets in small array then common tuples found*/
 901                                  if(k == (slength -1))
 902                                          sarray[0] = -1;
 903                         }
 904                 }
 905         }
 906
 907         return 0;
 908 }