Robust/src/ClassLibrary/JavaDSM/Work.java

   1 public class Work extends Thread {
   2   Task tasks;
   3         Object[] currentWorkList;
   4         int MY_MID;
   5         int NUM_THREADS;
   6
   7         Work (Task tasks, int num_threads, int mid, Object[] currentWorkList) {
   8                 this.tasks = tasks;
   9                 this.currentWorkList = currentWorkList;
  10                 NUM_THREADS = num_threads;
  11                 MY_MID = mid;
  12         }
  13
  14         public void run() {
  15     int workMID;
  16     atomic {
  17       workMID = MY_MID;
  18     }
  19
  20     int chk;
  21     int result;
  22     int i,j;
  23     int cc;
  24                 boolean isEmpty;
  25
  26     while(true) {
  27       atomic {
  28                                         isEmpty = tasks.isTodoListEmpty();              // flag > !keep assigning
  29
  30           if (!isEmpty) {
  31                               atomic {
  32                                 currentWorkList[workMID] = tasks.grabTask();    /* grab the work from work pool */
  33                                 }
  34             chk = 1;
  35           }
  36                         else {
  37             chk = Work.checkCurrentWorkList(this);
  38         }
  39       }
  40
  41       if(chk == 1) {    // still have work
  42         atomic {
  43           /* compute */
  44           tasks.execute(currentWorkList[workMID]);
  45           /* push into done list */
  46           tasks.done(currentWorkList[workMID]);
  47                                         currentWorkList[workMID] = null;
  48         }
  49
  50         atomic {
  51           cc = ((Drinker)tasks).ownTotal;
  52         }
  53
  54 //        System.out.println("CC = " + cc);
  55 //        sleep(1000000);
  56       }
  57       else if(chk  == -1) {    // finished all work
  58         break;
  59       }
  60       else {    // wait for other thread
  61                                 sleep(5000000);
  62       }
  63
  64     }
  65
  66     /* for debugging purpose */
  67     atomic {
  68       System.out.println("\n\nDoneSize = " + tasks.doneList.size());
  69     }
  70     System.out.println("\n\n\n I'm done");
  71
  72   }
  73
  74         public static int checkCurrentWorkList(Work mywork) {
  75     int i;
  76     int index = -1;
  77     int myID;
  78                 int num_threads;
  79     int status;
  80     boolean chk = false;
  81     Segment s;
  82
  83                 atomic {
  84             myID = mywork.MY_MID;
  85                         num_threads = mywork.NUM_THREADS;
  86                 }
  87
  88     for(i = 0 ; (i < num_threads) && (index < 0); i++) {
  89       if(myID == i) {
  90         continue;
  91       }
  92                         status = Thread.getStatus(i);
  93
  94       atomic {
  95
  96         s = (Segment)mywork.currentWorkList[i];
  97
  98         if(status == -1 && null != s) {
  99           mywork.currentWorkList[myID] = mywork.currentWorkList[i];
 100           mywork.currentWorkList[i] = null;
 101           index = 0;
 102         }
 103         else if(null != s) {
 104           chk = true;
 105         }
 106       }
 107
 108     }
 109
 110     if(index == 0)  // grabbed dead machine's work
 111       return 1;
 112     else if(i == num_threads && index < 0 && chk != true)  // wait for other machine's work
 113       return -1;
 114     else
 115       return 0; // others are still working wait until they finish work
 116   }
 117 }
 118