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