Mercurial Hosting > luan

comparison src/org/eclipse/jetty/io/nio/SelectChannelEndPoint.java @ 802:3428c60d7cfc

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