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=99f48f070727b627bf42f88a2230f3122d89e79d;hpb=bae799f54777dbf7bf3f222c61aeeb72b4b11c85;p=IRC.git

diff --git a/Robust/src/Runtime/DSTM/interface/dstmserver.c b/Robust/src/Runtime/DSTM/interface/dstmserver.c
index 99f48f07..672a955c 100644
--- a/Robust/src/Runtime/DSTM/interface/dstmserver.c
+++ b/Robust/src/Runtime/DSTM/interface/dstmserver.c
@@ -21,14 +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 */
-pthread_mutex_t threadnotify_mutex = PTHREAD_MUTEX_INITIALIZER;
+/**********************************************************
+ * 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 */
@@ -49,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;
 }
 
@@ -114,9 +128,10 @@ 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;
@@ -124,8 +139,6 @@ void *dstmAccept(void *acceptfd)
 	unsigned short objType, *versionarry, version;
 	unsigned int *oidarry, numoid, mid, threadid;
 	
-	int i;
-
 	transinfo.objlocked = NULL;
 	transinfo.objnotfound = NULL;
 	transinfo.modptr = NULL;
@@ -133,18 +146,12 @@ void *dstmAccept(void *acceptfd)
 	transinfo.numnotfound = 0;
 
 	/* Receive control messages from other machines */
-	if((retval = recv((int)acceptfd, &control, sizeof(char), 0)) <= 0) {
-		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 0x0 object from cooridnator\n");
-				pthread_exit(NULL);
-			}
+			recv_data((int)acceptfd, &oid, sizeof(unsigned int));
 			if((srcObj = mhashSearch(oid)) == NULL) {
 				printf("Error: Object 0x%x is not found in Main Object Store %s, %d\n", oid, __FILE__, __LINE__);
 				pthread_exit(NULL);
@@ -152,25 +159,17 @@ void *dstmAccept(void *acceptfd)
 			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;
 		
@@ -191,73 +190,70 @@ void *dstmAccept(void *acceptfd)
 			}
 			break;
 		case TRANS_PREFETCH:
-			if((val = prefetchReq((int)acceptfd)) != 0) {
-				printf("Error: In prefetchReq() %s, %d\n", __FILE__, __LINE__);
-				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 %s, %d\n",
-					retval, __FILE__, __LINE__);
-			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);
-			bzero(&buffer, RECEIVE_BUFFER_SIZE);
-			retval = recv((int)acceptfd, &buffer, size, 0);
-			numoid = *((unsigned int *) &buffer);
+			recv_data((int)acceptfd, &numoid, sizeof(unsigned int));
 			size = (sizeof(unsigned int) + sizeof(unsigned short)) * numoid + 2 * sizeof(unsigned int);
-			bzero(&buffer, RECEIVE_BUFFER_SIZE);
-			retval = recv((int)acceptfd, &buffer, size, 0);
-			if(retval <=0)
-				perror("dstmAccept(): error receiving THREAD_NOTIFY_REQUEST");
-			else if( retval != (2* sizeof(unsigned int) + (sizeof(unsigned int) + sizeof(unsigned short)) * numoid))
-				printf("dstmAccept(): incorrect msg size %d for THREAD_NOTIFY_REQUEST %s, %d\n", retval, 
-						__FILE__, __LINE__);
-			else {
-				oidarry = calloc(numoid, sizeof(unsigned int)); 
-				memcpy(oidarry, buffer, sizeof(unsigned int) * numoid);
-				size = sizeof(unsigned int) * numoid;
-				versionarry = calloc(numoid, sizeof(unsigned short));
-				memcpy(versionarry, buffer+size, sizeof(unsigned short) * numoid);
-				size += sizeof(unsigned short) * numoid;
-				mid = *((unsigned int *)(buffer+size));
-				size += sizeof(unsigned int);
-				threadid = *((unsigned int *)(buffer+size));
-				processReqNotify(numoid, oidarry, versionarry, mid, threadid);
+			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, buffer, sizeof(unsigned int) * numoid);
+			size = sizeof(unsigned int) * numoid;
+			versionarry = calloc(numoid, sizeof(unsigned short));
+			memcpy(versionarry, buffer+size, sizeof(unsigned short) * numoid);
+			size += sizeof(unsigned short) * numoid;
+			mid = *((unsigned int *)(buffer+size));
+			size += sizeof(unsigned int);
+			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);
-			bzero(&buffer, RECEIVE_BUFFER_SIZE);
-			retval = recv((int)acceptfd, &buffer, size, 0);
-			if(retval <= 0) 
-				perror("dstmAccept(): error receiving THREAD_NOTIFY_RESPONSE");
-			else if( retval != 2*sizeof(unsigned int) + sizeof(unsigned short))
-				printf("dstmAccept(): incorrect msg size %d for THREAD_NOTIFY_RESPONSE msg %s, %d\n", 
-						retval, __FILE__, __LINE__);
-			else {
-				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);
+			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("Error: dstmAccept() Unknown opcode %d at %s, %d\n", control, __FILE__, __LINE__);
 	}
