1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.io;
20
21 import java.io.Closeable;
22 import java.io.IOException;
23 import java.net.ConnectException;
24 import java.net.SocketTimeoutException;
25 import java.nio.channels.CancelledKeyException;
26 import java.nio.channels.SelectionKey;
27 import java.nio.channels.Selector;
28 import java.nio.channels.ServerSocketChannel;
29 import java.nio.channels.SocketChannel;
30 import java.util.ArrayDeque;
31 import java.util.ArrayList;
32 import java.util.Collections;
33 import java.util.Iterator;
34 import java.util.List;
35 import java.util.Queue;
36 import java.util.Set;
37 import java.util.concurrent.CountDownLatch;
38 import java.util.concurrent.TimeUnit;
39 import java.util.concurrent.atomic.AtomicBoolean;
40
41 import org.eclipse.jetty.util.component.AbstractLifeCycle;
42 import org.eclipse.jetty.util.component.ContainerLifeCycle;
43 import org.eclipse.jetty.util.component.Dumpable;
44 import org.eclipse.jetty.util.log.Log;
45 import org.eclipse.jetty.util.log.Logger;
46 import org.eclipse.jetty.util.thread.ExecutionStrategy;
47 import org.eclipse.jetty.util.thread.Locker;
48 import org.eclipse.jetty.util.thread.Scheduler;
49
50
51
52
53
54
55
56 public class ManagedSelector extends AbstractLifeCycle implements Runnable, Dumpable
57 {
58 private static final Logger LOG = Log.getLogger(ManagedSelector.class);
59
60 private final Locker _locker = new Locker();
61 private boolean _selecting = false;
62 private final Queue<Runnable> _actions = new ArrayDeque<>();
63 private final SelectorManager _selectorManager;
64 private final int _id;
65 private final ExecutionStrategy _strategy;
66 private Selector _selector;
67
68 public ManagedSelector(SelectorManager selectorManager, int id)
69 {
70 this(selectorManager, id, ExecutionStrategy.Factory.getDefault());
71 }
72
73 public ManagedSelector(SelectorManager selectorManager, int id, ExecutionStrategy.Factory executionFactory)
74 {
75 _selectorManager = selectorManager;
76 _id = id;
77 _strategy = executionFactory.newExecutionStrategy(new SelectorProducer(), selectorManager.getExecutor());
78 setStopTimeout(5000);
79 }
80
81 public ExecutionStrategy getExecutionStrategy()
82 {
83 return _strategy;
84 }
85
86 @Override
87 protected void doStart() throws Exception
88 {
89 super.doStart();
90 _selector = newSelector();
91 _selectorManager.execute(this);
92 }
93
94 protected Selector newSelector() throws IOException
95 {
96 return Selector.open();
97 }
98
99 public int size()
100 {
101 Selector s = _selector;
102 if (s == null)
103 return 0;
104 return s.keys().size();
105 }
106
107 @Override
108 protected void doStop() throws Exception
109 {
110 if (LOG.isDebugEnabled())
111 LOG.debug("Stopping {}", this);
112 CloseEndPoints close_endps = new CloseEndPoints();
113 submit(close_endps);
114 close_endps.await(getStopTimeout());
115 super.doStop();
116 CloseSelector close_selector = new CloseSelector();
117 submit(close_selector);
118 close_selector.await(getStopTimeout());
119
120 if (LOG.isDebugEnabled())
121 LOG.debug("Stopped {}", this);
122 }
123
124 public void submit(Runnable change)
125 {
126 if (LOG.isDebugEnabled())
127 LOG.debug("Queued change {} on {}", change, this);
128
129 Selector selector = null;
130 try (Locker.Lock lock = _locker.lock())
131 {
132 _actions.offer(change);
133 if (_selecting)
134 {
135 selector = _selector;
136
137 _selecting = false;
138 }
139 }
140 if (selector != null)
141 selector.wakeup();
142 }
143
144 @Override
145 public void run()
146 {
147 _strategy.execute();
148 }
149
150
151
152
153
154 public interface SelectableEndPoint extends EndPoint
155 {
156
157
158
159
160
161
162 Runnable onSelected();
163
164
165
166
167
168 void updateKey();
169 }
170
171 private class SelectorProducer implements ExecutionStrategy.Producer
172 {
173 private Set<SelectionKey> _keys = Collections.emptySet();
174 private Iterator<SelectionKey> _cursor = Collections.emptyIterator();
175
176 @Override
177 public Runnable produce()
178 {
179 while (true)
180 {
181 Runnable task = processSelected();
182 if (task != null)
183 return task;
184
185 Runnable action = runActions();
186 if (action != null)
187 return action;
188
189 update();
190
191 if (!select())
192 return null;
193 }
194 }
195
196 private Runnable runActions()
197 {
198 while (true)
199 {
200 Runnable action;
201 try (Locker.Lock lock = _locker.lock())
202 {
203 action = _actions.poll();
204 if (action == null)
205 {
206
207 _selecting = true;
208 return null;
209 }
210 }
211
212 if (action instanceof Product)
213 return action;
214
215
216 runChange(action);
217 }
218 }
219
220 private void runChange(Runnable change)
221 {
222 try
223 {
224 if (LOG.isDebugEnabled())
225 LOG.debug("Running change {}", change);
226 change.run();
227 }
228 catch (Throwable x)
229 {
230 LOG.debug("Could not run change " + change, x);
231 }
232 }
233
234 private boolean select()
235 {
236 try
237 {
238 Selector selector = _selector;
239 if (selector != null && selector.isOpen())
240 {
241 if (LOG.isDebugEnabled())
242 LOG.debug("Selector loop waiting on select");
243 int selected = selector.select();
244 if (LOG.isDebugEnabled())
245 LOG.debug("Selector loop woken up from select, {}/{} selected", selected, selector.keys().size());
246
247 try (Locker.Lock lock = _locker.lock())
248 {
249
250 _selecting = false;
251 }
252
253 _keys = selector.selectedKeys();
254 _cursor = _keys.iterator();
255
256 return true;
257 }
258 }
259 catch (Throwable x)
260 {
261 closeNoExceptions(_selector);
262 if (isRunning())
263 LOG.warn(x);
264 else
265 LOG.debug(x);
266 }
267 return false;
268 }
269
270 private Runnable processSelected()
271 {
272 while (_cursor.hasNext())
273 {
274 SelectionKey key = _cursor.next();
275 if (key.isValid())
276 {
277 Object attachment = key.attachment();
278 try
279 {
280 if (attachment instanceof SelectableEndPoint)
281 {
282
283 Runnable task = ((SelectableEndPoint)attachment).onSelected();
284 if (task != null)
285 return task;
286 }
287 else if (key.isConnectable())
288 {
289 Runnable task = processConnect(key, (Connect)attachment);
290 if (task != null)
291 return task;
292 }
293 else if (key.isAcceptable())
294 {
295 processAccept(key);
296 }
297 else
298 {
299 throw new IllegalStateException("key=" + key + ", att=" + attachment + ", iOps=" + key.interestOps() + ", rOps=" + key.readyOps());
300 }
301 }
302 catch (CancelledKeyException x)
303 {
304 LOG.debug("Ignoring cancelled key for channel {}", key.channel());
305 if (attachment instanceof EndPoint)
306 closeNoExceptions((EndPoint)attachment);
307 }
308 catch (Throwable x)
309 {
310 LOG.warn("Could not process key for channel " + key.channel(), x);
311 if (attachment instanceof EndPoint)
312 closeNoExceptions((EndPoint)attachment);
313 }
314 }
315 else
316 {
317 if (LOG.isDebugEnabled())
318 LOG.debug("Selector loop ignoring invalid key for channel {}", key.channel());
319 Object attachment = key.attachment();
320 if (attachment instanceof EndPoint)
321 closeNoExceptions((EndPoint)attachment);
322 }
323 }
324 return null;
325 }
326
327 private void update()
328 {
329 for (SelectionKey key : _keys)
330 updateKey(key);
331 _keys.clear();
332 }
333
334 private void updateKey(SelectionKey key)
335 {
336 Object attachment = key.attachment();
337 if (attachment instanceof SelectableEndPoint)
338 ((SelectableEndPoint)attachment).updateKey();
339 }
340 }
341
342 private interface Product extends Runnable
343 {
344 }
345
346 private Runnable processConnect(SelectionKey key, final Connect connect)
347 {
348 SocketChannel channel = (SocketChannel)key.channel();
349 try
350 {
351 key.attach(connect.attachment);
352 boolean connected = _selectorManager.finishConnect(channel);
353 if (LOG.isDebugEnabled())
354 LOG.debug("Connected {} {}", connected, channel);
355 if (connected)
356 {
357 if (connect.timeout.cancel())
358 {
359 key.interestOps(0);
360 return new CreateEndPoint(channel, key)
361 {
362 @Override
363 protected void failed(Throwable failure)
364 {
365 super.failed(failure);
366 connect.failed(failure);
367 }
368 };
369 }
370 else
371 {
372 throw new SocketTimeoutException("Concurrent Connect Timeout");
373 }
374 }
375 else
376 {
377 throw new ConnectException();
378 }
379 }
380 catch (Throwable x)
381 {
382 connect.failed(x);
383 return null;
384 }
385 }
386
387 private void processAccept(SelectionKey key)
388 {
389 ServerSocketChannel server = (ServerSocketChannel)key.channel();
390 SocketChannel channel = null;
391 try
392 {
393 while ((channel = server.accept()) != null)
394 {
395 _selectorManager.accepted(channel);
396 }
397 }
398 catch (Throwable x)
399 {
400 closeNoExceptions(channel);
401 LOG.warn("Accept failed for channel " + channel, x);
402 }
403 }
404
405 private void closeNoExceptions(Closeable closeable)
406 {
407 try
408 {
409 if (closeable != null)
410 closeable.close();
411 }
412 catch (Throwable x)
413 {
414 LOG.ignore(x);
415 }
416 }
417
418 private EndPoint createEndPoint(SocketChannel channel, SelectionKey selectionKey) throws IOException
419 {
420 EndPoint endPoint = _selectorManager.newEndPoint(channel, this, selectionKey);
421 _selectorManager.endPointOpened(endPoint);
422 Connection connection = _selectorManager.newConnection(channel, endPoint, selectionKey.attachment());
423 endPoint.setConnection(connection);
424 selectionKey.attach(endPoint);
425 _selectorManager.connectionOpened(connection);
426 if (LOG.isDebugEnabled())
427 LOG.debug("Created {}", endPoint);
428 return endPoint;
429 }
430
431 public void destroyEndPoint(final EndPoint endPoint)
432 {
433 final Connection connection = endPoint.getConnection();
434 submit(new Product()
435 {
436 @Override
437 public void run()
438 {
439 if (LOG.isDebugEnabled())
440 LOG.debug("Destroyed {}", endPoint);
441 if (connection != null)
442 _selectorManager.connectionClosed(connection);
443 _selectorManager.endPointClosed(endPoint);
444 }
445 });
446 }
447
448 @Override
449 public String dump()
450 {
451 return ContainerLifeCycle.dump(this);
452 }
453
454 @Override
455 public void dump(Appendable out, String indent) throws IOException
456 {
457 out.append(String.valueOf(this)).append(" id=").append(String.valueOf(_id)).append(System.lineSeparator());
458
459 Selector selector = _selector;
460 if (selector != null && selector.isOpen())
461 {
462 final ArrayList<Object> dump = new ArrayList<>(selector.keys().size() * 2);
463
464 DumpKeys dumpKeys = new DumpKeys(dump);
465 submit(dumpKeys);
466 dumpKeys.await(5, TimeUnit.SECONDS);
467
468 ContainerLifeCycle.dump(out, indent, dump);
469 }
470 }
471
472 @Override
473 public String toString()
474 {
475 Selector selector = _selector;
476 return String.format("%s id=%s keys=%d selected=%d",
477 super.toString(),
478 _id,
479 selector != null && selector.isOpen() ? selector.keys().size() : -1,
480 selector != null && selector.isOpen() ? selector.selectedKeys().size() : -1);
481 }
482
483 private class DumpKeys implements Runnable
484 {
485 private final CountDownLatch latch = new CountDownLatch(1);
486 private final List<Object> _dumps;
487
488 private DumpKeys(List<Object> dumps)
489 {
490 this._dumps = dumps;
491 }
492
493 @Override
494 public void run()
495 {
496 Selector selector = _selector;
497 if (selector != null && selector.isOpen())
498 {
499 Set<SelectionKey> keys = selector.keys();
500 _dumps.add(selector + " keys=" + keys.size());
501 for (SelectionKey key : keys)
502 {
503 try
504 {
505 _dumps.add(String.format("SelectionKey@%x{i=%d}->%s", key.hashCode(), key.interestOps(), key.attachment()));
506 }
507 catch (Throwable x)
508 {
509 LOG.ignore(x);
510 }
511 }
512 }
513 latch.countDown();
514 }
515
516 public boolean await(long timeout, TimeUnit unit)
517 {
518 try
519 {
520 return latch.await(timeout, unit);
521 }
522 catch (InterruptedException x)
523 {
524 return false;
525 }
526 }
527 }
528
529 class Acceptor implements Runnable
530 {
531 private final ServerSocketChannel _channel;
532
533 public Acceptor(ServerSocketChannel channel)
534 {
535 this._channel = channel;
536 }
537
538 @Override
539 public void run()
540 {
541 try
542 {
543 SelectionKey key = _channel.register(_selector, SelectionKey.OP_ACCEPT, null);
544 if (LOG.isDebugEnabled())
545 LOG.debug("{} acceptor={}", this, key);
546 }
547 catch (Throwable x)
548 {
549 closeNoExceptions(_channel);
550 LOG.warn(x);
551 }
552 }
553 }
554
555 class Accept implements Runnable, Closeable
556 {
557 private final SocketChannel channel;
558 private final Object attachment;
559
560 Accept(SocketChannel channel, Object attachment)
561 {
562 this.channel = channel;
563 this.attachment = attachment;
564 }
565
566 @Override
567 public void close()
568 {
569 LOG.debug("closed accept of {}", channel);
570 closeNoExceptions(channel);
571 }
572
573 @Override
574 public void run()
575 {
576 try
577 {
578 final SelectionKey key = channel.register(_selector, 0, attachment);
579 submit(new CreateEndPoint(channel, key));
580 }
581 catch (Throwable x)
582 {
583 closeNoExceptions(channel);
584 LOG.debug(x);
585 }
586 }
587 }
588
589 private class CreateEndPoint implements Product, Closeable
590 {
591 private final SocketChannel channel;
592 private final SelectionKey key;
593
594 public CreateEndPoint(SocketChannel channel, SelectionKey key)
595 {
596 this.channel = channel;
597 this.key = key;
598 }
599
600 @Override
601 public void run()
602 {
603 try
604 {
605 createEndPoint(channel, key);
606 }
607 catch (Throwable x)
608 {
609 LOG.debug(x);
610 failed(x);
611 }
612 }
613
614 @Override
615 public void close()
616 {
617 LOG.debug("closed creation of {}", channel);
618 closeNoExceptions(channel);
619 }
620
621 protected void failed(Throwable failure)
622 {
623 closeNoExceptions(channel);
624 LOG.debug(failure);
625 }
626 }
627
628 class Connect implements Runnable
629 {
630 private final AtomicBoolean failed = new AtomicBoolean();
631 private final SocketChannel channel;
632 private final Object attachment;
633 private final Scheduler.Task timeout;
634
635 Connect(SocketChannel channel, Object attachment)
636 {
637 this.channel = channel;
638 this.attachment = attachment;
639 this.timeout = ManagedSelector.this._selectorManager.getScheduler().schedule(new ConnectTimeout(this), ManagedSelector.this._selectorManager.getConnectTimeout(), TimeUnit.MILLISECONDS);
640 }
641
642 @Override
643 public void run()
644 {
645 try
646 {
647 channel.register(_selector, SelectionKey.OP_CONNECT, this);
648 }
649 catch (Throwable x)
650 {
651 failed(x);
652 }
653 }
654
655 private void failed(Throwable failure)
656 {
657 if (failed.compareAndSet(false, true))
658 {
659 timeout.cancel();
660 closeNoExceptions(channel);
661 ManagedSelector.this._selectorManager.connectionFailed(channel, failure, attachment);
662 }
663 }
664 }
665
666 private class ConnectTimeout implements Runnable
667 {
668 private final Connect connect;
669
670 private ConnectTimeout(Connect connect)
671 {
672 this.connect = connect;
673 }
674
675 @Override
676 public void run()
677 {
678 SocketChannel channel = connect.channel;
679 if (channel.isConnectionPending())
680 {
681 if (LOG.isDebugEnabled())
682 LOG.debug("Channel {} timed out while connecting, closing it", channel);
683 connect.failed(new SocketTimeoutException("Connect Timeout"));
684 }
685 }
686 }
687
688 private class CloseEndPoints implements Runnable
689 {
690 private final CountDownLatch _latch = new CountDownLatch(1);
691 private CountDownLatch _allClosed;
692
693 @Override
694 public void run()
695 {
696 List<EndPoint> end_points = new ArrayList<>();
697 for (SelectionKey key : _selector.keys())
698 {
699 if (key.isValid())
700 {
701 Object attachment = key.attachment();
702 if (attachment instanceof EndPoint)
703 end_points.add((EndPoint)attachment);
704 }
705 }
706
707 int size = end_points.size();
708 if (LOG.isDebugEnabled())
709 LOG.debug("Closing {} endPoints on {}", size, ManagedSelector.this);
710
711 _allClosed = new CountDownLatch(size);
712 _latch.countDown();
713
714 for (EndPoint endp : end_points)
715 submit(new EndPointCloser(endp, _allClosed));
716
717 if (LOG.isDebugEnabled())
718 LOG.debug("Closed {} endPoints on {}", size, ManagedSelector.this);
719 }
720
721 public boolean await(long timeout)
722 {
723 try
724 {
725 return _latch.await(timeout, TimeUnit.MILLISECONDS) &&
726 _allClosed.await(timeout, TimeUnit.MILLISECONDS);
727 }
728 catch (InterruptedException x)
729 {
730 return false;
731 }
732 }
733 }
734
735 private class EndPointCloser implements Product
736 {
737 private final EndPoint _endPoint;
738 private final CountDownLatch _latch;
739
740 private EndPointCloser(EndPoint endPoint, CountDownLatch latch)
741 {
742 _endPoint = endPoint;
743 _latch = latch;
744 }
745
746 @Override
747 public void run()
748 {
749 closeNoExceptions(_endPoint.getConnection());
750 _latch.countDown();
751 }
752 }
753
754 private class CloseSelector implements Runnable
755 {
756 private CountDownLatch _latch = new CountDownLatch(1);
757
758 @Override
759 public void run()
760 {
761 Selector selector = _selector;
762 _selector = null;
763 closeNoExceptions(selector);
764 _latch.countDown();
765 }
766
767 public boolean await(long timeout)
768 {
769 try
770 {
771 return _latch.await(timeout, TimeUnit.MILLISECONDS);
772 }
773 catch (InterruptedException x)
774 {
775 return false;
776 }
777 }
778 }
779 }