#include "dstm.h"
#include "ip.h"
-#include "clookup.h"
#include "machinepile.h"
#include "mlookup.h"
#include "llookup.h"
**********************************/
int numTransCommit = 0;
int numTransAbort = 0;
+int nchashSearch = 0;
+int nmhashSearch = 0;
+int nprehashSearch = 0;
+int nRemoteSend = 0;
void printhex(unsigned char *, int);
plistnode_t *createPiles(transrecord_t *);
fd=startlistening();
udpfd = udpInit();
+ pthread_attr_init(&attr);
+ pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
pthread_create(&udp_thread_Listen, &attr, udpListenBroadcast, (void*)udpfd);
if (master) {
- pthread_attr_init(&attr);
- pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
pthread_create(&thread_Listen, &attr, dstmListen, (void*)fd);
return 1;
} else {
return NULL;
}
tmp->cache = objstrCreate(1048576);
- tmp->lookupTable = chashCreate(HASH_SIZE, LOADFACTOR);
+ tmp->lookupTable = chashCreate(CHASH_SIZE, CLOADFACTOR);
#ifdef COMPILER
tmp->revertlist=NULL;
#endif
return NULL;
}
- if((objheader = (objheader_t *)chashSearch(record->lookupTable, oid)) != NULL){
+ if((objheader = chashSearch(record->lookupTable, oid)) != NULL){
+#ifdef TRANSSTATS
+ nchashSearch++;
+#endif
/* Search local transaction cache */
#ifdef COMPILER
return &objheader[1];
return objheader;
#endif
} else if ((objheader = (objheader_t *) mhashSearch(oid)) != NULL) {
+#ifdef TRANSSTATS
+ nmhashSearch++;
+#endif
/* Look up in machine lookup table and copy into cache*/
GETSIZE(size, objheader);
size += sizeof(objheader_t);
return objcopy;
#endif
} else if((tmp = (objheader_t *) prehashSearch(oid)) != NULL) {
+#ifdef TRANSSTATS
+ nprehashSearch++;
+#endif
/* Look up in prefetch cache */
GETSIZE(size, tmp);
size+=sizeof(objheader_t);
printf("Error: Object not found in Remote location %s, %d\n", __FILE__, __LINE__);
return NULL;
} else {
+#ifdef TRANSSTATS
+ nRemoteSend++;
+#endif
STATUS(objcopy)=0;
#ifdef COMPILER
return &objcopy[1];
if(curr->key == 0)
break;
- if ((headeraddr = chashSearch(record->lookupTable, curr->key)) == NULL) {
+ if ((headeraddr = (objheader_t *) chashSearch(record->lookupTable, curr->key)) == NULL) {
printf("Error: No such oid %s, %d\n", __FILE__, __LINE__);
return NULL;
}
thread_data_array[threadnum].replyctrl = &treplyctrl;
thread_data_array[threadnum].replyretry = &treplyretry;
thread_data_array[threadnum].rec = record;
- thread_data_array[threadnum].pilehead = pile_ptr;
/* If local do not create any extra connection */
if(pile->mid != myIpAddr) { /* Not local */
do {
if(treplyctrl == TRANS_ABORT) {
#ifdef TRANSSTATS
numTransAbort++;
+#endif
+#ifdef CHECKTB
+ char a[] = "Aborting";
+ TABORT1(a);
#endif
/* Free Resources */
objstrDelete(record->cache);
} else if(treplyctrl == TRANS_COMMIT) {
#ifdef TRANSSTATS
numTransCommit++;
+#endif
+#ifdef CHECKTB
+ char a[] = "Commiting";
+ TABORT1(a);
#endif
/* Free Resources */
objstrDelete(record->cache);
printf("transRequest(): socket create error\n");
pthread_exit(NULL);
}
-
+
/* Send bytes of data with TRANS_REQUEST control message */
send_data(sd, &(tdata->buffer->f), sizeof(fixed_data_t));
-
+
/* Send list of machines involved in the transaction */
{
int size=sizeof(unsigned int)*tdata->buffer->f.mcount;
objheader_t * header;
header = (objheader_t *) (((char *)newAddr) + offset);
oidToPrefetch = OID(header);
+#ifdef CHECKTA
+ char a[] = "object type";
+ TABORT8(__func__, a, TYPE(header));
+#endif
int size = 0;
GETSIZE(size, header);
size += sizeof(objheader_t);
offset += size;
}
}
+
recvcontrol = control;
+#ifdef CHECKTA
+ char a[] = "mid";
+ char c[] = "status";
+ TABORT5(__func__, a, c, tdata->mid, control);
+#endif
/* Update common data structure and increment count */
tdata->recvmsg[tdata->thread_id].rcv_status = recvcontrol;
pthread_cond_wait(tdata->threshold, tdata->lock);
}
pthread_mutex_unlock(tdata->lock);
+
+ if(*(tdata->replyctrl) == TRANS_COMMIT) {
+ int retval;
+ /* Update prefetch cache */
+ if((retval = updatePrefetchCache(tdata)) != 0) {
+ printf("Error: %s() in updating prefetch cache %s, %d\n", __func__, __FILE__, __LINE__);
+ return;
+ }
+
+ /* Invalidate objects in other machine cache */
+ if(tdata->buffer->f.nummod > 0) {
+ if((retval = invalidateObj(tdata)) != 0) {
+ printf("Error: %s() in invalidating Objects %s, %d\n", __func__, __FILE__, __LINE__);
+ return;
+ }
+ }
+ }
/* Send the final response such as TRANS_COMMIT or TRANS_ABORT
* to all participants in their respective socket */
/* Send Commit */
*(tdata->replyctrl) = TRANS_COMMIT;
*(tdata->replyretry) = 0;
- int retval;
- if((retval = updatePrefetchCache(tdata)) != 0) {
- printf("Error: %s() in updating prefetch cache %s, %d\n", __func__, __FILE__, __LINE__);
- return;
- }
-
- /* Invalidate objects in other machine cache */
- if(tdata->buffer->f.nummod > 0) {
- if((retval = invalidateObj(tdata)) != 0) {
- printf("Error: %s() in invalidating Objects %s, %d\n", __func__, __FILE__, __LINE__);
- return;
- }
- }
} else {
/* Send Abort in soft abort case followed by retry commiting transaction again*/
*(tdata->replyctrl) = TRANS_ABORT;
unsigned short version;
void *mobj;
objheader_t *headptr;
-
+
localtdata = (local_thread_data_array_t *) threadarg;
/* Counters and arrays to formulate decision on control message to be sent */
v_nomatch++;
/* Send TRANS_DISAGREE to Coordinator */
localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
+#ifdef CHECKTA
+ char a[] = "mid";
+ char b[] = "version mismatch";
+ char c[] = "object type";
+ TABORT7(__func__, b, a, c, localtdata->tdata->mid, TYPE(mobj));
+#endif
+ break;
}
} else {
//we're locked
v_nomatch++;
/* Send TRANS_DISAGREE to Coordinator */
localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_DISAGREE;
+#ifdef CHECKTA
+ char a[] = "mid";
+ char b[] = "version mismatch";
+ char c[] = "object type";
+ TABORT7(__func__, b, a, c, localtdata->tdata->mid, TYPE(mobj));
+#endif
+ break;
}
}
}
}
/* Condition to send TRANS_SOFT_ABORT */
if((v_matchlock > 0 && v_nomatch == 0) || (numoidnotfound > 0 && v_nomatch == 0)) {
+#ifdef CHECKTA
+ char a[] = "mid";
+ char b[] = "version mismatch";
+ char c[] = "object type";
+ TABORT7(__func__, b, a, c, localtdata->tdata->mid, TYPE(mobj));
+#endif
localtdata->tdata->recvmsg[localtdata->tdata->thread_id].rcv_status = TRANS_SOFT_ABORT;
}
if(*(localtdata->tdata->replyctrl) == TRANS_ABORT){
if(transAbortProcess(localtdata) != 0) {
printf("Error in transAbortProcess() %s,%d\n", __FILE__, __LINE__);
+ fflush(stdout);
pthread_exit(NULL);
}
} else if(*(localtdata->tdata->replyctrl) == TRANS_COMMIT) {
if(transComProcess(localtdata) != 0) {
printf("Error in transComProcess() %s,%d\n", __FILE__, __LINE__);
+ fflush(stdout);
pthread_exit(NULL);
}
}
if (localtdata->transinfo->objnotfound != NULL) {
free(localtdata->transinfo->objnotfound);
}
-
pthread_exit(NULL);
}
oid = *((unsigned int *)(recvbuffer + sizeof(char)));
/* TODO: For each object not found query DHT for new location and retrieve the object */
/* Throw an error */
- printf("OBJECT %x NOT FOUND.... THIS SHOULD NOT HAPPEN...TERMINATE PROGRAM\n", oid);
+ //printf("OBJECT %x NOT FOUND.... THIS SHOULD NOT HAPPEN...TERMINATE PROGRAM\n", oid);
// exit(-1);
} else {
printf("Error: in decoding the control value %d, %s, %d\n",control, __FILE__, __LINE__);
#else
myIpAddr = getMyIpAddr("eth0");
#endif
+
+#ifdef CHECKTA
+ printf("My ip address = %x", myIpAddr);
+#endif
myIndexInHostArray = findHost(myIpAddr);
if (myIndexInHostArray == -1)
{
if (connect(sock, (struct sockaddr *)&remoteAddr, sizeof(remoteAddr)) < 0){
printf("notifyAll():error %d connecting to %s:%d\n", errno,
inet_ntoa(remoteAddr.sin_addr), LISTEN_PORT);
- status = -1;
fflush(stdout);
+ status = -1;
} else {
bzero(msg, (1+sizeof(unsigned short) + 2*sizeof(unsigned int)));
msg[0] = THREAD_NOTIFY_RESPONSE;
projects / IRC.git / blobdiff