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

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <string.h>
   4 #include <pthread.h>
   5 #include <netdb.h>
   6 #include <fcntl.h>
   7 #include "dstm.h"
   8 #include "mlookup.h"
   9 #include "llookup.h"
  10
  11 #define LISTEN_PORT 2156
  12 #define BACKLOG 10 //max pending connections
  13 #define RECEIVE_BUFFER_SIZE 2048
  14
  15 extern int classsize[];
  16
  17 objstr_t *mainobjstore;
  18
  19 int dstmInit(void)
  20 {
  21         //Initialize main object store
  22         mainobjstore = objstrCreate(DEFAULT_OBJ_STORE_SIZE);
  23         if (mhashCreate(HASH_SIZE, LOADFACTOR))
  24                 return 1; //failure
  25
  26         if (lhashCreate(HASH_SIZE, LOADFACTOR))
  27                 return 1; //failure
  28
  29         //pthread_t threadListen;
  30         //pthread_create(&threadListen, NULL, dstmListen, NULL);
  31
  32         return 0;
  33 }
  34
  35 void *dstmListen()
  36 {
  37         int listenfd, acceptfd;
  38         struct sockaddr_in my_addr;
  39         struct sockaddr_in client_addr;
  40         socklen_t addrlength = sizeof(struct sockaddr);
  41         pthread_t thread_dstm_accept;
  42         int i;
  43
  44         listenfd = socket(AF_INET, SOCK_STREAM, 0);
  45         if (listenfd == -1)
  46         {
  47                 perror("socket");
  48                 exit(1);
  49         }
  50
  51         my_addr.sin_family = AF_INET;
  52         my_addr.sin_port = htons(LISTEN_PORT);
  53         my_addr.sin_addr.s_addr = INADDR_ANY;
  54         memset(&(my_addr.sin_zero), '\0', 8);
  55
  56         if (bind(listenfd, (struct sockaddr *)&my_addr, addrlength) == -1)
  57         {
  58                 perror("bind");
  59                 exit(1);
  60         }
  61
  62         if (listen(listenfd, BACKLOG) == -1)
  63         {
  64                 perror("listen");
  65                 exit(1);
  66         }
  67
  68         printf("Listening on port %d, fd = %d\n", LISTEN_PORT, listenfd);
  69         while(1)
  70         {
  71                 acceptfd = accept(listenfd, (struct sockaddr *)&client_addr, &addrlength);
  72                 pthread_create(&thread_dstm_accept, NULL, dstmAccept, (void *)acceptfd);
  73         }
  74         pthread_exit(NULL);
  75 }
  76
  77 void *dstmAccept(void *acceptfd)
  78 {
  79         int numbytes,i, val, retval;
  80         unsigned int oid;
  81         char buffer[RECEIVE_BUFFER_SIZE], control,ctrl;
  82         char *ptr;
  83         void *srcObj;
  84         objheader_t *h;
  85         trans_commit_data_t transinfo;
  86
  87         int fd_flags = fcntl((int)acceptfd, F_GETFD), size;
  88
  89         printf("Recieved connection: fd = %d\n", (int)acceptfd);
  90         if((retval = recv((int)acceptfd, &control, sizeof(char), 0)) <= 0) {
  91                 perror("Error in receiving control from coordinator\n");
  92                 return;
  93         }
  94         switch(control) {
  95                 case READ_REQUEST:
  96                         printf("DEBUG -> Recv READ_REQUEST from Coordinator\n");
  97                         if((retval = recv((int)acceptfd, &oid, sizeof(unsigned int), 0)) <= 0) {
  98                                 perror("Error receiving object from cooridnator\n");
  99                                 return;
 100                         }
 101                         srcObj = mhashSearch(oid);
 102                         h = (objheader_t *) srcObj;
 103                         size = sizeof(objheader_t) + sizeof(classsize[h->type]);
 104                         if (h == NULL) {
 105                                 ctrl = OBJECT_NOT_FOUND;
 106                                 if(send((int)acceptfd, &ctrl, sizeof(char), 0) < 0) {
 107                                         perror("Error sending control msg to coordinator\n");
 108                                 }
 109                         } else {
 110                                 //char responsemessage[sizeof(char)+sizeof(int)];
 111                                 /* Type */
 112                                 ctrl = OBJECT_FOUND;
 113                                 if(send((int)acceptfd, &ctrl, sizeof(char), 0) < 0) {
 114                                         perror("Error sending control msg to coordinator\n");
 115                                 }
 116
 117                                 //responsemessage[0]=OBJECT_FOUND;
 118                                 /* Size of object */
 119                                 //*((int *)(&responsemessage[1])) = sizeof(objheader_t) + classsize[h->type];
 120                                 //if(send((int)acceptfd, &responsemessage, sizeof(responsemessage), 0) < 0) {
 121                                 //      perror("Error sending control msg to coordinator\n");
 122                                 //}
 123
 124                                 /* Size of object */
 125                                 if(send((int)acceptfd, &size, sizeof(int), 0) < 0) {
 126                                         perror("Error sending size of object to coordinator\n");
 127                                 }
 128                                 if(send((int)acceptfd, h, size, 0) < 0) {
 129                                         perror("Error in sending object\n");
 130                                 }
 131                         }
 132                         break;
 133
 134                 case READ_MULT_REQUEST:
 135                         printf("DEBUG-> READ_MULT_REQUEST\n");
 136                         break;
 137
 138                 case MOVE_REQUEST:
 139                         printf("DEBUG -> MOVE_REQUEST\n");
 140                         break;
 141
 142                 case MOVE_MULT_REQUEST:
 143                         printf("DEBUG -> MOVE_MULT_REQUEST\n");
 144                         break;
 145
 146                 case TRANS_REQUEST:
 147                         printf("DEBUG -> Recv TRANS_REQUEST from Coordinator\n");
 148                         if((val = readClientReq((int)acceptfd, &transinfo)) != 0) {
 149                                 printf("Error in readClientReq\n");
 150                         }
 151                         break;
 152
 153                 default:
 154                         printf("DEBUG -> dstmAccept: Error Unknown opcode %d\n", control);
 155         }
 156         if (close((int)acceptfd) == -1)
 157         {
 158                 perror("close");
 159         }
 160         else
 161                 printf("Closed connection: fd = %d\n", (int)acceptfd);
 162
 163
 164         pthread_exit(NULL);
 165         printf("DEBUG -> Exiting dstmAccept\n");
 166 }
 167
 168 int readClientReq(int acceptfd, trans_commit_data_t *transinfo) {
 169         char *ptr, control, prevctrl, sendctrl, newctrl;
 170         void *modptr, *header;
 171         objheader_t *tmp_header;
 172         fixed_data_t fixed;
 173         int sum = 0, i, N, n, val, retval;
 174
 175         //Reads to process the TRANS_REQUEST protocol further
 176         // Read fixed_data
 177         N = sizeof(fixed) - 1;
 178         ptr = (char *)&fixed;;
 179         fixed.control = TRANS_REQUEST;
 180         do {
 181                 n = recv((int)acceptfd, (void *) ptr+1+sum, N-sum, 0);
 182         //      printf("DEBUG -> 1. Reading %d bytes \n", n);
 183                 sum += n;
 184         } while(sum < N && n != 0);
 185
 186         //printf("Machine count = %d\tnumread = %d\tnummod = %d\tsum_bytes = %d\n", fixed.mcount, fixed.numread, fixed.nummod, fixed.sum_bytes);
 187         // Read list of mids
 188         int mcount = fixed.mcount;
 189         N = mcount * sizeof(unsigned int);
 190         unsigned int listmid[mcount];
 191         ptr = (char *) listmid;
 192         sum = 0;
 193         do {
 194                 n = recv((int)acceptfd, (void *) ptr+sum, N-sum, 0);
 195         //      printf("DEBUG -> 2. Reading %d bytes cap = %d\n", n, N);
 196                 sum += n;
 197         } while(sum < N && n != 0);
 198
 199         // Read oid and version tuples
 200         int numread = fixed.numread;
 201         N = numread * (sizeof(unsigned int) + sizeof(short));
 202         char objread[N];
 203         if(numread != 0) { // If pile contains objects to be read
 204         //      N = numread * (sizeof(unsigned int) + sizeof(short));
 205         //      char objread[N];
 206                 sum = 0;
 207                 do {
 208                         n = recv((int)acceptfd, (void *) objread, N, 0);
 209                 //      printf("DEBUG -> 3. Reading %d bytes cap = %d\n", n, N);
 210                         sum += n;
 211                 } while(sum < N && n != 0);
 212 //              printf("DEBUG -> Recv objs from Coordinator %d %d %d %d\n", *objread, *(objread + 6), *(objread + 12), *(objread + 18));
 213         }
 214
 215         // Read modified objects
 216         if(fixed.nummod != 0) { // If pile contains modified objects
 217                 if ((modptr = objstrAlloc(mainobjstore, fixed.sum_bytes)) == NULL) {
 218                         printf("objstrAlloc error for modified objects %s, %d", __FILE__, __LINE__);
 219                         return 1;
 220                 }
 221                 sum = 0;
 222                 do { // Recv the objs that are modified at Coordinator
 223                         n = recv((int)acceptfd, modptr+sum, fixed.sum_bytes-sum, 0);
 224                 //      printf("DEBUG -> 4. Reading %d bytes cap = %d, oid = %d\n", n, fixed.sum_bytes, *((int *)modptr));
 225                         sum += n;
 226                 } while (sum < fixed.sum_bytes && n != 0);
 227         }
 228
 229         //Send control message as per all votes from all oids in the machine
 230         if((prevctrl = handleTransReq(acceptfd, &fixed, transinfo, listmid, objread, modptr)) == 0 ) {
 231                 printf("Handle req error\n");
 232         }
 233
 234         //Read for new control message from Coordiator
 235         if((retval = recv((int)acceptfd, &control, sizeof(char), 0)) <= 0 ) {
 236                 perror("Error in receiving control message");
 237                 return 1;
 238         }
 239
 240         switch(control) {
 241                 case TRANS_ABORT:
 242                         printf("DEBUG -> Recv TRANS_ABORT from Coordinator\n");
 243                         //send ack to coordinator
 244                         sendctrl = TRANS_SUCESSFUL;
 245                         if(send((int)acceptfd, &sendctrl, sizeof(char), 0) < 0) {
 246                                 perror("Error sending ACK to coordinator\n");
 247                                 return 1;
 248                         }
 249                         //Mark all ref counts as 1 and do garbage collection
 250                         ptr = modptr;
 251                         for(i = 0; i< fixed.nummod; i++) {
 252                                 tmp_header = (objheader_t *)ptr;
 253                                 tmp_header->rcount = 1;
 254                                 ptr += sizeof(objheader_t) + classsize[tmp_header->type];
 255                         }
 256                         //Unlock objects that was locked in this machine due to this transaction
 257                         for(i = 0; i< transinfo->numlocked; i++) {
 258                                 header = mhashSearch(transinfo->objlocked[i]);// find the header address
 259                                 ((objheader_t *)header)->status &= ~(LOCK);
 260                         }
 261                         ptr = NULL;
 262                         return 0;
 263
 264                 case TRANS_COMMIT:
 265                         printf("DEBUG -> Recv TRANS_COMMIT from Coordinator\n");
 266                         if((val = transCommitProcess(transinfo, (int)acceptfd)) != 0) {
 267                                 printf("Error in transCommitProcess %s, %d\n", __FILE__, __LINE__);
 268                         }
 269                         break;
 270                 case TRANS_ABORT_BUT_RETRY_COMMIT:
 271                         printf("DEBUG -> Recv TRANS_ABORT_BUT_RETRY_COMMIT from Coordinator\n");
 272                         //Process again after waiting for sometime and on prev control message sent
 273                         switch(prevctrl) {
 274                                 case TRANS_AGREE:
 275                                         sendctrl = TRANS_AGREE;
 276                                         if(send((int)acceptfd, &sendctrl, sizeof(char), 0) < 0) {
 277                                                 perror("Error sending ACK to coordinator\n");
 278                                         }
 279                                         sleep(5);
 280                                         break;
 281                                 case TRANS_SOFT_ABORT:
 282                                         if((newctrl = handleTransReq(acceptfd, &fixed, transinfo, listmid, objread, modptr)) == 0 ) {
 283                                                 printf("Handle req error\n");
 284                                         }
 285                                         if(newctrl == prevctrl){
 286                                                 //Send ABORT
 287                                                 newctrl = TRANS_DISAGREE;
 288                                                 if(send((int)acceptfd, &newctrl, sizeof(char), 0) < 0) {
 289                                                         perror("Error sending ACK to coordinator\n");
 290                                                 }
 291                                                 //Set the reference count of the object to 1 in mainstore for garbage collection
 292                                                 ptr = modptr;
 293                                                 for(i = 0; i< fixed.nummod; i++) {
 294                                                         tmp_header = (objheader_t *) ptr;
 295                                                         tmp_header->rcount = 1;
 296                                                         ptr += sizeof(objheader_t) + classsize[tmp_header->type];
 297                                                 }
 298                                                 //Unlock objects that was locked in this machine due to this transaction
 299                                                 for(i = 0; i< transinfo->numlocked; i++) {
 300                                                         ptr = mhashSearch(transinfo->objlocked[i]);// find the header address
 301                                                         ((objheader_t *)ptr)->status &= ~(LOCK);
 302                                                 }
 303                                                 return 0;
 304                                         } else {
 305                                                 //Send new control message
 306                                                 if(send((int)acceptfd, &newctrl, sizeof(char), 0) < 0) {
 307                                                         perror("Error sending ACK to coordinator\n");
 308                                                 }
 309                                         }
 310
 311                                         break;
 312                         }
 313
 314                         break;
 315                 case TRANS_ABORT_BUT_RETRY_COMMIT_WITH_RELOCATING:
 316                         //TODO expect another transrequest from client
 317                         printf("DEBUG -> Recv TRANS_ABORT_BUT_RETRY_COMMIT_WITH_RELOCATING from Coordinator\n");
 318                         break;
 319                 default:
 320                         printf("No response to TRANS_AGREE OR DISAGREE protocol\n");
 321                         //TODO Use fixed.trans_id  TID since Client may have died
 322                         break;
 323         }
 324         //Free memory
 325         printf("DEBUG -> Freeing...");
 326         fflush(stdout);
 327         if (transinfo->objmod != NULL) {
 328                 free(transinfo->objmod);
 329                 transinfo->objmod = NULL;
 330         }
 331         if (transinfo->objlocked != NULL) {
 332                 free(transinfo->objlocked);
 333                 transinfo->objlocked = NULL;
 334         }
 335         if (transinfo->objnotfound != NULL) {
 336                 free(transinfo->objnotfound);
 337                 transinfo->objnotfound = NULL;
 338         }
 339         return 0;
 340 }
 341
 342 //This function runs a decision after all objects are weighed under one of the 4 possibilities
 343 //and returns the appropriate control message to the Ccordinator
 344 char handleTransReq(int acceptfd, fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigned int *listmid, char *objread, void *modptr) {
 345         int val;
 346         short version;
 347         char control = 0, ctrlmissoid, *ptr;
 348         int i, j = 0;
 349         unsigned int oid;
 350         unsigned int *oidnotfound, *oidlocked, *oidmod;
 351
 352         oidnotfound = (unsigned int *) calloc(fixed->numread + fixed->nummod, sizeof(unsigned int));
 353         oidlocked = (unsigned int *) calloc(fixed->numread + fixed->nummod, sizeof(unsigned int));
 354         oidmod = (unsigned int *) calloc(fixed->nummod, sizeof(unsigned int));
 355         // Counters and arrays to formulate decision on control message to be sent
 356         int objnotfound = 0, objlocked = 0, objmod =0, v_nomatch = 0, v_matchlock = 0, v_matchnolock = 0;
 357         int objmodnotfound = 0, nummodfound = 0;
 358         void *mobj;
 359         objheader_t *headptr;
 360
 361         //Process each object present in the pile
 362         ptr = modptr;
 363         //printf("DEBUG -> Total objs involved in trans is %d\n",fixed->nummod + fixed->numread);
 364         fflush(stdout);
 365         //Process each oid in the machine pile/ group
 366         for (i = 0; i < fixed->numread + fixed->nummod; i++) {
 367                 if (i < fixed->numread) {//Object is read
 368                         int incr = sizeof(unsigned int) + sizeof(short);// Offset that points to next position in the objread array
 369                         incr *= i;
 370                         oid = *((unsigned int *)(objread + incr));
 371                         incr += sizeof(unsigned int);
 372                         version = *((short *)(objread + incr));
 373                 } else {//Obj is modified
 374                         headptr = (objheader_t *) ptr;
 375                         oid = headptr->oid;
 376                         oidmod[objmod] = oid;//Array containing modified oids
 377                         objmod++;
 378                         version = headptr->version;
 379                         ptr += sizeof(objheader_t) + classsize[headptr->type];
 380                 }
 381                 //Check if object is still present in the machine since the beginning of TRANS_REQUEST
 382                 if ((mobj = mhashSearch(oid)) == NULL) {// Obj not found
 383                         //Save the oids not found for later use
 384                         oidnotfound[objnotfound] = ((objheader_t *)mobj)->oid;
 385                         objnotfound++;
 386                 } else { // If obj found in machine (i.e. has not moved)
 387                         //Check if obj is locked
 388                         if ((((objheader_t *)mobj)->status & LOCK) == LOCK) {
 389                                 if (version == ((objheader_t *)mobj)->version) {      // If version match
 390                                         v_matchlock++;
 391                                 } else {//If versions don't match ..HARD ABORT
 392                                         v_nomatch++;
 393                                         //send TRANS_DISAGREE to Coordinator
 394                                         control = TRANS_DISAGREE;
 395                                         if((val = write(acceptfd, &control, sizeof(char))) <= 0) {
 396                                                 perror("Error in sending control to the Coordinator\n");
 397                                                 return 0;
 398                                         }
 399                                         printf("DEBUG -> Sending TRANS_DISAGREE\n");
 400                                         return control;
 401                                 }
 402                         } else {//Obj is not locked , so lock object
 403                                 ((objheader_t *)mobj)->status |= LOCK;
 404                                 //Save all object oids that are locked on this machine during this transaction request call
 405                                 oidlocked[objlocked] = ((objheader_t *)mobj)->oid;
 406                                 printf("DEBUG-> Obj locked are %d\n",((objheader_t *)mobj)->oid);
 407                                 objlocked++;
 408                                 if (version == ((objheader_t *)mobj)->version) { //If versions match
 409                                         v_matchnolock++;
 410                                 } else { //If versions don't match
 411                                         v_nomatch++;
 412                                         //send TRANS_DISAGREE to Coordinator
 413                                         control = TRANS_DISAGREE;
 414                                         if((val = write(acceptfd, &control, sizeof(char))) <= 0) {
 415                                                 perror("Error in sending control to the Coordinator\n");
 416                                                 return 0;
 417                                         }
 418                                         printf("DEBUG -> Sending TRANS_DISAGREE\n");
 419                                         return control;
 420                                 }
 421                         }
 422                 }
 423         }
 424
 425         printf("No of objs locked = %d\n", objlocked);
 426         printf("No of v_nomatch = %d\n", v_nomatch);
 427         printf("No of objs v_match but are did not have locks before = %d\n", v_matchnolock);
 428         printf("No of objs v_match but had locks before = %d\n", v_matchlock);
 429         printf("No of objs not found = %d\n", objnotfound);
 430         printf("No of objs modified but not found = %d\n", objmodnotfound);
 431
 432         //Decide what control message(s) to send
 433         if(v_matchnolock == fixed->numread + fixed->nummod) {
 434                 //send TRANS_AGREE to Coordinator
 435                 control = TRANS_AGREE;
 436                 if((val = write(acceptfd, &control, sizeof(char))) <= 0) {
 437                         perror("Error in sending control to Coordinator\n");
 438                         return 0;
 439                 }
 440                 printf("DEBUG -> Sending TRANS_AGREE\n");
 441         }
 442
 443         if((v_matchlock > 0 && v_nomatch == 0) || (objnotfound > 0 && v_nomatch == 0)) {
 444                 //send TRANS_SOFT_ABORT to Coordinator
 445                 control = TRANS_SOFT_ABORT;
 446                 if((val = write(acceptfd, &control, sizeof(char))) <=0 ) {
 447                         perror("Error in sending control back to coordinator\n");
 448                         return 0;
 449                 }
 450                 printf("DEBUG -> Sending TRANS_SOFT_ABORT\n");
 451                 //send number of oids not found and the missing oids
 452                 if((val = write(acceptfd, &objnotfound, sizeof(int))) <= 0) {
 453                         perror("Error in sending no of objects that are not found\n");
 454                         return 0;
 455                 }
 456                 if(objnotfound != 0) {
 457                         if((val = write(acceptfd, oidnotfound, (sizeof(unsigned int) * objnotfound))) <= 0) {
 458                                 perror("Error in sending objects that are not found\n");
 459                                 return 0;
 460                         }
 461                 }
 462         }
 463
 464         //Do the following when TRANS_DISAGREE is sent
 465         if(control == TRANS_DISAGREE) {
 466                 //Set the reference count of the object to 1 in mainstore for garbage collection
 467                 ptr = modptr;
 468                 for(i = 0; i< fixed->nummod; i++) {
 469                         headptr = (objheader_t *) ptr;
 470                         headptr->rcount = 1;
 471                         ptr += sizeof(objheader_t) + classsize[headptr->type];
 472                 }
 473                 //Unlock objects that was locked in the trans
 474                 for(i = 0; i< objlocked ; i++) {
 475                         mobj = mhashSearch(oidlocked[i]);// find the header address
 476                         ((objheader_t *)mobj)->status &= ~(LOCK);
 477                 }
 478         }
 479
 480         //Fill out the structure required for a trans commit process if pile receives a TRANS_COMMIT
 481         transinfo->objmod = oidmod;
 482         transinfo->objlocked = oidlocked;
 483         transinfo->objnotfound = oidnotfound;
 484         transinfo->modptr = modptr;
 485         transinfo->nummod = fixed->nummod;
 486         transinfo->numlocked = objlocked;
 487         transinfo->numnotfound = objnotfound;
 488
 489         return control;
 490 }
 491
 492 //Processes oids in the TRANS_COMMIT request at the participant end and sends an ack back
 493 int transCommitProcess(trans_commit_data_t *transinfo, int acceptfd) {
 494         objheader_t *header;
 495         int i = 0, offset = 0;
 496         char control;
 497         //Process each modified object saved in the mainobject store
 498         for(i=0; i<transinfo->nummod; i++) {
 499                 if((header = (objheader_t *) mhashSearch(transinfo->objmod[i])) == NULL) {
 500                         printf("mhashserach returns NULL\n");
 501                 }
 502                 //change reference count of older address and free space in objstr ??
 503                 header->rcount = 1; //Not sure what would be th val
 504                 //change ptr address in mhash table
 505                 mhashRemove(transinfo->objmod[i]);
 506                 mhashInsert(transinfo->objmod[i], (transinfo->modptr + offset));
 507                 offset += sizeof(objheader_t) + classsize[header->type];
 508                 //update object version
 509                 header = (objheader_t *) mhashSearch(transinfo->objmod[i]);
 510                 header->version += 1;
 511         }
 512         for(i=0; i<transinfo->numlocked; i++) {
 513                 //unlock objects
 514                 header = (objheader_t *) mhashSearch(transinfo->objlocked[i]);
 515                 header->status &= ~(LOCK);
 516         }
 517
 518         //TODO Update location lookup table
 519
 520         //send ack to coordinator
 521         control = TRANS_SUCESSFUL;
 522         if(send((int)acceptfd, &control, sizeof(char), 0) < 0) {
 523                 perror("Error sending ACK to coordinator\n");
 524         }
 525
 526         printf("DEBUG-> Completed the pending transaction\n");
 527         return 0;
 528 }
 529