1 //
2 // ========================================================================
3 // Copyright (c) 1995-2015 Mort Bay Consulting Pty. Ltd.
4 // ------------------------------------------------------------------------
5 // All rights reserved. This program and the accompanying materials
6 // are made available under the terms of the Eclipse Public License v1.0
7 // and Apache License v2.0 which accompanies this distribution.
8 //
9 // The Eclipse Public License is available at
10 // http://www.eclipse.org/legal/epl-v10.html
11 //
12 // The Apache License v2.0 is available at
13 // http://www.opensource.org/licenses/apache2.0.php
14 //
15 // You may elect to redistribute this code under either of these licenses.
16 // ========================================================================
17 //
18
19 package org.eclipse.jetty.io.nio;
20
21 import java.io.IOException;
22 import java.nio.channels.ClosedChannelException;
23 import java.nio.channels.SelectableChannel;
24 import java.nio.channels.SelectionKey;
25 import java.nio.channels.SocketChannel;
26 import java.util.Locale;
27
28 import org.eclipse.jetty.io.AsyncEndPoint;
29 import org.eclipse.jetty.io.Buffer;
30 import org.eclipse.jetty.io.ConnectedEndPoint;
31 import org.eclipse.jetty.io.Connection;
32 import org.eclipse.jetty.io.EofException;
33 import org.eclipse.jetty.io.nio.SelectorManager.SelectSet;
34 import org.eclipse.jetty.util.log.Log;
35 import org.eclipse.jetty.util.log.Logger;
36 import org.eclipse.jetty.util.thread.Timeout.Task;
37
38 /* ------------------------------------------------------------ */
39 /**
40 * An Endpoint that can be scheduled by {@link SelectorManager}.
41 */
42 public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPoint, ConnectedEndPoint
43 {
44 public static final Logger LOG=Log.getLogger("org.eclipse.jetty.io.nio");
45
46 private final boolean WORK_AROUND_JVM_BUG_6346658 = System.getProperty("os.name").toLowerCase(Locale.ENGLISH).contains("win");
47 private final SelectorManager.SelectSet _selectSet;
48 private final SelectorManager _manager;
49 private SelectionKey _key;
50 private final Runnable _handler = new Runnable()
51 {
52 public void run() { handle(); }
53 };
54
55 /** The desired value for {@link SelectionKey#interestOps()} */
56 private int _interestOps;
57
58 /**
59 * The connection instance is the handler for any IO activity on the endpoint.
60 * There is a different type of connection for HTTP, AJP, WebSocket and
61 * ProxyConnect. The connection may change for an SCEP as it is upgraded
62 * from HTTP to proxy connect or websocket.
63 */
64 private volatile AsyncConnection _connection;
65
66 private static final int STATE_NEEDS_DISPATCH=-1;
67 private static final int STATE_UNDISPATCHED=0;
68 private static final int STATE_DISPATCHED=1;
69 private static final int STATE_ASYNC=2;
70 private int _state;
71
72 private boolean _onIdle;
73
74 /** true if the last write operation succeed and wrote all offered bytes */
75 private volatile boolean _writable = true;
76
77
78 /** True if a thread has is blocked in {@link #blockReadable(long)} */
79 private boolean _readBlocked;
80
81 /** True if a thread has is blocked in {@link #blockWritable(long)} */
82 private boolean _writeBlocked;
83
84 /** true if {@link SelectSet#destroyEndPoint(SelectChannelEndPoint)} has not been called */
85 private boolean _open;
86
87 private volatile long _idleTimestamp;
88 private volatile boolean _checkIdle;
89
90 private boolean _ishut;
91
92 /* ------------------------------------------------------------ */
93 public SelectChannelEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key, int maxIdleTime)
94 throws IOException
95 {
96 super(channel, maxIdleTime);
97
98 _manager = selectSet.getManager();
99 _selectSet = selectSet;
100 _state=STATE_UNDISPATCHED;
101 _onIdle=false;
102 _open=true;
103 _key = key;
104
105 setCheckForIdle(true);
106 }
107
108 /* ------------------------------------------------------------ */
109 public SelectionKey getSelectionKey()
110 {
111 synchronized (this)
112 {
113 return _key;
114 }
115 }
116
117 /* ------------------------------------------------------------ */
118 public SelectorManager getSelectManager()
119 {
120 return _manager;
121 }
122
123 /* ------------------------------------------------------------ */
124 public Connection getConnection()
125 {
126 return _connection;
127 }
128
129 /* ------------------------------------------------------------ */
130 public void setConnection(Connection connection)
131 {
132 Connection old=_connection;
133 _connection=(AsyncConnection)connection;
134 if (old!=null && old!=_connection)
135 _manager.endPointUpgraded(this,old);
136 }
137
138 /* ------------------------------------------------------------ */
139 public long getIdleTimestamp()
140 {
141 return _idleTimestamp;
142 }
143
144 /* ------------------------------------------------------------ */
145 /** Called by selectSet to schedule handling
146 *
147 */
148 public void schedule()
149 {
150 synchronized (this)
151 {
152 // If there is no key, then do nothing
153 if (_key == null || !_key.isValid())
154 {
155 _readBlocked=false;
156 _writeBlocked=false;
157 this.notifyAll();
158 return;
159 }
160
161 // If there are threads dispatched reading and writing
162 if (_readBlocked || _writeBlocked)
163 {
164 // assert _dispatched;
165 if (_readBlocked && _key.isReadable())
166 _readBlocked=false;
167 if (_writeBlocked && _key.isWritable())
168 _writeBlocked=false;
169
170 // wake them up is as good as a dispatched.
171 this.notifyAll();
172
173 // we are not interested in further selecting
174 _key.interestOps(0);
175 if (_state<STATE_DISPATCHED)
176 updateKey();
177 return;
178 }
179
180 // Remove writeable op
181 if ((_key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE && (_key.interestOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE)
182 {
183 // Remove writeable op
184 _interestOps = _key.interestOps() & ~SelectionKey.OP_WRITE;
185 _key.interestOps(_interestOps);
186 _writable = true; // Once writable is in ops, only removed with dispatch.
187 }
188
189 // If dispatched, then deregister interest
190 if (_state>=STATE_DISPATCHED)
191 _key.interestOps(0);
192 else
193 {
194 // other wise do the dispatch
195 dispatch();
196 if (_state>=STATE_DISPATCHED && !_selectSet.getManager().isDeferringInterestedOps0())
197 {
198 _key.interestOps(0);
199 }
200 }
201 }
202 }
203
204 /* ------------------------------------------------------------ */
205 public void asyncDispatch()
206 {
207 synchronized(this)
208 {
209 switch(_state)
210 {
211 case STATE_NEEDS_DISPATCH:
212 case STATE_UNDISPATCHED:
213 dispatch();
214 break;
215
216 case STATE_DISPATCHED:
217 case STATE_ASYNC:
218 _state=STATE_ASYNC;
219 break;
220 }
221 }
222 }
223
224 /* ------------------------------------------------------------ */
225 public void dispatch()
226 {
227 synchronized(this)
228 {
229 if (_state<=STATE_UNDISPATCHED)
230 {
231 if (_onIdle)
232 _state = STATE_NEEDS_DISPATCH;
233 else
234 {
235 _state = STATE_DISPATCHED;
236 boolean dispatched = _manager.dispatch(_handler);
237 if(!dispatched)
238 {
239 _state = STATE_NEEDS_DISPATCH;
240 LOG.warn("Dispatched Failed! "+this+" to "+_manager);
241 updateKey();
242 }
243 }
244 }
245 }
246 }
247
248 /* ------------------------------------------------------------ */
249 /**
250 * Called when a dispatched thread is no longer handling the endpoint.
251 * The selection key operations are updated.
252 * @return If false is returned, the endpoint has been redispatched and
253 * thread must keep handling the endpoint.
254 */
255 protected boolean undispatch()
256 {
257 synchronized (this)
258 {
259 switch(_state)
260 {
261 case STATE_ASYNC:
262 _state=STATE_DISPATCHED;
263 return false;
264
265 default:
266 _state=STATE_UNDISPATCHED;
267 updateKey();
268 return true;
269 }
270 }
271 }
272
273 /* ------------------------------------------------------------ */
274 public void cancelTimeout(Task task)
275 {
276 getSelectSet().cancelTimeout(task);
277 }
278
279 /* ------------------------------------------------------------ */
280 public void scheduleTimeout(Task task, long timeoutMs)
281 {
282 getSelectSet().scheduleTimeout(task,timeoutMs);
283 }
284
285 /* ------------------------------------------------------------ */
286 public void setCheckForIdle(boolean check)
287 {
288 if (check)
289 {
290 _idleTimestamp=System.currentTimeMillis();
291 _checkIdle=true;
292 }
293 else
294 _checkIdle=false;
295 }
296
297 /* ------------------------------------------------------------ */
298 public boolean isCheckForIdle()
299 {
300 return _checkIdle;
301 }
302
303 /* ------------------------------------------------------------ */
304 protected void notIdle()
305 {
306 _idleTimestamp=System.currentTimeMillis();
307 }
308
309 /* ------------------------------------------------------------ */
310 public void checkIdleTimestamp(long now)
311 {
312 if (isCheckForIdle() && _maxIdleTime>0)
313 {
314 final long idleForMs=now-_idleTimestamp;
315
316 if (idleForMs>_maxIdleTime)
317 {
318 // Don't idle out again until onIdleExpired task completes.
319 setCheckForIdle(false);
320 _manager.dispatch(new Runnable()
321 {
322 public void run()
323 {
324 try
325 {
326 onIdleExpired(idleForMs);
327 }
328 finally
329 {
330 setCheckForIdle(true);
331 }
332 }
333 });
334 }
335 }
336 }
337
338 /* ------------------------------------------------------------ */
339 public void onIdleExpired(long idleForMs)
340 {
341 try
342 {
343 synchronized (this)
344 {
345 _onIdle=true;
346 }
347
348 _connection.onIdleExpired(idleForMs);
349 }
350 finally
351 {
352 synchronized (this)
353 {
354 _onIdle=false;
355 if (_state==STATE_NEEDS_DISPATCH)
356 dispatch();
357 }
358 }
359 }
360
361 /* ------------------------------------------------------------ */
362 @Override
363 public int fill(Buffer buffer) throws IOException
364 {
365 int fill=super.fill(buffer);
366 if (fill>0)
367 notIdle();
368 return fill;
369 }
370
371 /* ------------------------------------------------------------ */
372 @Override
373 public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
374 {
375 int l = super.flush(header, buffer, trailer);
376
377 // If there was something to write and it wasn't written, then we are not writable.
378 if (l==0 && ( header!=null && header.hasContent() || buffer!=null && buffer.hasContent() || trailer!=null && trailer.hasContent()))
379 {
380 synchronized (this)
381 {
382 _writable=false;
383 if (_state<STATE_DISPATCHED)
384 updateKey();
385 }
386 }
387 else if (l>0)
388 {
389 _writable=true;
390 notIdle();
391 }
392 return l;
393 }
394
395 /* ------------------------------------------------------------ */
396 /*
397 */
398 @Override
399 public int flush(Buffer buffer) throws IOException
400 {
401 int l = super.flush(buffer);
402
403 // If there was something to write and it wasn't written, then we are not writable.
404 if (l==0 && buffer!=null && buffer.hasContent())
405 {
406 synchronized (this)
407 {
408 _writable=false;
409 if (_state<STATE_DISPATCHED)
410 updateKey();
411 }
412 }
413 else if (l>0)
414 {
415 _writable=true;
416 notIdle();
417 }
418
419 return l;
420 }
421
422 /* ------------------------------------------------------------ */
423 /*
424 * Allows thread to block waiting for further events.
425 */
426 @Override
427 public boolean blockReadable(long timeoutMs) throws IOException
428 {
429 synchronized (this)
430 {
431 if (isInputShutdown())
432 throw new EofException();
433
434 long now=_selectSet.getNow();
435 long end=now+timeoutMs;
436 boolean check=isCheckForIdle();
437 setCheckForIdle(true);
438 try
439 {
440 _readBlocked=true;
441 while (!isInputShutdown() && _readBlocked)
442 {
443 try
444 {
445 updateKey();
446 this.wait(timeoutMs>0?(end-now):10000);
447 }
448 catch (InterruptedException e)
449 {
450 LOG.warn(e);
451 }
452 finally
453 {
454 now=_selectSet.getNow();
455 }
456
457 if (_readBlocked && timeoutMs>0 && now>=end)
458 return false;
459 }
460 }
461 finally
462 {
463 _readBlocked=false;
464 setCheckForIdle(check);
465 }
466 }
467 return true;
468 }
469
470 /* ------------------------------------------------------------ */
471 /*
472 * Allows thread to block waiting for further events.
473 */
474 @Override
475 public boolean blockWritable(long timeoutMs) throws IOException
476 {
477 synchronized (this)
478 {
479 if (isOutputShutdown())
480 throw new EofException();
481
482 long now=_selectSet.getNow();
483 long end=now+timeoutMs;
484 boolean check=isCheckForIdle();
485 setCheckForIdle(true);
486 try
487 {
488 _writeBlocked=true;
489 while (_writeBlocked && !isOutputShutdown())
490 {
491 try
492 {
493 updateKey();
494 this.wait(timeoutMs>0?(end-now):10000);
495 }
496 catch (InterruptedException e)
497 {
498 LOG.warn(e);
499 }
500 finally
501 {
502 now=_selectSet.getNow();
503 }
504 if (_writeBlocked && timeoutMs>0 && now>=end)
505 return false;
506 }
507 }
508 finally
509 {
510 _writeBlocked=false;
511 setCheckForIdle(check);
512 }
513 }
514 return true;
515 }
516
517 /* ------------------------------------------------------------ */
518 /**
519 * @see org.eclipse.jetty.io.AsyncEndPoint#scheduleWrite()
520 */
521 public void scheduleWrite()
522 {
523 if (_writable)
524 LOG.debug("Required scheduleWrite {}",this);
525
526 _writable=false;
527 updateKey();
528 }
529
530 /* ------------------------------------------------------------ */
531 public boolean isWritable()
532 {
533 return _writable;
534 }
535
536 /* ------------------------------------------------------------ */
537 public boolean hasProgressed()
538 {
539 return false;
540 }
541
542 /* ------------------------------------------------------------ */
543 /**
544 * Updates selection key. Adds operations types to the selection key as needed. No operations
545 * are removed as this is only done during dispatch. This method records the new key and
546 * schedules a call to doUpdateKey to do the keyChange
547 */
548 private void updateKey()
549 {
550 final boolean changed;
551 synchronized (this)
552 {
553 int current_ops=-1;
554 if (getChannel().isOpen())
555 {
556 boolean read_interest = _readBlocked || (_state<STATE_DISPATCHED && !_connection.isSuspended());
557 boolean write_interest= _writeBlocked || (_state<STATE_DISPATCHED && !_writable);
558
559 _interestOps =
560 ((!_socket.isInputShutdown() && read_interest ) ? SelectionKey.OP_READ : 0)
561 | ((!_socket.isOutputShutdown()&& write_interest) ? SelectionKey.OP_WRITE : 0);
562 try
563 {
564 current_ops = ((_key!=null && _key.isValid())?_key.interestOps():-1);
565 }
566 catch(Exception e)
567 {
568 _key=null;
569 LOG.ignore(e);
570 }
571 }
572 changed=_interestOps!=current_ops;
573 }
574
575 if(changed)
576 {
577 _selectSet.addChange(this);
578 _selectSet.wakeup();
579 }
580 }
581
582
583 /* ------------------------------------------------------------ */
584 /**
585 * Synchronize the interestOps with the actual key. Call is scheduled by a call to updateKey
586 */
587 void doUpdateKey()
588 {
589 synchronized (this)
590 {
591 if (getChannel().isOpen())
592 {
593 if (_interestOps>0)
594 {
595 if (_key==null || !_key.isValid())
596 {
597 SelectableChannel sc = (SelectableChannel)getChannel();
598 if (sc.isRegistered())
599 {
600 updateKey();
601 }
602 else
603 {
604 try
605 {
606 _key=((SelectableChannel)getChannel()).register(_selectSet.getSelector(),_interestOps,this);
607 }
608 catch (Exception e)
609 {
610 LOG.ignore(e);
611 if (_key!=null && _key.isValid())
612 {
613 _key.cancel();
614 }
615
616 if (_open)
617 {
618 _selectSet.destroyEndPoint(this);
619 }
620 _open=false;
621 _key = null;
622 }
623 }
624 }
625 else
626 {
627 _key.interestOps(_interestOps);
628 }
629 }
630 else
631 {
632 if (_key!=null && _key.isValid())
633 _key.interestOps(0);
634 else
635 _key=null;
636 }
637 }
638 else
639 {
640 if (_key!=null && _key.isValid())
641 _key.cancel();
642
643 if (_open)
644 {
645 _open=false;
646 _selectSet.destroyEndPoint(this);
647 }
648 _key = null;
649 }
650 }
651 }
652
653 /* ------------------------------------------------------------ */
654 /*
655 */
656 protected void handle()
657 {
658 boolean dispatched=true;
659 try
660 {
661 while(dispatched)
662 {
663 try
664 {
665 while(true)
666 {
667 final AsyncConnection next = (AsyncConnection)_connection.handle();
668 if (next!=_connection)
669 {
670 LOG.debug("{} replaced {}",next,_connection);
671 Connection old=_connection;
672 _connection=next;
673 _manager.endPointUpgraded(this,old);
674 continue;
675 }
676 break;
677 }
678 }
679 catch (ClosedChannelException e)
680 {
681 LOG.ignore(e);
682 }
683 catch (EofException e)
684 {
685 LOG.debug("EOF", e);
686 try{close();}
687 catch(IOException e2){LOG.ignore(e2);}
688 }
689 catch (IOException e)
690 {
691 LOG.warn(e.toString());
692 try{close();}
693 catch(IOException e2){LOG.ignore(e2);}
694 }
695 catch (Throwable e)
696 {
697 LOG.warn("handle failed", e);
698 try{close();}
699 catch(IOException e2){LOG.ignore(e2);}
700 }
701 finally
702 {
703 if (!_ishut && isInputShutdown() && isOpen())
704 {
705 _ishut=true;
706 try
707 {
708 _connection.onInputShutdown();
709 }
710 catch(Throwable x)
711 {
712 LOG.warn("onInputShutdown failed", x);
713 try{close();}
714 catch(IOException e2){LOG.ignore(e2);}
715 }
716 finally
717 {
718 updateKey();
719 }
720 }
721 dispatched=!undispatch();
722 }
723 }
724 }
725 finally
726 {
727 if (dispatched)
728 {
729 dispatched=!undispatch();
730 while (dispatched)
731 {
732 LOG.warn("SCEP.run() finally DISPATCHED");
733 dispatched=!undispatch();
734 }
735 }
736 }
737 }
738
739 /* ------------------------------------------------------------ */
740 /*
741 * @see org.eclipse.io.nio.ChannelEndPoint#close()
742 */
743 @Override
744 public void close() throws IOException
745 {
746 // On unix systems there is a JVM issue that if you cancel before closing, it can
747 // cause the selector to block waiting for a channel to close and that channel can
748 // block waiting for the remote end. But on windows, if you don't cancel before a
749 // close, then the selector can block anyway!
750 // https://bugs.eclipse.org/bugs/show_bug.cgi?id=357318
751 if (WORK_AROUND_JVM_BUG_6346658)
752 {
753 try
754 {
755 SelectionKey key = _key;
756 if (key!=null)
757 key.cancel();
758 }
759 catch (Throwable e)
760 {
761 LOG.ignore(e);
762 }
763 }
764
765 try
766 {
767 super.close();
768 }
769 catch (IOException e)
770 {
771 LOG.ignore(e);
772 }
773 finally
774 {
775 updateKey();
776 }
777 }
778
779 /* ------------------------------------------------------------ */
780 @Override
781 public String toString()
782 {
783 // Do NOT use synchronized (this)
784 // because it's very easy to deadlock when debugging is enabled.
785 // We do a best effort to print the right toString() and that's it.
786 SelectionKey key = _key;
787 String keyString = "";
788 if (key != null)
789 {
790 if (key.isValid())
791 {
792 if (key.isReadable())
793 keyString += "r";
794 if (key.isWritable())
795 keyString += "w";
796 }
797 else
798 {
799 keyString += "!";
800 }
801 }
802 else
803 {
804 keyString += "-";
805 }
806 return String.format("SCEP@%x{l(%s)<->r(%s),s=%d,open=%b,ishut=%b,oshut=%b,rb=%b,wb=%b,w=%b,i=%d%s}-{%s}",
807 hashCode(),
808 _socket.getRemoteSocketAddress(),
809 _socket.getLocalSocketAddress(),
810 _state,
811 isOpen(),
812 isInputShutdown(),
813 isOutputShutdown(),
814 _readBlocked,
815 _writeBlocked,
816 _writable,
817 _interestOps,
818 keyString,
819 _connection);
820 }
821
822 /* ------------------------------------------------------------ */
823 public SelectSet getSelectSet()
824 {
825 return _selectSet;
826 }
827
828 /* ------------------------------------------------------------ */
829 /**
830 * Don't set the SoTimeout
831 * @see org.eclipse.jetty.io.nio.ChannelEndPoint#setMaxIdleTime(int)
832 */
833 @Override
834 public void setMaxIdleTime(int timeMs) throws IOException
835 {
836 _maxIdleTime=timeMs;
837 }
838
839 }