@@ -265,7 +261,6 @@ void *dstmAccept(void *acceptfd)
 	/* Close connection */
 	if (close((int)acceptfd) == -1)
 		perror("close");
-	
 	pthread_exit(NULL);
 }
 
@@ -277,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 */
@@ -320,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 */
@@ -378,10 +359,8 @@ int processClientReq(fixed_data_t *fixed, trans_commit_data_t *transinfo,
 		return 1;
 	}
 
-	do {
-		retval = recv((int)acceptfd, &control, sizeof(char), 0);
-	} while(retval < sizeof(char));
-
+	recv_data((int)acceptfd, &control, sizeof(char));
+	
 	/* Process the new control message */
 	switch(control) {
 		case TRANS_ABORT:
@@ -398,17 +377,7 @@ int processClientReq(fixed_data_t *fixed, trans_commit_data_t *transinfo,
 
 			/* Send ack to Coordinator */
 			sendctrl = TRANS_UNSUCESSFUL;
-			if(send((int)acceptfd, &sendctrl, sizeof(char), MSG_NOSIGNAL) < sizeof(char)) {
-				perror("Error: In sending ACK to coordinator\n");
-				if (transinfo->objlocked != NULL) {
-					free(transinfo->objlocked);
-				}
-				if (transinfo->objnotfound != NULL) {
-					free(transinfo->objnotfound);
-				}
-
-				return 1;
-			}
+			send_data((int)acceptfd, &sendctrl, sizeof(char));
 			break;
 
 		case TRANS_COMMIT:
@@ -470,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);
@@ -500,7 +469,7 @@ char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigne
 					/* Send TRANS_DISAGREE to Coordinator */
 					control = TRANS_DISAGREE;
 					if (objlocked > 0) {
-						for(j = 0; j < objlocked; j++) {
+					  for(j = 0; j < objlocked; j++) {
 							if((headptr = mhashSearch(oidlocked[j])) == NULL) {
 								printf("mhashSearch returns NULL at %s, %d\n", __FILE__, __LINE__);
 								return 0;
@@ -509,10 +478,7 @@ char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigne
 						}
 						free(oidlocked);
 					}
-					if((val = send(acceptfd, &control, sizeof(char), MSG_NOSIGNAL)) < sizeof(char)) {
-						perror("Error in sending control to the Coordinator\n");
-						return 0;
-					}
+					send_data(acceptfd, &control, sizeof(char));
 					return control;
 				}
 			} else {/* If Obj is not locked then lock object */
@@ -537,11 +503,7 @@ char handleTransReq(fixed_data_t *fixed, trans_commit_data_t *transinfo, unsigne
 					}
 
 					/* 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;
-					}
-					
+					send_data(acceptfd, &control, sizeof(char));
 					return control;
 				}
 			}
@@ -570,34 +532,22 @@ char decideCtrlMessage(fixed_data_t *fixed, trans_commit_data_t *transinfo, int
 	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;
 
 		/* Send control message */
-		if((val = send(acceptfd, &control, sizeof(char),MSG_NOSIGNAL)) < sizeof(char)) {
-			perror("Error in sending TRANS_SOFT_ABORT control\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);
-			if((val = send(acceptfd, msg, sizeof(int) ,MSG_NOSIGNAL)) < sizeof(int)) {
-				perror("Error in sending objects that are not found\n");
-				return 0;
-			}
+			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);
 		}
 	}
 
@@ -655,11 +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");
-		return 1;
-	}
-	
+	send_data((int)acceptfd, &control, sizeof(char));
 	return 0;
 }
 
