|
16
|
1 package fdl;
|
|
|
2
|
|
|
3 import java.io.IOException;
|
|
|
4 import java.nio.ByteBuffer;
|
|
|
5 import java.nio.CharBuffer;
|
|
|
6 import java.nio.channels.SelectionKey;
|
|
|
7 import java.nio.channels.SocketChannel;
|
|
|
8 import java.nio.charset.CharacterCodingException;
|
|
|
9 import java.nio.charset.Charset;
|
|
|
10 import java.nio.charset.CharsetDecoder;
|
|
|
11
|
|
|
12 public class TupleSpace implements PSXQueueInterface{
|
|
|
13 static final boolean debug = true;
|
|
|
14 static final int CAPSIZE = 4194304;
|
|
|
15 public Tuple[] tuple_space;
|
|
|
16
|
|
|
17 public TupleSpace(Tuple[] _tuple_space) {
|
|
|
18 super();
|
|
|
19 // 読みこんだデータを格納するためのリストの初期化
|
|
|
20 tuple_space = _tuple_space;
|
|
|
21 }
|
|
|
22
|
|
|
23 public TupleSpace() {
|
|
|
24 // TODO Auto-generated constructor stub
|
|
|
25 }
|
|
|
26
|
|
|
27 protected String Out(ByteBuffer command, ByteBuffer data) throws IOException {
|
|
|
28 Tuple tmpTuple;
|
|
|
29 int id;
|
|
|
30 int datasize;
|
|
|
31 char idc = (char)command.getShort(LINDA_ID_OFFSET);
|
|
|
32 command.rewind();
|
|
|
33 id = (int)idc;
|
|
|
34 String sendtext = "none";
|
|
|
35
|
|
|
36 datasize = command.getInt(LINDA_DATA_LENGTH_OFFSET);
|
|
|
37 command.rewind();
|
|
|
38
|
|
|
39 System.out.println("*** out command : id = "+id +" ***");
|
|
|
40
|
|
|
41 while((tuple_space[id] != null) &&
|
|
|
42 ((tuple_space[id].mode == PSX_WAIT_RD)||(tuple_space[id].mode == PSX_RD))) {
|
|
|
43 command.put(LINDA_MODE_OFFSET, (byte)'a');
|
|
|
44 command.rewind();
|
|
|
45 command.putInt(LINDA_SEQ_OFFSET, tuple_space[id].getSeq());
|
|
|
46 command.rewind();
|
|
|
47 //if(debug){
|
|
|
48 //int sendsize = tmpTuple.getdataLength()+LINDA_HEADER_SIZE;
|
|
|
49 //System.out.println("send size "+sendsize+" : mode = "+(char)mode);
|
|
|
50 //}
|
|
|
51 //ByteBuffer sendcommand = tmpTuple.getCommand();
|
|
|
52 //ByteBuffer senddata = tmpTuple.getData();
|
|
|
53 send(tuple_space[id].ch, command, data);
|
|
|
54
|
|
|
55 sendtext = getdataString(data);
|
|
|
56
|
|
|
57
|
|
|
58 //後処理
|
|
|
59 tmpTuple = tuple_space[id];
|
|
|
60 tuple_space[id] = tmpTuple.next;
|
|
|
61 tmpTuple = null;
|
|
|
62 }
|
|
|
63 if(tuple_space[id] != null && tuple_space[id].mode == PSX_IN) {
|
|
|
64 command.put(LINDA_MODE_OFFSET, (byte)'a');
|
|
|
65 command.rewind();
|
|
|
66 command.putInt(LINDA_SEQ_OFFSET, tuple_space[id].getSeq());
|
|
|
67 command.rewind();
|
|
|
68
|
|
|
69 if(debug){
|
|
|
70 int sendsize = datasize+LINDA_HEADER_SIZE;
|
|
|
71 System.out.println("send size "+sendsize+" : mode = "+(char)'a');
|
|
|
72 }
|
|
|
73 //ByteBuffer sendcommand = tmpTuple.getCommand();
|
|
|
74 //ByteBuffer senddata = tmpTuple.getData();
|
|
|
75 send(tuple_space[id].ch, command, data);
|
|
|
76
|
|
|
77 sendtext = getdataString(data);
|
|
|
78
|
|
|
79 //後処理
|
|
|
80 tmpTuple = tuple_space[id];
|
|
|
81 tuple_space[id] = tmpTuple.next;
|
|
|
82 tmpTuple = null;
|
|
|
83 } else if ((tuple_space[id] == null)|| (tuple_space[id].getMode() == PSX_OUT)) {
|
|
|
84 if((tmpTuple = tuple_space[id]) == null) {
|
|
|
85 tmpTuple = tuple_space[id] = new Tuple();
|
|
|
86 tmpTuple.next = null;
|
|
|
87 }
|
|
|
88 else {
|
|
|
89 while(tmpTuple.next != null) tmpTuple = tmpTuple.next;
|
|
|
90 tmpTuple.next = new Tuple();
|
|
|
91 tmpTuple = tmpTuple.next;
|
|
|
92 tmpTuple.next = null;
|
|
|
93 }
|
|
|
94
|
|
|
95 tmpTuple.setMode('o');
|
|
|
96 int seq = command.getInt(LINDA_SEQ_OFFSET);
|
|
|
97 command.rewind();
|
|
|
98 tmpTuple.setSeq(seq);
|
|
|
99 tmpTuple.setData(data);
|
|
|
100 tmpTuple.setDataLength(datasize);
|
|
|
101 if(debug){
|
|
|
102 System.out.println("data inserted len = "+tmpTuple.getdataLength()+" : id = "+id);
|
|
|
103 }
|
|
|
104 }
|
|
|
105 else {
|
|
|
106 System.out.println("Incorrect mode :"+(char)tuple_space[id].getMode());
|
|
|
107 command.clear();
|
|
|
108 data.clear();
|
|
|
109 System.exit(1);
|
|
|
110 }
|
|
|
111 return sendtext;
|
|
|
112 }
|
|
|
113
|
|
|
114 protected void Wait_Rd(SelectionKey key, ByteBuffer command, int mode) {
|
|
|
115 Tuple tmpTuple;
|
|
|
116 int id;
|
|
|
117
|
|
|
118 char idc = (char)command.getShort(LINDA_ID_OFFSET);
|
|
|
119 command.rewind();
|
|
|
120 id = (int)idc;
|
|
|
121
|
|
|
122 if (debug)
|
|
|
123 System.out.println("*** "+(char)mode+" command : id = "+ id +" ***\n");
|
|
|
124
|
|
|
125 tmpTuple = new Tuple();
|
|
|
126 tmpTuple.setMode(mode);
|
|
|
127 int seq = command.getInt(LINDA_SEQ_OFFSET);
|
|
|
128 command.rewind();
|
|
|
129 tmpTuple.setSeq(seq);
|
|
|
130 tmpTuple.ch = (SocketChannel) key.channel();
|
|
|
131 tmpTuple.setDataLength(0);
|
|
|
132 ByteBuffer buff = ByteBuffer.allocate(0);
|
|
|
133 tmpTuple.setData(buff);
|
|
|
134 tmpTuple.next = tuple_space[id];
|
|
|
135 tuple_space[id] = tmpTuple;
|
|
|
136 if(debug){
|
|
|
137 System.out.println("data inserted insert seq = "+seq +", id = "+id);
|
|
|
138 }
|
|
|
139 }
|
|
|
140
|
|
|
141 protected String In_Rd(SelectionKey key, ByteBuffer command, int mode)
|
|
|
142 throws IOException {
|
|
|
143 Tuple tmpTuple = read_in_1(key, command, mode);
|
|
|
144
|
|
|
145 if (tmpTuple!=null) {
|
|
|
146 //send
|
|
|
147 ByteBuffer sendcommand = tmpTuple.getCommand();
|
|
|
148 ByteBuffer senddata = tmpTuple.getData();
|
|
|
149 send(key,sendcommand, senddata);
|
|
|
150 }
|
|
|
151 String sendtext = getdataString(tmpTuple.getData());
|
|
|
152 return sendtext;
|
|
|
153 }
|
|
|
154
|
|
|
155 private Tuple read_in_1(SelectionKey key, ByteBuffer command, int mode) {
|
|
|
156 Tuple tmpTuple;
|
|
|
157 int id;
|
|
|
158 //id = command.getInt(LINDA_ID_OFFSET);
|
|
|
159 //int mode = command.getInt(LINDA_MODE_OFFSET);
|
|
|
160 Tuple temp = null;
|
|
|
161
|
|
|
162 char idc = (char)command.getShort(LINDA_ID_OFFSET);
|
|
|
163 command.rewind();
|
|
|
164 id = (int)idc;
|
|
|
165
|
|
|
166
|
|
|
167 System.out.println("*** "+(char)mode+" command : id = "+ id +" ***\n");
|
|
|
168
|
|
|
169 tmpTuple = tuple_space[id];
|
|
|
170
|
|
|
171 //wを無視
|
|
|
172 while(tmpTuple != null && tmpTuple.next != null && (tmpTuple.mode == 'w')){
|
|
|
173 temp = tmpTuple;
|
|
|
174 tmpTuple = tmpTuple.next;
|
|
|
175 }
|
|
|
176
|
|
|
177 if (tmpTuple != null && (tmpTuple.mode == 'o')){
|
|
|
178 //tmpTuple = new Tuple((SocketChannel)key.channel());
|
|
|
179 int seq = command.getInt(LINDA_SEQ_OFFSET);
|
|
|
180 command.rewind();
|
|
|
181 tmpTuple.setCommand('a', seq);
|
|
|
182
|
|
|
183 if(debug){
|
|
|
184 int sendsize = tmpTuple.getdataLength()+LINDA_HEADER_SIZE;
|
|
|
185 System.out.println("send size "+sendsize+" : mode = "+(char)tmpTuple.getMode());
|
|
|
186 }
|
|
|
187
|
|
|
188
|
|
|
189
|
|
|
190 //INの場合はremoveする
|
|
|
191 if(mode == PSX_IN) {
|
|
|
192 if(tmpTuple.data != null){
|
|
|
193 //ByteBuffer buff = ByteBuffer.allocate(0);
|
|
|
194 //tmpTuple.setData(buff);
|
|
|
195 tmpTuple.data = null;
|
|
|
196 }
|
|
|
197 if(temp != null){
|
|
|
198 temp.next = tmpTuple.next;
|
|
|
199 }
|
|
|
200 else {
|
|
|
201 tuple_space[id] = tmpTuple.next;
|
|
|
202 }
|
|
|
203 }
|
|
|
204 } else {
|
|
|
205 if(tmpTuple == null) {
|
|
|
206 //ServerSocketChannel sc = (ServerSocketChannel)key.channel();
|
|
|
207 tmpTuple = tuple_space[id] = new Tuple((SocketChannel)key.channel());
|
|
|
208 tmpTuple.next = null;
|
|
|
209 }else {
|
|
|
210 while(tmpTuple.next !=null) tmpTuple =tmpTuple.next;
|
|
|
211 tmpTuple.next= new Tuple((SocketChannel)key.channel());
|
|
|
212 tmpTuple = tmpTuple.next;
|
|
|
213 tmpTuple.next = null;
|
|
|
214 }
|
|
|
215
|
|
|
216 tmpTuple.setMode(mode);
|
|
|
217 int seq2 = command.getInt(LINDA_SEQ_OFFSET);
|
|
|
218 command.rewind();
|
|
|
219 tmpTuple.setSeq(seq2);
|
|
|
220 tmpTuple.ch = (SocketChannel) key.channel();
|
|
|
221 tmpTuple.setDataLength(0);
|
|
|
222 ByteBuffer buff = ByteBuffer.allocate(0);
|
|
|
223 buff.rewind();
|
|
|
224 tmpTuple.setData(buff);
|
|
|
225 tmpTuple = null;
|
|
|
226
|
|
|
227 if(debug){
|
|
|
228 System.out.println("data inserted insert seq = "+seq2 +", id = "+id);
|
|
|
229 }
|
|
|
230 }
|
|
|
231 return tmpTuple;
|
|
|
232 }
|
|
|
233
|
|
|
234 protected String Check(SelectionKey key, ByteBuffer command) throws IOException {
|
|
|
235 String sendtext;
|
|
|
236 ByteBuffer data = check1(command);
|
|
|
237 send(key, command, data);
|
|
|
238
|
|
|
239 sendtext = getdataString(data);
|
|
|
240
|
|
|
241 return sendtext;
|
|
|
242 }
|
|
|
243
|
|
|
244 private ByteBuffer check1(ByteBuffer command) {
|
|
|
245 ByteBuffer data;
|
|
|
246 Tuple tmpTuple;
|
|
|
247 int id;
|
|
|
248 char idc = (char)command.getShort(LINDA_ID_OFFSET);
|
|
|
249 command.rewind();
|
|
|
250 id = (int)idc;
|
|
|
251
|
|
|
252 tmpTuple = tuple_space[id];
|
|
|
253 while(tmpTuple != null && tmpTuple.next != null && (tmpTuple.mode == 'w')){
|
|
|
254 tmpTuple = tmpTuple.next;
|
|
|
255 }
|
|
|
256 if (tmpTuple != null && (tmpTuple.mode == 'o')) {
|
|
|
257 command.putInt(LINDA_DATA_LENGTH_OFFSET, tmpTuple.datalen);
|
|
|
258 command.rewind();
|
|
|
259 data = tmpTuple.getData();
|
|
|
260 }else {
|
|
|
261 //means no out tuple
|
|
|
262 command.putInt(LINDA_DATA_LENGTH_OFFSET, 0);
|
|
|
263 command.rewind();
|
|
|
264 data = ByteBuffer.allocate(0);
|
|
|
265 }
|
|
|
266 return data;
|
|
|
267 }
|
|
|
268
|
|
|
269 public void send(SocketChannel ch, ByteBuffer command, ByteBuffer data)
|
|
|
270 throws IOException {
|
|
|
271 if (debug) {
|
|
|
272 if (command == null) {
|
|
|
273 System.out.println("Manager_run: command is null");
|
|
|
274 }
|
|
|
275 if (data == null) {
|
|
|
276 System.out.println("Manager_run: data is null");
|
|
|
277 }
|
|
|
278 }
|
|
|
279 int send_size = LINDA_HEADER_SIZE;
|
|
|
280 int count = 0;
|
|
|
281
|
|
|
282 //command Send
|
|
|
283 command.rewind();
|
|
|
284 while(send_size > 0){
|
|
|
285 count = ch.write(command);
|
|
|
286 if(count < 0){
|
|
|
287 System.out.println("Write Falied! close ch:"+ch);
|
|
|
288 ch.close();
|
|
|
289 return;
|
|
|
290 }
|
|
|
291 send_size -= count;
|
|
|
292 }
|
|
|
293
|
|
|
294 //data Send
|
|
|
295 data.rewind();
|
|
|
296 if(data != null) {
|
|
|
297 data.rewind();
|
|
|
298 ch.write(data);
|
|
|
299 }
|
|
|
300 }
|
|
|
301
|
|
|
302 public void send(SelectionKey key, ByteBuffer command, ByteBuffer data)
|
|
|
303 throws IOException {
|
|
|
304 SocketChannel ch = (SocketChannel)key.channel();
|
|
|
305 send(ch,command,data);
|
|
|
306 }
|
|
|
307
|
|
|
308 private String getdataString(ByteBuffer data) {
|
|
|
309 String sendtext;
|
|
|
310 data.rewind();
|
|
|
311 //set sendtext
|
|
|
312 //CharBuffer chardata = data.asCharBuffer();
|
|
|
313
|
|
|
314 //Decode UTF-8 to System Encoding(UTF-16)
|
|
|
315 Charset charset = Charset.forName("UTF-8");
|
|
|
316 CharsetDecoder decoder = charset.newDecoder();
|
|
|
317 CharBuffer cb = null;
|
|
|
318 try {
|
|
|
319 cb = decoder.decode(data);
|
|
|
320 } catch (CharacterCodingException e) {
|
|
|
321 e.printStackTrace();
|
|
|
322 }
|
|
|
323 cb.rewind();
|
|
|
324
|
|
|
325 sendtext = cb.toString();
|
|
|
326 return sendtext;
|
|
|
327 }
|
|
|
328
|
|
|
329 } |
