View Javadoc

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