@@ -669,122 +615,217 @@ 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;
+	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;
 
-	/* Repeatedly recv one oid and offset pair sent for prefetch */
-	while(numbytes = recv((int)acceptfd, &length, sizeof(int), 0) != 0) {
-		count++;
-		if(length == -1)
-			break;
-		index = 0;  
-		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;
-		sum = 0;
-		/* Recv the offset values per oid */ 
-		do {
-			n = recv((int)acceptfd, (void *)ptr+sum, N-sum, 0); 
-			sum += n; 
-		} while(sum < N && n != 0);	
-
-		bzero(&buffer, PRE_BUF_SIZE);
-		/* 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);
-			*((unsigned int *)(buffer+index)) = oid;
-			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 = (objheader_t *)mobj;
-			GETSIZE(size, header);
-			size += sizeof(objheader_t);
-			*(buffer + index) = OBJECT_FOUND;
-			index += sizeof(char);
-			*((unsigned int *)(buffer+index)) = oid;
-			index += sizeof(unsigned int);
-			*((int *)(buffer+index)) = size;
-			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;
+	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(isArray == 1) {
-					int elementsize = classsize[TYPE(header)];
-					objoid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + sizeof(struct ArrayObject) + (elementsize*offset[i])));
+			}
+
+			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 {
-					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);
-					*((unsigned int *)(buffer+index)) = objoid;
-					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);
-					*((unsigned int *)(buffer+index)) = objoid;
-					index += sizeof(unsigned int);
-					*((int *)(buffer+index)) = size;
-					index += sizeof(int);
-					memcpy(buffer+index, header, size);
-					index += size;
-					isArray = 0;
-					continue;
+					//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
 
-		/* Check for overflow in the buffer */
-		if (index >= PRE_BUF_SIZE) {
-			printf("Error: Buffer array overflow %s, %d\n", __FILE__, __LINE__);
-			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;
+			/*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;
+}
 
-		//Send buffer size 
-		if((numbytes = send(acceptfd, &index, sizeof(unsigned int), MSG_NOSIGNAL)) < sizeof(unsigned int)) {
-			perror("Error: in sending PREFETCH RESPONSE to Coordinator\n");
-			return 1;
-		}
+int sendPrefetchResponse(int sd, char *control, char *sendbuffer, int *size) {
+	int numbytes = 0;
 
-		/* Send the entire buffer with its size and oids found and not found */
-		if(send((int)acceptfd, &buffer, index, MSG_NOSIGNAL) < sizeof(index)) {
-			perror("Error: sending oids found\n");
-			return 1;
-		}
-	}
+	send_data(sd, control, sizeof(char));
+	/* Send the buffer with its size */
+	int length = *(size);
+	send_data(sd, sendbuffer, length);
+	free(sendbuffer);
 	return 0;
 }
 
@@ -796,7 +837,7 @@ void processReqNotify(unsigned int numoid, unsigned int *oidarry, unsigned short
 	int sd;
 	struct sockaddr_in remoteAddr;
 	int bytesSent;
-	int status, size;
+	int size;
 
 	int i = 0;
 	while(i < numoid) {
@@ -832,7 +873,8 @@ checkversion:
 					if (connect(sd, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0){
 						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;
@@ -841,18 +883,8 @@ checkversion:
 						*((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("Error: processReqNotify(): error, sent %d bytes %s, %d\n", 
-									bytesSent, __FILE__, __LINE__);
-							status = -1;
-						} else {
-							status = 0;
-						}
-
+						size = 1+ 2*sizeof(unsigned int) + sizeof(unsigned short);
+						send_data(sd, msg, size);
 					}
 					close(sd);
 				}
@@ -866,4 +898,3 @@ checkversion:
 	free(oidarry);
 	free(versionarry);
 }
-