abortreaders option finished

diff --git a/Robust/src/Runtime/DSTM/interface/abortreaders.c b/Robust/src/Runtime/DSTM/interface/abortreaders.c
index a71aae2d8f37ba57dae47457b4e145135b1e5ea9..73e9033abf4eb3047228c6b6a8c93171ca1904bb 100644 (file)
--- a/Robust/src/Runtime/DSTM/interface/abortreaders.c
+++ b/Robust/src/Runtime/DSTM/interface/abortreaders.c
@@ -73,7 +73,87 @@ void removetransaction(unsigned int oidarray[], unsigned int numoids) {
   pthread_mutex_unlock(&aborttablelock);
 }
 
-void removeaborttransaction(unsigned int oidarray[], unsigned int numoids, struct transrecord * trans) {
+void removethisreadtransaction(unsigned char* oidverread, unsigned int numoids, struct transrecord * trans) {
+  int i,j;
+  pthread_mutex_lock(&aborttablelock);
+  for(i=0;i<numoids;i++) {
+    unsigned int oid=*((unsigned int *)oidverread);
+    struct readerlist * rl=chashSearch(aborttable, oid);
+    struct readerlist *first=rl;
+    oidverread+=(sizeof(unsigned int)+sizeof(unsigned short));
+    while(1) {
+      for(j=0;j<READERSIZE;j++) {
+       if (rl->array[j]==trans) {
+         rl->array[j]=NULL;
+         if ((--rl->numreaders)==0) {
+           if (first==rl) {
+             chashRemove2(table, oid);
+             if (rl->next!=NULL) 
+               chashInsert(table, oid, rl->next);
+             rl->next=freelist;
+             freelist=rl;
+           } else {
+             first->next=rl->next;
+             rl->next=freelist;
+             freelist=rl;
+           }
+         }
+         goto nextitem;
+       }
+      }
+      first=rl;
+      rl=rl->next;
+    }
+  nextitem:
+  }
+  pthread_mutex_unlock(&aborttablelock);
+}
+
+void removetransactionhash(chashtable_t *table, struct transrecord *trans) {
+  chashlistnode_t *ptr=table->table;
+  unsigned int size=table->size;
+  int i,j;
+  pthread_mutex_lock(&aborttablelock);
+  for(i=0;i<size;i++) {
+    chashlistnode_t *curr=&ptr[i];
+    do {
+      unsigned int oid=curr->key;
+      if (oid==0)
+       break;
+      struct readerlist * rl=chashSearch(aborttable, oid);
+      struct readerlist *first=rl;
+      while(1) {
+       for(j=0;j<READERSIZE;j++) {
+         if (rl->array[j]==trans) {
+           rl->array[j]=NULL;
+           if ((--rl->numreaders)==0) {
+             if (first==rl) {
+               chashRemove2(table, oid);
+               if (rl->next!=NULL) 
+                 chashInsert(table, oid, rl->next);
+               rl->next=freelist;
+               freelist=rl;
+             } else {
+               first->next=rl->next;
+               rl->next=freelist;
+               freelist=rl;
+             }
+           }
+           goto nextitem;
+         }
+       }
+       first=rl;
+       rl=rl->next;
+      }
+    nextitem:
+      curr=curr->next;
+    } while(curr!=NULL);
+  }
+  pthread_mutex_unlock(&aborttablelock);
+}
+
+
+void removethistransaction(unsigned int oidarray[], unsigned int numoids, struct transrecord * trans) {
   int i,j;
   pthread_mutex_lock(&aborttablelock);
   for(i=0;i<numoids;i++) {

diff --git a/Robust/src/Runtime/DSTM/interface/abortreaders.h b/Robust/src/Runtime/DSTM/interface/abortreaders.h
index c0115c10af9f9f9cee220b4172ee69144580b395..a3801cc230fca006c4cc3331140aee04e639e345 100644 (file)
--- a/Robust/src/Runtime/DSTM/interface/abortreaders.h
+++ b/Robust/src/Runtime/DSTM/interface/abortreaders.h
@@ -13,6 +13,7 @@ struct readerlist {
 void initreaderlist();
 void addtransaction(unsigned int oid, struct transrecord * trans);
 void removetransaction(unsigned int oidarray[], unsigned int numoids);
-void removeaborttransaction(unsigned int oidarray[], unsigned int numoids, struct transrecord * trans);
-
+void removethistransaction(unsigned int oidarray[], unsigned int numoids, struct transrecord * trans);
+void removethisreadtransaction(unsigned char* oidverread, unsigned int numoids, struct transrecord * trans);
+void removetransactionhash(chashtable_t *table, struct transrecord *trans);
 #endif

diff --git a/Robust/src/Runtime/DSTM/interface/addUdpEnhance.c b/Robust/src/Runtime/DSTM/interface/addUdpEnhance.c
index 8280d15f08517e8f2fb88682db4b99ec34783394..5c9363a0f0e82080eac4fd5ae5a75b1133d11be8 100644 (file)
--- a/Robust/src/Runtime/DSTM/interface/addUdpEnhance.c
+++ b/Robust/src/Runtime/DSTM/interface/addUdpEnhance.c
@@ -6,6 +6,9 @@
 #include <netinet/tcp.h>
 #include "addUdpEnhance.h"
 #include "prelookup.h"
+#ifdef ABORTREADERS
+#include "abortreaders.h"
+#endif
 
 /************************
  * Global Variables *
@@ -197,6 +200,9 @@ int invalidateFromPrefetchCache(char *buffer) {
     int numObjsRecv = *((short *)(buffer+offset)) / sizeof(unsigned int);
     offset += sizeof(short);
     int i;
+#ifdef ABORTREADERS
+    removetransaction((unsigned int *)(buffer+offset), numObjsRecv);
+#endif
     for(i = 0; i < numObjsRecv; i++) {
       unsigned int oid;
       oid = *((unsigned int *)(buffer+offset));

diff --git a/Robust/src/Runtime/DSTM/interface/dstm.h b/Robust/src/Runtime/DSTM/interface/dstm.h
index d501be66291c852108d50e5d1a96f02f8e3a02c5..356e4490212cc06f1cf4af2e941c288d42a4660a 100644 (file)
--- a/Robust/src/Runtime/DSTM/interface/dstm.h
+++ b/Robust/src/Runtime/DSTM/interface/dstm.h
@@ -282,7 +282,7 @@ void addHost(unsigned int);
 void mapObjMethod(unsigned short);
 
 void randomdelay();
-transrecord_t *transStart();
+__attribute__((malloc)) transrecord_t *transStart();
 __attribute__((pure)) objheader_t *transRead(transrecord_t *, unsigned int);
 objheader_t *transCreateObj(transrecord_t *, unsigned int); //returns oid header
 int transCommit(transrecord_t *record); //return 0 if successful

diff --git a/Robust/src/Runtime/DSTM/interface/trans.c b/Robust/src/Runtime/DSTM/interface/trans.c
index 0b64b92389e0c14b4963068c1e940c57824270bb..85564b8b05127ad1449518bbb7c49598ed723c7a 100644 (file)
--- a/Robust/src/Runtime/DSTM/interface/trans.c
+++ b/Robust/src/Runtime/DSTM/interface/trans.c
@@ -320,7 +320,7 @@ void randomdelay() {
 }
 
 /* This function initializes things required in the transaction start*/
-transrecord_t *transStart() {
+__attribute__((malloc)) transrecord_t *transStart() {
   transrecord_t *tmp;
   if((tmp = calloc(1, sizeof(transrecord_t))) == NULL) {
     printf("%s() Calloc error at line %d, %s\n", __func__, __LINE__, __FILE__);
@@ -328,9 +328,9 @@ transrecord_t *transStart() {
   }
   tmp->cache = objstrCreate(1048576);
   tmp->lookupTable = chashCreate(CHASH_SIZE, CLOADFACTOR);
-#ifdef COMPILER
-  tmp->revertlist=NULL;
-#endif
+  //#ifdef COMPILER
+  //  tmp->revertlist=NULL; //Not necessary...already null
+  //#endif
   return tmp;
 }
 
@@ -741,6 +741,8 @@ int transCommit(transrecord_t *record) {
            free(listmid);
            return 1;
          }
+
+
          /* Invalidate objects in other machine cache */
          if(tosend[i].f.nummod > 0) {
            if((retval = invalidateObj(&(tosend[i]))) != 0) {
@@ -750,12 +752,31 @@ int transCommit(transrecord_t *record) {
              return 1;
            }
          }
+#ifdef ABORTREADERS
+         removetransaction(tosend[i].oidmod,tosend[i].f.nummod);
+         removethisreadtransaction(tosend[i].oidmod,tosend[i].f.numread, record);
+#endif
+       }
+#ifdef ABORTREADERS
+       else if (!treplyretry) {
+         removethistransaction(tosend[i].oidmod,tosend[i].f.nummod,record);
+         removethisreadtransaction(tosend[i].oidmod,tosend[i].f.numread,record);
        }
+#endif
 #endif
        send_data(sd, &finalResponse, sizeof(char));
       } else {
        /* Complete local processing */
        doLocalProcess(finalResponse, &(tosend[i]), &transinfo, record);
+#ifdef ABORTREADERS
+       if(finalResponse == TRANS_COMMIT) {
+         removetransaction(tosend[i].oidmod,tosend[i].f.nummod);
+         removethisreadtransaction(tosend[i].oidmod,tosend[i].f.numread, record);
+       } else if (!treplyretry) {
+         removethistransaction(tosend[i].oidmod,tosend[i].f.nummod,record);
+         removethisreadtransaction(tosend[i].oidmod,tosend[i].f.numread,record);
+       }
+#endif
       }
     }
 
@@ -1779,6 +1800,9 @@ int notifyAll(threadlist_t **head, unsigned int oid, unsigned int version) {
 }
 
 void transAbort(transrecord_t *trans) {
+#ifdef ABORTREADERS
+  removetransactionhash(trans->lookupTable, trans);
+#endif
   objstrDelete(trans->cache);
   chashDelete(trans->lookupTable);
   free(trans);

diff --git a/Robust/src/buildscript b/Robust/src/buildscript
index bd4928ed58a52a641033fc7201b9c638175bae31..bde78a2b9a98585155eb007c93282b2779cc7dad 100755 (executable)
--- a/Robust/src/buildscript
+++ b/Robust/src/buildscript
@@ -50,6 +50,7 @@ echo -justanalyze exit after compiler analyses complete
 echo -help help
 }
 
+ABORTREADERS=false;
 ROBUSTROOT=~/research/Robust/src
 DSMRUNTIME=$ROBUSTROOT/Runtime/DSTM/interface/
 REPAIRROOT=~/research/Repair/RepairCompiler/
@@ -98,6 +99,7 @@ then
 EXITAFTERANALYSIS=true
 elif [[ $1 = '-abortreaders' ]]
 then
+ABORTREADERS=true
 EXTRAOPTIONS="$EXTRAOPTIONS -DABORTREADERS"
 JAVAOPTS="$JAVAOPTS -abortreaders"
 elif [[ $1 = '-robustroot' ]]
@@ -426,6 +428,11 @@ fi
 FILES="$FILES $DSMRUNTIME/trans.c $DSMRUNTIME/mcpileq.c $DSMRUNTIME/objstr.c $DSMRUNTIME/dstm.c $DSMRUNTIME/mlookup.c $DSMRUNTIME/clookup.c $DSMRUNTIME/llookup.c $DSMRUNTIME/threadnotify.c $DSMRUNTIME/dstmserver.c $DSMRUNTIME/plookup.c $DSMRUNTIME/ip.c $DSMRUNTIME/queue.c $DSMRUNTIME/prelookup.c $DSMRUNTIME/machinepile.c $ROBUSTROOT/Runtime/localobjects.c $ROBUSTROOT/Runtime/thread.c $DSMRUNTIME/sockpool.c $DSMRUNTIME/addUdpEnhance.c $DSMRUNTIME/signal.c $DSMRUNTIME/gCollect.c $DSMRUNTIME/addPrefetchEnhance.c $DSMRUNTIME/dsmlock.c $DSMRUNTIME/prefetch.c"
 fi
 
+if $ABORTREADERS
+then
+FILES="$FILES $DSMRUNTIME/abortreaders.c"
+fi
+
 if $FASTCHECK
 then
 FILES="$FILES $ROBUSTROOT/Runtime/localobjects.c"