bug fixes

[IRC.git] / Robust / src / Runtime / thread.c

diff --git a/Robust/src/Runtime/thread.c b/Robust/src/Runtime/thread.c
index 235b4ed930feb21fbd20ecac8d33bf9a7a369068..4dc88de3f10ac4a9fa1ff9dd6d6d4f6a35e073eb 100644 (file)
--- a/Robust/src/Runtime/thread.c
+++ b/Robust/src/Runtime/thread.c
@@ -6,6 +6,8 @@
 #include "thread.h"
 #include "option.h"
 #include <signal.h>
+#include <DSTM/interface/dstm.h>
+#include <DSTM/interface/llookup.h>
 
 #include <stdio.h>
 int threadcount;
@@ -15,8 +17,16 @@ pthread_cond_t gccond;
 pthread_mutex_t objlock;
 pthread_cond_t objcond;
 pthread_key_t threadlocks;
+pthread_mutex_t threadnotifylock;
+pthread_cond_t threadnotifycond;
+pthread_key_t oidval;
 
 void threadexit() {
+  objheader_t* ptr;
+  void *value;
+  transrecord_t * trans;
+  unsigned int oidvalue;
+
 #ifdef THREADS
   struct ___Object___ *ll=pthread_getspecific(threadlocks);
   while(ll!=NULL) {
@@ -35,6 +45,23 @@ void threadexit() {
   threadcount--;
   pthread_cond_signal(&gccond);
   pthread_mutex_unlock(&gclistlock);
+#ifdef DSTM
+  /* Add transaction to check if thread finished for join operation */
+  value = pthread_getspecific(oidval);
+  oidvalue = *((unsigned int *)value);
+  goto transstart;
+transstart:
+  {
+    transrecord_t * trans = transStart();
+    ptr = transRead(trans, oidvalue);
+    struct ___Thread___ *p = (struct ___Thread___ *) ptr;
+    p->___threadDone___ = 1;
+    *((unsigned int *)&((struct ___Object___ *) p)->___localcopy___) |=DIRTY;
+    if(transCommit(trans) != 0) {
+      goto transstart;
+    }
+  }
+#endif 
   pthread_exit(NULL);
 }
 
@@ -98,6 +125,54 @@ void CALL11(___Thread______sleep____J, long long ___millis___, long long ___mill
 #endif
 }
 
+#ifdef DSTM
+/* Add thread join capability */
+void CALL01(___Thread______join____, struct ___Thread___ * ___this___) {
+  unsigned int *oidarray;
+  unsigned short *versionarray, version;
+  transrecord_t *trans;
+  objheader_t *ptr;
+  /* Add transaction to check if thread finished for join operation */
+  goto transstart;
+transstart:
+  trans = transStart();
+  ptr = transRead(trans, (unsigned int) VAR(___this___));
+  struct ___Thread___ *p = (struct ___Thread___ *) ptr;
+  if(p->___threadDone___ == 1) {
+         transAbort(trans);
+         return;
+  } else {
+         version = (ptr-1)->version;
+         if((oidarray = calloc(1, sizeof(unsigned int))) == NULL) {
+                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
+                 return;
+         }
+
+         oidarray[0] = (unsigned int) VAR(___this___);
+
+         if((versionarray = calloc(1, sizeof(unsigned short))) == NULL) {
+                 printf("Calloc error %s, %d\n", __FILE__, __LINE__);
+                 free(oidarray);
+                 return;
+         }
+         versionarray[0] = version;
+         /* Request Notification */
+#ifdef PRECISE_GC
+         struct listitem *tmp=stopforgc((struct garbagelist *)___params___);
+#endif
+         reqNotify(oidarray, versionarray, 1); 
+#ifdef PRECISE_GC
+         restartaftergc(tmp);
+#endif
+         free(oidarray);
+         free(versionarray);
+         transAbort(trans);
+         goto transstart;
+  }
+  return;
+}
+#endif
+
 #ifdef THREADS
 void CALL01(___Thread______nativeCreate____, struct ___Thread___ * ___this___) {
   pthread_t thread;
@@ -127,7 +202,15 @@ void CALL12(___Thread______start____I, int ___mid___, struct ___Thread___ * ___t
 #endif
 
 #ifdef DSTM
+void globalDestructor(void *value) {
+       free(value);
+       pthread_setspecific(oidval, NULL);
+}
+
 void initDSMthread(int *ptr) {
+  objheader_t *tmp;    
+  transrecord_t * trans;
+  void *threadData;
   int oid=ptr[0];
   int type=ptr[1];
   free(ptr);
@@ -137,25 +220,42 @@ void initDSMthread(int *ptr) {
 #else
   ((void (*)(void *))virtualtable[type*MAXCOUNT+RUNMETHOD])(oid);
 #endif
+  threadData = calloc(1, sizeof(unsigned int));
+  *((unsigned int *) threadData) = oid;
+  pthread_setspecific(oidval, threadData);
   pthread_mutex_lock(&gclistlock);
   threadcount--;
   pthread_cond_signal(&gccond);
   pthread_mutex_unlock(&gclistlock);
+  /* Add transaction to check if thread finished for join operation */
+  goto transstart;
+transstart:
+  {
+    transrecord_t * trans = transStart();
+    tmp  = transRead(trans, (unsigned int) oid);
+    ((struct ___Thread___ *)tmp)->___threadDone___ = 1;
+    *((unsigned int *)&((struct ___Object___ *) tmp)->___localcopy___) |=DIRTY;
+    if(transCommit(trans)!= 0) {
+      goto transstart;
+    }
+  }
+  pthread_exit(NULL);
 }
 
 void startDSMthread(int oid, int objType) {
-  pthread_t thread;
-  int retval;
-  pthread_attr_t nattr;
+       pthread_t thread;
+       int retval;
+       pthread_attr_t nattr;
 
-  pthread_mutex_lock(&gclistlock);
-  threadcount++;
-  pthread_mutex_unlock(&gclistlock);
+       pthread_mutex_lock(&gclistlock);
+       threadcount++;
+       pthread_mutex_unlock(&gclistlock);
   pthread_attr_init(&nattr);
   pthread_attr_setdetachstate(&nattr, PTHREAD_CREATE_DETACHED);
   int * ptr=malloc(sizeof(int)*2);
   ptr[0]=oid;
   ptr[1]=objType;
+  pthread_key_create(&oidval, globalDestructor);
   do {
     retval=pthread_create(&thread, &nattr, (void * (*)(void *)) &initDSMthread,  ptr);
     if (retval!=0)