1 // ========================================================================
2 // Copyright (c) 2004-2009 Mort Bay Consulting Pty. Ltd.
3 // ------------------------------------------------------------------------
4 // All rights reserved. This program and the accompanying materials
5 // are made available under the terms of the Eclipse Public License v1.0
6 // and Apache License v2.0 which accompanies this distribution.
7 // The Eclipse Public License is available at
8 // http://www.eclipse.org/legal/epl-v10.html
9 // The Apache License v2.0 is available at
10 // http://www.opensource.org/licenses/apache2.0.php
11 // You may elect to redistribute this code under either of these licenses.
12 // ========================================================================
13
14 package org.eclipse.jetty.io.nio;
15
16 import java.io.IOException;
17 import java.nio.channels.ClosedChannelException;
18 import java.nio.channels.SelectableChannel;
19 import java.nio.channels.SelectionKey;
20 import java.nio.channels.SocketChannel;
21
22 import org.eclipse.jetty.io.AsyncEndPoint;
23 import org.eclipse.jetty.io.Buffer;
24 import org.eclipse.jetty.io.ConnectedEndPoint;
25 import org.eclipse.jetty.io.Connection;
26 import org.eclipse.jetty.io.EofException;
27 import org.eclipse.jetty.io.nio.SelectorManager.SelectSet;
28 import org.eclipse.jetty.util.log.Log;
29
30 /* ------------------------------------------------------------ */
31 /**
32 * An Endpoint that can be scheduled by {@link SelectorManager}.
33 */
34 public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPoint, ConnectedEndPoint
35 {
36 private final SelectorManager.SelectSet _selectSet;
37 private final SelectorManager _manager;
38 private final Runnable _handler = new Runnable()
39 {
40 public void run() { handle(); }
41 };
42
43 private volatile Connection _connection;
44 private boolean _dispatched = false;
45 private boolean _redispatched = false;
46 private volatile boolean _writable = true;
47
48 private SelectionKey _key;
49 private int _interestOps;
50 private boolean _readBlocked;
51 private boolean _writeBlocked;
52 private boolean _open;
53 private volatile long _idleTimestamp;
54
55 /* ------------------------------------------------------------ */
56 public SelectChannelEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key, int maxIdleTime)
57 throws IOException
58 {
59 super(channel, maxIdleTime);
60
61 _manager = selectSet.getManager();
62 _selectSet = selectSet;
63 _dispatched = false;
64 _redispatched = false;
65 _open=true;
66 _key = key;
67
68 _connection = _manager.newConnection(channel,this);
69
70 scheduleIdle();
71 }
72
73 /* ------------------------------------------------------------ */
74 public SelectChannelEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key)
75 throws IOException
76 {
77 super(channel);
78
79 _manager = selectSet.getManager();
80 _selectSet = selectSet;
81 _dispatched = false;
82 _redispatched = false;
83 _open=true;
84 _key = key;
85
86 _connection = _manager.newConnection(channel,this);
87
88 scheduleIdle();
89 }
90 /* ------------------------------------------------------------ */
91 public SelectionKey getSelectionKey()
92 {
93 synchronized (this)
94 {
95 return _key;
96 }
97 }
98
99 /* ------------------------------------------------------------ */
100 public SelectorManager getSelectManager()
101 {
102 return _manager;
103 }
104
105 /* ------------------------------------------------------------ */
106 public Connection getConnection()
107 {
108 return _connection;
109 }
110
111 /* ------------------------------------------------------------ */
112 public void setConnection(Connection connection)
113 {
114 Connection old=_connection;
115 _connection=connection;
116 _manager.endPointUpgraded(this,old);
117 }
118
119 /* ------------------------------------------------------------ */
120 /** Called by selectSet to schedule handling
121 *
122 */
123 public void schedule()
124 {
125 synchronized (this)
126 {
127 // If there is no key, then do nothing
128 if (_key == null || !_key.isValid())
129 {
130 _readBlocked=false;
131 _writeBlocked=false;
132 this.notifyAll();
133 return;
134 }
135
136 // If there are threads dispatched reading and writing
137 if (_readBlocked || _writeBlocked)
138 {
139 // assert _dispatched;
140 if (_readBlocked && _key.isReadable())
141 _readBlocked=false;
142 if (_writeBlocked && _key.isWritable())
143 _writeBlocked=false;
144
145 // wake them up is as good as a dispatched.
146 this.notifyAll();
147
148 // we are not interested in further selecting
149 _key.interestOps(0);
150 return;
151 }
152
153 // Otherwise if we are still dispatched
154 if (!isReadyForDispatch())
155 {
156 // we are not interested in further selecting
157 _key.interestOps(0);
158 return;
159 }
160
161
162 // Remove writeable op
163 if ((_key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE && (_key.interestOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE)
164 {
165 // Remove writeable op
166 _interestOps = _key.interestOps() & ~SelectionKey.OP_WRITE;
167 _key.interestOps(_interestOps);
168 _writable = true; // Once writable is in ops, only removed with dispatch.
169 }
170
171 if (_dispatched)
172 _key.interestOps(0);
173 else
174 dispatch();
175 }
176 }
177
178 /* ------------------------------------------------------------ */
179 public void dispatch()
180 {
181 synchronized(this)
182 {
183 if (_dispatched)
184 _redispatched=true;
185 else
186 {
187 _dispatched = _manager.dispatch(_handler);
188 if(!_dispatched)
189 {
190 Log.warn("Dispatched Failed!");
191 updateKey();
192 }
193 }
194 }
195 }
196
197 /* ------------------------------------------------------------ */
198 /**
199 * Called when a dispatched thread is no longer handling the endpoint.
200 * The selection key operations are updated.
201 * @return If false is returned, the endpoint has been redispatched and
202 * thread must keep handling the endpoint.
203 */
204 private boolean undispatch()
205 {
206 synchronized (this)
207 {
208 if (_redispatched)
209 {
210 _redispatched=false;
211 return false;
212 }
213 _dispatched = false;
214 updateKey();
215 }
216 return true;
217 }
218
219 /* ------------------------------------------------------------ */
220 public void scheduleIdle()
221 {
222 _idleTimestamp=System.currentTimeMillis();
223 }
224
225 /* ------------------------------------------------------------ */
226 public void cancelIdle()
227 {
228 _idleTimestamp=0;
229 }
230
231 /* ------------------------------------------------------------ */
232 public void checkIdleTimestamp(long now)
233 {
234 if (_idleTimestamp!=0 && _maxIdleTime!=0 && now>(_idleTimestamp+_maxIdleTime))
235 {
236 idleExpired();
237 }
238 }
239
240 /* ------------------------------------------------------------ */
241 protected void idleExpired()
242 {
243 try
244 {
245 close();
246 }
247 catch (IOException e)
248 {
249 Log.ignore(e);
250 }
251 }
252
253 /* ------------------------------------------------------------ */
254 /*
255 */
256 @Override
257 public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
258 {
259 int l = super.flush(header, buffer, trailer);
260 if (!(_writable=l!=0))
261 {
262 synchronized (this)
263 {
264 if (!_dispatched)
265 updateKey();
266 }
267 }
268 return l;
269 }
270
271 /* ------------------------------------------------------------ */
272 /*
273 */
274 @Override
275 public int flush(Buffer buffer) throws IOException
276 {
277 int l = super.flush(buffer);
278 if (!(_writable=l!=0))
279 {
280 synchronized (this)
281 {
282 if (!_dispatched)
283 updateKey();
284 }
285 }
286 return l;
287 }
288
289 /* ------------------------------------------------------------ */
290 public boolean isReadyForDispatch()
291 {
292 synchronized (this)
293 {
294 return !(_dispatched || getConnection().isSuspended());
295 }
296 }
297
298 /* ------------------------------------------------------------ */
299 /*
300 * Allows thread to block waiting for further events.
301 */
302 @Override
303 public boolean blockReadable(long timeoutMs) throws IOException
304 {
305 synchronized (this)
306 {
307 long start=_selectSet.getNow();
308 try
309 {
310 _readBlocked=true;
311 while (isOpen() && _readBlocked)
312 {
313 try
314 {
315 updateKey();
316 this.wait(timeoutMs);
317
318 timeoutMs -= _selectSet.getNow()-start;
319 if (_readBlocked && timeoutMs<=0)
320 return false;
321 }
322 catch (InterruptedException e)
323 {
324 Log.warn(e);
325 }
326 }
327 }
328 finally
329 {
330 _readBlocked=false;
331 }
332 }
333 return true;
334 }
335
336 /* ------------------------------------------------------------ */
337 /*
338 * Allows thread to block waiting for further events.
339 */
340 @Override
341 public boolean blockWritable(long timeoutMs) throws IOException
342 {
343 synchronized (this)
344 {
345 long start=_selectSet.getNow();
346 try
347 {
348 _writeBlocked=true;
349 while (isOpen() && _writeBlocked)
350 {
351 try
352 {
353 updateKey();
354 this.wait(timeoutMs);
355
356 timeoutMs -= _selectSet.getNow()-start;
357 if (_writeBlocked && timeoutMs<=0)
358 return false;
359 }
360 catch (InterruptedException e)
361 {
362 Log.warn(e);
363 }
364 }
365 }
366 finally
367 {
368 _writeBlocked=false;
369 if (_idleTimestamp!=-1)
370 scheduleIdle();
371 }
372 }
373 return true;
374 }
375
376 /* ------------------------------------------------------------ */
377 public void setWritable(boolean writable)
378 {
379 _writable=writable;
380 }
381
382 /* ------------------------------------------------------------ */
383 public void scheduleWrite()
384 {
385 _writable=false;
386 updateKey();
387 }
388
389 /* ------------------------------------------------------------ */
390 /**
391 * Updates selection key. Adds operations types to the selection key as needed. No operations
392 * are removed as this is only done during dispatch. This method records the new key and
393 * schedules a call to doUpdateKey to do the keyChange
394 */
395 private void updateKey()
396 {
397 synchronized (this)
398 {
399 int ops=-1;
400 if (getChannel().isOpen())
401 {
402 _interestOps =
403 ((!_dispatched || _readBlocked) ? SelectionKey.OP_READ : 0)
404 | ((!_writable || _writeBlocked) ? SelectionKey.OP_WRITE : 0);
405 try
406 {
407 ops = ((_key!=null && _key.isValid())?_key.interestOps():-1);
408 }
409 catch(Exception e)
410 {
411 _key=null;
412 Log.ignore(e);
413 }
414 }
415
416 if(_interestOps == ops && getChannel().isOpen())
417 return;
418
419 }
420 _selectSet.addChange(this);
421 _selectSet.wakeup();
422 }
423
424 /* ------------------------------------------------------------ */
425 /**
426 * Synchronize the interestOps with the actual key. Call is scheduled by a call to updateKey
427 */
428 void doUpdateKey()
429 {
430 synchronized (this)
431 {
432 if (getChannel().isOpen())
433 {
434 if (_interestOps>0)
435 {
436 if (_key==null || !_key.isValid())
437 {
438 SelectableChannel sc = (SelectableChannel)getChannel();
439 if (sc.isRegistered())
440 {
441 updateKey();
442 }
443 else
444 {
445 try
446 {
447 _key=((SelectableChannel)getChannel()).register(_selectSet.getSelector(),_interestOps,this);
448 }
449 catch (Exception e)
450 {
451 Log.ignore(e);
452 if (_key!=null && _key.isValid())
453 {
454 _key.cancel();
455 }
456 cancelIdle();
457
458 if (_open)
459 {
460 _selectSet.destroyEndPoint(this);
461 }
462 _open=false;
463 _key = null;
464 }
465 }
466 }
467 else
468 {
469 _key.interestOps(_interestOps);
470 }
471 }
472 else
473 {
474 if (_key!=null && _key.isValid())
475 _key.interestOps(0);
476 else
477 _key=null;
478 }
479 }
480 else
481 {
482 if (_key!=null && _key.isValid())
483 _key.cancel();
484
485 cancelIdle();
486 if (_open)
487 {
488 _selectSet.destroyEndPoint(this);
489 }
490 _open=false;
491 _key = null;
492 }
493 }
494 }
495
496 /* ------------------------------------------------------------ */
497 /*
498 */
499 private void handle()
500 {
501 boolean dispatched=true;
502 try
503 {
504 while(dispatched)
505 {
506 try
507 {
508 while(true)
509 {
510 final Connection next = _connection.handle();
511 if (next!=_connection)
512 {
513 _connection=next;
514 continue;
515 }
516 break;
517 }
518 }
519 catch (ClosedChannelException e)
520 {
521 Log.ignore(e);
522 }
523 catch (EofException e)
524 {
525 Log.debug("EOF", e);
526 try{close();}
527 catch(IOException e2){Log.ignore(e2);}
528 }
529 catch (IOException e)
530 {
531 Log.warn(e.toString());
532 Log.debug(e);
533 try{close();}
534 catch(IOException e2){Log.ignore(e2);}
535 }
536 catch (Throwable e)
537 {
538 Log.warn("handle failed", e);
539 try{close();}
540 catch(IOException e2){Log.ignore(e2);}
541 }
542 dispatched=!undispatch();
543 }
544 }
545 finally
546 {
547 if (dispatched)
548 {
549 dispatched=!undispatch();
550 while (dispatched)
551 {
552 Log.warn("SCEP.run() finally DISPATCHED");
553 dispatched=!undispatch();
554 }
555 }
556 }
557 }
558
559 /* ------------------------------------------------------------ */
560 /*
561 * @see org.eclipse.io.nio.ChannelEndPoint#close()
562 */
563 @Override
564 public void close() throws IOException
565 {
566 try
567 {
568 super.close();
569 }
570 catch (IOException e)
571 {
572 Log.ignore(e);
573 }
574 finally
575 {
576 updateKey();
577 }
578 }
579
580 /* ------------------------------------------------------------ */
581 @Override
582 public String toString()
583 {
584 synchronized(this)
585 {
586 return "SCEP@" + hashCode() + "\t[d=" + _dispatched + ",io=" + _interestOps+
587 ",w=" + _writable + ",rb=" + _readBlocked + ",wb=" + _writeBlocked + "]";
588 }
589 }
590
591 /* ------------------------------------------------------------ */
592 public SelectSet getSelectSet()
593 {
594 return _selectSet;
595 }
596
597 /* ------------------------------------------------------------ */
598 /**
599 * Don't set the SoTimeout
600 * @see org.eclipse.jetty.io.nio.ChannelEndPoint#setMaxIdleTime(int)
601 */
602 @Override
603 public void setMaxIdleTime(int timeMs) throws IOException
604 {
605 _maxIdleTime=timeMs;
606 }
607
608
609
610 }