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 *)objheader, 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].replyctrl = &treplyctrl;
 239                 thread_data_array[threadnum].replyretry = &treplyretry;
 240                 thread_data_array[threadnum].rec = record;
 241                 /* If local do not create any extra connection */
 242                 if(pile->local != 1) {
 243                         rc = pthread_create(&thread[threadnum], NULL, transRequest, (void *) &thread_data_array[threadnum]);
 244                         if (rc) {
 245                                 perror("Error in pthread create\n");
 246                                 return 1;
 247                         }
 248                 } else {
 249                         /*Unset the pile->local flag*/
 250                         pile->local = 0;
 251                         /*Set flag to identify that Local machine is involved*/
 252                         ltdata->tdata = &thread_data_array[threadnum];
 253                         ltdata->transinfo = &transinfo;
 254                         val = pthread_create(&thread[threadnum], NULL, handleLocalReq, (void *) ltdata);
 255                         if (val) {
 256                                 perror("Error in pthread create\n");
 257                                 return 1;
 258                         }
 259                 }
 260                 threadnum++;
 261                 pile = pile->next;
 262         }
 263
 264         /* Free attribute and wait for the other threads */
 265         pthread_attr_destroy(&attr);
 266         for (i = 0 ;i < pilecount ; i++) {
 267                 rc = pthread_join(thread[i], NULL);
 268                 if (rc)
 269                 {
 270                         printf("ERROR return code from pthread_join() is %d\n", rc);
 271                         return 1;
 272                 }
 273         }
 274
 275         /* Free resources */
 276         pthread_cond_destroy(&tcond);
 277         pthread_mutex_destroy(&tlock);
 278         free(tosend);
 279         free(listmid);
 280         pDelete(pile);
 281         free(thread_data_array);
 282         free(ltdata);
 283
 284         /* Retry trans commit procedure if not sucessful in the first try */
 285         if(treplyretry == 1) {
 286                 /* wait a random amount of time */
 287                 randomdelay();
 288                 //sleep(1);
 289                 /* Retry the commiting transaction again */
 290                 transCommit(record);
 291         }
 292
 293         return 0;
 294 }
 295
 296 /* This function sends information involved in the transaction request and
 297  * accepts a response from particpants.
 298  * It calls decideresponse() to decide on what control message
 299  * to send next and sends the message using sendResponse()*/
 300 void *transRequest(void *threadarg) {
 301         int sd, i, n;
 302         struct sockaddr_in serv_addr;
 303         struct hostent *server;
 304         thread_data_array_t *tdata;
 305         objheader_t *headeraddr;
 306         char buffer[RECEIVE_BUFFER_SIZE], control, recvcontrol;
 307         char machineip[16], retval;
 308
 309         tdata = (thread_data_array_t *) threadarg;
 310
 311         /* Send Trans Request */
 312         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 313                 perror("Error in socket for TRANS_REQUEST\n");
 314                 return NULL;
 315         }
 316         bzero((char*) &serv_addr, sizeof(serv_addr));
 317         serv_addr.sin_family = AF_INET;
 318         serv_addr.sin_port = htons(LISTEN_PORT);
 319         midtoIP(tdata->mid,machineip);
 320         machineip[15] = '\0';
 321         serv_addr.sin_addr.s_addr = inet_addr(machineip);
 322         /* Open Connection */
 323         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
 324                 perror("Error in connect for TRANS_REQUEST\n");
 325                 return NULL;
 326         }
 327
 328         printf("DEBUG-> trans.c Sending TRANS_REQUEST to mid %s\n", machineip);
 329         /* Send bytes of data with TRANS_REQUEST control message */
 330         if (send(sd, &(tdata->buffer->f), sizeof(fixed_data_t),MSG_NOSIGNAL) < sizeof(fixed_data_t)) {
 331                 perror("Error sending fixed bytes for thread\n");
 332                 return NULL;
 333         }
 334         /* Send list of machines involved in the transaction */
 335         {
 336           int size=sizeof(unsigned int)*tdata->pilecount;
 337           if (send(sd, tdata->buffer->listmid, size, MSG_NOSIGNAL) < size) {
 338             perror("Error sending list of machines for thread\n");
 339             return NULL;
 340           }
 341         }
 342         /* Send oids and version number tuples for objects that are read */
 343         {
 344           int size=(sizeof(unsigned int)+sizeof(short))*tdata->buffer->f.numread;
 345           if (send(sd, tdata->buffer->objread, size, MSG_NOSIGNAL) < size) {
 346             perror("Error sending tuples for thread\n");
 347             return NULL;
 348           }
 349         }
 350         /* Send objects that are modified */
 351         for(i = 0; i < tdata->buffer->f.nummod ; i++) {
 352           int size;
 353           headeraddr = chashSearch(tdata->rec->lookupTable, tdata->buffer->oidmod[i]);
 354           size=sizeof(objheader_t)+classsize[headeraddr->type];
 355           if (send(sd, headeraddr, size, MSG_NOSIGNAL)  < size) {
 356             perror("Error sending obj modified for thread\n");
 357             return NULL;
 358           }
 359         }
 360
 361         /* Read control message from Participant */
 362         if((n = read(sd, &control, sizeof(char))) <= 0) {
 363                 perror("Error in reading control message from Participant\n");
 364                 return NULL;
 365         }
 366         recvcontrol = control;
 367
 368         /* Update common data structure and increment count */
 369         tdata->recvmsg[tdata->thread_id].rcv_status = recvcontrol;
 370
 371         /* Lock and update count */
 372         //Thread sleeps until all messages from pariticipants are received by coordinator
 373         pthread_mutex_lock(tdata->lock);
 374
 375         (*(tdata->count))++; /* keeps track of no of messages received by the coordinator */
 376
 377         /* Wake up the threads and invoke decideResponse (once) */
 378         if(*(tdata->count) == tdata->pilecount) {
 379                 if (decideResponse(tdata) != 0) {
 380                         printf("decideResponse returned error %s,%d\n", __FILE__, __LINE__);
 381                         pthread_mutex_unlock(tdata->lock);
 382                         close(sd);
 383                         return NULL;
 384                 }
 385                 pthread_cond_broadcast(tdata->threshold);
 386         } else {
 387                 pthread_cond_wait(tdata->threshold, tdata->lock);
 388         }
 389         pthread_mutex_unlock(tdata->lock);
 390
 391         /* Send the final response such as TRANS_COMMIT or TRANS_ABORT t
 392          * to all participants in their respective socket */
 393         if (sendResponse(tdata, sd) == 0) {
 394                 printf("sendResponse returned error %s,%d\n", __FILE__, __LINE__);
 395                 pthread_mutex_unlock(tdata->lock);
 396                 close(sd);
 397                 return NULL;
 398         }
 399
 400         /* Close connection */
 401         close(sd);
 402         pthread_exit(NULL);
 403 }
 404
 405 /* This function decides the reponse that needs to be sent to
 406  * all Participant machines involved in the transaction commit */
 407 int decideResponse(thread_data_array_t *tdata) {
 408         char control;
 409         int i, transagree = 0, transdisagree = 0, transsoftabort = 0; /* Counters to formulate decision of what
 410                                                                          message to send */
 411
 412         //Check common data structure
 413         for (i = 0 ; i < tdata->pilecount ; i++) {
 414                 /*Switch on response from Participant */
 415                 control = tdata->recvmsg[i].rcv_status; /* tdata: keeps track of all participant responses
 416                                                            written onto the shared array */
 417                 switch(control) {
 418                         case TRANS_DISAGREE:
 419                                 printf("DEBUG-> trans.c Recv TRANS_DISAGREE\n");
 420                                 transdisagree++;
 421                                 break;
 422
 423                         case TRANS_AGREE:
 424                                 printf("DEBUG-> trans.c Recv TRANS_AGREE\n");
 425                                 transagree++;
 426                                 break;
 427
 428                         case TRANS_SOFT_ABORT:
 429                                 printf("DEBUG-> trans.c Recv TRANS_SOFT_ABORT\n");
 430                                 transsoftabort++;
 431                                 break;
 432                         default:
 433                                 printf("Participant sent unknown message in %s, %d\n", __FILE__, __LINE__);
 434                                 return -1;
 435                 }
 436         }
 437
 438         /* Decide what control message to send to Participant */
 439         if(transdisagree > 0) {
 440                 /* Send Abort */
 441                 *(tdata->replyctrl) = TRANS_ABORT;
 442                 printf("DEBUG-> trans.c Sending TRANS_ABORT\n");
 443                 objstrDelete(tdata->rec->cache);
 444                 chashDelete(tdata->rec->lookupTable);
 445                 free(tdata->rec);
 446         } else if(transagree == tdata->pilecount){
 447                 /* Send Commit */
 448                 *(tdata->replyctrl) = TRANS_COMMIT;
 449                 printf("DEBUG-> trans.c Sending TRANS_COMMIT\n");
 450                 objstrDelete(tdata->rec->cache);
 451                 chashDelete(tdata->rec->lookupTable);
 452                 free(tdata->rec);
 453         } else if(transsoftabort > 0 && transdisagree == 0) {
 454                 /* Send Abort in soft abort case followed by retry commiting transaction again*/
 455                 *(tdata->replyctrl) = TRANS_ABORT;
 456                 *(tdata->replyretry) = 1;
 457                 printf("DEBUG-> trans.c Sending TRANS_ABORT\n");
 458         } else {
 459                 printf("DEBUG -> %s, %d: Error: undecided response\n", __FILE__, __LINE__);
 460                 return -1;
 461         }
 462
 463         return 0;
 464 }
 465 /* This function sends the final response to all threads in their respective socket id */
 466 char sendResponse(thread_data_array_t *tdata, int sd) {
 467         int n, N, sum, oidcount = 0;
 468         char *ptr, retval = 0;
 469         unsigned int *oidnotfound;
 470
 471         /* If the decided response is due to a soft abort and missing objects at the Participant's side */
 472         if(tdata->recvmsg[tdata->thread_id].rcv_status == TRANS_SOFT_ABORT) {
 473                 /* Read list of objects missing */
 474                 if((read(sd, &oidcount, sizeof(int)) != 0) && (oidcount != 0)) {
 475                         N = oidcount * sizeof(unsigned int);
 476                         if((oidnotfound = calloc(oidcount, sizeof(unsigned int))) == NULL) {
 477                                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
 478                         }
 479                         ptr = (char *) oidnotfound;
 480                         do {
 481                                 n = read(sd, ptr+sum, N-sum);
 482                                 sum += n;
 483                         } while(sum < N && n !=0);
 484                 }
 485                 retval =  TRANS_SOFT_ABORT;
 486         }
 487         /* If the decided response is TRANS_ABORT */
 488         if(*(tdata->replyctrl) == TRANS_ABORT) {
 489                 retval = TRANS_ABORT;
 490         } else if(*(tdata->replyctrl) == TRANS_COMMIT) { /* If the decided response is TRANS_COMMIT */
 491                 retval = TRANS_COMMIT;
 492         }
 493         /* Send response to the Participant */
 494         if (send(sd, tdata->replyctrl, sizeof(char),MSG_NOSIGNAL) < sizeof(char)) {
 495                 perror("Error sending ctrl message for participant\n");
 496         }
 497
 498         return retval;
 499 }
 500
 501 /* This function opens a connection, places an object read request to the
 502  * remote machine, reads the control message and object if available  and
 503  * copies the object and its header to the local cache.
 504  * TODO replace mnum and midtoIP() with MACHINE_IP address later */
 505
 506 void *getRemoteObj(transrecord_t *record, unsigned int mnum, unsigned int oid) {
 507         int sd, size, val;
 508         struct sockaddr_in serv_addr;
 509         struct hostent *server;
 510         char control;
 511         char machineip[16];
 512         objheader_t *h;
 513         void *objcopy;
 514
 515         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
 516                 perror("Error in socket\n");
 517                 return NULL;
 518         }
 519         bzero((char*) &serv_addr, sizeof(serv_addr));
 520         serv_addr.sin_family = AF_INET;
 521         serv_addr.sin_port = htons(LISTEN_PORT);
 522         //serv_addr.sin_addr.s_addr = inet_addr(MACHINE_IP);
 523         midtoIP(mnum,machineip);
 524         machineip[15] = '\0';
 525         serv_addr.sin_addr.s_addr = inet_addr(machineip);
 526         /* Open connection */
 527         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
 528                 perror("Error in connect\n");
 529                 return NULL;
 530         }
 531         char readrequest[sizeof(char)+sizeof(unsigned int)];
 532         readrequest[0] = READ_REQUEST;
 533         *((unsigned int *)(&readrequest[1])) = oid;
 534         if (send(sd, &readrequest, sizeof(readrequest), MSG_NOSIGNAL) < sizeof(readrequest)) {
 535                 perror("Error sending message\n");
 536                 return NULL;
 537         }
 538
 539 #ifdef DEBUG1
 540         printf("DEBUG -> ready to rcv ...\n");
 541 #endif
 542         /* Read response from the Participant */
 543         if((val = read(sd, &control, sizeof(char))) <= 0) {
 544                 perror("No control response for getRemoteObj sent\n");
 545                 return NULL;
 546         }
 547         switch(control) {
 548                 case OBJECT_NOT_FOUND:
 549                         printf("DEBUG -> Control OBJECT_NOT_FOUND received\n");
 550                         return NULL;
 551                 case OBJECT_FOUND:
 552                         /* Read object if found into local cache */
 553                         if((val = read(sd, &size, sizeof(int))) <= 0) {
 554                                 perror("No size is read from the participant\n");
 555                                 return NULL;
 556                         }
 557                         objcopy = objstrAlloc(record->cache, size);
 558                         if((val = read(sd, objcopy, size)) <= 0) {
 559                                 perror("No objects are read from the remote participant\n");
 560                                 return NULL;
 561                         }
 562                         /* Insert into cache's lookup table */
 563                         chashInsert(record->lookupTable, oid, objcopy);
 564                         break;
 565                 default:
 566                         printf("Error in recv request from participant on a READ_REQUEST %s, %d\n",__FILE__, __LINE__);
 567                         return NULL;
 568         }
 569         /* Close connection */
 570         close(sd);
 571         return objcopy;
 572 }
 573
 574 /*This function handles the local trans requests involved in a transaction commiting process
 575  * makes a decision if the local machine sends AGREE or DISAGREE or SOFT_ABORT
 576  * Activates the other nonlocal threads that are waiting for the decision and the
 577  * based on common decision by all groups involved in the transaction it
 578  * either commits or aborts the transaction.
 579  * It also frees the calloced memory resources
 580  */
 581
 582 void *handleLocalReq(void *threadarg) {
 583         int val, i = 0;
 584         short version;
 585         char control = 0, *ptr;
 586         unsigned int oid;
 587         unsigned int *oidnotfound = NULL, *oidlocked = NULL, *oidmod = NULL;
 588         void *mobj, *modptr;
 589         objheader_t *headptr;
 590         local_thread_data_array_t *localtdata;
 591
 592         localtdata = (local_thread_data_array_t *) threadarg;
 593
 594         /* Counters and arrays to formulate decision on control message to be sent */
 595         oidnotfound = (unsigned int *) calloc((localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod), sizeof(unsigned int));
 596         oidlocked = (unsigned int *) calloc((localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod), sizeof(unsigned int));
 597         oidmod = (unsigned int *) calloc(localtdata->tdata->buffer->f.nummod, sizeof(unsigned int));
 598         int objnotfound = 0, objlocked = 0, objmod =0, v_nomatch = 0, v_matchlock = 0, v_matchnolock = 0;
 599         int objmodnotfound = 0, nummodfound = 0;
 600
 601         /* modptr points to the beginning of the object store
 602          * created at the Pariticipant */
 603         if ((modptr = objstrAlloc(mainobjstore, localtdata->tdata->buffer->f.sum_bytes)) == NULL) {
 604                 printf("objstrAlloc error for modified objects %s, %d\n", __FILE__, __LINE__);
 605                 return NULL;
 606         }
 607
 608         ptr = modptr;
 609
 610         /* Process each oid in the machine pile/ group per thread */
 611         for (i = 0; i < localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod; i++) {
 612                 if (i < localtdata->tdata->buffer->f.numread) {//Objs only read and not modified
 613                         int incr = sizeof(unsigned int) + sizeof(short);// Offset that points to next position in the objread array
 614                         incr *= i;
 615                         oid = *((unsigned int *)(localtdata->tdata->buffer->objread + incr));
 616                         incr += sizeof(unsigned int);
 617                         version = *((short *)(localtdata->tdata->buffer->objread + incr));
 618                 } else {//Objs modified
 619                         headptr = (objheader_t *) ptr;
 620                         oid = headptr->oid;
 621                         oidmod[objmod] = oid;//Array containing modified oids
 622                         objmod++;
 623                         version = headptr->version;
 624                         ptr += sizeof(objheader_t) + classsize[headptr->type];
 625                 }
 626
 627                 /* Check if object is still present in the machine since the beginning of TRANS_REQUEST */
 628
 629                 /* Save the oids not found and number of oids not found for later use */
 630                 if ((mobj = mhashSearch(oid)) == NULL) {/* Obj not found */
 631                         /* Save the oids not found and number of oids not found for later use */
 632
 633                         oidnotfound[objnotfound] = ((objheader_t *)mobj)->oid;
 634                         objnotfound++;
 635                 } else { /* If Obj found in machine (i.e. has not moved) */
 636                         /* Check if Obj is locked by any previous transaction */
 637                         if ((((objheader_t *)mobj)->status & LOCK) == LOCK) {
 638                                 if (version == ((objheader_t *)mobj)->version) {      /* If not locked then match versions */
 639                                         v_matchlock++;
 640                                 } else {/* If versions don't match ...HARD ABORT */
 641                                         v_nomatch++;
 642                                         /* Send TRANS_DISAGREE to Coordinator */
 643                                         localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
 644                                         printf("DEBUG -> Sending TRANS_DISAGREE\n");
 645                                         //return tdata->recvmsg[tdata->thread_id].rcv_status;
 646                                 }
 647                         } else {/* If Obj is not locked then lock object */
 648                                 ((objheader_t *)mobj)->status |= LOCK;
 649                                 //TODO Remove this for Testing
 650                                 randomdelay();
 651
 652                                 /* Save all object oids that are locked on this machine during this transaction request call */
 653                                 oidlocked[objlocked] = ((objheader_t *)mobj)->oid;
 654                                 objlocked++;
 655                                 if (version == ((objheader_t *)mobj)->version) { /* Check if versions match */
 656                                         v_matchnolock++;
 657                                 } else { /* If versions don't match ...HARD ABORT */
 658                                         v_nomatch++;
 659                                         /* Send TRANS_DISAGREE to Coordinator */
 660                                         localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
 661                                         printf("DEBUG -> Sending TRANS_DISAGREE\n");
 662                                 //      return tdata->recvmsg[tdata->thread_id].rcv_status;
 663                                 }
 664                         }
 665                 }
 666         }
 667
 668         /*Decide the response to be sent to the Coordinator( the local machine in this case)*/
 669
 670         /* Condition to send TRANS_AGREE */
 671         if(v_matchnolock == localtdata->tdata->buffer->f.numread + localtdata->tdata->buffer->f.nummod) {
 672                 localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_AGREE;
 673                 printf("DEBUG -> Sending TRANS_AGREE\n");
 674         }
 675         /* Condition to send TRANS_SOFT_ABORT */
 676         if((v_matchlock > 0 && v_nomatch == 0) || (objnotfound > 0 && v_nomatch == 0)) {
 677                 localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_SOFT_ABORT;
 678                 printf("DEBUG -> Sending TRANS_SOFT_ABORT\n");
 679                 /* Send number of oids not found and the missing oids if objects are missing in the machine */
 680                 /* TODO Remember to store the oidnotfound for later use
 681                 if(objnotfound != 0) {
 682                         int size = sizeof(unsigned int)* objnotfound;
 683                 }
 684                 */
 685         }
 686
 687         /* Fill out the trans_commit_data_t data structure. This is required for a trans commit process
 688          * if Participant receives a TRANS_COMMIT */
 689         localtdata->transinfo->objmod = oidmod;
 690         localtdata->transinfo->objlocked = oidlocked;
 691         localtdata->transinfo->objnotfound = oidnotfound;
 692         localtdata->transinfo->modptr = modptr;
 693         localtdata->transinfo->nummod = localtdata->tdata->buffer->f.nummod;
 694         localtdata->transinfo->numlocked = objlocked;
 695         localtdata->transinfo->numnotfound = objnotfound;
 696
 697         /*Set flag to show that common data structure for this individual thread has been written to */
 698         //*(tdata->localstatus) |= LM_UPDATED;
 699
 700         /* Lock and update count */
 701         //Thread sleeps until all messages from pariticipants are received by coordinator
 702         pthread_mutex_lock(localtdata->tdata->lock);
 703         (*(localtdata->tdata->count))++; /* keeps track of no of messages received by the coordinator */
 704
 705         /* Wake up the threads and invoke decideResponse (once) */
 706         if(*(localtdata->tdata->count) == localtdata->tdata->pilecount) {
 707                 if (decideResponse(localtdata->tdata) != 0) {
 708                         printf("decideResponse returned error %s,%d\n", __FILE__, __LINE__);
 709                         pthread_mutex_unlock(localtdata->tdata->lock);
 710                         return NULL;
 711                 }
 712                 pthread_cond_broadcast(localtdata->tdata->threshold);
 713         } else {
 714                 pthread_cond_wait(localtdata->tdata->threshold, localtdata->tdata->lock);
 715         }
 716         pthread_mutex_unlock(localtdata->tdata->lock);
 717
 718         /*Based on DecideResponse(), Either COMMIT or ABORT the operation*/
 719         if(*(localtdata->tdata->replyctrl) == TRANS_ABORT){
 720                 if(transAbortProcess(modptr,oidlocked, localtdata->transinfo->numlocked, localtdata->transinfo->nummod, localtdata->tdata->buffer->f.numread) != 0) {
 721                         printf("Error in transAbortProcess() %s,%d\n", __FILE__, __LINE__);
 722                         return NULL;
 723                 }
 724         }else if(*(localtdata->tdata->replyctrl) == TRANS_COMMIT){
 725                 if(transComProcess(localtdata->transinfo) != 0) {
 726                         printf("Error in transComProcess() %s,%d\n", __FILE__, __LINE__);
 727                         return NULL;
 728                 }
 729         }
 730
 731         /* Free memory */
 732         printf("DEBUG -> Freeing...\n");
 733         fflush(stdout);
 734         if (localtdata->transinfo->objmod != NULL) {
 735                 free(localtdata->transinfo->objmod);
 736                 localtdata->transinfo->objmod = NULL;
 737         }
 738         if (localtdata->transinfo->objlocked != NULL) {
 739                 free(localtdata->transinfo->objlocked);
 740                 localtdata->transinfo->objlocked = NULL;
 741         }
 742         if (localtdata->transinfo->objnotfound != NULL) {
 743                 free(localtdata->transinfo->objnotfound);
 744                 localtdata->transinfo->objnotfound = NULL;
 745         }
 746
 747         pthread_exit(NULL);
 748 }
 749 /* This function completes the ABORT process if the transaction is aborting
 750  */
 751 int transAbortProcess(void *modptr, unsigned int *objlocked, int numlocked, int nummod, int numread) {
 752         char *ptr;
 753         int i;
 754         objheader_t *tmp_header;
 755         void *header;
 756
 757         printf("DEBUG -> Recv TRANS_ABORT\n");
 758         /* Set all ref counts as 1 and do garbage collection */
 759         ptr = modptr;
 760         for(i = 0; i< nummod; i++) {
 761                 tmp_header = (objheader_t *)ptr;
 762                 tmp_header->rcount = 1;
 763                 ptr += sizeof(objheader_t) + classsize[tmp_header->type];
 764         }
 765         /* Unlock objects that was locked due to this transaction */
 766         for(i = 0; i< numlocked; i++) {
 767                 header = mhashSearch(objlocked[i]);// find the header address
 768                 ((objheader_t *)header)->status &= ~(LOCK);
 769         }
 770         //TODO/* Unset the bit for local objects */
 771
 772         /* Send ack to Coordinator */
 773         printf("DEBUG-> TRANS_SUCCESSFUL\n");
 774
 775         /*Free the pointer */
 776         ptr = NULL;
 777         return 0;
 778 }
 779
 780 /*This function completes the COMMIT process is the transaction is commiting
 781  */
 782  int transComProcess(trans_commit_data_t *transinfo) {
 783          objheader_t *header;
 784          int i = 0, offset = 0;
 785          char control;
 786
 787          printf("DEBUG -> Recv TRANS_COMMIT\n");
 788          /* Process each modified object saved in the mainobject store */
 789          for(i=0; i<transinfo->nummod; i++) {
 790                  if((header = (objheader_t *) mhashSearch(transinfo->objmod[i])) == NULL) {
 791                          printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__);
 792                  }
 793                  /* Change reference count of older address and free space in objstr ?? */
 794                  header->rcount = 1; //TODO Not sure what would be the val
 795
 796                  /* Change ptr address in mhash table */
 797                  mhashRemove(transinfo->objmod[i]);
 798                  mhashInsert(transinfo->objmod[i], (transinfo->modptr + offset));
 799                  offset += sizeof(objheader_t) + classsize[header->type];
 800
 801                  /* Update object version number */
 802                  header = (objheader_t *) mhashSearch(transinfo->objmod[i]);
 803                  header->version += 1;
 804          }
 805
 806          /* Unlock locked objects */
 807          for(i=0; i<transinfo->numlocked; i++) {
 808                  header = (objheader_t *) mhashSearch(transinfo->objlocked[i]);
 809                  header->status &= ~(LOCK);
 810          }
 811
 812          //TODO Update location lookup table
 813          //TODO/* Unset the bit for local objects */
 814
 815          /* Send ack to Coordinator */
 816          printf("DEBUG-> TRANS_SUCESSFUL\n");
 817          return 0;
 818  }
 819
 820 /*This function makes piles to prefetch records and prefetches the oids from remote machines */
 821 int transPrefetch(transrecord_t *record, int *arrayofoffset[], short numoids){
 822         int i, k = 0, rc;
 823         int arraylength[numoids];
 824         unsigned int machinenumber;
 825         objheader_t *tmp, *objheader;
 826         void *objcopy;
 827         int size;
 828         pthread_attr_t attr;
 829
 830         /* Given tuple find length of tuple*/
 831         for(i = 0; i < numoids ; i++) {
 832                 arraylength[i] = arrayLength(arrayofoffset[i]);
 833         }
 834         /* Check for similar tuples or  other special case tuples that can be combined to a
 835          * prefetch message*/
 836         if(checkPrefetchTuples(arrayofoffset, arraylength, numoids) != 0) {
 837                 printf("Error on checkPrefetchTuples at %s, %d\n", __FILE__, __LINE__);
 838                 return 1;
 839         }
 840
 841         /* Check if part of prefetch request available locally */
 842         for(i = 0; i < numoids ; i++) {
 843                 if(arrayofoffset[i][0] != -1) {
 844                         if((objheader = (objheader_t *) mhashSearch(arrayofoffset[i][k])) != NULL) {
 845                                 /* Look up in machine lookup table  and copy  into cache*/
 846                                 //tmp = mhashSearch(oid);
 847                                 size = sizeof(objheader_t)+classsize[tmp->type];
 848                                 objcopy = objstrAlloc(record->cache, size);
 849                                 memcpy(objcopy, (void *)objheader, size);
 850                                 /* Insert into cache's lookup table */
 851                                 chashInsert(record->lookupTable, objheader->oid, objcopy);
 852                                 /* Find the offset field*/
 853                                 if(foundLocal(arrayofoffset, record, arrayofoffset[i], tmp, arrayofoffset[i][k+1], i) != 0 ) {
 854                                         printf("Error in foundLocal() %s, %d\n", __FILE__, __LINE__);
 855                                         return 1;
 856                                 }
 857                         } else
 858                                 continue;
 859                 } else
 860                         continue;
 861         }
 862
 863         /* Initialize and set thread attributes
 864          * Spawn a thread for each prefetch request sent*/
 865         pthread_t thread[numoids];
 866         pthread_attr_init(&attr);
 867         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
 868
 869         /* Create  Machine Piles to send prefetch requests use threads*/
 870         for( i = 0 ; i< numoids ; i++) {
 871                 if(arrayofoffset[i][0] == -1)
 872                         continue;
 873                 else{
 874                         /* For each Pile in the machine send TRANS_PREFETCH */
 875                         //makePiles(arrayofoffset, numoids);
 876                         /* Fill thread data structure */
 877                         rc = pthread_create(&thread[i] , &attr, sendPrefetchReq, (void *) arrayofoffset[i]);
 878                         if (rc) {
 879                                 perror("Error in pthread create at transPrefetch()\n");
 880                                 return 1;
 881                         }
 882
 883                 }
 884         }
 885
 886         /* Free attribute and wait to join other threads */
 887         for (i = 0 ;i < numoids ; i++) {
 888                 rc = pthread_join(thread[i], NULL);
 889                 if (rc) {
 890                         perror("Error pthread_join() in transPrefetch()\n");
 891                         return 1;
 892                 }
 893         }
 894         pthread_attr_destroy(&attr);
 895
 896         return 0;
 897
 898 }
 899
 900 int checkPrefetchTuples(int *arrayofoffset[], int *arraylength, short numoids) {
 901         int i, j, k, matchfound = 0, count = 0, slength, length;
 902         int *sarray, *larray;
 903
 904         /* Check if the prefetch oids are same and have same offsets
 905          * for case x.a.b and y.a.b where x and y have same oid's
 906          * or if a.b.c is a subset of x.b.c.d*/
 907         /* check for case where the generated request a.y.z or x.y.z.g then
 908          * prefetch needs to be generated for x.y.z.g  if oid of a and x are same*/
 909         for(i = 0; i < numoids -1 ; i++) {
 910                 if(arrayofoffset[i][0] == -1)
 911                         continue;
 912                 for( j = 0; j < (numoids - (i+1)); j++) {
 913                         if(arrayofoffset[j][0] == -1)
 914                                 continue;
 915                         /* If the oids of the tuples match */
 916                         if((arrayofoffset[i][0] == arrayofoffset[j][0]) && (i != j )) {
 917                                 /*Find the  smallest length of two arrays, find ptrs to smallest and long array */
 918                                 if(arraylength[i] > arraylength[j]) {
 919                                         slength = arraylength[j];
 920                                         sarray = arrayofoffset[j];
 921                                         larray = arrayofoffset[i];
 922                                 } else {
 923                                         slength = arraylength[i];
 924                                         sarray = arrayofoffset[i];
 925                                         larray = arrayofoffset[j];
 926                                 }
 927                                 /* From first offset until end of tuple compare all offsets of sarray and larray
 928                                  * if not a match then break */
 929                                 for(k = 1 ; k < slength -1 ; k++) {
 930                                         if(sarray[k] != larray[k]) {
 931                                                 break;
 932                                         }
 933                                 }
 934                                 /* If number of same offsets encountered is same as
 935                                  * no of offsets in small array then common tuples found*/
 936                                  if(k == (slength -1))
 937                                          sarray[0] = -1;
 938                         }
 939                 }
 940         }
 941
 942         return 0;
 943 }
 944
 945 /* This function goes through an array and finds out until what offets
 946  *  * can it make a new pile
 947  *   * -1 indicates a special character to mark end of oid, offset tuple*/
 948 int foundLocal(int *arrayofoffset[], transrecord_t *record, int *array, objheader_t *header, int offset, int index) {
 949         int i, j, counter, len;
 950         unsigned int oid;
 951         void *objcopy;
 952         objheader_t *tmp, *objheader;
 953         int size;
 954
 955         /* The oid corresponding to the first offset */
 956         oid = (int) (header + sizeof(objheader_t) + offset);
 957         /* Repeat for all offset values in the array */
 958         for(i = 1; array[i] != -1 ; ) {
 959                 /* If oid  is found locally  then insert into cache and continue
 960                  * with next offset to find the corresponding oid */
 961                 if((objheader = (objheader_t*) mhashSearch(oid)) != NULL) {
 962                         tmp = mhashSearch(oid);
 963                         size = sizeof(objheader_t)+classsize[tmp->type];
 964                         objcopy = objstrAlloc(record->cache, size);
 965                         memcpy(objcopy, (void *)tmp, size);
 966                         /* Insert into cache's lookup table */
 967                         chashInsert(record->lookupTable, objheader->oid, objcopy);
 968                         oid = (int) (tmp + sizeof(objheader_t) + array[i+1]);
 969                 } else {
 970                         /*If oid not found locally then
 971                          *assign the latest oid found as the first element
 972                          *of array and copy left over offsets as other
 973                          *array elements to build the new prefetch array*/
 974
 975                         arrayofoffset[index][0] = (int) oid;
 976                         counter = i;
 977                         len = arrayLength((int *)array[index]);
 978                         for( j = 1 ; counter <= len; j++,counter++) {
 979                                 arrayofoffset[index][j] = array[counter + 1];
 980                         }
 981                         break;
 982                 }
 983                 i++;
 984         }
 985
 986         /* If all offsets are found locally, then do not treat this as valid prefetch tuple */
 987         if(i == arrayLength((int *)array[index]) - 1) {
 988                 arrayofoffset[index][0] = -1; //Mark beginning of array as -1
 989         }
 990
 991         return 0;
 992 }
 993
 994 void *sendPrefetchReq(void *prefetchtuple) {
 995         int sd, i;
 996         struct sockaddr_in serv_addr;
 997         struct hostent *server;
 998         char buffer[RECEIVE_BUFFER_SIZE], control, recvcontrol;
 999         char machineip[16], retval;
1000
1001
1002         /* Send Trans Prefetch Request */
1003         if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
1004                 perror("Error in socket for TRANS_REQUEST\n");
1005                 return NULL;
1006         }
1007         bzero((char*) &serv_addr, sizeof(serv_addr));
1008         serv_addr.sin_family = AF_INET;
1009         serv_addr.sin_port = htons(LISTEN_PORT);
1010         //midtoIP(tdata->mid,machineip);
1011 //      machineip[15] = '\0';
1012 //      serv_addr.sin_addr.s_addr = inet_addr(machineip);
1013
1014         /* Open Connection */
1015         if (connect(sd, (struct sockaddr *) &serv_addr, sizeof(struct sockaddr)) < 0) {
1016                 perror("Error in connect for TRANS_REQUEST\n");
1017                 return NULL;
1018         }
1019
1020         close(sd);
1021         pthread_exit(NULL);
1022
1023 }