6 extern sockPoolHashTable_t *transPrefetchSockPool;
7 extern unsigned int myIpAddr;
8 extern sockPoolHashTable_t *transPResponseSocketPool;
9 extern pthread_mutex_t prefetchcache_mutex;
10 extern prehashtable_t pflookup;
12 // Function for new prefetch call
13 void rangePrefetch(unsigned int oid, short numoffset, short *offsets) {
14 /* Allocate memory in prefetch queue and push the block there */
15 int qnodesize = sizeof(unsigned int) + sizeof(unsigned short) + numoffset * sizeof(short);
16 char *node = (char *) getmemory(qnodesize);
22 ((unsigned int *)node)[0] = oid;
23 index = index + (sizeof(unsigned int));
24 *((short *)(node+index)) = numoffset;
25 index = index + (sizeof(short));
26 memcpy(node+index, offsets, numoffset * sizeof(short));
31 void *transPrefetchNew() {
33 /* Read from prefetch queue */
34 void *node = gettail();
36 /* Check tuples if they are found locally */
37 perMcPrefetchList_t* pilehead = checkIfLocal(node);
40 // Get sock from shared pool
41 int sd = getSock2(transPrefetchSockPool, pilehead->mid);
43 /* Send Prefetch Request */
44 perMcPrefetchList_t *ptr = pilehead;
46 sendRangePrefetchReq(ptr, sd);
50 /* Deallocated pilehead */
51 proPrefetchQDealloc(pilehead);
53 // Deallocate the prefetch queue pile node
58 int getsize(short *ptr, int n) {
59 int sum = 0, newsum, i;
60 for (i = n-1; i >= 0; i--) {
61 newsum = (1 + ptr[i])+((1 + ptr[i])*sum);
68 perMcPrefetchList_t *checkIfLocal(char *ptr) {
69 unsigned int oid = *(GET_OID(ptr));
70 short numoffset = *(GET_NUM_OFFSETS(ptr));
71 short *offsetarray = GET_OFFSETS(ptr);
73 unsigned int dfsList[numoffset];
77 perMcPrefetchList_t *head = NULL;
81 for(i = 0; i<numoffset; i++) {
85 //Start searching the dfsList
88 if((retval = getNextOid(offsetarray, dfsList, &top, &depth, &odep, oid)) != 0) {
89 printf("%s() Error: Getting new oid at %s, %d\n", __func__, __FILE__, __LINE__);
92 dfsList[top] = odep.oid;
96 objheader_t *objhead = searchObj(dfsList[top]);
97 if(objhead == NULL) { //oid not found
98 if(dfsList[top] == 0) { //oid is null
102 int machinenum = lhashSearch(dfsList[top]);
103 insertPrefetch(machinenum, dfsList[top], numoffset-(depth), &offsetarray[depth], &head);
106 while((dfsList[top+1] == *(offsetarray + depth + 1)) && (depth >= 0)) {
114 //return if no more paths to explore
115 if(top < 0 || depth < 0) {
118 //If more paths to explore, proceed down the tree
122 header = searchObj(dfsList[prev]);
124 printf("%s() Error Object not found at %s , %d\n", __func__, __FILE__, __LINE__);
128 if(TYPE(header) > NUMCLASSES) {
129 dfsList[top] = getNextArrayOid(offsetarray, dfsList, &top, &depth);
130 } else { //linked list
131 dfsList[top] = getNextPointerOid(offsetarray, dfsList, &top, &depth);
135 } else { // increment and go down the tree
139 if(depth >= numoffset) { //reached the end of the path
143 while((dfsList[top + 1] == *(offsetarray + depth + 1)) && (depth >= 0)) {
150 //return if no more paths to explore
151 if(top < 0 || depth < 0) {
154 //If more paths to explore, go down the tree
157 objheader_t * header;
158 header = searchObj(dfsList[prev]);
160 printf("%s() Error Object not found at %s , %d\n", __func__, __FILE__, __LINE__);
164 if(TYPE(header) > NUMCLASSES) {
165 dfsList[top] = getNextArrayOid(offsetarray, dfsList, &top, &depth);
166 } else { //linked list
167 dfsList[top] = getNextPointerOid(offsetarray, dfsList, &top, &depth);
178 objheader_t *searchObj(unsigned int oid) {
179 objheader_t *header = NULL;
181 if ((header = (objheader_t *)mhashSearch(oid)) != NULL) {
183 } else if ((header = (objheader_t *) prehashSearch(oid)) != NULL) {
186 //printf("Error: Cannot find header %s, %d\n", __func__, __LINE__);
191 /* Delete perMcPrefetchList_t and everything it points to */
192 void proPrefetchQDealloc(perMcPrefetchList_t *node) {
193 perMcPrefetchList_t *prefetchpile_ptr;
194 perMcPrefetchList_t *prefetchpile_next_ptr;
195 objOffsetPile_t *objpile_ptr;
196 objOffsetPile_t *objpile_next_ptr;
198 prefetchpile_ptr = node;
199 while (prefetchpile_ptr != NULL) {
200 prefetchpile_next_ptr = prefetchpile_ptr;
201 while(prefetchpile_ptr->list != NULL) {
202 //offsets aren't owned by us, so we don't free them.
203 objpile_ptr = prefetchpile_ptr->list;
204 prefetchpile_ptr->list = objpile_ptr->next;
207 prefetchpile_ptr = prefetchpile_next_ptr->next;
208 free(prefetchpile_next_ptr);
212 void insertPrefetch(int mid, unsigned int oid, short numoffset, short *offsets, perMcPrefetchList_t **head) {
213 perMcPrefetchList_t *ptr;
214 objOffsetPile_t *objnode;
215 objOffsetPile_t **tmp;
217 //Loop through the machines
218 for(; 1; head=&((*head)->next)) {
220 if ((*head)==NULL||(tmid=(*head)->mid)>mid) {
221 perMcPrefetchList_t * tmp = (perMcPrefetchList_t *) malloc(sizeof(perMcPrefetchList_t));
223 objnode = malloc(sizeof(objOffsetPile_t));
224 objnode->offsets = offsets;
226 objnode->numoffset = numoffset;
227 objnode->next = NULL;
239 for(tmp=&((*head)->list); 1; tmp=&((*tmp)->next)) {
243 if ((*tmp)==NULL||((toid=(*tmp)->oid)>oid)) {
244 objnode = (objOffsetPile_t *) malloc(sizeof(objOffsetPile_t));
245 objnode->offsets = offsets;
247 objnode->numoffset = numoffset;
248 objnode->next = *tmp;
257 int onumoffset=(*tmp)->numoffset;
258 short * ooffset=(*tmp)->offsets;
260 for(i=0; i<numoffset; i++) {
262 //We've matched, let's just extend the current prefetch
263 (*tmp)->numoffset=numoffset;
264 (*tmp)->offsets=offsets;
267 if (ooffset[i]<offsets[i]) {
269 } else if (ooffset[i]>offsets[i]) {
270 //Place item before the current one
271 objnode = (objOffsetPile_t *) malloc(sizeof(objOffsetPile_t));
272 objnode->offsets = offsets;
274 objnode->numoffset = numoffset;
275 objnode->next = *tmp;
280 //if we get to the end, we're already covered by this prefetch
288 void sendRangePrefetchReq(perMcPrefetchList_t *mcpilenode, int sd) {
291 objOffsetPile_t *tmp;
293 /* Send TRANS_PREFETCH control message */
294 control = TRANS_PREFETCH;
295 send_data(sd, &control, sizeof(char));
297 /* Send Oids and offsets in pairs */
298 tmp = mcpilenode->list;
300 len = sizeof(int) + sizeof(unsigned int) + sizeof(unsigned int) + ((tmp->numoffset) * sizeof(short));
301 char oidnoffset[len];
302 char *buf=oidnoffset;
303 *((int*)buf) = tmp->numoffset;
305 *((unsigned int *)buf) = tmp->oid;
306 buf+=sizeof(unsigned int);
307 *((unsigned int *)buf) = myIpAddr;
308 buf += sizeof(unsigned int);
309 memcpy(buf, tmp->offsets, (tmp->numoffset)*sizeof(short));
310 send_data(sd, oidnoffset, len);
314 /* Send a special char -1 to represent the end of sending oids + offset pair to remote machine */
316 send_data(sd, &endpair, sizeof(int));
320 int getRangePrefetchResponse(int sd) {
322 recv_data(sd, &length, sizeof(int));
323 int size = length - sizeof(int);
324 char recvbuffer[size];
325 recv_data(sd, recvbuffer, size);
326 char control = *((char *) recvbuffer);
328 if(control == OBJECT_FOUND) {
329 oid = *((unsigned int *)(recvbuffer + sizeof(char)));
330 size = size - (sizeof(char) + sizeof(unsigned int));
331 pthread_mutex_lock(&prefetchcache_mutex);
333 if((ptr = prefetchobjstrAlloc(size)) == NULL) {
334 printf("%s() Error: objstrAlloc error for copying into prefetch cache in line %d at %s\n",
335 __func__, __LINE__, __FILE__);
336 pthread_mutex_unlock(&prefetchcache_mutex);
339 pthread_mutex_unlock(&prefetchcache_mutex);
340 memcpy(ptr, recvbuffer + sizeof(char) + sizeof(unsigned int), size);
343 /* Insert into prefetch hash lookup table */
345 if((oldptr = prehashSearch(oid)) != NULL) {
346 if(((objheader_t *)oldptr)->version <= ((objheader_t *)ptr)->version) {
348 prehashInsert(oid, ptr);
351 prehashInsert(oid, ptr);
353 pthread_mutex_lock(&pflookup.lock);
354 pthread_cond_broadcast(&pflookup.cond);
355 pthread_mutex_unlock(&pflookup.lock);
356 } else if(control == OBJECT_NOT_FOUND) {
357 oid = *((unsigned int *)(recvbuffer + sizeof(char)));
358 //printf("%s() Error: OBJ NOT FOUND.. THIS SHOULD NOT HAPPEN\n", __func__);
360 printf("%s() Error: in Decoding the control value %d, %s\n", __func__, __LINE__, __FILE__);
365 int rangePrefetchReq(int acceptfd) {
366 int numoffset, sd = -1;
367 unsigned int baseoid, mid = -1;
371 recv_data(acceptfd, &numoffset, sizeof(int));
374 recv_data(acceptfd, &oidmid, 2*sizeof(unsigned int));
375 baseoid = oidmid.oid;
376 if(mid != oidmid.mid) {
378 freeSockWithLock(transPResponseSocketPool, mid, sd);
380 sd = getSockWithLock(transPResponseSocketPool, mid);
382 short offsetsarry[numoffset];
383 recv_data(acceptfd, offsetsarry, numoffset*sizeof(short));
386 if((retval = dfsOffsetTree(baseoid, offsetsarry, sd, numoffset)) != 0) {
387 printf("%s() Error: in dfsOffsetTree() at line %d in %s()\n",
388 __func__, __LINE__, __FILE__);
395 freeSockWithLock(transPResponseSocketPool, mid, sd);
399 int dfsOffsetTree(unsigned int baseoid, short * offsetarray, int sd, int numoffset) {
401 unsigned int dfsList[numoffset];
408 for(i = 0; i<numoffset; i++) {
412 //Start searching the dfsList
415 if((retval = getNextOid(offsetarray, dfsList, &top, &depth, &odep, baseoid)) != 0) {
416 printf("%s() Error: Getting new oid at %s, %d\n", __func__, __FILE__, __LINE__);
419 dfsList[top] = odep.oid;
423 objheader_t *objhead = searchObj(dfsList[top]);
424 if(objhead == NULL) { //null oid or oid not found
426 if((retval = sendOidNotFound(dfsList[top], sd)) != 0) {
427 printf("%s() Error in sendOidNotFound() at line %d in %s()\n", __func__, __LINE__, __FILE__);
431 while((dfsList[top+1] == *(offsetarray + depth + 1)) && (depth >= 0)) {
439 //return if no more paths to explore
440 if(top < 0 || depth < 0) {
443 //If more paths to explore, proceed down the tree
447 header = searchObj(dfsList[prev]);
449 printf("%s() Error Object not found at %s , %d\n", __func__, __FILE__, __LINE__);
454 if(TYPE(header) > NUMCLASSES) {
455 dfsList[top] = getNextArrayOid(offsetarray, dfsList, &top, &depth);
456 } else { //linked list
457 dfsList[top] = getNextPointerOid(offsetarray, dfsList, &top, &depth);
461 } else { // increment and go down the tree
462 //Send Object id found
463 if((retval = sendOidFound(OID(objhead), sd)) != 0) {
464 printf("%s() Error in sendOidFound() at line %d in %s()\n", __func__, __LINE__, __FILE__);
470 if(depth >= numoffset) { //reached the end of the path
474 while((dfsList[top + 1] == *(offsetarray + depth + 1)) && (depth >= 0)) {
481 //return if no more paths to explore
482 if(top < 0 || depth < 0) {
485 //If more paths to explore, go down the tree
488 objheader_t * header;
489 header = searchObj(dfsList[prev]);
491 printf("%s() Error Object not found at %s , %d\n", __func__, __FILE__, __LINE__);
496 if(TYPE(header) > NUMCLASSES) {
497 dfsList[top] = getNextArrayOid(offsetarray, dfsList, &top, &depth);
498 } else { //linked list
499 dfsList[top] = getNextPointerOid(offsetarray, dfsList, &top, &depth);
510 int getNextOid(short * offsetarray, unsigned int *dfsList, int *top, int *depth, oidAtDepth_t *odep, unsigned int baseoid) {
515 int prev = (*top) - 2;
516 unsigned int oid = *(dfsList+prev);
517 objheader_t * header = searchObj(oid);
523 int range = GET_RANGE(*(offsetarray+(*depth) + 1));
524 short stride = GET_STRIDE(*(offsetarray+(*depth) + 1));
525 stride++; //Note bit pattern 000 => stride = 1 etc
527 if(TYPE(header) > NUMCLASSES) {
528 int elementsize = classsize[TYPE(header)];
529 struct ArrayObject *ao = (struct ArrayObject *) (((char *)header) + sizeof(objheader_t));
530 int length = ao->___length___;
531 //check is stride is +ve or -ve
533 if(GET_STRIDEINC(*(offsetarray+ (*depth) + 1))) {
538 int startelement = *(offsetarray + (*depth));
539 if(startelement < 0 || startelement >=length) {
540 printf("%s() Error: Offset out of range at %d\n", __func__, __LINE__);
545 int index = *(dfsList+(*top)+1);
546 odep->oid = *((unsigned int *)(((char *)ao) + sizeof(struct ArrayObject) \
547 + (elementsize * (startelement + (sign*stride*index)))));
548 odep->depth = *(depth);
549 } else { //linked list
552 if(range > 0) { //go to the next offset
553 startelement = *((int *)(offsetarray + (*depth) + 2));
555 } else if(range == 0) {
556 startelement = *((int *)(offsetarray + (*depth)));
562 odep->oid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + startelement));
563 odep->depth = *depth;
570 unsigned int getNextArrayOid(short *offsetarray, unsigned int *dfsList, int *top, int* depth) {
571 int prev = (*top) - 2;
572 unsigned int oid = *(dfsList + prev);
573 if(oid == 0) { //null oid
576 objheader_t *header = searchObj(oid);
578 printf("%s() Error: Object not found at %s , %d\n", __func__, __FILE__, __LINE__);
581 short stride = GET_STRIDE(*(offsetarray+(*depth) + 1));
582 stride++; //Note bit pattern 000 => stride = 1 etc
583 //check is stride is +ve or -ve
585 if(GET_STRIDEINC(*(offsetarray+ (*depth) + 1))) {
590 int elementsize = classsize[TYPE(header)];
591 struct ArrayObject *ao = (struct ArrayObject *) (((char *)header) + sizeof(objheader_t));
592 int length = ao->___length___;
593 int startelement = *(offsetarray + (*depth));
594 if(startelement < 0 || startelement >=length) {
595 printf("%s() Error: Offset out of range at %d\n", __func__, __LINE__);
598 int index = *(dfsList + *top + 1);
599 oid = *((unsigned int *)(((char *)ao) + sizeof(struct ArrayObject) \
600 + (elementsize * (startelement + (sign*stride*index)))));
605 unsigned int getNextPointerOid(short *offsetarray, unsigned int *dfsList, int *top, int* depth) {
607 if(*(dfsList + *top + 1) > 1) { //tells which offset to calculate the oid from
608 //(if range > 1 then use available oid to compute next oid else go to previous oid)
613 unsigned int oid = *(dfsList + prev);
614 if(oid == 0) { //null oid
617 objheader_t *header = searchObj(oid);
619 printf("%s() Error: Object not found at %s , %d\n", __func__, __FILE__, __LINE__);
622 int startelement = *(offsetarray + *depth);
623 oid = *((unsigned int *)(((char *)header) + sizeof(objheader_t) + startelement));
624 //TODO add optimization for checking if this oid has already not been found
629 int sendOidFound(unsigned int oid, int sd) {
631 if((header = (objheader_t *) mhashSearch(oid)) != NULL) {
633 } else if((header = (objheader_t *) prehashSearch(oid))!=NULL) {
636 printf("%s() Error: THIS SHOULD NOT HAPPEN at line %d in %s()\n", __func__, __LINE__, __FILE__);
642 GETSIZE(objsize, header);
643 int size = sizeof(int) + sizeof(char) + sizeof(unsigned int) + sizeof(objheader_t) + objsize;
644 char sendbuffer[size];
645 *((int *)(sendbuffer + incr)) = size;
647 *((char *)(sendbuffer + incr)) = OBJECT_FOUND;
648 incr += sizeof(char);
649 *((unsigned int *)(sendbuffer + incr)) = oid;
650 incr += sizeof(unsigned int);
651 memcpy(sendbuffer + incr, header, objsize + sizeof(objheader_t));
653 char control = TRANS_PREFETCH_RESPONSE;
654 sendPrefetchResponse(sd, &control, sendbuffer, &size);
658 int sendOidNotFound(unsigned int oid, int sd) {
659 int size = sizeof(int) + sizeof(char) + sizeof(unsigned int);
660 char sendbuffer[size];
661 *((int *)sendbuffer) = size;
662 *((char *)(sendbuffer + sizeof(int))) = OBJECT_NOT_FOUND;
663 *((unsigned int *)(sendbuffer + sizeof(int) + sizeof(unsigned int))) = oid;
664 char control = TRANS_PREFETCH_RESPONSE;
665 sendPrefetchResponse(sd, &control, sendbuffer, &size);