Mercurial > hg > RemoteEditor > REPSessionManager
annotate rep/SessionManager.java @ 370:46c95d8885ad
fix put_ack in single server test.
| author | kono |
|---|---|
| date | Tue, 21 Oct 2008 22:40:30 +0900 |
| parents | 7d21eb209e27 |
| children | 97aa54c9afe6 |
| rev | line source |
|---|---|
| 266 | 1 |
| 0 | 2 package rep; |
| 3 | |
| 4 import java.io.IOException; | |
| 5 import java.net.InetSocketAddress; | |
| 267 | 6 import java.nio.channels.ClosedChannelException; |
| 0 | 7 import java.nio.channels.SelectionKey; |
| 83 | 8 import java.util.LinkedList; |
| 144 | 9 import java.util.List; |
| 231 | 10 import java.util.Set; |
| 178 | 11 import java.util.concurrent.BlockingQueue; |
| 192 | 12 import java.util.concurrent.LinkedBlockingQueue; |
| 0 | 13 |
| 346 | 14 import org.xml.sax.SAXException; |
| 15 | |
| 198 | 16 |
| 353 | 17 |
| 337 | 18 import rep.channel.REPLogger; |
| 123 | 19 import rep.channel.REPServerSocketChannel; |
| 133 | 20 import rep.channel.REPSocketChannel; |
| 144 | 21 import rep.handler.PacketSet; |
| 146 | 22 import rep.handler.REPHandler; |
| 158 | 23 import rep.channel.REPSelector; |
| 56 | 24 import rep.xml.SessionXMLDecoder; |
| 45 | 25 import rep.xml.SessionXMLEncoder; |
| 198 | 26 import rep.channel.REPSelectionKey; |
| 264 | 27 |
| 198 | 28 /* |
| 264 | 29 +-------+--------+--------+-------+--------+---------+------+ |
| 30 | cmd | session| editor | seqid | lineno | textsiz | text | | |
| 31 | | id | id | | | | | | |
| 32 +-------+--------+--------+-------+--------+---------+------+ | |
| 33 o---------- header section (network order) ----------o | |
| 34 | |
| 35 int cmd; kind of command | |
| 36 int sid; session ID : uniqu to editing file | |
| 37 int eid; editor ID : owner editor ID = 1。Session に対して unique | |
| 308 | 38 -1 session manager command |
| 39 -2 merge command | |
| 264 | 40 int seqno; Sequence number : sequence number はエディタごとに管理 |
| 41 int lineno; line number | |
| 42 int textsize; textsize : bytesize | |
| 43 byte[] text; | |
| 198 | 44 */ |
| 1 | 45 |
| 250 | 46 public class SessionManager implements SessionManagerEventListener{ |
| 337 | 47 static public REPLogger logger = REPLogger.singleton(); |
| 48 | |
| 363 | 49 SessionList sessionList; |
| 280 | 50 private SessionManagerGUI gui; |
| 363 | 51 // Main nio.Selector of this server |
| 198 | 52 private REPSelector<REPCommand> selector; |
| 363 | 53 // Known Session Manager List, At most one parent. No parent means master. |
| 319 | 54 SessionManagerList smList; |
| 363 | 55 // Known Editor list. Connected Editor has a channel. |
| 56 // Session Manager Channel may have dummy editors. | |
| 356 | 57 EditorList editorList; |
| 363 | 58 // Commands for busy editor are kept in this queue. |
| 212 | 59 private List<PacketSet> waitingCommandInMerge; |
| 363 | 60 // Command from gui. Synchronization is required. |
| 61 private BlockingQueue<SessionManagerEvent> waitingEventQueue | |
| 62 = new LinkedBlockingQueue<SessionManagerEvent>();; | |
| 63 // host name of this server. One of connecting SocketChannel's hostname | |
| 319 | 64 String myHost; |
| 363 | 65 // Single threaded write queueu. To avoid dead lock with too many writes. |
| 317 | 66 private LinkedList<PacketSet> writeQueue = new LinkedList<PacketSet>(); |
| 336 | 67 private int receive_port; |
| 68 private int parent_port; | |
| 101 | 69 static final int DEFAULT_PORT = 8766; |
| 363 | 70 // Queue limit for debugging purpose. |
| 332 | 71 private static final int packetLimit = 200; |
| 358 | 72 |
| 363 | 73 // globalSessionID = SessionManagerID * MAXID + localSessionID |
| 358 | 74 private static final int MAXID = 10000; |
| 353 | 75 SessionXMLDecoder decoder = new SessionXMLDecoder(); |
| 358 | 76 SessionXMLEncoder encoder = new SessionXMLEncoder(); |
| 363 | 77 // SocketChannel for our parent. At most one parent is allowed. |
| 355 | 78 private Forwarder sm_join_channel; |
| 363 | 79 // Routing table for session and session manager. |
| 369 | 80 private RoutingTable routingTable = new RoutingTable(this); |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
81 private SessionManagerEvent execAfterConnect = null;; |
| 358 | 82 |
| 316 | 83 public static void main(String[] args) throws InterruptedException, IOException { |
| 84 | |
| 85 int port = DEFAULT_PORT; | |
| 86 int port_s = DEFAULT_PORT; | |
| 87 //System.setProperty("file.encoding", "UTF-8"); | |
| 88 if(args.length > 0){ | |
| 363 | 89 if (args.length!=2) { |
| 90 logger.writeLog("Usage: sessionManager our_port parent_port"); | |
| 91 return; | |
| 92 } | |
| 316 | 93 port = Integer.parseInt(args[0]); |
| 94 port_s = Integer.parseInt(args[1]); | |
| 95 } | |
| 96 SessionManager sm = new SessionManager(); | |
| 336 | 97 sm.setReceivePort(port); |
| 98 sm.setParentPort(port_s); | |
| 363 | 99 // Ok start main loop |
| 316 | 100 sm.init(port,new SessionManagerGUIimpl(sm)); |
| 101 } | |
| 102 | |
| 336 | 103 public void setReceivePort(int port) { |
| 104 receive_port = port; | |
| 105 } | |
| 106 | |
| 280 | 107 public void init(int port, SessionManagerGUI gui) throws IOException, InterruptedException { |
| 108 this.gui = gui; | |
| 109 init(port); | |
| 110 mainLoop(); | |
| 111 } | |
| 112 | |
| 113 private void init(int port) throws InterruptedException, IOException { | |
| 363 | 114 selector = REPSelector.<REPCommand>create(); |
| 186 | 115 REPServerSocketChannel<REPCommand> ssc = REPServerSocketChannel.<REPCommand>open(new REPCommandPacker()); |
| 363 | 116 ssc.configureBlocking(false); // Selector requires this |
| 117 ssc.socket().setReuseAddress(true); //reuse address 必須 | |
| 212 | 118 //getAllByNameで取れた全てのアドレスに対してbindする |
| 119 ssc.socket().bind(new InetSocketAddress(port)); | |
| 349 | 120 ssc.register(selector, SelectionKey.OP_ACCEPT, |
| 121 new Forwarder(this)); | |
| 6 | 122 |
| 358 | 123 sessionList = new SessionList(); |
| 7 | 124 smList = new SessionManagerList(); |
| 356 | 125 editorList = new EditorList(); |
| 212 | 126 waitingCommandInMerge = new LinkedList<PacketSet>(); |
| 228 | 127 |
| 215 | 128 |
| 155 | 129 } |
| 313 | 130 |
| 131 /* | |
| 363 | 132 * The main loop. |
| 133 * Check incoming events and waiting writes. | |
| 134 * Do select and call select() to check in coming packets. | |
| 313 | 135 * We wrote everything in one thread, but we can assign |
| 136 * one thread for each communication channel and GUI event. | |
| 137 */ | |
| 231 | 138 public void mainLoop() throws IOException { |
| 0 | 139 while(true){ |
| 328 | 140 checkWaitingCommandInMerge(); |
| 313 | 141 if (checkInputEvent() || |
| 328 | 142 checkWaitingWrite()) { |
| 313 | 143 // try to do fair execution for waiting task |
| 144 if(selector.selectNow() > 0) select(); | |
| 145 continue; | |
| 300 | 146 } |
| 313 | 147 // now we can wait for input packet or event |
| 233 | 148 selector.select(); |
| 144 | 149 select(); |
| 150 } | |
| 151 } | |
| 152 | |
| 363 | 153 /* |
| 154 * Synchronize GUI event in the main loop. | |
| 155 */ | |
| 313 | 156 private boolean checkInputEvent() { |
| 157 SessionManagerEvent e; | |
| 158 if((e = waitingEventQueue.poll())!=null){ | |
| 334 | 159 e.exec(this); |
| 313 | 160 return true; |
| 161 } | |
| 162 return false; | |
| 163 } | |
| 164 | |
| 363 | 165 /* |
| 166 * Write a packet during the main loop. | |
| 167 */ | |
| 313 | 168 private boolean checkWaitingWrite() throws IOException { |
| 317 | 169 PacketSet p = writeQueue.poll(); |
| 170 if (p!=null) { | |
| 327 | 171 p.channel.write(p.command); |
| 317 | 172 return true; |
| 313 | 173 } |
| 174 return false; | |
| 175 } | |
| 176 | |
| 308 | 177 /** |
| 178 * Check waiting command in merge | |
| 179 * @return true if there is a processed waiting command | |
| 180 * @throws IOException | |
| 181 */ | |
| 346 | 182 private void checkWaitingCommandInMerge() { |
| 328 | 183 List<PacketSet> w = waitingCommandInMerge; |
| 184 waitingCommandInMerge = new LinkedList<PacketSet>(); | |
| 185 for(PacketSet p: w) { | |
| 186 Editor e = p.getEditor(); | |
| 353 | 187 if(e.isMerging()) { // still merging do nothing |
| 328 | 188 waitingCommandInMerge.add(p); |
| 189 } else { | |
| 346 | 190 try { |
| 353 | 191 if (sessionManage(e, p.command)) { // we don't need this |
| 192 assert false; | |
| 193 return; | |
| 194 } | |
| 195 e.manage(p.command); | |
| 347 | 196 } catch (Exception e1) { |
| 349 | 197 // should be e.close()? |
| 346 | 198 close(p.channel); |
| 199 } | |
| 328 | 200 } |
| 201 } | |
| 202 } | |
| 363 | 203 |
| 204 /* | |
| 205 * If we have waiting write commands, further sent commands also | |
| 206 * wait to avoid out of order packet sending. | |
| 207 */ | |
| 208 public boolean hasWaitingCommand(REPSocketChannel<REPCommand>c) { | |
| 209 for(PacketSet p:waitingCommandInMerge) { | |
| 210 if (p.channel==c) { | |
| 211 return true; | |
| 212 } | |
| 213 } | |
| 214 return false; | |
| 215 } | |
| 328 | 216 |
| 363 | 217 /* |
| 218 * Close a channel in case of exception or close. | |
| 219 */ | |
| 346 | 220 private void close(REPSocketChannel<REPCommand> channel) { |
| 221 REPSelectionKey<REPCommand>key = channel.keyFor1(selector); | |
| 222 REPHandler handler = (REPHandler)key.attachment(); | |
| 223 key.cancel(); | |
| 224 handler.cancel(channel); | |
| 225 // we have to remove session/enditor | |
| 226 } | |
| 227 | |
| 228 | |
| 363 | 229 /* |
| 230 * Do select operation on the Selector. Each key has a forwarder. | |
| 231 * A forwarder can be a firstConnector, a forwarder for Session Manager | |
| 232 * or an Editor. | |
| 233 */ | |
| 144 | 234 private void select() throws IOException { |
| 231 | 235 |
| 236 Set<REPSelectionKey<REPCommand>> keys = selector.selectedKeys1(); | |
| 237 for(REPSelectionKey<REPCommand> key : keys){ | |
| 144 | 238 if(key.isAcceptable()){ |
| 363 | 239 /* |
| 240 * Incoming connection. We don't know which, editor or | |
| 241 * session manager. Assign FirstConnector to distinguish. | |
| 242 */ | |
| 199 | 243 REPSocketChannel<REPCommand> channel = key.accept(new REPCommandPacker()); |
| 337 | 244 logger.writeLog("SessionManager.select() : key.isAcceptable : channel = " + channel); |
| 363 | 245 registerChannel(channel, new FirstConnector(this)); |
| 144 | 246 channel = null; |
| 247 }else if(key.isReadable()){ | |
| 363 | 248 /* |
| 249 * Incoming packets are handled by a various forwarder. | |
| 250 * A hadler throw IOException() in case of a trouble to | |
| 251 * close the channel. | |
| 252 */ | |
| 212 | 253 REPHandler handler = (REPHandler)(key.attachment()); |
| 267 | 254 try { |
| 255 handler.handle(key); | |
| 358 | 256 } catch (IOException e) { |
| 267 | 257 key.cancel(); |
| 308 | 258 handler.cancel(key.channel1()); |
| 267 | 259 } |
| 0 | 260 } |
| 261 } | |
| 262 } | |
| 1 | 263 |
| 355 | 264 void registerChannel(REPSocketChannel<REPCommand> channel,Forwarder handler) throws IOException { |
| 2 | 265 if(channel == null) { |
| 266 return; | |
| 267 } | |
| 349 | 268 handler.setChannel(channel); |
| 2 | 269 channel.configureBlocking(false); |
| 355 | 270 channel.register(selector, SelectionKey.OP_READ, handler); |
| 2 | 271 } |
| 272 | |
| 363 | 273 /* |
| 274 * After loop detection, we give up session manager join. | |
| 275 */ | |
| 276 private void cancel_sm_join() { | |
| 364 | 277 logger.writeLog("Loop detected "+this); |
| 355 | 278 removeChannel(sm_join_channel); |
| 279 sm_join_channel=null; | |
| 280 } | |
| 281 | |
| 282 | |
| 363 | 283 private void removeChannel(Forwarder channel) { |
| 284 REPSelectionKey<REPCommand> key = channel.channel.keyFor1(selector); | |
| 355 | 285 key.cancel(); |
| 286 try { | |
| 364 | 287 channel.channel.close1(); |
| 355 | 288 } catch (IOException e) { |
| 289 } | |
| 349 | 290 } |
| 291 | |
| 320 | 292 |
| 319 | 293 void updateGUI() { |
| 212 | 294 //リストのコピーをGUIに渡す |
| 358 | 295 LinkedList<Session> sList = new LinkedList<Session>(sessionList.values()); |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
296 LinkedList<Editor> eList; |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
297 if (false) { |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
298 // local editor only |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
299 eList = new LinkedList<Editor>(); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
300 for(Editor e:editorList.values()) { |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
301 if (getSMID(e.eid)==smList.sessionManagerID()) { |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
302 eList.add(e); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
303 } |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
304 } |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
305 } else { |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
306 eList = new LinkedList<Editor>(editorList.values()); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
307 } |
| 212 | 308 //GUIに反映 |
| 309 Runnable doRun = new DoGUIUpdate(sList, eList, gui); | |
| 279 | 310 gui.invokeLater(doRun); |
| 212 | 311 } |
| 312 | |
| 83 | 313 |
| 139 | 314 |
| 353 | 315 void setMyHostName(String localHostName) { |
| 308 | 316 myHost = localHostName + receive_port; |
| 164 | 317 setHostToEditor(myHost); |
| 318 } | |
| 319 | |
| 320 private void setHostToEditor(String myHost2) { | |
| 358 | 321 for(Editor editor : editorList.values()){ |
| 322 if (editor.channel!=null) | |
| 323 editor.setHost(myHost2); | |
| 164 | 324 } |
| 76 | 325 } |
| 0 | 326 |
| 355 | 327 |
| 328 /** | |
| 363 | 329 * GUI から、呼ばれて、Session Managerに接続する。 |
| 355 | 330 * Host 名のSession Manager に SM_JOIN する。自分は、Session を持っていては |
| 331 * ならない。複数のSession Managerにjoinすることは出来ない。(NATを実装するまでは)。 | |
| 332 * @param host | |
| 333 */ | |
| 364 | 334 public void connectSession(String host, int port) { |
| 355 | 335 if (sm_join_channel!=null) return; |
| 336 if (!sessionList.isEmpty()) return; | |
| 337 if (!smList.isMaster()) return; | |
| 363 | 338 /* |
| 339 * IPv6 対応では、複数のアドレスを取って、それのすべてに接続を試す必要が | |
| 340 * ある。 | |
| 341 */ | |
| 1 | 342 InetSocketAddress addr = new InetSocketAddress(host, port); |
| 343 try { | |
| 186 | 344 REPSocketChannel<REPCommand> sessionchannel = REPSocketChannel.<REPCommand>create(new REPCommandPacker()); |
| 1 | 345 sessionchannel.connect(addr); |
| 337 | 346 while(!sessionchannel.finishConnect()); |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
347 Forwarder sm = new FirstConnector(this); |
| 355 | 348 registerChannel(sessionchannel, sm); |
| 349 | 349 sm_join(sm); |
| 1 | 350 }catch (IOException e) { |
| 351 } | |
| 352 } | |
| 364 | 353 |
| 354 public void connectSession(String host) { | |
| 355 connectSession(host,parent_port); | |
| 356 } | |
| 77 | 357 |
| 363 | 358 /** |
| 359 * channel に SMCMD_SM_JOIN command を送る。 | |
| 360 * @param channel | |
| 361 */ | |
| 349 | 362 private void sm_join(Forwarder channel){ |
| 355 | 363 sm_join_channel = channel; |
| 122 | 364 //SM_JOINコマンドを生成。 |
| 77 | 365 REPCommand command = new REPCommand(); |
| 366 command.setCMD(REP.SMCMD_SM_JOIN); | |
| 349 | 367 command.setEID(-1); // request Parent SessionManagerID |
| 368 command.setSID(-1); // request SessionManagerID | |
| 79 | 369 |
| 122 | 370 //hostnameをセット。 |
| 349 | 371 setMyHostName(channel.getLocalHostName()); |
| 82 | 372 |
| 355 | 373 String string = myHost; |
| 77 | 374 command.setString(string); |
| 375 | |
| 122 | 376 //SM_JOINコマンドを送信。 |
| 349 | 377 channel.send(command); |
| 363 | 378 // ack を受け取ったら、SessionManagerのListに追加。ここではやらない。 |
| 77 | 379 } |
| 349 | 380 |
| 361 | 381 /* |
| 382 * Select Session from Manager button | |
| 383 * selected editor is joined editor directly connected to this session | |
| 384 * manager. | |
| 385 */ | |
| 316 | 386 public void selectSession(SelectButtonEvent event) throws IOException { |
| 250 | 387 int sid = event.getSID(); |
| 358 | 388 Session session = sessionList.get(sid); |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
389 if (session==null) throw new IOException(); |
| 320 | 390 Editor editor = (Editor)event.getEditor(); |
| 324 | 391 if (editor.hasSession()) return; |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
392 // assert(getSMID(editor.eid)==smList.sessionManagerID()); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
393 // assert(editor.channel!=null); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
394 editor.setSID(sid); // mark as selected |
| 358 | 395 selectSession(sid, session, editor.getEID(), editor); |
| 396 } | |
| 397 | |
| 361 | 398 /* |
| 399 * Select Session Protocol handler | |
| 363 | 400 * called from GUI or incoming SMCMD_SELECT command. |
| 361 | 401 */ |
| 358 | 402 private void selectSession(int sid, Session session, int eid, Forwarder editor) { |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
403 if(session.hasOwner() && editor.channel!=null){ |
| 363 | 404 // we have selected session. |
| 107 | 405 REPCommand sendCommand = new REPCommand(); |
| 358 | 406 if (editor.isDirect()&&editor.getEID()==eid) { |
| 363 | 407 // Found directly connected joined editor. Send join_ack(). |
| 361 | 408 session.addForwarder(editor); |
|
359
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
409 sendUpdate(session.getSID()); |
| 358 | 410 sendCommand.setCMD(REP.SMCMD_JOIN_ACK); |
| 411 } else { | |
| 363 | 412 // We have a session, but joined editor is on the other sm. |
| 358 | 413 // SELECT_ACK is sent to the session ring to |
| 363 | 414 // find out the joined editor. |
| 358 | 415 sendCommand.setCMD(REP.SMCMD_SELECT_ACK); |
| 361 | 416 // Do not directly addForwarder(forwarder). It may be |
| 417 // shared among sessions. | |
| 418 Forwarder f = new Editor(this, false, makeID(editorList.newEid())); | |
| 419 f.setChannel(editor.channel); // incoming channel | |
| 420 f.setHost(myHost); | |
| 421 f.setSID(sid); | |
| 363 | 422 session.addForwarder(f); // f.next is set up here. |
| 358 | 423 } |
| 370 | 424 sendCommand.setSID(sid); |
| 425 sendCommand.string = session.getName(); | |
| 426 sendCommand.setEID(eid); | |
| 427 editor.send(sendCommand); | |
| 107 | 428 }else { |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
429 Forwarder next; |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
430 if (editor.channel==null) { |
| 369 | 431 // we don't have this editor, search the editor first. |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
432 next = routingTable.toSessionManager(getSMID(eid)); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
433 } else { |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
434 // session searching continue... |
| 367 | 435 next = routingTable.toSessionManager(getSMID(sid)); |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
436 // create dummy editor for this session |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
437 Forwarder f = new Editor(this, false, makeID(editorList.newEid())); |
| 369 | 438 f.setChannel(next.channel); // incoming channel |
| 367 | 439 f.setNext(f); |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
440 f.setHost(myHost); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
441 f.setSID(sid); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
442 session.setFirstForwarder(f); |
| 369 | 443 session.addForwarder(editor); |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
444 } |
| 363 | 445 // pass the select command to the next path. |
| 107 | 446 REPCommand command = new REPCommand(); |
| 447 command.setCMD(REP.SMCMD_SELECT); | |
| 448 command.setSID(sid); | |
| 358 | 449 command.setEID(eid); |
| 178 | 450 command.setString(editor.getHost()); |
| 358 | 451 next.send(command); |
| 107 | 452 } |
| 8 | 453 } |
| 122 | 454 |
| 363 | 455 /* |
| 456 * Create and send UPDATE command. | |
| 457 */ | |
|
359
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
458 private void sendUpdate(int sid) { |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
459 REPCommand command = makeREPCommandWithSessionList(REP.SMCMD_UPDATE); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
460 command.setSID(sid); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
461 command.setEID(REP.SM_EID.id); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
462 smList.sendToMaster(command); |
| 358 | 463 } |
| 464 | |
| 363 | 465 /* |
| 466 * Create new editor in this sessin manager. A dummy editor | |
| 467 * is created also. | |
| 468 */ | |
|
359
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
469 public Editor newEditor(REPSocketChannel<REPCommand> channel) { |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
470 int eid = makeID(editorList.newEid()); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
471 Editor editor = new Editor(this, eid, channel); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
472 editorList.add(editor); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
473 return editor; |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
474 } |
| 363 | 475 |
| 476 /* | |
| 477 * Create new session. | |
| 478 */ | |
|
359
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
479 public Session newSession(Forwarder master) { |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
480 int sid= makeID(sessionList.newSessionID()); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
481 Session session = new Session(sid, master); |
| 360 | 482 sessionList.put(sid, session); |
|
359
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
483 return session; |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
484 } |
| 358 | 485 |
| 144 | 486 public void addWaitingCommand(PacketSet set) { |
| 212 | 487 waitingCommandInMerge.add(set); |
| 144 | 488 } |
| 148 | 489 |
| 363 | 490 /* |
| 491 * Synchronize GUI command in this session manager. | |
| 492 */ | |
| 222 | 493 public void buttonPressed(SessionManagerEvent event) { |
| 494 try { | |
| 308 | 495 waitingEventQueue.put(event); |
| 222 | 496 } catch (InterruptedException e) {} |
| 497 selector.wakeup(); | |
| 498 } | |
| 281 | 499 |
| 363 | 500 /* |
| 501 * Execute incoming event during the initialization for | |
| 502 * testing purpose. | |
| 503 */ | |
| 281 | 504 public void syncExec(SessionManagerEvent event) { |
| 505 try { | |
| 308 | 506 waitingEventQueue.put(event); |
| 281 | 507 } catch (InterruptedException e) { |
| 508 } | |
| 509 } | |
| 222 | 510 |
| 363 | 511 /* |
| 512 * GUI command interface for close session. | |
| 513 */ | |
| 259 | 514 public void closeSession(SessionManagerEvent event) { |
| 515 Session session = ((CloseButtonEvent) event).getSession(); | |
| 516 session.closeSession(); | |
| 517 sessionList.remove(session); | |
| 518 updateGUI(); | |
| 519 } | |
| 520 | |
| 363 | 521 /* |
| 522 * Remove editors which has the cannel. | |
| 523 */ | |
| 274 | 524 public void remove(REPSocketChannel<REPCommand> channel) { |
| 358 | 525 int i = 0; |
| 526 for(Session s:sessionList.values()) { | |
| 274 | 527 if (s.deleteEditor(channel)) { |
| 358 | 528 i++; |
| 274 | 529 } |
| 530 } | |
| 358 | 531 assert(i==1); |
| 274 | 532 // can be other session manager? what should I do? |
| 533 } | |
| 534 | |
| 317 | 535 |
| 536 public void addWriteQueue(PacketSet packetSet) { | |
| 324 | 537 writeQueue.addLast(packetSet); |
| 323 | 538 assert(writeQueue.size()<packetLimit) ; |
| 317 | 539 } |
| 540 | |
| 318 | 541 |
| 542 public void remove(Editor editor) { | |
| 358 | 543 Session s = sessionList.get(editor.getSID()); |
| 362 | 544 if (s==null) { |
| 545 assert(false); | |
| 546 editorList.remove(editor); | |
| 547 } else if (editor.isMaster()) { | |
| 358 | 548 removeSession(s); |
| 549 } else { | |
| 550 s.deleteForwarder(editor); | |
| 551 editorList.remove(editor); | |
| 318 | 552 } |
| 341 | 553 updateGUI(); |
| 554 } | |
| 555 | |
| 556 private void removeSession(Session s0) { | |
| 358 | 557 s0.remove(this); |
| 341 | 558 sessionList.remove(s0); |
|
359
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
559 sendUpdate(s0.getSID()); |
| 318 | 560 } |
| 561 | |
| 334 | 562 public void setParentPort(int port) { |
| 563 parent_port = port; | |
| 564 } | |
| 565 public int getParentPort() { | |
| 566 return parent_port; | |
| 567 } | |
| 568 | |
| 569 public int getPort() { | |
| 570 return receive_port; | |
| 571 } | |
| 572 | |
| 353 | 573 |
| 355 | 574 boolean sessionManage(Forwarder forwarder, REPCommand command) throws ClosedChannelException, |
| 358 | 575 IOException { |
| 355 | 576 switch(command.cmd){ |
| 353 | 577 |
| 578 // Session Manager Command | |
| 579 | |
| 580 case SMCMD_JOIN: | |
| 581 { | |
| 358 | 582 // first connection or forwarded command |
| 367 | 583 routingTable.add(forwarder,getSMID(command.eid)); |
| 358 | 584 if(isMaster()) { |
| 585 REPCommand ackCommand = new REPCommand(); | |
| 586 ackCommand.setCMD(REP.SMCMD_JOIN_ACK); | |
| 587 ackCommand.setEID(command.eid); | |
| 360 | 588 ackCommand.setSID(command.sid); |
| 358 | 589 ackCommand.string = command.string; |
| 590 smList.sendToSlaves(ackCommand); | |
| 591 registEditor(forwarder,ackCommand); | |
| 592 } else { | |
| 593 smList.sendToMaster(command); | |
| 594 } | |
| 360 | 595 updateGUI(); |
| 353 | 596 } |
| 597 | |
| 598 break; | |
| 599 | |
| 361 | 600 case SMCMD_PUT_ACK: |
| 366 | 601 if (forwarder.isDirect()) { |
| 602 // send put_ack to the editor now. | |
| 603 command.setCMD(REP.SMCMD_PUT_ACK); | |
| 604 command.string = command.string; | |
| 605 command.setEID(command.eid); | |
| 606 command.setSID(command.sid); | |
| 607 forwarder.send(command); | |
| 608 } | |
| 353 | 609 case SMCMD_JOIN_ACK: |
| 358 | 610 registEditor(forwarder,command); |
| 360 | 611 updateGUI(); |
| 353 | 612 break; |
| 613 | |
| 614 case SMCMD_PUT: | |
| 615 { | |
| 358 | 616 // first connection or forwarded command |
| 367 | 617 routingTable.add(forwarder,getSMID(command.eid)); |
| 370 | 618 REPCommand ack = new REPCommand(command); ack.setCMD(REP.SMCMD_PUT_ACK); |
| 619 if(isMaster()) { | |
| 620 smList.sendToSlaves(ack); | |
| 621 registEditor(forwarder,ack); | |
| 358 | 622 } else { |
| 623 smList.sendToMaster(command); | |
| 361 | 624 // registEditor will be done by SMCMD_PUT_ACK |
| 625 } | |
| 370 | 626 if (forwarder.isDirect()) { |
| 627 forwarder.send(ack); | |
| 628 } | |
| 360 | 629 updateGUI(); |
| 353 | 630 |
| 631 } | |
| 632 break; | |
| 358 | 633 |
| 353 | 634 case SMCMD_SELECT: |
| 361 | 635 { |
| 636 Session session = sessionList.get(command.sid); | |
| 637 if (session==null) { | |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
638 session = new Session(command.sid, command.string,null); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
639 sessionList.put(command.sid,session); |
| 361 | 640 } |
| 641 selectSession(command.sid, session, command.eid, forwarder); | |
| 642 } | |
| 643 break; | |
| 353 | 644 case SMCMD_SELECT_ACK: |
| 645 { | |
| 369 | 646 // Sessionが見つかったので、select したeditorに教える。 |
| 358 | 647 Session session = sessionList.get(command.sid); |
| 369 | 648 Forwarder next = session.getNextEditor(forwarder); |
| 649 if (next==null) next = session.getFirstForwarder(); | |
| 650 if (next.getEID()==command.eid) { | |
| 651 // select したeditor を見つける | |
| 652 command.cmd=REP.SMCMD_JOIN_ACK; | |
| 653 next.send(command); | |
| 654 } else if (next.isDirect()) { | |
| 655 // | |
| 656 next.manage(command); | |
| 657 } else { | |
| 658 next.send(command); | |
| 659 } | |
| 353 | 660 } |
| 358 | 661 break; |
| 662 | |
| 353 | 663 case SMCMD_SM_JOIN: |
| 664 { | |
| 355 | 665 // SM_JOIN中にSMCMD_SM_JOINが来たら、これはループなので、 |
| 666 ///自分のSM_JOINを取り消す。 | |
| 667 if (sm_join_channel!=null) cancel_sm_join(); | |
| 353 | 668 // SMCMD_SM_JOIN は、master まで上昇する。 |
| 669 // masterでなければ、自分のparentに転送する。 | |
| 355 | 670 if(isMaster()) { |
| 353 | 671 // master であれば、SessionManager IDを決めて、 |
| 672 // 自分のsmList に登録 | |
| 355 | 673 Forwarder sm; |
| 674 int psid = command.eid; | |
| 675 if (forwarder.sid!=-1) { | |
| 676 // すでに channelはSessionManager Idを持っていて、 | |
| 677 // direct link ではないので、 | |
| 678 // channel を持たないForwarderとして登録する | |
| 679 sm = new Forwarder(this); | |
| 680 } else { | |
| 681 sm = forwarder; | |
| 682 } | |
| 367 | 683 int sid = smList.addNewSessionManager(sm,command); |
| 684 routingTable.add(forwarder,sid); | |
| 685 | |
| 355 | 686 REPCommand sendCommand = makeREPCommandWithSessionList(REP.SMCMD_SM_JOIN_ACK); |
| 687 // command.eid==smList.sesionManagerID() の場合は、 | |
| 688 // 待っている自分の下のsessionManagerにsidをassignする必要がある。 | |
| 353 | 689 sendCommand.setSID(sid); // new Session manager ID |
| 690 // 複数のSM_JOIN_ACKを識別するには、最初にSM_JOINを受け付けた | |
| 691 // Session manager IDを使う。 | |
| 355 | 692 sendCommand.setEID(psid); |
| 693 send_sm_join_ack(psid, sid, sendCommand); | |
| 694 } else { | |
| 695 if (forwarder.sid==-1) { | |
| 696 // direct link の場合は、識別のために、EIDに直上の | |
| 697 // smid を入れておく。 | |
| 698 command.setEID(smList.sessionManagerID()); | |
| 699 } | |
| 700 smList.sendToMaster(command); | |
| 353 | 701 } |
| 702 } | |
| 703 break; | |
| 704 | |
| 355 | 705 case SMCMD_SM_JOIN_ACK: |
| 706 send_sm_join_ack(command.eid, command.sid, command); | |
| 353 | 707 break; |
| 708 | |
| 709 case SMCMD_UPDATE: | |
|
359
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
710 if (!isMaster()) { |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
711 command.setString(mergeUpdate(command)); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
712 // 上に知らせる |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
713 smList.sendToMaster(command); |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
714 break; |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
715 } |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
716 // fall thru |
|
fa041bae35f1
all code written for distributed session except gather.
kono
parents:
358
diff
changeset
|
717 command.setCMD(REP.SMCMD_UPDATE_ACK); |
| 353 | 718 case SMCMD_UPDATE_ACK: |
| 358 | 719 command.setString(mergeUpdate(command)); |
| 720 // 下に知らせる | |
| 355 | 721 smList.sendToSlaves(command); |
| 358 | 722 updateGUI(); |
| 353 | 723 break; |
| 724 default: | |
| 725 return false; | |
| 726 } | |
| 727 return true; | |
| 728 } | |
| 729 | |
| 355 | 730 |
| 363 | 731 /** |
| 732 * UPDATE/UPDATE_ACKにより送られてきたSessionの情報を追加する | |
| 733 * @param command | |
| 734 * @return | |
| 735 * @throws IOException | |
| 736 */ | |
| 358 | 737 private String mergeUpdate(REPCommand command) throws IOException { |
| 738 SessionList receivedSessionList; | |
| 739 try { | |
| 740 receivedSessionList = decoder.decode(command.string); | |
| 741 } catch (SAXException e) { | |
| 742 throw new IOException(); | |
| 743 } | |
| 363 | 744 // 受け取った情報と自分の情報を混ぜる。 |
| 745 sessionList.merge(receivedSessionList); | |
| 358 | 746 //XMLを生成。送信コマンドにセット。 |
| 747 return encoder.sessionListToXML(sessionList); | |
| 748 | |
| 749 } | |
| 750 | |
| 751 /* | |
| 752 * id has SessionManager ID part | |
| 753 */ | |
| 754 private int makeID(int newid) { | |
| 755 return newid+smList.sessionManagerID()*MAXID; | |
| 756 } | |
| 757 | |
| 758 private int getSMID(int id) { | |
| 759 return id/MAXID; | |
| 760 } | |
| 761 | |
| 762 | |
| 361 | 763 /** |
| 764 * Register Editor to our editorList. No connection is made. | |
| 765 * @param forwarder Editor to be add | |
| 766 * @param command | |
| 767 */ | |
| 367 | 768 public void registEditor(Forwarder forwarder,REPCommand command) { |
| 358 | 769 // make ack for PUT/JOIN. Do not send this to the editor, |
| 770 // before select. After select, ack is sent to the editor. | |
| 771 Editor editor; | |
| 367 | 772 if (getSMID(command.eid)==smList.sessionManagerID()) { |
| 773 if (forwarder.isDirect()) { | |
| 774 editor = (Editor)forwarder; | |
| 775 } else | |
| 776 return; | |
| 358 | 777 } else { |
| 778 editor = new Editor(this, command.cmd==REP.SMCMD_PUT_ACK, command.eid); | |
| 779 } | |
| 780 editor.setName(command.string); | |
| 781 editor.setSID(command.sid); | |
| 782 if (!editorList.hasEid(command.eid)) { | |
| 783 editorList.add(editor); | |
| 784 } | |
| 367 | 785 if (command.cmd==REP.SMCMD_PUT_ACK) { |
| 786 Session session = new Session(command.sid, command.string, editor); | |
| 787 sessionList.put(command.sid, session); | |
| 788 } | |
| 358 | 789 // we don't join ack to the direct linked editor. We |
| 790 // have to wait select command | |
| 791 } | |
| 792 | |
| 793 | |
| 355 | 794 void send_sm_join_ack(int psid, int sid,REPCommand sendCommand) { |
| 795 if (psid==smList.sessionManagerID()) { | |
| 796 // 直下のsessionManagerにIDを割り振る必要がある。 | |
| 797 smList.assignSessionManagerIDtoWaitingSM(sid); | |
| 798 // ここで smList に一つだけ追加されるので | |
| 358 | 799 // 待っている最初のsm一つにだけ、sm_join_ackが新たに送られる。 |
| 355 | 800 } |
| 801 smList.sendToSlaves(sendCommand); | |
| 802 } | |
| 803 | |
| 804 | |
| 805 private REPCommand makeREPCommandWithSessionList(REP cmd) { | |
| 806 //SessionListからXMLを生成。 | |
| 807 //joinしてきたSessionManagerに対してACKを送信。 | |
| 808 REPCommand sendCommand = new REPCommand(); | |
| 809 sendCommand.setCMD(cmd); | |
| 358 | 810 sendCommand.setString(encoder.sessionListToXML(sessionList)); |
| 355 | 811 return sendCommand; |
| 812 } | |
| 813 | |
| 814 | |
| 815 public boolean isMaster() { | |
| 816 return smList.isMaster(); | |
| 817 } | |
| 818 | |
| 819 | |
| 820 public void setSessionManagerID(int sid) { | |
| 821 smList.setSessionManagerID(sid); | |
| 822 } | |
| 823 | |
| 358 | 824 |
| 825 public Session getSession(int sid) { | |
| 826 return sessionList.get(sid); | |
| 827 } | |
| 828 | |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
829 public void execAfterConnect(SessionManagerEvent sessionManagerEvent) { |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
830 execAfterConnect = sessionManagerEvent; |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
831 } |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
832 |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
833 public void afterConnect() { |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
834 if (execAfterConnect!=null) execAfterConnect.exec(this); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
835 execAfterConnect = null; |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
836 } |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
837 |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
838 public void setParent(Forwarder fw) { |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
839 smList.setParent(fw); |
|
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
840 } |
| 367 | 841 |
| 842 public String toString() { | |
| 843 int myId = 0; | |
| 844 if (smList!=null) myId = smList.sessionManagerID(); | |
| 845 return "rep.SessionManager-"+myId+"@"+myHost+":"+receive_port; | |
| 846 } | |
|
365
c432755c3555
distributed session debug continue... SELECT/SELECT_ACK loop
kono
parents:
364
diff
changeset
|
847 |
| 364 | 848 |
| 0 | 849 } |