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.CancelledKeyException;
18 import java.nio.channels.Channel;
19 import java.nio.channels.ClosedSelectorException;
20 import java.nio.channels.SelectableChannel;
21 import java.nio.channels.SelectionKey;
22 import java.nio.channels.Selector;
23 import java.nio.channels.ServerSocketChannel;
24 import java.nio.channels.SocketChannel;
25 import java.util.ArrayList;
26 import java.util.List;
27 import java.util.concurrent.ConcurrentHashMap;
28 import java.util.concurrent.ConcurrentLinkedQueue;
29 import java.util.concurrent.ConcurrentMap;
30 import java.util.concurrent.CountDownLatch;
31 import java.util.concurrent.TimeUnit;
32
33 import javax.management.RuntimeErrorException;
34
35 import org.eclipse.jetty.io.ConnectedEndPoint;
36 import org.eclipse.jetty.io.Connection;
37 import org.eclipse.jetty.io.EndPoint;
38 import org.eclipse.jetty.util.TypeUtil;
39 import org.eclipse.jetty.util.component.AbstractLifeCycle;
40 import org.eclipse.jetty.util.component.AggregateLifeCycle;
41 import org.eclipse.jetty.util.component.Dumpable;
42 import org.eclipse.jetty.util.log.Log;
43 import org.eclipse.jetty.util.thread.Timeout;
44 import org.eclipse.jetty.util.thread.Timeout.Task;
45
46
47 /* ------------------------------------------------------------ */
48 /**
49 * The Selector Manager manages and number of SelectSets to allow
50 * NIO scheduling to scale to large numbers of connections.
51 * <p>
52 * This class works around a number of know JVM bugs. For details
53 * see http://wiki.eclipse.org/Jetty/Feature/JVM_NIO_Bug
54 */
55 public abstract class SelectorManager extends AbstractLifeCycle implements Dumpable
56 {
57 // TODO Tune these by approx system speed.
58 private static final int __JVMBUG_THRESHHOLD=Integer.getInteger("org.eclipse.jetty.io.nio.JVMBUG_THRESHHOLD",0).intValue();
59 private static final int __MONITOR_PERIOD=Integer.getInteger("org.eclipse.jetty.io.nio.MONITOR_PERIOD",1000).intValue();
60 private static final int __MAX_SELECTS=Integer.getInteger("org.eclipse.jetty.io.nio.MAX_SELECTS",25000).intValue();
61 private static final int __BUSY_PAUSE=Integer.getInteger("org.eclipse.jetty.io.nio.BUSY_PAUSE",50).intValue();
62 private static final int __BUSY_KEY=Integer.getInteger("org.eclipse.jetty.io.nio.BUSY_KEY",-1).intValue();
63 private static final int __IDLE_TICK=Integer.getInteger("org.eclipse.jetty.io.nio.IDLE_TICK",400).intValue();
64
65 private int _maxIdleTime;
66 private int _lowResourcesMaxIdleTime;
67 private long _lowResourcesConnections;
68 private SelectSet[] _selectSet;
69 private int _selectSets=1;
70 private volatile int _set;
71 private boolean _deferringInterestedOps0=true;
72
73 /* ------------------------------------------------------------ */
74 /**
75 * @param maxIdleTime The maximum period in milli seconds that a connection may be idle before it is closed.
76 * @see #setLowResourcesMaxIdleTime(long)
77 */
78 public void setMaxIdleTime(long maxIdleTime)
79 {
80 _maxIdleTime=(int)maxIdleTime;
81 }
82
83 /* ------------------------------------------------------------ */
84 /**
85 * @param selectSets number of select sets to create
86 */
87 public void setSelectSets(int selectSets)
88 {
89 long lrc = _lowResourcesConnections * _selectSets;
90 _selectSets=selectSets;
91 _lowResourcesConnections=lrc/_selectSets;
92 }
93
94 /* ------------------------------------------------------------ */
95 /**
96 * @return the max idle time
97 */
98 public long getMaxIdleTime()
99 {
100 return _maxIdleTime;
101 }
102
103 /* ------------------------------------------------------------ */
104 /**
105 * @return the number of select sets in use
106 */
107 public int getSelectSets()
108 {
109 return _selectSets;
110 }
111
112 /* ------------------------------------------------------------ */
113 /**
114 * @param i
115 * @return The select set
116 */
117 public SelectSet getSelectSet(int i)
118 {
119 return _selectSet[i];
120 }
121
122 /* ------------------------------------------------------------ */
123 /** Register a channel
124 * @param channel
125 * @param att Attached Object
126 */
127 public void register(SocketChannel channel, Object att)
128 {
129 // The ++ increment here is not atomic, but it does not matter.
130 // so long as the value changes sometimes, then connections will
131 // be distributed over the available sets.
132
133 int s=_set++;
134 s=s%_selectSets;
135 SelectSet[] sets=_selectSet;
136 if (sets!=null)
137 {
138 SelectSet set=sets[s];
139 set.addChange(channel,att);
140 set.wakeup();
141 }
142 }
143
144
145 /* ------------------------------------------------------------ */
146 /** Register a channel
147 * @param channel
148 */
149 public void register(SocketChannel channel)
150 {
151 // The ++ increment here is not atomic, but it does not matter.
152 // so long as the value changes sometimes, then connections will
153 // be distributed over the available sets.
154
155 int s=_set++;
156 s=s%_selectSets;
157 SelectSet[] sets=_selectSet;
158 if (sets!=null)
159 {
160 SelectSet set=sets[s];
161 set.addChange(channel);
162 set.wakeup();
163 }
164 }
165
166 /* ------------------------------------------------------------ */
167 /** Register a {@link ServerSocketChannel}
168 * @param acceptChannel
169 */
170 public void register(ServerSocketChannel acceptChannel)
171 {
172 int s=_set++;
173 s=s%_selectSets;
174 SelectSet set=_selectSet[s];
175 set.addChange(acceptChannel);
176 set.wakeup();
177 }
178
179 /* ------------------------------------------------------------ */
180 /**
181 * @return the lowResourcesConnections
182 */
183 public long getLowResourcesConnections()
184 {
185 return _lowResourcesConnections*_selectSets;
186 }
187
188 /* ------------------------------------------------------------ */
189 /**
190 * Set the number of connections, which if exceeded places this manager in low resources state.
191 * This is not an exact measure as the connection count is averaged over the select sets.
192 * @param lowResourcesConnections the number of connections
193 * @see #setLowResourcesMaxIdleTime(long)
194 */
195 public void setLowResourcesConnections(long lowResourcesConnections)
196 {
197 _lowResourcesConnections=(lowResourcesConnections+_selectSets-1)/_selectSets;
198 }
199
200 /* ------------------------------------------------------------ */
201 /**
202 * @return the lowResourcesMaxIdleTime
203 */
204 public long getLowResourcesMaxIdleTime()
205 {
206 return _lowResourcesMaxIdleTime;
207 }
208
209 /* ------------------------------------------------------------ */
210 /**
211 * @param lowResourcesMaxIdleTime the period in ms that a connection is allowed to be idle when this SelectSet has more connections than {@link #getLowResourcesConnections()}
212 * @see #setMaxIdleTime(long)
213 */
214 public void setLowResourcesMaxIdleTime(long lowResourcesMaxIdleTime)
215 {
216 _lowResourcesMaxIdleTime=(int)lowResourcesMaxIdleTime;
217 }
218
219 /* ------------------------------------------------------------ */
220 /**
221 * @param acceptorID
222 * @throws IOException
223 */
224 public void doSelect(int acceptorID) throws IOException
225 {
226 SelectSet[] sets= _selectSet;
227 if (sets!=null && sets.length>acceptorID && sets[acceptorID]!=null)
228 sets[acceptorID].doSelect();
229 }
230
231 /* ------------------------------------------------------------------------------- */
232 public abstract boolean dispatch(Runnable task);
233
234 /* ------------------------------------------------------------ */
235 /* (non-Javadoc)
236 * @see org.eclipse.component.AbstractLifeCycle#doStart()
237 */
238 @Override
239 protected void doStart() throws Exception
240 {
241 _selectSet = new SelectSet[_selectSets];
242 for (int i=0;i<_selectSet.length;i++)
243 _selectSet[i]= new SelectSet(i);
244
245 super.doStart();
246 }
247
248
249 /* ------------------------------------------------------------------------------- */
250 @Override
251 protected void doStop() throws Exception
252 {
253 SelectSet[] sets= _selectSet;
254 _selectSet=null;
255 if (sets!=null)
256 {
257 for (SelectSet set : sets)
258 {
259 if (set!=null)
260 set.stop();
261 }
262 }
263 super.doStop();
264 }
265
266 /* ------------------------------------------------------------ */
267 /**
268 * @param endpoint
269 */
270 protected abstract void endPointClosed(SelectChannelEndPoint endpoint);
271
272 /* ------------------------------------------------------------ */
273 /**
274 * @param endpoint
275 */
276 protected abstract void endPointOpened(SelectChannelEndPoint endpoint);
277
278 /* ------------------------------------------------------------ */
279 protected abstract void endPointUpgraded(ConnectedEndPoint endpoint,Connection oldConnection);
280
281 /* ------------------------------------------------------------------------------- */
282 protected abstract Connection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint);
283
284 /* ------------------------------------------------------------ */
285 /**
286 * Create a new end point
287 * @param channel
288 * @param selectSet
289 * @param sKey the selection key
290 * @return the new endpoint {@link SelectChannelEndPoint}
291 * @throws IOException
292 */
293 protected abstract SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectorManager.SelectSet selectSet, SelectionKey sKey) throws IOException;
294
295 /* ------------------------------------------------------------------------------- */
296 protected void connectionFailed(SocketChannel channel,Throwable ex,Object attachment)
297 {
298 Log.warn(ex+","+channel+","+attachment);
299 Log.debug(ex);
300 }
301
302 /* ------------------------------------------------------------ */
303 public String dump()
304 {
305 return AggregateLifeCycle.dump(this);
306 }
307
308 /* ------------------------------------------------------------ */
309 public void dump(Appendable out, String indent) throws IOException
310 {
311 out.append(String.valueOf(this)).append("\n");
312 AggregateLifeCycle.dump(out,indent,TypeUtil.asList(_selectSet));
313 }
314
315
316 /* ------------------------------------------------------------------------------- */
317 /* ------------------------------------------------------------------------------- */
318 /* ------------------------------------------------------------------------------- */
319 public class SelectSet implements Dumpable
320 {
321 private final int _setID;
322 private final Timeout _timeout;
323
324 private final ConcurrentLinkedQueue<Object> _changes = new ConcurrentLinkedQueue<Object>();
325
326 private Selector _selector;
327
328 private volatile Thread _selecting;
329 private int _jvmBug;
330 private int _selects;
331 private long _monitorStart;
332 private long _monitorNext;
333 private boolean _pausing;
334 private SelectionKey _busyKey;
335 private int _busyKeyCount;
336 private long _log;
337 private int _paused;
338 private int _jvmFix0;
339 private int _jvmFix1;
340 private int _jvmFix2;
341 private volatile long _idleTick;
342 private ConcurrentMap<SelectChannelEndPoint,Object> _endPoints = new ConcurrentHashMap<SelectChannelEndPoint, Object>();
343
344 /* ------------------------------------------------------------ */
345 SelectSet(int acceptorID) throws Exception
346 {
347 _setID=acceptorID;
348
349 _idleTick = System.currentTimeMillis();
350 _timeout = new Timeout(this);
351 _timeout.setDuration(0L);
352
353 // create a selector;
354 _selector = Selector.open();
355 _monitorStart=System.currentTimeMillis();
356 _monitorNext=_monitorStart+__MONITOR_PERIOD;
357 _log=_monitorStart+60000;
358 }
359
360 /* ------------------------------------------------------------ */
361 public void addChange(Object change)
362 {
363 _changes.add(change);
364 }
365
366 /* ------------------------------------------------------------ */
367 public void addChange(SelectableChannel channel, Object att)
368 {
369 if (att==null)
370 addChange(channel);
371 else if (att instanceof EndPoint)
372 addChange(att);
373 else
374 addChange(new ChannelAndAttachment(channel,att));
375 }
376
377 /* ------------------------------------------------------------ */
378 /**
379 * Select and dispatch tasks found from changes and the selector.
380 *
381 * @throws IOException
382 */
383 public void doSelect() throws IOException
384 {
385 try
386 {
387 _selecting=Thread.currentThread();
388 final Selector selector=_selector;
389
390 // Make any key changes required
391 Object change;
392 int changes=_changes.size();
393 while (changes-->0 && (change=_changes.poll())!=null)
394 {
395 Channel ch=null;
396 SelectionKey key=null;
397
398 try
399 {
400 if (change instanceof EndPoint)
401 {
402 // Update the operations for a key.
403 SelectChannelEndPoint endpoint = (SelectChannelEndPoint)change;
404 ch=endpoint.getChannel();
405 endpoint.doUpdateKey();
406 }
407 else if (change instanceof ChannelAndAttachment)
408 {
409 // finish accepting/connecting this connection
410 final ChannelAndAttachment asc = (ChannelAndAttachment)change;
411 final SelectableChannel channel=asc._channel;
412 ch=channel;
413 final Object att = asc._attachment;
414
415 if ((channel instanceof SocketChannel) && ((SocketChannel)channel).isConnected())
416 {
417 key = channel.register(selector,SelectionKey.OP_READ,att);
418 SelectChannelEndPoint endpoint = createEndPoint((SocketChannel)channel,key);
419 key.attach(endpoint);
420 endpoint.schedule();
421 }
422 else if (channel.isOpen())
423 {
424 key = channel.register(selector,SelectionKey.OP_CONNECT,att);
425 }
426 }
427 else if (change instanceof SocketChannel)
428 {
429 // Newly registered channel
430 final SocketChannel channel=(SocketChannel)change;
431 ch=channel;
432 key = channel.register(selector,SelectionKey.OP_READ,null);
433 SelectChannelEndPoint endpoint = createEndPoint(channel,key);
434 key.attach(endpoint);
435 endpoint.schedule();
436 }
437 else if (change instanceof Runnable)
438 {
439 dispatch((Runnable)change);
440 }
441 else
442 throw new IllegalArgumentException(change.toString());
443 }
444 catch (Throwable e)
445 {
446 if (e instanceof ThreadDeath)
447 throw (ThreadDeath)e;
448
449 if (isRunning())
450 Log.warn(e);
451 else
452 Log.debug(e);
453
454 if(ch!=null && key==null)
455 {
456 try
457 {
458 ch.close();
459 }
460 catch(IOException e2)
461 {
462 Log.debug(e2);
463 }
464 }
465 }
466 }
467
468
469 // Do and instant select to see if any connections can be handled.
470 int selected=selector.selectNow();
471 _selects++;
472
473 long now=System.currentTimeMillis();
474
475 // if no immediate things to do
476 if (selected==0 && selector.selectedKeys().isEmpty())
477 {
478 // If we are in pausing mode
479 if (_pausing)
480 {
481 try
482 {
483 Thread.sleep(__BUSY_PAUSE); // pause to reduce impact of busy loop
484 }
485 catch(InterruptedException e)
486 {
487 Log.ignore(e);
488 }
489 now=System.currentTimeMillis();
490 }
491
492 // workout how long to wait in select
493 _timeout.setNow(now);
494 long to_next_timeout=_timeout.getTimeToNext();
495
496 long wait = _changes.size()==0?__IDLE_TICK:0L;
497 if (wait > 0 && to_next_timeout >= 0 && wait > to_next_timeout)
498 wait = to_next_timeout;
499
500 // If we should wait with a select
501 if (wait>0)
502 {
503 long before=now;
504 selected=selector.select(wait);
505 _selects++;
506 now = System.currentTimeMillis();
507 _timeout.setNow(now);
508
509 if (__JVMBUG_THRESHHOLD>0)
510 checkJvmBugs(before, now, wait, selected);
511 }
512 }
513
514 // have we been destroyed while sleeping
515 if (_selector==null || !selector.isOpen())
516 return;
517
518 // Look for things to do
519 for (SelectionKey key: selector.selectedKeys())
520 {
521 try
522 {
523 if (!key.isValid())
524 {
525 key.cancel();
526 SelectChannelEndPoint endpoint = (SelectChannelEndPoint)key.attachment();
527 if (endpoint != null)
528 endpoint.doUpdateKey();
529 continue;
530 }
531
532 Object att = key.attachment();
533 if (att instanceof SelectChannelEndPoint)
534 {
535 if (key.isReadable()||key.isWritable())
536 ((SelectChannelEndPoint)att).schedule();
537 }
538 else if (key.isConnectable())
539 {
540 // Complete a connection of a registered channel
541 SocketChannel channel = (SocketChannel)key.channel();
542 boolean connected=false;
543 try
544 {
545 connected=channel.finishConnect();
546 }
547 catch(Exception e)
548 {
549 connectionFailed(channel,e,att);
550 }
551 finally
552 {
553 if (connected)
554 {
555 key.interestOps(SelectionKey.OP_READ);
556 SelectChannelEndPoint endpoint = createEndPoint(channel,key);
557 key.attach(endpoint);
558 endpoint.schedule();
559 }
560 else
561 {
562 key.cancel();
563 }
564 }
565 }
566 else
567 {
568 // Wrap readable registered channel in an endpoint
569 SocketChannel channel = (SocketChannel)key.channel();
570 SelectChannelEndPoint endpoint = createEndPoint(channel,key);
571 key.attach(endpoint);
572 if (key.isReadable())
573 endpoint.schedule();
574 }
575 key = null;
576 }
577 catch (CancelledKeyException e)
578 {
579 Log.ignore(e);
580 }
581 catch (Exception e)
582 {
583 if (isRunning())
584 Log.warn(e);
585 else
586 Log.ignore(e);
587
588 if (key != null && !(key.channel() instanceof ServerSocketChannel) && key.isValid())
589 key.cancel();
590 }
591 }
592
593 // Everything always handled
594 selector.selectedKeys().clear();
595
596 now=System.currentTimeMillis();
597 _timeout.setNow(now);
598 Task task = _timeout.expired();
599 while (task!=null)
600 {
601 if (task instanceof Runnable)
602 dispatch((Runnable)task);
603 task = _timeout.expired();
604 }
605
606 // Idle tick
607 if (now-_idleTick>__IDLE_TICK)
608 {
609 _idleTick=now;
610
611 final long idle_now=((_lowResourcesConnections>0 && selector.keys().size()>_lowResourcesConnections))
612 ?(now+_maxIdleTime-_lowResourcesMaxIdleTime)
613 :now;
614
615 dispatch(new Runnable()
616 {
617 public void run()
618 {
619 for (SelectChannelEndPoint endp:_endPoints.keySet())
620 {
621 endp.checkIdleTimestamp(idle_now);
622 }
623 }
624 });
625 }
626 }
627 catch (ClosedSelectorException e)
628 {
629 if (isRunning())
630 Log.warn(e);
631 else
632 Log.ignore(e);
633 }
634 catch (CancelledKeyException e)
635 {
636 Log.ignore(e);
637 }
638 finally
639 {
640 _selecting=null;
641 }
642 }
643
644 /* ------------------------------------------------------------ */
645 private void checkJvmBugs(long before, long now, long wait, int selected)
646 throws IOException
647 {
648 Selector selector = _selector;
649 if (selector==null)
650 return;
651
652 // Look for JVM bugs over a monitor period.
653 // http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6403933
654 // http://bugs.sun.com/view_bug.do?bug_id=6693490
655 if (now>_monitorNext)
656 {
657 _selects=(int)(_selects*__MONITOR_PERIOD/(now-_monitorStart));
658 _pausing=_selects>__MAX_SELECTS;
659 if (_pausing)
660 _paused++;
661
662 _selects=0;
663 _jvmBug=0;
664 _monitorStart=now;
665 _monitorNext=now+__MONITOR_PERIOD;
666 }
667
668 if (now>_log)
669 {
670 if (_paused>0)
671 Log.debug(this+" Busy selector - injecting delay "+_paused+" times");
672
673 if (_jvmFix2>0)
674 Log.debug(this+" JVM BUG(s) - injecting delay"+_jvmFix2+" times");
675
676 if (_jvmFix1>0)
677 Log.debug(this+" JVM BUG(s) - recreating selector "+_jvmFix1+" times, cancelled keys "+_jvmFix0+" times");
678
679 else if(Log.isDebugEnabled() && _jvmFix0>0)
680 Log.debug(this+" JVM BUG(s) - cancelled keys "+_jvmFix0+" times");
681 _paused=0;
682 _jvmFix2=0;
683 _jvmFix1=0;
684 _jvmFix0=0;
685 _log=now+60000;
686 }
687
688 // If we see signature of possible JVM bug, increment count.
689 if (selected==0 && wait>10 && (now-before)<(wait/2))
690 {
691 // Increment bug count and try a work around
692 _jvmBug++;
693 if (_jvmBug>(__JVMBUG_THRESHHOLD))
694 {
695 try
696 {
697 if (_jvmBug==__JVMBUG_THRESHHOLD+1)
698 _jvmFix2++;
699
700 Thread.sleep(__BUSY_PAUSE); // pause to avoid busy loop
701 }
702 catch(InterruptedException e)
703 {
704 Log.ignore(e);
705 }
706 }
707 else if (_jvmBug==__JVMBUG_THRESHHOLD)
708 {
709 renewSelector();
710 }
711 else if (_jvmBug%32==31) // heuristic attempt to cancel key 31,63,95,... loops
712 {
713 // Cancel keys with 0 interested ops
714 int cancelled=0;
715 for (SelectionKey k: selector.keys())
716 {
717 if (k.isValid()&&k.interestOps()==0)
718 {
719 k.cancel();
720 cancelled++;
721 }
722 }
723 if (cancelled>0)
724 _jvmFix0++;
725
726 return;
727 }
728 }
729 else if (__BUSY_KEY>0 && selected==1 && _selects>__MAX_SELECTS)
730 {
731 // Look for busy key
732 SelectionKey busy = selector.selectedKeys().iterator().next();
733 if (busy==_busyKey)
734 {
735 if (++_busyKeyCount>__BUSY_KEY && !(busy.channel() instanceof ServerSocketChannel))
736 {
737 final SelectChannelEndPoint endpoint = (SelectChannelEndPoint)busy.attachment();
738 Log.warn("Busy Key "+busy.channel()+" "+endpoint);
739 busy.cancel();
740 if (endpoint!=null)
741 {
742 dispatch(new Runnable()
743 {
744 public void run()
745 {
746 try
747 {
748 endpoint.close();
749 }
750 catch (IOException e)
751 {
752 Log.ignore(e);
753 }
754 }
755 });
756 }
757 }
758 }
759 else
760 _busyKeyCount=0;
761 _busyKey=busy;
762 }
763 }
764
765 /* ------------------------------------------------------------ */
766 private void renewSelector()
767 {
768 try
769 {
770 synchronized (this)
771 {
772 Selector selector=_selector;
773 if (selector==null)
774 return;
775 final Selector new_selector = Selector.open();
776 for (SelectionKey k: selector.keys())
777 {
778 if (!k.isValid() || k.interestOps()==0)
779 continue;
780
781 final SelectableChannel channel = k.channel();
782 final Object attachment = k.attachment();
783
784 if (attachment==null)
785 addChange(channel);
786 else
787 addChange(channel,attachment);
788 }
789 _selector.close();
790 _selector=new_selector;
791 }
792 }
793 catch(IOException e)
794 {
795 throw new RuntimeException("recreating selector",e);
796 }
797 }
798
799 /* ------------------------------------------------------------ */
800 public SelectorManager getManager()
801 {
802 return SelectorManager.this;
803 }
804
805 /* ------------------------------------------------------------ */
806 public long getNow()
807 {
808 return _timeout.getNow();
809 }
810
811 /* ------------------------------------------------------------ */
812 /**
813 * @param task The task to timeout. If it implements Runnable, then
814 * expired will be called from a dispatched thread.
815 *
816 * @param timeoutMs
817 */
818 public void scheduleTimeout(Timeout.Task task, long timeoutMs)
819 {
820 if (!(task instanceof Runnable))
821 throw new IllegalArgumentException("!Runnable");
822 _timeout.schedule(task, timeoutMs);
823 }
824
825 /* ------------------------------------------------------------ */
826 public void cancelTimeout(Timeout.Task task)
827 {
828 task.cancel();
829 }
830
831 /* ------------------------------------------------------------ */
832 public void wakeup()
833 {
834 try
835 {
836 Selector selector = _selector;
837 if (selector!=null)
838 selector.wakeup();
839 }
840 catch(Exception e)
841 {
842 renewSelector();
843 }
844 }
845
846 /* ------------------------------------------------------------ */
847 private SelectChannelEndPoint createEndPoint(SocketChannel channel, SelectionKey sKey) throws IOException
848 {
849 SelectChannelEndPoint endp = newEndPoint(channel,this,sKey);
850 endPointOpened(endp);
851 _endPoints.put(endp,this);
852 return endp;
853 }
854
855 /* ------------------------------------------------------------ */
856 public void destroyEndPoint(SelectChannelEndPoint endp)
857 {
858 _endPoints.remove(endp);
859 endPointClosed(endp);
860 }
861
862 /* ------------------------------------------------------------ */
863 Selector getSelector()
864 {
865 return _selector;
866 }
867
868 /* ------------------------------------------------------------ */
869 void stop() throws Exception
870 {
871 // Spin for a while waiting for selector to complete
872 // to avoid unneccessary closed channel exceptions
873 try
874 {
875 for (int i=0;i<100 && _selecting!=null;i++)
876 {
877 wakeup();
878 Thread.sleep(10);
879 }
880 }
881 catch(Exception e)
882 {
883 Log.ignore(e);
884 }
885
886 // close endpoints and selector
887 synchronized (this)
888 {
889 for (SelectionKey key:_selector.keys())
890 {
891 if (key==null)
892 continue;
893 Object att=key.attachment();
894 if (att instanceof EndPoint)
895 {
896 EndPoint endpoint = (EndPoint)att;
897 try
898 {
899 endpoint.close();
900 }
901 catch(IOException e)
902 {
903 Log.ignore(e);
904 }
905 }
906 }
907
908
909 _timeout.cancelAll();
910 try
911 {
912 if (_selector != null)
913 _selector.close();
914 }
915 catch (IOException e)
916 {
917 Log.ignore(e);
918 }
919 _selector=null;
920 }
921 }
922
923 /* ------------------------------------------------------------ */
924 public String dump()
925 {
926 return AggregateLifeCycle.dump(this);
927 }
928
929 /* ------------------------------------------------------------ */
930 public void dump(Appendable out, String indent) throws IOException
931 {
932 out.append(String.valueOf(this)).append(" id=").append(String.valueOf(_setID)).append("\n");
933
934 Thread selecting = _selecting;
935
936 Object where = "not selecting";
937 StackTraceElement[] trace =selecting==null?null:selecting.getStackTrace();
938 if (trace!=null)
939 {
940 for (StackTraceElement t:trace)
941 if (t.getClassName().startsWith("org.eclipse.jetty."))
942 {
943 where=t;
944 break;
945 }
946 }
947
948 Selector selector=_selector;
949 final ArrayList<Object> dump = new ArrayList<Object>(selector.keys().size()*2);
950 dump.add(where);
951
952 final CountDownLatch latch = new CountDownLatch(1);
953
954 addChange(new Runnable(){
955 public void run()
956 {
957 dumpKeyState(dump);
958 latch.countDown();
959 }
960 });
961
962 try
963 {
964 latch.await(5,TimeUnit.SECONDS);
965 }
966 catch(InterruptedException e)
967 {
968 Log.ignore(e);
969 }
970 AggregateLifeCycle.dump(out,indent,dump);
971 }
972
973 /* ------------------------------------------------------------ */
974 public void dumpKeyState(List<Object> dumpto)
975 {
976 Selector selector=_selector;
977 dumpto.add(selector+" keys="+selector.keys().size());
978 for (SelectionKey key: selector.keys())
979 {
980 if (key.isValid())
981 dumpto.add(key.attachment()+" "+key.interestOps()+" "+key.readyOps());
982 else
983 dumpto.add(key.attachment()+" - - ");
984 }
985 }
986 }
987
988 /* ------------------------------------------------------------ */
989 private static class ChannelAndAttachment
990 {
991 final SelectableChannel _channel;
992 final Object _attachment;
993
994 public ChannelAndAttachment(SelectableChannel channel, Object attachment)
995 {
996 super();
997 _channel = channel;
998 _attachment = attachment;
999 }
1000 }
1001
1002 /* ------------------------------------------------------------ */
1003 public boolean isDeferringInterestedOps0()
1004 {
1005 return _deferringInterestedOps0;
1006 }
1007
1008 /* ------------------------------------------------------------ */
1009 public void setDeferringInterestedOps0(boolean deferringInterestedOps0)
1010 {
1011 _deferringInterestedOps0 = deferringInterestedOps0;
1012 }
1013
1014 }