1 //
2 // ========================================================================
3 // Copyright (c) 1995-2012 Mort Bay Consulting Pty. Ltd.
4 // ------------------------------------------------------------------------
5 // All rights reserved. This program and the accompanying materials
6 // are made available under the terms of the Eclipse Public License v1.0
7 // and Apache License v2.0 which accompanies this distribution.
8 //
9 // The Eclipse Public License is available at
10 // http://www.eclipse.org/legal/epl-v10.html
11 //
12 // The Apache License v2.0 is available at
13 // http://www.opensource.org/licenses/apache2.0.php
14 //
15 // You may elect to redistribute this code under either of these licenses.
16 // ========================================================================
17 //
18
19 package org.eclipse.jetty.server;
20
21 import java.io.IOException;
22 import java.net.Socket;
23 import java.nio.ByteBuffer;
24 import java.util.ArrayList;
25 import java.util.Collection;
26 import java.util.LinkedHashMap;
27 import java.util.List;
28 import java.util.Map;
29 import java.util.concurrent.CountDownLatch;
30 import java.util.concurrent.Executor;
31 import java.util.concurrent.Future;
32 import java.util.concurrent.TimeUnit;
33
34 import org.eclipse.jetty.io.ArrayByteBufferPool;
35 import org.eclipse.jetty.io.ByteBufferPool;
36 import org.eclipse.jetty.io.Connection;
37 import org.eclipse.jetty.io.ssl.SslConnection;
38 import org.eclipse.jetty.util.FutureCallback;
39 import org.eclipse.jetty.util.annotation.ManagedAttribute;
40 import org.eclipse.jetty.util.annotation.ManagedObject;
41 import org.eclipse.jetty.util.component.ContainerLifeCycle;
42 import org.eclipse.jetty.util.component.Dumpable;
43 import org.eclipse.jetty.util.log.Log;
44 import org.eclipse.jetty.util.log.Logger;
45 import org.eclipse.jetty.util.ssl.SslContextFactory;
46 import org.eclipse.jetty.util.thread.Scheduler;
47 import org.eclipse.jetty.util.thread.TimerScheduler;
48
49 /**
50 * <p>An abstract implementation of {@link Connector} that provides a {@link ConnectionFactory} mechanism
51 * for creating {@link Connection} instances for various protocols (HTTP, SSL, SPDY, etc).</p>
52 *
53 * <h2>Connector Services</h2>
54 * The abstract connector manages the dependent services needed by all specific connector instances:
55 * <ul>
56 * <li>The {@link Executor} service is used to run all active tasks needed by this connector such as accepting connections
57 * or handle HTTP requests. The default is to use the {@link Server#getThreadPool()} as an executor.
58 * </li>
59 * <li>The {@link Scheduler} service is used to monitor the idle timeouts of all connections and is also made available
60 * to the connections to time such things as asynchronous request timeouts. The default is to use a new
61 * {@link TimerScheduler} instance.
62 * </li>
63 * <li>The {@link ByteBufferPool} service is made available to all connections to be used to acquire and release
64 * {@link ByteBuffer} instances from a pool. The default is to use a new {@link ArrayByteBufferPool} instance.
65 * </li>
66 * </ul>
67 * These services are managed as aggregate beans by the {@link ContainerLifeCycle} super class and
68 * may either be managed or unmanaged beans.
69 *
70 * <h2>Connection Factories</h2>
71 * The connector keeps a collection of {@link ConnectionFactory} instances, each of which are known by their
72 * protocol name. The protocol name may be a real protocol (eg http/1.1 or spdy/3) or it may be a private name
73 * that represents a special connection factory. For example, the name "SSL-http/1.1" is used for
74 * an {@link SslConnectionFactory} that has been instantiated with the {@link HttpConnectionFactory} as it's
75 * next protocol.
76 *
77 * <h4>Configuring Connection Factories</h4>
78 * The collection of available {@link ConnectionFactory} may be constructor injected or modified with the
79 * methods {@link #addConnectionFactory(ConnectionFactory)}, {@link #removeConnectionFactory(String)} and
80 * {@link #setConnectionFactories(Collection)}. Only a single {@link ConnectionFactory} instance may be configured
81 * per protocol name, so if two factories with the same {@link ConnectionFactory#getProtocol()} are set, then
82 * the second will replace the first.
83 * <p>
84 * The protocol factory used for newly accepted connections is specified by
85 * the method {@link #setDefaultProtocol(String)} or defaults to the protocol of the first configured factory.
86 * <p>
87 * Each Connection factory type is responsible for the configuration of the protocols that it accepts. Thus to
88 * configure the HTTP protocol, you pass a {@link HttpChannelConfig} instance to the {@link HttpConnectionFactory}
89 * (or the SPDY factories that can also provide HTTP Semantics). Similarly the {@link SslConnectionFactory} is
90 * configured by passing it a {@link SslContextFactory} and a next protocol name.
91 *
92 * <h4>Connection Factory Operation</h4>
93 * {@link ConnectionFactory}s may simply create a {@link Connection} instance to support a specific
94 * protocol. For example, the {@link HttpConnectionFactory} will create a {@link HttpConnection} instance
95 * that can handle http/1.1, http/1.0 and http/0.9.
96 * <p>
97 * {@link ConnectionFactory}s may also create a chain of {@link Connection} instances, using other {@link ConnectionFactory} instances.
98 * For example, the {@link SslConnectionFactory} is configured with a next protocol name, so that once it has accepted
99 * a connection and created an {@link SslConnection}, it then used the next {@link ConnectionFactory} from the
100 * connector using the {@link #getConnectionFactory(String)} method, to create a {@link Connection} instance that
101 * will handle the unecrypted bytes from the {@link SslConnection}. If the next protocol is "http/1.1", then the
102 * {@link SslConnectionFactory} will have a protocol name of "SSL-http/1.1" and lookup "http/1.1" for the protocol
103 * to run over the SSL connection.
104 * <p>
105 * {@link ConnectionFactory}s may also create temporary {@link Connection} instances that will exchange bytes
106 * over the connection to determine what is the next protocol to use. For example the NPN protocol is an extension
107 * of SSL to allow a protocol to be specified during the SSL handshake. NPN is used by the SPDY protocol to
108 * negotiate the version of SPDY or HTTP that the client and server will speak. Thus to accept a SPDY connection, the
109 * connector will be configured with {@link ConnectionFactory}s for "SSL-NPN", "NPN", "spdy/3", "spdy/2", "http/1.1"
110 * with the default protocol being "SSL-NPN". Thus a newly accepted connection uses "SSL-NPN", which specifies a
111 * SSLConnectionFactory with "NPN" as the next protocol. Thus an SslConnection instance is created chained to an NPNConnection
112 * instance. The NPN connection then negotiates with the client to determined the next protocol, which could be
113 * "spdy/3", "spdy/2" or the default of "http/1.1". Once the next protocol is determined, the NPN connection
114 * calls {@link #getConnectionFactory(String)} to create a connection instance that will replace the NPN connection as
115 * the connection chained to the SSLConnection.
116 * <p>
117 * <h2>Acceptors</h2>
118 * The connector will execute a number of acceptor tasks to the {@link Exception} service passed to the constructor.
119 * The acceptor tasks run in a loop while the connector is running and repeatedly call the abstract {@link #accept(int)} method.
120 * The implementation of the accept method must:
121 * <nl>
122 * <li>block waiting for new connections
123 * <li>accept the connection (eg socket accept)
124 * <li>perform any configuration of the connection (eg. socket linger times)
125 * <li>call the {@link #getDefaultConnectionFactory()} {@link ConnectionFactory#newConnection(Connector, org.eclipse.jetty.io.EndPoint)}
126 * method to create a new Connection instance.
127 * <li>call the {@link #connectionOpened(Connection)} method to signal a new connection has been created.
128 * <li>arrange for the {@link #connectionClosed(Connection)} method to be called once the connection is closed.
129 * </nl>
130 * The default number of acceptor tasks is the minimum of 1 and half the number of available CPUs. Having more acceptors may reduce
131 * the latency for servers that see a high rate of new connections (eg HTTP/1.0 without keep-alive). Typically the default is
132 * sufficient for modern persistent protocols (HTTP/1.1, SPDY etc.)
133 */
134 @ManagedObject("Abstract implementation of the Connector Interface")
135 public abstract class AbstractConnector extends ContainerLifeCycle implements Connector, Dumpable
136 {
137 protected final Logger LOG = Log.getLogger(getClass());
138 // Order is important on server side, so we use a LinkedHashMap
139 private final Map<String, ConnectionFactory> _factories = new LinkedHashMap<>();
140 private final Server _server;
141 private final Executor _executor;
142 private final Scheduler _scheduler;
143 private final ByteBufferPool _byteBufferPool;
144 private final Thread[] _acceptors;
145 private volatile CountDownLatch _stopping;
146 private long _idleTimeout = 200000;
147 private String _defaultProtocol;
148 private ConnectionFactory _defaultConnectionFactory;
149
150 /**
151 * @param server The server this connector will be added to. Must not be null.
152 * @param executor An executor for this connector or null to use the servers executor
153 * @param scheduler A scheduler for this connector or null to a new {@link TimerScheduler} instance.
154 * @param pool A buffer pool for this connector or null to use a default {@link ByteBufferPool}
155 * @param acceptors the number of acceptor threads to use, or 0 for a default value.
156 * @param factories The Connection Factories to use.
157 */
158 public AbstractConnector(
159 Server server,
160 Executor executor,
161 Scheduler scheduler,
162 ByteBufferPool pool,
163 int acceptors,
164 ConnectionFactory... factories)
165 {
166 _server=server;
167 _executor=executor!=null?executor:_server.getThreadPool();
168 _scheduler=scheduler!=null?scheduler:new TimerScheduler();
169 _byteBufferPool = pool!=null?pool:new ArrayByteBufferPool();
170
171 addBean(_server,false);
172 addBean(_executor);
173 if (executor==null)
174 unmanage(_executor); // inherited from server
175 addBean(_scheduler);
176 addBean(_byteBufferPool);
177
178 for (ConnectionFactory factory:factories)
179 addConnectionFactory(factory);
180
181 if (acceptors<=0)
182 acceptors=Math.max(1,(Runtime.getRuntime().availableProcessors()) / 2);
183 if (acceptors > 2 * Runtime.getRuntime().availableProcessors())
184 LOG.warn("Acceptors should be <= 2*availableProcessors: " + this);
185 _acceptors = new Thread[acceptors];
186 }
187
188
189 @Override
190 public Server getServer()
191 {
192 return _server;
193 }
194
195 @Override
196 public Executor getExecutor()
197 {
198 return _executor;
199 }
200
201 @Override
202 public ByteBufferPool getByteBufferPool()
203 {
204 return _byteBufferPool;
205 }
206
207 @Override
208 public long getIdleTimeout()
209 {
210 return _idleTimeout;
211 }
212
213 /**
214 * <p>Sets the maximum Idle time for a connection, which roughly translates to the {@link Socket#setSoTimeout(int)}
215 * call, although with NIO implementations other mechanisms may be used to implement the timeout.</p>
216 * <p>The max idle time is applied:</p>
217 * <ul>
218 * <li>When waiting for a new message to be received on a connection</li>
219 * <li>When waiting for a new message to be sent on a connection</li>
220 * </ul>
221 * <p>This value is interpreted as the maximum time between some progress being made on the connection.
222 * So if a single byte is read or written, then the timeout is reset.</p>
223 *
224 * @param idleTimeout the idle timeout
225 */
226 public void setIdleTimeout(long idleTimeout)
227 {
228 _idleTimeout = idleTimeout;
229 }
230
231 /**
232 * @return Returns the number of acceptor threads.
233 */
234 @ManagedAttribute("number of acceptor threads")
235 public int getAcceptors()
236 {
237 return _acceptors.length;
238 }
239
240 @Override
241 protected void doStart() throws Exception
242 {
243 _defaultConnectionFactory = getConnectionFactory(_defaultProtocol);
244 if(_defaultConnectionFactory==null)
245 throw new IllegalStateException("No protocol factory for default protocol: "+_defaultProtocol);
246
247 super.doStart();
248
249 _stopping=new CountDownLatch(_acceptors.length);
250 for (int i = 0; i < _acceptors.length; i++)
251 getExecutor().execute(new Acceptor(i));
252
253 LOG.info("Started {}", this);
254 }
255
256
257 protected void interruptAcceptors()
258 {
259 for (Thread thread : _acceptors)
260 {
261 if (thread != null)
262 thread.interrupt();
263 }
264 }
265
266 @Override
267 public <C> Future<C> shutdown(C c)
268 {
269 return new FutureCallback<C>(c);
270 }
271
272 @Override
273 protected void doStop() throws Exception
274 {
275 // Tell the acceptors we are stopping
276 interruptAcceptors();
277
278 // If we have a stop timeout
279 long stopTimeout = getStopTimeout();
280 if (stopTimeout > 0 && _stopping!=null)
281 _stopping.await(stopTimeout,TimeUnit.MILLISECONDS);
282 _stopping=null;
283
284 super.doStop();
285
286 LOG.info("Stopped {}", this);
287 }
288
289 public void join() throws InterruptedException
290 {
291 join(0);
292 }
293
294 public void join(long timeout) throws InterruptedException
295 {
296 for (Thread thread : _acceptors)
297 if (thread != null)
298 thread.join(timeout);
299 }
300
301 protected abstract void accept(int acceptorID) throws IOException, InterruptedException;
302
303
304 /* ------------------------------------------------------------ */
305 /**
306 * @return Is the connector accepting new connections
307 */
308 protected boolean isAccepting()
309 {
310 return isRunning();
311 }
312
313 @Override
314 public ConnectionFactory getConnectionFactory(String protocol)
315 {
316 synchronized (_factories)
317 {
318 return _factories.get(protocol.toLowerCase());
319 }
320 }
321
322 @Override
323 public <T extends ConnectionFactory> T getConnectionFactory(Class<T> factoryType)
324 {
325 synchronized (_factories)
326 {
327 for (ConnectionFactory f : _factories.values())
328 if (factoryType.isAssignableFrom(f.getClass()))
329 return (T)f;
330 return null;
331 }
332 }
333
334 public void addConnectionFactory(ConnectionFactory factory)
335 {
336 synchronized (_factories)
337 {
338 ConnectionFactory old=_factories.remove(factory.getProtocol());
339 if (old!=null)
340 removeBean(old);
341 _factories.put(factory.getProtocol().toLowerCase(), factory);
342 addBean(factory);
343 if (_defaultProtocol==null)
344 _defaultProtocol=factory.getProtocol();
345 }
346 }
347
348 public ConnectionFactory removeConnectionFactory(String protocol)
349 {
350 synchronized (_factories)
351 {
352 ConnectionFactory factory= _factories.remove(protocol.toLowerCase());
353 removeBean(factory);
354 return factory;
355 }
356 }
357
358 @Override
359 public Collection<ConnectionFactory> getConnectionFactories()
360 {
361 synchronized (_factories)
362 {
363 return _factories.values();
364 }
365 }
366
367 public void setConnectionFactories(Collection<ConnectionFactory> factories)
368 {
369 synchronized (_factories)
370 {
371 List<ConnectionFactory> existing = new ArrayList<>(_factories.values());
372 for (ConnectionFactory factory: existing)
373 removeConnectionFactory(factory.getProtocol());
374 for (ConnectionFactory factory: factories)
375 if (factory!=null)
376 addConnectionFactory(factory);
377 }
378 }
379
380
381 @Override
382 public List<String> getProtocols()
383 {
384 synchronized (_factories)
385 {
386 return new ArrayList<>(_factories.keySet());
387 }
388 }
389
390 public void clearConnectionFactories()
391 {
392 synchronized (_factories)
393 {
394 _factories.clear();
395 }
396 }
397
398 public String getDefaultProtocol()
399 {
400 return _defaultProtocol;
401 }
402
403 public void setDefaultProtocol(String defaultProtocol)
404 {
405 _defaultProtocol = defaultProtocol.toLowerCase();
406 if (isRunning())
407 _defaultConnectionFactory=getConnectionFactory(_defaultProtocol);
408 }
409
410 @Override
411 public ConnectionFactory getDefaultConnectionFactory()
412 {
413 if (isStarted())
414 return _defaultConnectionFactory;
415 return getConnectionFactory(_defaultProtocol);
416 }
417
418 private class Acceptor implements Runnable
419 {
420 private final int _acceptor;
421
422 private Acceptor(int id)
423 {
424 _acceptor = id;
425 }
426
427 @Override
428 public void run()
429 {
430 Thread current = Thread.currentThread();
431 String name = current.getName();
432 current.setName(name + "-acceptor-" + _acceptor + "-" + AbstractConnector.this);
433
434 synchronized (AbstractConnector.this)
435 {
436 _acceptors[_acceptor] = current;
437 }
438
439 try
440 {
441 while (isAccepting())
442 {
443 try
444 {
445 accept(_acceptor);
446 }
447 catch (Throwable e)
448 {
449 if (isAccepting())
450 LOG.warn(e);
451 else
452 LOG.debug(e);
453 }
454 }
455 }
456 finally
457 {
458 current.setName(name);
459
460 synchronized (AbstractConnector.this)
461 {
462 _acceptors[_acceptor] = null;
463 }
464 _stopping.countDown();
465 }
466 }
467 }
468
469 // protected void connectionOpened(Connection connection)
470 // {
471 // _stats.connectionOpened();
472 // connection.onOpen();
473 // }
474 //
475 // protected void connectionClosed(Connection connection)
476 // {
477 // connection.onClose();
478 // long duration = System.currentTimeMillis() - connection.getEndPoint().getCreatedTimeStamp();
479 // _stats.connectionClosed(duration, connection.getMessagesIn(), connection.getMessagesOut());
480 // }
481 //
482 // public void connectionUpgraded(Connection oldConnection, Connection newConnection)
483 // {
484 // oldConnection.onClose();
485 // _stats.connectionUpgraded(oldConnection.getMessagesIn(), oldConnection.getMessagesOut());
486 // newConnection.onOpen();
487 // }
488
489 @Override
490 public Scheduler getScheduler()
491 {
492 return _scheduler;
493 }
494
495 @Override
496 public String toString()
497 {
498 return String.format("%s@%x{%s}",
499 getClass().getSimpleName(),
500 hashCode(),
501 getDefaultProtocol());
502 }
503 }