X-Git-Url: http://demsky.eecs.uci.edu/git/?a=blobdiff_plain;f=Robust%2Fsrc%2FRuntime%2FDSTM%2Finterface%2Fdstmserver.c;h=672a955c6644129f2aa68786ba907b1be50f7cf9;hb=dd72e8b5c40b43ef6b796f7b481528c5d6af380e;hp=5928799c780694b28b17ce4ab84ff96090424741;hpb=159c180e5e02e77fd5d21c1ffc4a21b56b8258cc;p=IRC.git diff --git a/Robust/src/Runtime/DSTM/interface/dstmserver.c b/Robust/src/Runtime/DSTM/interface/dstmserver.c index 5928799c..672a955c 100644 --- a/Robust/src/Runtime/DSTM/interface/dstmserver.c +++ b/Robust/src/Runtime/DSTM/interface/dstmserver.c @@ -21,13 +21,20 @@ #define LISTEN_PORT 2156 #define BACKLOG 10 //max pending connections #define RECEIVE_BUFFER_SIZE 2048 -#define PRE_BUF_SIZE 2048 extern int classsize[]; objstr_t *mainobjstore; pthread_mutex_t mainobjstore_mutex; pthread_mutexattr_t mainobjstore_mutex_attr; /* Attribute for lock to make it a recursive lock */ +/********************************************************** + * Global variables to map socketid and remote mid + * to resuse sockets + **************************************************/ +midSocketInfo_t sockArray[NUM_MACHINES]; +int sockCount; //number of connections with all remote machines(one socket per mc) +int sockIdFound; //track if socket file descriptor is already established +pthread_mutex_t sockLock = PTHREAD_MUTEX_INITIALIZER; //lock to prevent global sock variables to be inconsistent /* This function initializes the main objects store and creates the * global machine and location lookup table */ @@ -48,6 +55,14 @@ int dstmInit(void) if (notifyhashCreate(N_HASH_SIZE, N_LOADFACTOR)) return 1; //failure + //Initialize mid to socketid mapping array + int t; + sockCount = 0; + for(t = 0; t < NUM_MACHINES; t++) { + sockArray[t].mid = 0; + sockArray[t].sockid = 0; + } + return 0; } @@ -113,16 +128,17 @@ void *dstmListen() * and accordingly calls other functions to process new requests */ void *dstmAccept(void *acceptfd) { - int val, retval, size; + int val, retval, size, sum, sockid; unsigned int oid; - char buffer[RECEIVE_BUFFER_SIZE], control,ctrl; + char *buffer; + char control,ctrl; char *ptr; void *srcObj; objheader_t *h; trans_commit_data_t transinfo; unsigned short objType, *versionarry, version; unsigned int *oidarry, numoid, mid, threadid; - + transinfo.objlocked = NULL; transinfo.objnotfound = NULL; transinfo.modptr = NULL; @@ -130,136 +146,121 @@ void *dstmAccept(void *acceptfd) transinfo.numnotfound = 0; /* Receive control messages from other machines */ - if((retval = recv((int)acceptfd, &control, sizeof(char), 0)) <= 0) { - if (retval == 0) { - pthread_exit(NULL); // Testing connection - } - perror("Error in receiving control from coordinator\n"); - pthread_exit(NULL); - } - + recv_data((int)acceptfd, &control, sizeof(char)); + switch(control) { case READ_REQUEST: /* Read oid requested and search if available */ - if((retval = recv((int)acceptfd, &oid, sizeof(unsigned int), 0)) <= 0) { - perror("Error receiving object from cooridnator\n"); - pthread_exit(NULL); - } + recv_data((int)acceptfd, &oid, sizeof(unsigned int)); if((srcObj = mhashSearch(oid)) == NULL) { - printf("Object not found in Main Object Store %s %d\n", __FILE__, __LINE__); + printf("Error: Object 0x%x is not found in Main Object Store %s, %d\n", oid, __FILE__, __LINE__); pthread_exit(NULL); } h = (objheader_t *) srcObj; GETSIZE(size, h); size += sizeof(objheader_t); + sockid = (int) acceptfd; if (h == NULL) { ctrl = OBJECT_NOT_FOUND; - if(send((int)acceptfd, &ctrl, sizeof(char), MSG_NOSIGNAL) < sizeof(char)) { - perror("Error sending control msg to coordinator\n"); - pthread_exit(NULL); - } + send_data(sockid, &ctrl, sizeof(char)); } else { /* Type */ char msg[]={OBJECT_FOUND, 0, 0, 0, 0}; *((int *)&msg[1])=size; - if(send((int)acceptfd, &msg, sizeof(msg), MSG_NOSIGNAL) < sizeof(msg)) { - perror("Error sending size of object to coordinator\n"); - pthread_exit(NULL); - } - if(send((int)acceptfd, h, size, MSG_NOSIGNAL) < size) { - perror("Error in sending object\n"); - pthread_exit(NULL); - } + send_data(sockid, &msg, sizeof(msg)); + send_data(sockid, h, size); } break; case READ_MULT_REQUEST: - printf("DEBUG-> READ_MULT_REQUEST\n"); break; case MOVE_REQUEST: - printf("DEBUG -> MOVE_REQUEST\n"); break; case MOVE_MULT_REQUEST: - printf("DEBUG -> MOVE_MULT_REQUEST\n"); break; case TRANS_REQUEST: /* Read transaction request */ - printf("DEBUG -> Recv TRANS_REQUEST\n"); if((val = readClientReq(&transinfo, (int)acceptfd)) != 0) { - printf("Error in readClientReq\n"); + printf("Error: In readClientReq() %s, %d\n", __FILE__, __LINE__); pthread_exit(NULL); } break; case TRANS_PREFETCH: - printf("DEBUG -> Recv TRANS_PREFETCH\n"); - if((val = prefetchReq((int)acceptfd)) != 0) { - printf("Error in transPrefetch\n"); - pthread_exit(NULL); - } + do { + if((val = prefetchReq((int)acceptfd)) != 0) { + printf("Error: In prefetchReq() %s, %d\n", __FILE__, __LINE__); + break; + } + recv_data((int)acceptfd, &control, sizeof(char)); + } while (control == TRANS_PREFETCH); + break; + case TRANS_PREFETCH_RESPONSE: + //do { + if((val = getPrefetchResponse((int) acceptfd)) != 0) { + printf("Error: In getPrefetchResponse() %s, %d\n", __FILE__, __LINE__); + pthread_exit(NULL); + } + //} while (control == TRANS_PREFETCH_RESPONSE); break; case START_REMOTE_THREAD: - retval = recv((int)acceptfd, &oid, sizeof(unsigned int), 0); - if (retval <= 0) - perror("dstmAccept(): error receiving START_REMOTE_THREAD msg"); - else if (retval != sizeof(unsigned int)) - printf("dstmAccept(): incorrect msg size %d for START_REMOTE_THREAD\n", - retval); - else - { - objType = getObjType(oid); - startDSMthread(oid, objType); - } + recv_data((int)acceptfd, &oid, sizeof(unsigned int)); + objType = getObjType(oid); + startDSMthread(oid, objType); break; case THREAD_NOTIFY_REQUEST: - size = sizeof(unsigned int); - retval = recv((int)acceptfd, ptr, size, 0); - numoid = *((unsigned int *) ptr); + recv_data((int)acceptfd, &numoid, sizeof(unsigned int)); size = (sizeof(unsigned int) + sizeof(unsigned short)) * numoid + 2 * sizeof(unsigned int); - retval = recv((int)acceptfd, ptr, size, 0); + if((buffer = calloc(1,size)) == NULL) { + printf("%s() Calloc error at %s, %d\n", __func__, __FILE__, __LINE__); + pthread_exit(NULL); + } + + recv_data((int)acceptfd, buffer, size); + oidarry = calloc(numoid, sizeof(unsigned int)); - memcpy(oidarry, ptr, sizeof(unsigned int) * numoid); + memcpy(oidarry, buffer, sizeof(unsigned int) * numoid); size = sizeof(unsigned int) * numoid; versionarry = calloc(numoid, sizeof(unsigned short)); - memcpy(versionarry, ptr+size, sizeof(unsigned short) * numoid); + memcpy(versionarry, buffer+size, sizeof(unsigned short) * numoid); size += sizeof(unsigned short) * numoid; - mid = *((unsigned int *)(ptr+size)); + mid = *((unsigned int *)(buffer+size)); size += sizeof(unsigned int); - threadid = *((unsigned int *)(ptr+size)); + threadid = *((unsigned int *)(buffer+size)); processReqNotify(numoid, oidarry, versionarry, mid, threadid); + free(buffer); break; case THREAD_NOTIFY_RESPONSE: size = sizeof(unsigned short) + 2 * sizeof(unsigned int); - retval = recv((int)acceptfd, ptr, size, 0); - if(retval <= 0) - perror("dstmAccept(): error receiving THREAD_NOTIFY_RESPONSE msg"); - else if( retval != 2*sizeof(unsigned int) + sizeof(unsigned short)) - printf("dstmAccept(): incorrect smsg size %d for THREAD_NOTIFY_RESPONSE msg\n", retval); - else { - oid = *((unsigned int *)ptr); - size = sizeof(unsigned int); - version = *((unsigned short *)(ptr+size)); - size += sizeof(unsigned short); - threadid = *((unsigned int *)(ptr+size)); - threadNotify(oid,version,threadid); + if((buffer = calloc(1,size)) == NULL) { + printf("%s() Calloc error at %s, %d\n", __func__, __FILE__, __LINE__); + pthread_exit(NULL); } - break; + recv_data((int)acceptfd, buffer, size); + + oid = *((unsigned int *)buffer); + size = sizeof(unsigned int); + version = *((unsigned short *)(buffer+size)); + size += sizeof(unsigned short); + threadid = *((unsigned int *)(buffer+size)); + threadNotify(oid,version,threadid); + free(buffer); + break; default: - printf("DEBUG -> dstmAccept: Error Unknown opcode %d\n", control); + printf("Error: dstmAccept() Unknown opcode %d at %s, %d\n", control, __FILE__, __LINE__); } /* Close connection */ if (close((int)acceptfd) == -1) perror("close"); - pthread_exit(NULL); } @@ -271,41 +272,30 @@ int readClientReq(trans_commit_data_t *transinfo, int acceptfd) { unsigned int *oidmod, oid; fixed_data_t fixed; objheader_t *headaddr; - int sum = 0, i, N, n, val; + int sum, i, size, n, val; oidmod = NULL; /* Read fixed_data_t data structure */ - N = sizeof(fixed) - 1; + size = sizeof(fixed) - 1; ptr = (char *)&fixed;; fixed.control = TRANS_REQUEST; - do { - n = recv((int)acceptfd, (void *) ptr+1+sum, N-sum, 0); - sum += n; - } while(sum < N && n != 0); + recv_data((int)acceptfd, ptr+1, size); /* Read list of mids */ int mcount = fixed.mcount; - N = mcount * sizeof(unsigned int); + size = mcount * sizeof(unsigned int); unsigned int listmid[mcount]; ptr = (char *) listmid; - sum = 0; - do { - n = recv((int)acceptfd, (void *) ptr+sum, N-sum, 0); - sum += n; - } while(sum < N && n != 0); - + recv_data((int)acceptfd, ptr, size); + /* Read oid and version tuples for those objects that are not modified in the transaction */ int numread = fixed.numread; - N = numread * (sizeof(unsigned int) + sizeof(short)); - char objread[N]; + size = numread * (sizeof(unsigned int) + sizeof(unsigned short)); + char objread[size]; if(numread != 0) { //If pile contains more than one object to be read, // keep reading all objects - sum = 0; - do { - n = recv((int)acceptfd, (void *) objread, N, 0); - sum += n; - } while(sum < N && n != 0); + recv_data((int)acceptfd, objread, size); } /* Read modified objects */ @@ -314,11 +304,8 @@ int readClientReq(trans_commit_data_t *transinfo, int acceptfd) { printf("calloc error for modified objects %s, %d\n", __FILE__, __LINE__); return 1; } - sum = 0; - do { // Recv the objs that are modified by the Coordinator - n = recv((int)acceptfd, (char *) modptr+sum, fixed.sum_bytes-sum, 0); - sum += n; - } while (sum < fixed.sum_bytes && n != 0); + size = fixed.sum_bytes; + recv_data((int)acceptfd, modptr, size); } /* Create an array of oids for modified objects */ @@ -340,7 +327,7 @@ int readClientReq(trans_commit_data_t *transinfo, int acceptfd) { /*Process the information read */ if((val = processClientReq(&fixed, transinfo, listmid, objread, modptr, oidmod, acceptfd)) != 0) { - printf("Error in processClientReq %s, %d\n", __FILE__, __LINE__); + printf("Error: In processClientReq() %s, %d\n", __FILE__, __LINE__); /* Free resources */ if(oidmod != NULL) { free(oidmod); @@ -361,23 +348,19 @@ int readClientReq(trans_commit_data_t *transinfo, int acceptfd) { * Following this it also receives a new control message from the co-ordinator and processes this message*/ int processClientReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigned int *listmid, char *objread, void *modptr, unsigned int *oidmod, int acceptfd) { - char *ptr, control, sendctrl; + char control, sendctrl, retval; objheader_t *tmp_header; void *header; - int i = 0, val, retval; + int i = 0, val; /* Send reply to the Coordinator */ if((retval = handleTransReq(fixed, transinfo, listmid, objread, modptr,acceptfd)) == 0 ) { - printf("Handle Trans Req error %s, %d\n", __FILE__, __LINE__); - return 1; - } - - /* Read new control message from Coordiator */ - if((retval = recv((int)acceptfd, &control, sizeof(char), 0)) <= 0 ) { - perror("Error in receiving control message\n"); + printf("Error: In handleTransReq() %s, %d\n", __FILE__, __LINE__); return 1; } + recv_data((int)acceptfd, &control, sizeof(char)); + /* Process the new control message */ switch(control) { case TRANS_ABORT: @@ -385,30 +368,22 @@ int processClientReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, free(modptr); /* Unlock objects that was locked due to this transaction */ for(i = 0; i< transinfo->numlocked; i++) { - header = mhashSearch(transinfo->objlocked[i]);// find the header address + if((header = mhashSearch(transinfo->objlocked[i])) == NULL) { + printf("mhashSearch returns NULL at %s, %d\n", __FILE__, __LINE__);// find the header address + return 1; + } STATUS(((objheader_t *)header)) &= ~(LOCK); } /* Send ack to Coordinator */ - sendctrl = TRANS_SUCESSFUL; - if(send((int)acceptfd, &sendctrl, sizeof(char), MSG_NOSIGNAL) < sizeof(char)) { - perror("Error sending ACK to coordinator\n"); - if (transinfo->objlocked != NULL) { - free(transinfo->objlocked); - } - if (transinfo->objnotfound != NULL) { - free(transinfo->objnotfound); - } - - return 1; - } - ptr = NULL; + sendctrl = TRANS_UNSUCESSFUL; + send_data((int)acceptfd, &sendctrl, sizeof(char)); break; case TRANS_COMMIT: /* Invoke the transCommit process() */ if((val = transCommitProcess(modptr, oidmod, transinfo->objlocked, fixed->nummod, transinfo->numlocked, (int)acceptfd)) != 0) { - printf("Error in transCommitProcess %s, %d\n", __FILE__, __LINE__); + printf("Error: In transCommitProcess() %s, %d\n", __FILE__, __LINE__); /* Free memory */ if (transinfo->objlocked != NULL) { free(transinfo->objlocked); @@ -421,11 +396,9 @@ int processClientReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, break; case TRANS_ABORT_BUT_RETRY_COMMIT_WITH_RELOCATING: - //TODO expect another transrequest from client - printf("DEBUG -> Recv TRANS_ABORT_BUT_RETRY_COMMIT_WITH_RELOCATING\n"); break; default: - printf("No response to TRANS_AGREE OR DISAGREE protocol\n"); + printf("Error: No response to TRANS_AGREE OR DISAGREE protocol %s, %d\n", __FILE__, __LINE__); //TODO Use fixed.trans_id TID since Client may have died break; } @@ -444,7 +417,7 @@ int processClientReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, /* This function increments counters while running a voting decision on all objects involved * in TRANS_REQUEST and If a TRANS_DISAGREE sends the response immediately back to the coordinator */ char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigned int *listmid, char *objread, void *modptr, int acceptfd) { - int val, i = 0; + int val, i = 0, j; unsigned short version; char control = 0, *ptr; unsigned int oid; @@ -466,7 +439,7 @@ char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigne /* Process each oid in the machine pile/ group per thread */ for (i = 0; i < fixed->numread + fixed->nummod; i++) { if (i < fixed->numread) {//Objs only read and not modified - int incr = sizeof(unsigned int) + sizeof(short);// Offset that points to next position in the objread array + int incr = sizeof(unsigned int) + sizeof(unsigned short);// Offset that points to next position in the objread array incr *= i; oid = *((unsigned int *)(objread + incr)); incr += sizeof(unsigned int); @@ -495,10 +468,17 @@ char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigne v_nomatch++; /* Send TRANS_DISAGREE to Coordinator */ control = TRANS_DISAGREE; - if((val = send(acceptfd, &control, sizeof(char), MSG_NOSIGNAL)) < sizeof(char)) { - perror("Error in sending control to the Coordinator\n"); - return 0; + if (objlocked > 0) { + for(j = 0; j < objlocked; j++) { + if((headptr = mhashSearch(oidlocked[j])) == NULL) { + printf("mhashSearch returns NULL at %s, %d\n", __FILE__, __LINE__); + return 0; + } + STATUS(headptr) &= ~(LOCK); + } + free(oidlocked); } + send_data(acceptfd, &control, sizeof(char)); return control; } } else {/* If Obj is not locked then lock object */ @@ -511,15 +491,19 @@ char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigne } else { /* If versions don't match ...HARD ABORT */ v_nomatch++; control = TRANS_DISAGREE; - /* Send TRANS_DISAGREE to Coordinator */ - if((val = send(acceptfd, &control, sizeof(char), MSG_NOSIGNAL)) < sizeof(char)) { - perror("Error in sending control to the Coordinator\n"); - return 0; - } if (objlocked > 0) { - STATUS(((objheader_t *)mobj)) &= ~(LOCK); + for(j = 0; j < objlocked; j++) { + if((headptr = mhashSearch(oidlocked[j])) == NULL) { + printf("mhashSearch returns NULL at %s, %d\n", __FILE__, __LINE__); + return 0; + } + STATUS(headptr) &= ~(LOCK); + } free(oidlocked); } + + /* Send TRANS_DISAGREE to Coordinator */ + send_data(acceptfd, &control, sizeof(char)); return control; } } @@ -527,49 +511,43 @@ char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigne } /* Decide what control message to send to Coordinator */ - if ((val = decideCtrlMessage(fixed, transinfo, &v_matchnolock, &v_matchlock, &v_nomatch, &objnotfound, &objlocked, + if ((control = decideCtrlMessage(fixed, transinfo, &v_matchnolock, &v_matchlock, &v_nomatch, &objnotfound, &objlocked, modptr, oidnotfound, oidlocked, acceptfd)) == 0) { - printf("Error in decideCtrlMessage %s, %d\n", __FILE__, __LINE__); + printf("Error: In decideCtrlMessage() %s, %d\n", __FILE__, __LINE__); return 0; } - return val; + return control; } /* This function decides what control message such as TRANS_AGREE, TRANS_DISAGREE or TRANS_SOFT_ABORT * to send to Coordinator based on the votes of oids involved in the transaction */ -int decideCtrlMessage(fixed_data_t *fixed, trans_commit_data_t *transinfo, int *v_matchnolock, int *v_matchlock, +char decideCtrlMessage(fixed_data_t *fixed, trans_commit_data_t *transinfo, int *v_matchnolock, int *v_matchlock, int *v_nomatch, int *objnotfound, int *objlocked, void *modptr, unsigned int *oidnotfound, unsigned int *oidlocked, int acceptfd) { int val; char control = 0; + /* Condition to send TRANS_AGREE */ if(*(v_matchnolock) == fixed->numread + fixed->nummod) { control = TRANS_AGREE; /* Send control message */ - if((val = send(acceptfd, &control, sizeof(char), MSG_NOSIGNAL)) < sizeof(char)) { - perror("Error in sending control to Coordinator\n"); - return 0; - } + send_data(acceptfd, &control, sizeof(char)); } /* Condition to send TRANS_SOFT_ABORT */ if((*(v_matchlock) > 0 && *(v_nomatch) == 0) || (*(objnotfound) > 0 && *(v_nomatch) == 0)) { control = TRANS_SOFT_ABORT; - char msg[]={TRANS_SOFT_ABORT, 0,0,0,0}; - *((int*)&msg[1])= *(objnotfound); /* Send control message */ - if((val = send(acceptfd, &msg, sizeof(msg),MSG_NOSIGNAL)) < sizeof(msg)) { - perror("Error in sending no of objects that are not found\n"); - return 0; - } + send_data(acceptfd, &control, sizeof(char)); + /* Send number of oids not found and the missing oids if objects are missing in the machine */ if(*(objnotfound) != 0) { + int msg[1]; + msg[0] = *(objnotfound); + send_data(acceptfd, &msg, sizeof(int)); int size = sizeof(unsigned int)* *(objnotfound); - if((val = send(acceptfd, oidnotfound, size ,MSG_NOSIGNAL)) < size) { - perror("Error in sending objects that are not found\n"); - return 0; - } + send_data(acceptfd, oidnotfound, size); } } @@ -597,12 +575,12 @@ int transCommitProcess(void *modptr, unsigned int *oidmod, unsigned int *oidlock /* Process each modified object saved in the mainobject store */ for(i = 0; i < nummod; i++) { if((header = (objheader_t *) mhashSearch(oidmod[i])) == NULL) { - printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__); + printf("Error: mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__); return 1; } GETSIZE(tmpsize,header); pthread_mutex_lock(&mainobjstore_mutex); - memcpy(header, (char *)modptr + offset, tmpsize + sizeof(objheader_t)); + memcpy((char*)header + sizeof(objheader_t), ((char *)modptr + sizeof(objheader_t) + offset), tmpsize); header->version += 1; /* If threads are waiting on this object to be updated, notify them */ if(header->notifylist != NULL) { @@ -618,7 +596,7 @@ int transCommitProcess(void *modptr, unsigned int *oidmod, unsigned int *oidlock /* Unlock locked objects */ for(i = 0; i < numlocked; i++) { if((header = (objheader_t *) mhashSearch(oidlocked[i])) == NULL) { - printf("mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__); + printf("Error: mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__); return 1; } STATUS(header) &= ~(LOCK); @@ -627,10 +605,7 @@ int transCommitProcess(void *modptr, unsigned int *oidmod, unsigned int *oidlock /* Send ack to coordinator */ control = TRANS_SUCESSFUL; - if(send((int)acceptfd, &control, sizeof(char), MSG_NOSIGNAL) < sizeof(char)) { - perror("Error sending ACK to coordinator\n"); - } - + send_data((int)acceptfd, &control, sizeof(char)); return 0; } @@ -640,146 +615,252 @@ int transCommitProcess(void *modptr, unsigned int *oidmod, unsigned int *oidlock * then use offset values to prefetch references to other objects */ int prefetchReq(int acceptfd) { - int i, length, sum, n, numbytes, numoffset, N, objnotfound = 0, size, count = 0; - int isArray = 0; - unsigned int oid, index = 0; - char *ptr, buffer[PRE_BUF_SIZE]; - void *mobj; - unsigned int objoid; - char control; - objheader_t * header; - int bytesRecvd; - - /* Repeatedly recv the oid and offset pairs sent for prefetch */ - while(numbytes = recv((int)acceptfd, &length, sizeof(int), 0) != 0) { - count++; - if(length == -1) - break; - sum = 0; - index = sizeof(unsigned int); // Index starts with sizeof unsigned int because the - // first 4 bytes are saved to send the - // size of the buffer (that is computed at the end of the loop) - bytesRecvd = 0; - do { - bytesRecvd += recv((int)acceptfd, (char *)&oid +bytesRecvd, - sizeof(unsigned int) - bytesRecvd, 0); - } while (bytesRecvd < sizeof(unsigned int)); - numoffset = (length - (sizeof(int) + sizeof(unsigned int)))/ sizeof(short); - N = numoffset * sizeof(short); - short offset[numoffset]; - ptr = (char *)&offset; - /* Recv the offset values per oid */ - do { - n = recv((int)acceptfd, (void *)ptr+sum, N-sum, 0); - sum += n; - } while(sum < N && n != 0); - - /* Process each oid */ - if ((mobj = mhashSearch(oid)) == NULL) {/* Obj not found */ - /* Save the oids not found in buffer for later use */ - *(buffer + index) = OBJECT_NOT_FOUND; - index += sizeof(char); - memcpy(buffer+index, &oid, sizeof(unsigned int)); - index += sizeof(unsigned int); - } else { /* If Obj found in machine (i.e. has not moved) */ - /* send the oid, it's size, it's header and data */ - header = mobj; - GETSIZE(size, header); - size += sizeof(objheader_t); - *(buffer + index) = OBJECT_FOUND; - index += sizeof(char); - memcpy(buffer+index, &oid, sizeof(unsigned int)); - index += sizeof(unsigned int); - memcpy(buffer+index, &size, sizeof(int)); - index += sizeof(int); - memcpy(buffer + index, header, size); - index += size; - /* Calculate the oid corresponding to the offset value */ - for(i = 0 ; i< numoffset ; i++) { - /* Check for arrays */ - if(TYPE(header) > NUMCLASSES) { - isArray = 1; - } - if(isArray == 1) { - int elementsize = classsize[TYPE(header)]; - objoid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + sizeof(struct ArrayObject) + (elementsize*offset[i]))); - } else { - objoid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + offset[i])); - } - if((header = mhashSearch(objoid)) == NULL) { - /* Obj not found, send oid */ - *(buffer + index) = OBJECT_NOT_FOUND; - index += sizeof(char); - memcpy(buffer+index, &oid, sizeof(unsigned int)); - index += sizeof(unsigned int); - break; - } else {/* Obj Found */ - /* send the oid, it's size, it's header and data */ - GETSIZE(size, header); - size+=sizeof(objheader_t); - *(buffer + index) = OBJECT_FOUND; - index += sizeof(char); - memcpy(buffer+index, &oid, sizeof(unsigned int)); - index += sizeof(unsigned int); - memcpy(buffer+index, &size, sizeof(int)); - index += sizeof(int); - memcpy(buffer+index, header, size); - index += size; - isArray = 0; - continue; - } - } - } - /* Check for overflow in the buffer */ - if (index >= PRE_BUF_SIZE) { - printf("Char buffer is overflowing\n"); - return 1; - } - /* Send Prefetch response control message only once*/ - if(count == 1) { - control = TRANS_PREFETCH_RESPONSE; - if((numbytes = send(acceptfd, &control, sizeof(char), MSG_NOSIGNAL)) < sizeof(char)) { - perror("Error in sending PREFETCH RESPONSE to Coordinator\n"); - return 1; - } - } - - /* Add the buffer size into buffer as a parameter */ - *((unsigned int *)buffer)=index; - /* Send the entire buffer with its size and oids found and not found */ - if(send((int)acceptfd, &buffer, index, MSG_NOSIGNAL) < sizeof(index -1)) { - perror("Error sending oids found\n"); - return 1; - } - } - return 0; + int i, size, objsize, numbytes = 0, isArray = 0, numoffset = 0; + int length, sd = -1; + char *recvbuffer, *sendbuffer, control; + unsigned int oid, mid; + short *offsetarry; + objheader_t *header; + struct sockaddr_in remoteAddr; + + do { + recv_data((int)acceptfd, &length, sizeof(int)); + if(length != -1) { + size = length - sizeof(int); + if((recvbuffer = calloc(1, size)) == NULL) { + printf("Calloc error at %s,%d\n", __FILE__, __LINE__); + return -1; + } + recv_data((int)acceptfd, recvbuffer, size); + oid = *((unsigned int *) recvbuffer); + mid = *((unsigned int *) (recvbuffer + sizeof(unsigned int))); + size = size - (2 * sizeof(unsigned int)); + numoffset = size / sizeof(short); + if((offsetarry = calloc(numoffset, sizeof(short))) == NULL) { + printf("Calloc error at %s,%d\n", __FILE__, __LINE__); + free(recvbuffer); + return -1; + } + memcpy(offsetarry, recvbuffer + (2 * sizeof(unsigned int)), size); + free(recvbuffer); +#if 0 + pthread_mutex_lock(&sockLock); + sockIdFound = 0; + pthread_mutex_unlock(&sockLock); + for(i = 0; i < NUM_MACHINES; i++) { + if(sockArray[i].mid == mid) { + sd = sockArray[i].sockid; + pthread_mutex_lock(&sockLock); + sockIdFound = 1; + pthread_mutex_unlock(&sockLock); + break; + } + } + + if(sockIdFound == 0) { + if(sockCount < NUM_MACHINES) { + +#endif + /* Create socket to send information */ + if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0){ + perror("prefetchReq():socket()"); + return -1; + } + bzero(&remoteAddr, sizeof(remoteAddr)); + remoteAddr.sin_family = AF_INET; + remoteAddr.sin_port = htons(LISTEN_PORT); + remoteAddr.sin_addr.s_addr = htonl(mid); + + if (connect(sd, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0){ + printf("Error: prefetchReq():error %d connecting to %s:%d\n", errno, + inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT); + close(sd); + return -1; + } + +#if 0 + sockArray[sockCount].mid = mid; + sockArray[sockCount].sockid = sd; + pthread_mutex_lock(&sockLock); + sockCount++; + pthread_mutex_unlock(&sockLock); + } else { + //TODO Fix for connecting to more than 2 machines && close socket + printf("%s(): Error: Currently works for only 2 machines\n", __func__); + return -1; + } + } +#endif + + /*Process each oid */ + if ((header = mhashSearch(oid)) == NULL) {/* Obj not found */ + /* Save the oids not found in buffer for later use */ + size = sizeof(int) + sizeof(char) + sizeof(unsigned int) ; + if((sendbuffer = calloc(1, size)) == NULL) { + printf("Calloc error at %s,%d\n", __FILE__, __LINE__); + free(offsetarry); + close(sd); + return -1; + } + *((int *) sendbuffer) = size; + *((char *)(sendbuffer + sizeof(int))) = OBJECT_NOT_FOUND; + *((unsigned int *)(sendbuffer + sizeof(int) + sizeof(char))) = oid; + + control = TRANS_PREFETCH_RESPONSE; + if(sendPrefetchResponse(sd, &control, sendbuffer, &size) != 0) { + free(offsetarry); + printf("Error: %s() in sending prefetch response at %s, %d\n", + __func__, __FILE__, __LINE__); + close(sd); + return -1; + } + } else { /* Object Found */ + int incr = 0; + GETSIZE(objsize, header); + size = sizeof(int) + sizeof(char) + sizeof(unsigned int) + sizeof(objheader_t) + objsize; + if((sendbuffer = calloc(1, size)) == NULL) { + printf("Calloc error at %s,%d\n", __FILE__, __LINE__); + free(offsetarry); + close(sd); + return -1; + } + *((int *) (sendbuffer + incr)) = size; + incr += sizeof(int); + *((char *)(sendbuffer + incr)) = OBJECT_FOUND; + incr += sizeof(char); + *((unsigned int *)(sendbuffer+incr)) = oid; + incr += sizeof(unsigned int); + memcpy(sendbuffer + incr, header, objsize + sizeof(objheader_t)); + + control = TRANS_PREFETCH_RESPONSE; + if(sendPrefetchResponse(sd, &control, sendbuffer, &size) != 0) { + free(offsetarry); + printf("Error: %s() in sending prefetch response at %s, %d\n", + __func__, __FILE__, __LINE__); + close(sd); + return -1; + } + + /* Calculate the oid corresponding to the offset value */ + for(i = 0 ; i< numoffset ; i++) { + /* Check for arrays */ + if(TYPE(header) > NUMCLASSES) { + isArray = 1; + } + if(isArray == 1) { + int elementsize = classsize[TYPE(header)]; + struct ArrayObject *ao = (struct ArrayObject *) (((char *)header) + sizeof(objheader_t)); + unsigned short length = ao->___length___; + /* Check if array out of bounds */ + if(offsetarry[i]< 0 || offsetarry[i] >= length) { + break; + } + oid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + sizeof(struct ArrayObject) + (elementsize*offsetarry[i]))); + } else { + oid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + offsetarry[i])); + } + + if((header = mhashSearch(oid)) == NULL) { + size = sizeof(int) + sizeof(char) + sizeof(unsigned int) ; + if((sendbuffer = calloc(1, size)) == NULL) { + printf("Calloc error at %s,%d\n", __FILE__, __LINE__); + free(offsetarry); + close(sd); + return -1; + } + *((int *) sendbuffer) = size; + *((char *)(sendbuffer + sizeof(int))) = OBJECT_NOT_FOUND; + *((unsigned int *)(sendbuffer + sizeof(int) + sizeof(char))) = oid; + + control = TRANS_PREFETCH_RESPONSE; + if(sendPrefetchResponse(sd, &control, sendbuffer, &size) != 0) { + free(offsetarry); + printf("Error: %s() in sending prefetch response at %s, %d\n", + __FILE__, __LINE__); + close(sd); + return -1; + } + break; + } else {/* Obj Found */ + int incr = 0; + GETSIZE(objsize, header); + size = sizeof(int) + sizeof(char) + sizeof(unsigned int) + sizeof(objheader_t) + objsize; + if((sendbuffer = calloc(1, size)) == NULL) { + printf("Calloc error at %s,%d\n", __func__, __FILE__, __LINE__); + free(offsetarry); + close(sd); + return -1; + } + *((int *) (sendbuffer + incr)) = size; + incr += sizeof(int); + *((char *)(sendbuffer + incr)) = OBJECT_FOUND; + incr += sizeof(char); + *((unsigned int *)(sendbuffer+incr)) = oid; + incr += sizeof(unsigned int); + memcpy(sendbuffer + incr, header, objsize + sizeof(objheader_t)); + + control = TRANS_PREFETCH_RESPONSE; + if(sendPrefetchResponse(sd, &control, sendbuffer, &size) != 0) { + free(offsetarry); + printf("Error: %s() in sending prefetch response at %s, %d\n", + __func__, __FILE__, __LINE__); + close(sd); + return -1; + } + } + isArray = 0; + } + free(offsetarry); + } + close(sd); + } + } while (length != -1); + return 0; +} + +int sendPrefetchResponse(int sd, char *control, char *sendbuffer, int *size) { + int numbytes = 0; + + send_data(sd, control, sizeof(char)); + /* Send the buffer with its size */ + int length = *(size); + send_data(sd, sendbuffer, length); + free(sendbuffer); + return 0; } void processReqNotify(unsigned int numoid, unsigned int *oidarry, unsigned short *versionarry, unsigned int mid, unsigned int threadid) { objheader_t *header; unsigned int oid; unsigned short newversion; - char msg[1+ 2 * sizeof(unsigned int) + sizeof(unsigned short)]; + char msg[1+ 2 * sizeof(unsigned int) + sizeof(unsigned short)]; int sd; struct sockaddr_in remoteAddr; int bytesSent; - int status, size, retry = 0; + int size; int i = 0; while(i < numoid) { oid = *(oidarry + i); if((header = (objheader_t *) mhashSearch(oid)) == NULL) { - printf("processReqNotify(): Object is not found in mlookup %s, %d\n", __FILE__, __LINE__); + printf("Error: mhashsearch returns NULL at %s, %d\n", __FILE__, __LINE__); return; } else { /* Check to see if versions are same */ checkversion: if ((STATUS(header) & LOCK) != LOCK) { + //FIXME make locking atomic STATUS(header) |= LOCK; - if(header->version == *(versionarry + i)) { + newversion = header->version; + if(newversion == *(versionarry + i)) { //Add to the notify list - insNode(header->notifylist, threadid, mid); + if((header->notifylist = insNode(header->notifylist, threadid, mid)) == NULL) { + printf("Error: Obj notify list points to NULL %s, %d\n", __FILE__, __LINE__); + return; + } + STATUS(header) &= ~(LOCK); } else { + STATUS(header) &= ~(LOCK); if ((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0){ perror("processReqNotify():socket()"); return; @@ -790,40 +871,30 @@ checkversion: remoteAddr.sin_addr.s_addr = htonl(mid); if (connect(sd, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0){ - printf("processReqNotify():error %d connecting to %s:%d\n", errno, + printf("Error: processReqNotify():error %d connecting to %s:%d\n", errno, inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT); - status = -1; + close(sd); + return; } else { //Send Update notification msg[0] = THREAD_NOTIFY_RESPONSE; - msg[1] = oid; + *((unsigned int *)&msg[1]) = oid; size = sizeof(unsigned int); *((unsigned short *)(&msg[1]+size)) = newversion; size += sizeof(unsigned short); *((unsigned int *)(&msg[1]+size)) = threadid; - bytesSent = send(sd, msg, 1+ 2*sizeof(unsigned int) + sizeof(unsigned short), 0); - if (bytesSent < 0){ - perror("processReqNotify():send()"); - status = -1; - } else if (bytesSent != 1 + sizeof(unsigned short) + 2*sizeof(unsigned int)){ - printf("processReqNotify(): error, sent %d bytes\n", bytesSent); - status = -1; - } else { - status = 0; - } - + size = 1+ 2*sizeof(unsigned int) + sizeof(unsigned short); + send_data(sd, msg, size); } close(sd); } - STATUS(header) &= ~(LOCK); } else { randomdelay(); - printf("DEBUG-> processReqNotify() Object is still locked\n"); goto checkversion; } } + i++; } free(oidarry); free(versionarry); } -