Mercurial > hg > FederatedLinda

comparison src/fdl/TupleSpace.java @ 16:cccf34386cad

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
*** empty log message ***
author kono
date Mon, 18 Aug 2008 06:17:54 +0900
parents
children 609b288f47f9
comparison
equal deleted inserted replaced
15:aced4bfc15af 16:cccf34386cad
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 }