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