Add two benchmarks: MapReduce & PERT
[IRC.git] / Robust / src / Benchmarks / MapReduce / Java / Master.java
1 //package mapreduce;
2
3 /*import java.io.FileInputStream;
4 import java.io.FileOutputStream;
5 import java.util.Vector;*/
6
7 public class Master {
8
9     int m;
10     int r;
11     int[] mworkerStates; // array of map worker's state
12     // 0: idle  1: process  2: finished 3: fail
13     int[] rworkerStates; // array of reduce worker's state
14     Vector[] interoutputs; // array of string vector containing
15     // paths of intermediate outputs from
16     // map worker
17
18     Splitter splitter;
19
20     String outputfile;  // path of final output file
21
22     boolean partial;
23
24     public Master(int m, int r, Splitter splitter) {
25         this.m = m;
26         this.r = r;
27
28         mworkerStates = new int[m];
29         rworkerStates = new int[r];
30         for(int i = 0; i < m; ++i) {
31             mworkerStates[i] = 0;
32         }
33         for(int i = 0; i < r; ++i) {
34             rworkerStates[i] = 0;
35         }
36
37         interoutputs = new Vector[r];
38         for(int i = 0; i < r; ++i) {
39             interoutputs[i] = null;
40         }
41
42         this.splitter = splitter;
43         this.outputfile = new String("/home/jzhou/mapreduce/output.dat");
44
45         this.partial = false;
46     }
47
48     public int getR() {
49         return this.r;
50     }
51
52     public String getOutputFile() {
53         return this.outputfile;
54     }
55
56     public boolean isPartial() {
57         return this.partial;
58     }
59
60     public void setPartial(boolean partial) {
61         this.partial = partial || this.partial;
62     }
63
64     public void split() {
65         splitter.split();
66     }
67
68     public MapWorker[] assignMap() {
69         String[] contentsplits = splitter.getSlices();
70         MapWorker[] mworkers = new MapWorker[contentsplits.length];
71         for(int i = 0; i < contentsplits.length; ++i) {
72             //System.printString("*************************\n");
73             //System.printString(contentsplits[i] + "\n");
74             //System.printString("*************************\n");
75             MapWorker mworker = new MapWorker(splitter.getFilename(), contentsplits[i], r, i);
76             mworkerStates[i] = 1;
77             mworkers[i] = mworker;
78         }
79         return mworkers;
80     }
81
82     public void setMapFinish(int i) {
83         mworkerStates[i] = 2;
84     }
85
86     public void setMapFail(int i) {
87         mworkerStates[i] = 3;
88     }
89
90     public boolean isMapFinish() {
91         for(int i = 0; i < mworkerStates.length; ++i) {
92             if(mworkerStates[i] == 1) {
93                 return false;
94             }
95         }
96
97         return true;
98     }
99
100     public void addInterOutput(String interoutput) {
101         int start = interoutput.indexOf('_');
102         int end = interoutput.indexOf('.');
103         int index = Integer.parseInt(interoutput.substring(start + 1, end));
104         //System.printString(interoutput.subString(start + 1, end) + "\n");
105         if(interoutputs[index] == null) {
106             interoutputs[index] = new Vector();
107         }
108         interoutputs[index].addElement(interoutput);
109     }
110
111     public ReduceWorker[] assignReduce() {
112         ReduceWorker[] rworkers = new ReduceWorker[interoutputs.length];
113         for(int i = 0; i < interoutputs.length; ++i) {
114             ReduceWorker rworker = new ReduceWorker(interoutputs[i], i);
115             rworkerStates[i] = 1;
116             rworkers[i] = rworker;
117         }
118         return rworkers;
119     }
120
121     public void setReduceFinish(int i) {
122         rworkerStates[i] = 2;
123     }
124
125     public void setReduceFail(int i) {
126         rworkerStates[i] = 3;
127     }
128
129     public boolean isReduceFinish() {
130         for(int i = 0; i < rworkerStates.length; ++i) {
131             if(rworkerStates[i] == 1) {
132                 return false;
133             }
134         }
135
136         return true;
137     }
138
139     public void collectROutput(String file) {
140         //try{
141             FileInputStream iStream = new FileInputStream(file);
142             FileOutputStream oStream = new FileOutputStream(outputfile, true);
143             byte[] b = new byte[1024 * 100];
144             int length = iStream.read(b);
145             if(length < 0) {
146                 System./*out.println*/printString("Error! Can not read from intermediate output file from reduce worker: " + file + "\n");
147                 System.exit(-1);
148             }
149             //System.printString(new String(b, 0, length) + "\n");
150             oStream.write(b, 0, length);
151             iStream.close();
152             oStream.close();
153         /*} catch(Exception e) {
154             e.printStackTrace();
155             System.exit(-1);
156         }*/
157     }
158 }