Mercurial Hosting > luan
annotate src/org/eclipse/jetty/io/nio/SelectorManager.java @ 948:f5aefdc4a81a
simplify SelectChannelConnector
| author | Franklin Schmidt <fschmidt@gmail.com> |
|---|---|
| date | Tue, 11 Oct 2016 22:16:29 -0600 |
| parents | 1d24b6e422fa |
| children | e9088af3787f |
| rev | line source |
|---|---|
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
1 // |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
2 // ======================================================================== |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
3 // Copyright (c) 1995-2014 Mort Bay Consulting Pty. Ltd. |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
4 // ------------------------------------------------------------------------ |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
5 // All rights reserved. This program and the accompanying materials |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
6 // are made available under the terms of the Eclipse Public License v1.0 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
7 // and Apache License v2.0 which accompanies this distribution. |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
8 // |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
9 // The Eclipse Public License is available at |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
10 // http://www.eclipse.org/legal/epl-v10.html |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
11 // |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
12 // The Apache License v2.0 is available at |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
13 // http://www.opensource.org/licenses/apache2.0.php |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
14 // |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
15 // You may elect to redistribute this code under either of these licenses. |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
16 // ======================================================================== |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
17 // |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
18 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
19 package org.eclipse.jetty.io.nio; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
20 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
21 import java.io.IOException; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
22 import java.nio.channels.CancelledKeyException; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
23 import java.nio.channels.Channel; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
24 import java.nio.channels.ClosedSelectorException; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
25 import java.nio.channels.SelectableChannel; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
26 import java.nio.channels.SelectionKey; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
27 import java.nio.channels.Selector; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
28 import java.nio.channels.ServerSocketChannel; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
29 import java.nio.channels.SocketChannel; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
30 import java.util.ArrayList; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
31 import java.util.List; |
|
944
1d24b6e422fa
simplify SelectorManager
Franklin Schmidt <fschmidt@gmail.com>
parents:
943
diff
changeset
|
32 import java.util.Collections; |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
33 import java.util.Set; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
34 import java.util.concurrent.ConcurrentHashMap; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
35 import java.util.concurrent.ConcurrentLinkedQueue; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
36 import java.util.concurrent.ConcurrentMap; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
37 import java.util.concurrent.CountDownLatch; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
38 import java.util.concurrent.TimeUnit; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
39 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
40 import org.eclipse.jetty.io.AsyncEndPoint; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
41 import org.eclipse.jetty.io.ConnectedEndPoint; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
42 import org.eclipse.jetty.io.Connection; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
43 import org.eclipse.jetty.io.EndPoint; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
44 import org.eclipse.jetty.util.TypeUtil; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
45 import org.eclipse.jetty.util.component.AbstractLifeCycle; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
46 import org.eclipse.jetty.util.component.AggregateLifeCycle; |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
47 import org.eclipse.jetty.util.component.Dumpable; |
|
820
8e9db0bbf4f9
remove org.eclipse.jetty.util.log and upgrade slf4j
Franklin Schmidt <fschmidt@gmail.com>
parents:
802
diff
changeset
|
48 import org.slf4j.Logger; |
|
8e9db0bbf4f9
remove org.eclipse.jetty.util.log and upgrade slf4j
Franklin Schmidt <fschmidt@gmail.com>
parents:
802
diff
changeset
|
49 import org.slf4j.LoggerFactory; |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
50 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
51 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
52 /* ------------------------------------------------------------ */ |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
53 /** |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
54 * The Selector Manager manages and number of SelectSets to allow |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
55 * NIO scheduling to scale to large numbers of connections. |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
56 * <p> |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
57 */ |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
58 public abstract class SelectorManager extends AbstractLifeCycle implements Dumpable |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
59 { |
| 865 | 60 public static final Logger LOG=LoggerFactory.getLogger("org.eclipse.jetty.io.nio"); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
61 |
| 865 | 62 private static final int __MONITOR_PERIOD=Integer.getInteger("org.eclipse.jetty.io.nio.MONITOR_PERIOD",1000).intValue(); |
| 63 private static final int __MAX_SELECTS=Integer.getInteger("org.eclipse.jetty.io.nio.MAX_SELECTS",100000).intValue(); | |
| 64 private static final int __BUSY_PAUSE=Integer.getInteger("org.eclipse.jetty.io.nio.BUSY_PAUSE",50).intValue(); | |
| 65 private static final int __IDLE_TICK=Integer.getInteger("org.eclipse.jetty.io.nio.IDLE_TICK",400).intValue(); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
66 |
| 865 | 67 private int _maxIdleTime; |
| 68 private long _lowResourcesConnections; | |
| 69 private SelectSet[] _selectSet; | |
| 70 private int _selectSets=1; | |
| 71 private volatile int _set=0; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
72 |
| 865 | 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 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
82 |
| 865 | 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 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
93 |
| 865 | 94 /* ------------------------------------------------------------ */ |
| 95 /** Register a channel | |
| 96 * @param channel | |
| 97 */ | |
| 98 public void register(SocketChannel channel) | |
| 99 { | |
| 100 // The ++ increment here is not atomic, but it does not matter. | |
| 101 // so long as the value changes sometimes, then connections will | |
| 102 // be distributed over the available sets. | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
103 |
|
948
f5aefdc4a81a
simplify SelectChannelConnector
Franklin Schmidt <fschmidt@gmail.com>
parents:
944
diff
changeset
|
104 int s = _set++; |
| 865 | 105 if (s<0) |
| 106 s=-s; | |
| 107 s=s%_selectSets; | |
|
948
f5aefdc4a81a
simplify SelectChannelConnector
Franklin Schmidt <fschmidt@gmail.com>
parents:
944
diff
changeset
|
108 SelectSet[] sets = _selectSet; |
| 865 | 109 if (sets!=null) |
| 110 { | |
| 111 SelectSet set=sets[s]; | |
| 112 set.addChange(channel); | |
| 113 set.wakeup(); | |
| 114 } | |
| 115 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
116 |
| 865 | 117 /* ------------------------------------------------------------ */ |
| 118 /** Register a {@link ServerSocketChannel} | |
| 119 * @param acceptChannel | |
| 120 */ | |
| 121 public void register(ServerSocketChannel acceptChannel) | |
| 122 { | |
| 123 int s=_set++; | |
| 124 if (s<0) | |
| 125 s=-s; | |
| 126 s=s%_selectSets; | |
| 127 SelectSet set=_selectSet[s]; | |
| 128 set.addChange(acceptChannel); | |
| 129 set.wakeup(); | |
| 130 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
131 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
132 |
| 865 | 133 /* ------------------------------------------------------------ */ |
| 134 /** | |
| 135 * @return the lowResourcesConnections | |
| 136 */ | |
| 137 public long getLowResourcesConnections() | |
| 138 { | |
| 139 return _lowResourcesConnections*_selectSets; | |
| 140 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
141 |
| 865 | 142 /* ------------------------------------------------------------ */ |
| 143 /** | |
| 144 * Set the number of connections, which if exceeded places this manager in low resources state. | |
| 145 * This is not an exact measure as the connection count is averaged over the select sets. | |
| 146 * @param lowResourcesConnections the number of connections | |
| 147 * @see #setLowResourcesMaxIdleTime(long) | |
| 148 */ | |
| 149 public void setLowResourcesConnections(long lowResourcesConnections) | |
| 150 { | |
| 151 _lowResourcesConnections=(lowResourcesConnections+_selectSets-1)/_selectSets; | |
| 152 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
153 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
154 |
| 865 | 155 public abstract void execute(Runnable task); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
156 |
| 865 | 157 /* ------------------------------------------------------------ */ |
| 158 /* (non-Javadoc) | |
| 159 * @see org.eclipse.component.AbstractLifeCycle#doStart() | |
| 160 */ | |
| 161 @Override | |
| 162 protected void doStart() throws Exception | |
| 163 { | |
| 164 _selectSet = new SelectSet[_selectSets]; | |
| 165 for (int i=0;i<_selectSet.length;i++) | |
| 166 _selectSet[i]= new SelectSet(i); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
167 |
| 865 | 168 super.doStart(); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
169 |
| 865 | 170 // start a thread to Select |
|
948
f5aefdc4a81a
simplify SelectChannelConnector
Franklin Schmidt <fschmidt@gmail.com>
parents:
944
diff
changeset
|
171 for (int i=0;i<_selectSets;i++) |
| 865 | 172 { |
| 173 final int id=i; | |
| 174 execute(new Runnable() | |
| 175 { | |
| 176 public void run() | |
| 177 { | |
| 178 String name=Thread.currentThread().getName(); | |
| 179 try | |
| 180 { | |
| 181 SelectSet[] sets=_selectSet; | |
| 182 if (sets==null) | |
| 183 return; | |
| 184 SelectSet set=sets[id]; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
185 |
| 865 | 186 Thread.currentThread().setName(name+" Selector"+id); |
| 187 LOG.debug("Starting {} on {}",Thread.currentThread(),this); | |
| 188 while (isRunning()) | |
| 189 { | |
| 190 try | |
| 191 { | |
| 192 set.doSelect(); | |
| 193 } | |
| 194 catch(IOException e) | |
| 195 { | |
| 196 LOG.trace("",e); | |
| 197 } | |
| 198 catch(Exception e) | |
| 199 { | |
| 200 LOG.warn("",e); | |
| 201 } | |
| 202 } | |
| 203 } | |
| 204 finally | |
| 205 { | |
| 206 LOG.debug("Stopped {} on {}",Thread.currentThread(),this); | |
| 207 Thread.currentThread().setName(name); | |
| 208 } | |
| 209 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
210 |
| 865 | 211 }); |
| 212 } | |
| 213 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
214 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
215 |
| 865 | 216 /* ------------------------------------------------------------------------------- */ |
| 217 @Override | |
| 218 protected void doStop() throws Exception | |
| 219 { | |
| 220 SelectSet[] sets= _selectSet; | |
| 221 _selectSet=null; | |
| 222 if (sets!=null) | |
| 223 { | |
| 224 for (SelectSet set : sets) | |
| 225 { | |
| 226 if (set!=null) | |
| 227 set.stop(); | |
| 228 } | |
| 229 } | |
| 230 super.doStop(); | |
| 231 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
232 |
| 865 | 233 /* ------------------------------------------------------------------------------- */ |
| 234 public abstract AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
235 |
| 865 | 236 public String dump() |
| 237 { | |
| 238 return AggregateLifeCycle.dump(this); | |
| 239 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
240 |
| 865 | 241 public void dump(Appendable out, String indent) throws IOException |
| 242 { | |
| 243 AggregateLifeCycle.dumpObject(out,this); | |
| 244 AggregateLifeCycle.dump(out,indent,TypeUtil.asList(_selectSet)); | |
| 245 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
246 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
247 |
| 865 | 248 public class SelectSet implements Dumpable |
| 249 { | |
| 250 private final int _setID; | |
| 943 | 251 private volatile long _now = System.currentTimeMillis(); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
252 |
| 865 | 253 private final ConcurrentLinkedQueue<Object> _changes = new ConcurrentLinkedQueue<Object>(); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
254 |
| 865 | 255 private volatile Selector _selector; |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
256 |
| 865 | 257 private volatile Thread _selecting; |
| 258 private int _busySelects; | |
| 259 private long _monitorNext; | |
| 260 private boolean _pausing; | |
| 261 private boolean _paused; | |
| 262 private volatile long _idleTick; | |
| 263 private ConcurrentMap<SelectChannelEndPoint,Object> _endPoints = new ConcurrentHashMap<SelectChannelEndPoint, Object>(); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
264 |
| 865 | 265 SelectSet(int acceptorID) throws Exception |
| 266 { | |
| 267 _setID=acceptorID; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
268 |
| 865 | 269 _idleTick = System.currentTimeMillis(); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
270 |
| 865 | 271 // create a selector; |
| 272 _selector = Selector.open(); | |
| 273 _monitorNext=System.currentTimeMillis()+__MONITOR_PERIOD; | |
| 274 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
275 |
| 865 | 276 public void addChange(Object change) |
| 277 { | |
| 278 _changes.add(change); | |
| 279 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
280 |
| 865 | 281 /* ------------------------------------------------------------ */ |
| 282 /** | |
| 283 * Select and dispatch tasks found from changes and the selector. | |
| 284 * | |
| 285 * @throws IOException | |
| 286 */ | |
| 287 public void doSelect() throws IOException | |
| 288 { | |
| 289 try | |
| 290 { | |
| 291 _selecting=Thread.currentThread(); | |
| 292 final Selector selector=_selector; | |
| 293 // Stopped concurrently ? | |
| 294 if (selector == null) | |
| 295 return; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
296 |
| 865 | 297 // Make any key changes required |
| 298 Object change; | |
| 299 int changes=_changes.size(); | |
| 300 while (changes-->0 && (change=_changes.poll())!=null) | |
| 301 { | |
| 302 Channel ch=null; | |
| 303 SelectionKey key=null; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
304 |
| 865 | 305 try |
| 306 { | |
| 307 if (change instanceof EndPoint) | |
| 308 { | |
| 309 // Update the operations for a key. | |
| 310 SelectChannelEndPoint endpoint = (SelectChannelEndPoint)change; | |
| 311 ch=endpoint.getChannel(); | |
| 312 endpoint.doUpdateKey(); | |
| 313 } | |
| 314 else if (change instanceof SocketChannel) | |
| 315 { | |
| 316 // Newly registered channel | |
| 317 final SocketChannel channel=(SocketChannel)change; | |
| 318 ch=channel; | |
| 319 key = channel.register(selector,SelectionKey.OP_READ,null); | |
| 320 SelectChannelEndPoint endpoint = createEndPoint(channel,key); | |
| 321 key.attach(endpoint); | |
| 322 endpoint.schedule(); | |
| 323 } | |
| 324 else if (change instanceof Runnable) | |
| 325 { | |
| 326 execute((Runnable)change); | |
| 327 } | |
| 328 else | |
| 329 throw new IllegalArgumentException(change.toString()); | |
| 330 } | |
| 331 catch (CancelledKeyException e) | |
| 332 { | |
| 333 LOG.trace("",e); | |
| 334 } | |
| 335 catch (Throwable e) | |
| 336 { | |
| 337 if (isRunning()) | |
| 338 LOG.warn("",e); | |
| 339 else | |
| 340 LOG.debug("",e); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
341 |
| 865 | 342 try |
| 343 { | |
| 344 if (ch!=null) | |
| 345 ch.close(); | |
| 346 } | |
| 347 catch(IOException e2) | |
| 348 { | |
| 349 LOG.debug("",e2); | |
| 350 } | |
| 351 } | |
| 352 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
353 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
354 |
| 865 | 355 // Do and instant select to see if any connections can be handled. |
| 356 int selected=selector.selectNow(); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
357 |
| 943 | 358 _now = System.currentTimeMillis(); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
359 |
| 865 | 360 // if no immediate things to do |
| 361 if (selected==0 && selector.selectedKeys().isEmpty()) | |
| 362 { | |
| 363 // If we are in pausing mode | |
| 364 if (_pausing) | |
| 365 { | |
| 366 try | |
| 367 { | |
| 368 Thread.sleep(__BUSY_PAUSE); // pause to reduce impact of busy loop | |
| 369 } | |
| 370 catch(InterruptedException e) | |
| 371 { | |
| 372 LOG.trace("",e); | |
| 373 } | |
| 943 | 374 _now = System.currentTimeMillis(); |
| 865 | 375 } |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
376 |
| 865 | 377 // workout how long to wait in select |
| 378 long wait = _changes.size()==0?__IDLE_TICK:0L; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
379 |
| 865 | 380 // If we should wait with a select |
| 381 if (wait>0) | |
| 382 { | |
| 943 | 383 long before = _now; |
| 865 | 384 selector.select(wait); |
| 943 | 385 _now = System.currentTimeMillis(); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
386 |
| 865 | 387 // If we are monitoring for busy selector |
| 388 // and this select did not wait more than 1ms | |
| 943 | 389 if (__MONITOR_PERIOD>0 && _now-before <=1) |
| 865 | 390 { |
| 391 // count this as a busy select and if there have been too many this monitor cycle | |
| 392 if (++_busySelects>__MAX_SELECTS) | |
| 393 { | |
| 394 // Start injecting pauses | |
| 395 _pausing=true; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
396 |
| 865 | 397 // if this is the first pause |
| 398 if (!_paused) | |
| 399 { | |
| 400 // Log and dump some status | |
| 401 _paused=true; | |
| 402 LOG.warn("Selector {} is too busy, pausing!",this); | |
| 403 } | |
| 404 } | |
| 405 } | |
| 406 } | |
| 407 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
408 |
| 865 | 409 // have we been destroyed while sleeping |
| 410 if (_selector==null || !selector.isOpen()) | |
| 411 return; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
412 |
| 865 | 413 // Look for things to do |
| 414 for (SelectionKey key: selector.selectedKeys()) | |
| 415 { | |
| 416 SocketChannel channel=null; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
417 |
| 865 | 418 try |
| 419 { | |
| 420 if (!key.isValid()) | |
| 421 { | |
| 422 key.cancel(); | |
| 423 SelectChannelEndPoint endpoint = (SelectChannelEndPoint)key.attachment(); | |
| 424 if (endpoint != null) | |
| 425 endpoint.doUpdateKey(); | |
| 426 continue; | |
| 427 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
428 |
| 865 | 429 Object att = key.attachment(); |
| 430 if (att instanceof SelectChannelEndPoint) | |
| 431 { | |
| 432 if (key.isReadable()||key.isWritable()) | |
| 433 ((SelectChannelEndPoint)att).schedule(); | |
| 434 } | |
| 435 else if (key.isConnectable()) | |
| 436 { | |
| 437 // Complete a connection of a registered channel | |
| 438 channel = (SocketChannel)key.channel(); | |
| 439 boolean connected=false; | |
| 440 try | |
| 441 { | |
| 442 connected=channel.finishConnect(); | |
| 443 } | |
| 444 catch(Exception e) | |
| 445 { | |
|
948
f5aefdc4a81a
simplify SelectChannelConnector
Franklin Schmidt <fschmidt@gmail.com>
parents:
944
diff
changeset
|
446 LOG.warn(e+","+channel+","+att); |
|
f5aefdc4a81a
simplify SelectChannelConnector
Franklin Schmidt <fschmidt@gmail.com>
parents:
944
diff
changeset
|
447 LOG.debug("",e); |
| 865 | 448 } |
| 449 finally | |
| 450 { | |
| 451 if (connected) | |
| 452 { | |
| 453 key.interestOps(SelectionKey.OP_READ); | |
| 454 SelectChannelEndPoint endpoint = createEndPoint(channel,key); | |
| 455 key.attach(endpoint); | |
| 456 endpoint.schedule(); | |
| 457 } | |
| 458 else | |
| 459 { | |
| 460 key.cancel(); | |
| 461 channel.close(); | |
| 462 } | |
| 463 } | |
| 464 } | |
| 465 else | |
| 466 { | |
| 467 // Wrap readable registered channel in an endpoint | |
| 468 channel = (SocketChannel)key.channel(); | |
| 469 SelectChannelEndPoint endpoint = createEndPoint(channel,key); | |
| 470 key.attach(endpoint); | |
| 471 if (key.isReadable()) | |
| 472 endpoint.schedule(); | |
| 473 } | |
| 474 key = null; | |
| 475 } | |
| 476 catch (CancelledKeyException e) | |
| 477 { | |
| 478 LOG.trace("",e); | |
| 479 } | |
| 480 catch (Exception e) | |
| 481 { | |
| 482 if (isRunning()) | |
| 483 LOG.warn("",e); | |
| 484 else | |
| 485 LOG.trace("",e); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
486 |
| 865 | 487 try |
| 488 { | |
| 489 if (channel!=null) | |
| 490 channel.close(); | |
| 491 } | |
| 492 catch(IOException e2) | |
| 493 { | |
| 494 LOG.debug("",e2); | |
| 495 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
496 |
| 865 | 497 if (key != null && !(key.channel() instanceof ServerSocketChannel) && key.isValid()) |
| 498 key.cancel(); | |
| 499 } | |
| 500 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
501 |
| 865 | 502 // Everything always handled |
| 503 selector.selectedKeys().clear(); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
504 |
| 943 | 505 _now = System.currentTimeMillis(); |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
506 |
| 865 | 507 // Idle tick |
| 943 | 508 if (_now-_idleTick>__IDLE_TICK) |
| 865 | 509 { |
| 943 | 510 _idleTick = _now; |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
511 |
| 865 | 512 final long idle_now=((_lowResourcesConnections>0 && selector.keys().size()>_lowResourcesConnections)) |
| 943 | 513 ?(_now+_maxIdleTime) |
| 514 :_now; | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
515 |
| 865 | 516 execute(new Runnable() |
| 517 { | |
| 518 public void run() | |
| 519 { | |
| 520 for (SelectChannelEndPoint endp:_endPoints.keySet()) | |
| 521 { | |
| 522 endp.checkIdleTimestamp(idle_now); | |
| 523 } | |
| 524 } | |
| 525 public String toString() {return "Idle-"+super.toString();} | |
| 526 }); | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
527 |
| 865 | 528 } |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
529 |
| 865 | 530 // Reset busy select monitor counts |
| 943 | 531 if (__MONITOR_PERIOD>0 && _now>_monitorNext) |
| 865 | 532 { |
| 533 _busySelects=0; | |
| 534 _pausing=false; | |
| 943 | 535 _monitorNext=_now+__MONITOR_PERIOD; |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
536 |
| 865 | 537 } |
| 538 } | |
| 539 catch (ClosedSelectorException e) | |
| 540 { | |
| 541 if (isRunning()) | |
| 542 LOG.warn("",e); | |
| 543 else | |
| 544 LOG.trace("",e); | |
| 545 } | |
| 546 catch (CancelledKeyException e) | |
| 547 { | |
| 548 LOG.trace("",e); | |
| 549 } | |
| 550 finally | |
| 551 { | |
| 552 _selecting=null; | |
| 553 } | |
| 554 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
555 |
| 865 | 556 public SelectorManager getManager() |
| 557 { | |
| 558 return SelectorManager.this; | |
| 559 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
560 |
| 865 | 561 public long getNow() |
| 562 { | |
| 943 | 563 return _now; |
| 865 | 564 } |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
565 |
| 865 | 566 public void wakeup() |
| 567 { | |
|
944
1d24b6e422fa
simplify SelectorManager
Franklin Schmidt <fschmidt@gmail.com>
parents:
943
diff
changeset
|
568 Selector selector = _selector; |
|
1d24b6e422fa
simplify SelectorManager
Franklin Schmidt <fschmidt@gmail.com>
parents:
943
diff
changeset
|
569 if (selector!=null) |
|
1d24b6e422fa
simplify SelectorManager
Franklin Schmidt <fschmidt@gmail.com>
parents:
943
diff
changeset
|
570 selector.wakeup(); |
| 865 | 571 } |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
572 |
| 865 | 573 private SelectChannelEndPoint createEndPoint(SocketChannel channel, SelectionKey sKey) throws IOException |
| 574 { | |
|
948
f5aefdc4a81a
simplify SelectChannelConnector
Franklin Schmidt <fschmidt@gmail.com>
parents:
944
diff
changeset
|
575 SelectChannelEndPoint endp = new SelectChannelEndPoint(channel,this,sKey, _maxIdleTime); |
|
f5aefdc4a81a
simplify SelectChannelConnector
Franklin Schmidt <fschmidt@gmail.com>
parents:
944
diff
changeset
|
576 endp.setConnection(getManager().newConnection(channel,endp, sKey.attachment())); |
| 865 | 577 LOG.debug("created {}",endp); |
| 578 _endPoints.put(endp,this); | |
| 579 return endp; | |
| 580 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
581 |
| 865 | 582 public void destroyEndPoint(SelectChannelEndPoint endp) |
| 583 { | |
| 584 LOG.debug("destroyEndPoint {}",endp); | |
| 585 _endPoints.remove(endp); | |
|
948
f5aefdc4a81a
simplify SelectChannelConnector
Franklin Schmidt <fschmidt@gmail.com>
parents:
944
diff
changeset
|
586 endp.getConnection().onClose(); |
| 865 | 587 } |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
588 |
| 865 | 589 Selector getSelector() |
| 590 { | |
| 591 return _selector; | |
| 592 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
593 |
| 865 | 594 void stop() throws Exception |
| 595 { | |
| 596 // Spin for a while waiting for selector to complete | |
| 597 // to avoid unneccessary closed channel exceptions | |
| 598 try | |
| 599 { | |
| 600 for (int i=0;i<100 && _selecting!=null;i++) | |
| 601 { | |
| 602 wakeup(); | |
| 603 Thread.sleep(10); | |
| 604 } | |
| 605 } | |
| 606 catch(Exception e) | |
| 607 { | |
|
944
1d24b6e422fa
simplify SelectorManager
Franklin Schmidt <fschmidt@gmail.com>
parents:
943
diff
changeset
|
608 LOG.warn("",e); |
| 865 | 609 } |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
610 |
| 865 | 611 // close endpoints and selector |
| 612 synchronized (this) | |
| 613 { | |
| 614 Selector selector=_selector; | |
| 615 for (SelectionKey key:selector.keys()) | |
| 616 { | |
| 617 if (key==null) | |
| 618 continue; | |
| 619 Object att=key.attachment(); | |
| 620 if (att instanceof EndPoint) | |
| 621 { | |
| 622 EndPoint endpoint = (EndPoint)att; | |
| 623 try | |
| 624 { | |
| 625 endpoint.close(); | |
| 626 } | |
| 627 catch(IOException e) | |
| 628 { | |
| 629 LOG.trace("",e); | |
| 630 } | |
| 631 } | |
| 632 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
633 |
| 865 | 634 try |
| 635 { | |
| 636 selector=_selector; | |
| 637 if (selector != null) | |
| 638 selector.close(); | |
| 639 } | |
| 640 catch (IOException e) | |
| 641 { | |
| 642 LOG.trace("",e); | |
| 643 } | |
|
944
1d24b6e422fa
simplify SelectorManager
Franklin Schmidt <fschmidt@gmail.com>
parents:
943
diff
changeset
|
644 _selector = null; |
| 865 | 645 } |
| 646 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
647 |
| 865 | 648 public String dump() |
| 649 { | |
| 650 return AggregateLifeCycle.dump(this); | |
| 651 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
652 |
| 865 | 653 public void dump(Appendable out, String indent) throws IOException |
| 654 { | |
| 655 out.append(String.valueOf(this)).append(" id=").append(String.valueOf(_setID)).append("\n"); | |
|
944
1d24b6e422fa
simplify SelectorManager
Franklin Schmidt <fschmidt@gmail.com>
parents:
943
diff
changeset
|
656 AggregateLifeCycle.dump(out,indent,Collections.emptyList()); |
| 865 | 657 } |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
658 |
| 865 | 659 public String toString() |
| 660 { | |
| 661 Selector selector=_selector; | |
| 662 return String.format("%s keys=%d selected=%d", | |
| 663 super.toString(), | |
| 664 selector != null && selector.isOpen() ? selector.keys().size() : -1, | |
| 665 selector != null && selector.isOpen() ? selector.selectedKeys().size() : -1); | |
| 666 } | |
| 667 } | |
|
802
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
668 |
|
3428c60d7cfc
replace jetty jars with source
Franklin Schmidt <fschmidt@gmail.com>
parents:
diff
changeset
|
669 } |