distri-mapreduce/TaskServerHeartThread.java at master · ruobing-wang/distri-mapreduce · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
package map_reduce_demo;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;

public class TaskServerHeartThread extends Thread{
	TaskClientState tcs;
	WorkerCount wc;
	WorkState ws;
	TaskHeart th;
	public TaskServerHeartThread(TaskClientState Tcs,WorkState Ws,WorkerCount w){
		tcs = Tcs;
		ws = Ws;
		wc = w;
	}
	public void run(){
		try {
			DataInputStream dis = new DataInputStream(tcs.GetSocket().getInputStream());
			while(true){
				int x = dis.readInt();
				synchronized(tcs){
					if(x==1){
						synchronized(tcs){
							tcs.SetasReady();
						}
					}
					else if(x==2){
						synchronized(tcs){
							tcs.SetasBusy();
						}
					}
					else if(x==3){
		        		final int buffersize = 4096 * 5;
		    			long size = dis.readLong();
		    			byte[] buf = new byte[buffersize];
		    			File f = new File("bible" + (wc.getmt()-1) + ".mpd");
		    			DataOutputStream bos = new DataOutputStream(new FileOutputStream(f) );
		    			while(true){
		    				int read;
		    				read = dis.read(buf,0,min(size,buffersize));
		    				size = size - read;
		    				bos.write(buf,0,read);
		    				if(size <= 0 ){
		    					break;
		    				}
		    			bos.flush();
		    			}
		    			bos.close();
						synchronized(tcs){
							tcs.SubaMW();
							if(tcs.getmwc() == 0){
								tcs.SetasReady();
							}
						}
						synchronized(wc){
							wc.Submt();
							if(wc.misEmpty()){
								ws.SetasRReady();
								synchronized(ws){
									ws.notifyAll();
								}
							}
						}
					}
					else if(x==4){
						synchronized(tcs){
							tcs.SetasBusy();
						}
					}
					else if(x==5){
		        		final int buffersize = 4096 * 5;
		    			long size = dis.readLong();
		    			byte[] buf = new byte[buffersize];
		    			File f = new File("bible" + (wc.getrt()-1) + ".rdc");
		    			DataOutputStream bos = new DataOutputStream(new FileOutputStream(f) );
		    			while(true){
		    				int read;
		    				read = dis.read(buf,0,min(size,buffersize));
		    				size = size - read;
		    				bos.write(buf,0,read);
		    				if(size <= 0 ){
		    					break;
		    				}
		    			bos.flush();
		    			}
		    			bos.close();
						synchronized(tcs){
							tcs.SubaRW();
							if(tcs.getmwc() == 0){
								tcs.SetasReady();
							}
						}
						synchronized(wc){
							wc.Subrt();
							if(wc.misEmpty()){
								ws.SetasFinished();
								synchronized(ws){
									ws.notifyAll();
								}
							}
						}
					}
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	private int min(long size,int y) {
		if(size<y)
			return (int) size;
		else
			return y;
	}
}