View Javadoc

1   /*
2    * ====================================================================
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *   http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing,
14   * software distributed under the License is distributed on an
15   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16   * KIND, either express or implied.  See the License for the
17   * specific language governing permissions and limitations
18   * under the License.
19   * ====================================================================
20   *
21   * This software consists of voluntary contributions made by many
22   * individuals on behalf of the Apache Software Foundation.  For more
23   * information on the Apache Software Foundation, please see
24   * <http://www.apache.org/>.
25   *
26   */
27  
28  package org.apache.http.impl.nio.reactor;
29  
30  import java.io.IOException;
31  import java.io.InterruptedIOException;
32  import java.net.Socket;
33  import java.nio.channels.Channel;
34  import java.nio.channels.ClosedChannelException;
35  import java.nio.channels.ClosedSelectorException;
36  import java.nio.channels.SelectableChannel;
37  import java.nio.channels.SelectionKey;
38  import java.nio.channels.Selector;
39  import java.util.ArrayList;
40  import java.util.Date;
41  import java.util.List;
42  import java.util.concurrent.ThreadFactory;
43  
44  import org.apache.http.annotation.ThreadSafe;
45  import org.apache.http.nio.params.NIOReactorPNames;
46  import org.apache.http.nio.reactor.IOEventDispatch;
47  import org.apache.http.nio.reactor.IOReactor;
48  import org.apache.http.nio.reactor.IOReactorException;
49  import org.apache.http.nio.reactor.IOReactorExceptionHandler;
50  import org.apache.http.nio.reactor.IOReactorStatus;
51  import org.apache.http.params.BasicHttpParams;
52  import org.apache.http.params.CoreConnectionPNames;
53  import org.apache.http.params.HttpParams;
54  import org.apache.http.util.Args;
55  import org.apache.http.util.Asserts;
56  
57  /**
58   * Generic implementation of {@link IOReactor} that can run multiple
59   * {@link BaseIOReactor} instance in separate worker threads and distribute
60   * newly created I/O session equally across those I/O reactors for a more
61   * optimal resource utilization and a better I/O performance. Usually it is
62   * recommended to have one worker I/O reactor per physical CPU core.
63   * <p>
64   * <strong>Important note about exception handling</strong>
65   * <p>
66   * Protocol specific exceptions as well as those I/O exceptions thrown in the
67   * course of interaction with the session's channel are to be expected are to be
68   * dealt with by specific protocol handlers. These exceptions may result in
69   * termination of an individual session but should not affect the I/O reactor
70   * and all other active sessions. There are situations, however, when the I/O
71   * reactor itself encounters an internal problem such as an I/O exception in
72   * the underlying NIO classes or an unhandled runtime exception. Those types of
73   * exceptions are usually fatal and will cause the I/O reactor to shut down
74   * automatically.
75   * <p>
76   * There is a possibility to override this behavior and prevent I/O reactors
77   * from shutting down automatically in case of a runtime exception or an I/O
78   * exception in internal classes. This can be accomplished by providing a custom
79   * implementation of the {@link IOReactorExceptionHandler} interface.
80   * <p>
81   * If an I/O reactor is unable to automatically recover from an I/O or a runtime
82   * exception it will enter the shutdown mode. First off, it cancel all pending
83   * new session requests. Then it will attempt to close all active I/O sessions
84   * gracefully giving them some time to flush pending output data and terminate
85   * cleanly. Lastly, it will forcibly shut down those I/O sessions that still
86   * remain active after the grace period. This is a fairly complex process, where
87   * many things can fail at the same time and many different exceptions can be
88   * thrown in the course of the shutdown process. The I/O reactor will record all
89   * exceptions thrown during the shutdown process, including the original one
90   * that actually caused the shutdown in the first place, in an audit log. One
91   * can obtain the audit log using {@link #getAuditLog()}, examine exceptions
92   * thrown by the I/O reactor prior and in the course of the reactor shutdown
93   * and decide whether it is safe to restart the I/O reactor.
94   *
95   * @since 4.0
96   */
97  @SuppressWarnings("deprecation")
98  @ThreadSafe // public methods only
99  public abstract class AbstractMultiworkerIOReactor implements IOReactor {
100 
101     protected volatile IOReactorStatus status;
102 
103     /**
104      * @deprecated (4.2)
105      */
106     @Deprecated
107     protected final HttpParams params;
108     protected final IOReactorConfig config;
109     protected final Selector selector;
110     protected final long selectTimeout;
111     protected final boolean interestOpsQueueing;
112 
113     private final int workerCount;
114     private final ThreadFactory threadFactory;
115     private final BaseIOReactor[] dispatchers;
116     private final Worker[] workers;
117     private final Thread[] threads;
118     private final Object statusLock;
119 
120     protected IOReactorExceptionHandler exceptionHandler;
121     protected List<ExceptionEvent> auditLog;
122 
123     private int currentWorker = 0;
124 
125     /**
126      * Creates an instance of AbstractMultiworkerIOReactor with the given configuration.
127      *
128      * @param config I/O reactor configuration.
129      * @param threadFactory the factory to create threads.
130      *   Can be <code>null</code>.
131      * @throws IOReactorException in case if a non-recoverable I/O error.
132      *
133      * @since 4.2
134      */
135     public AbstractMultiworkerIOReactor(
136             final IOReactorConfig config,
137             final ThreadFactory threadFactory) throws IOReactorException {
138         super();
139         this.config = config != null ? config : IOReactorConfig.DEFAULT;
140         this.params = new BasicHttpParams();
141         try {
142             this.selector = Selector.open();
143         } catch (final IOException ex) {
144             throw new IOReactorException("Failure opening selector", ex);
145         }
146         this.selectTimeout = this.config.getSelectInterval();
147         this.interestOpsQueueing = this.config.isInterestOpQueued();
148         this.statusLock = new Object();
149         if (threadFactory != null) {
150             this.threadFactory = threadFactory;
151         } else {
152             this.threadFactory = new DefaultThreadFactory();
153         }
154         this.workerCount = this.config.getIoThreadCount();
155         this.dispatchers = new BaseIOReactor[workerCount];
156         this.workers = new Worker[workerCount];
157         this.threads = new Thread[workerCount];
158         this.status = IOReactorStatus.INACTIVE;
159     }
160 
161     /**
162      * Creates an instance of AbstractMultiworkerIOReactor with default configuration.
163      *
164      * @throws IOReactorException in case if a non-recoverable I/O error.
165      *
166      * @since 4.2
167      */
168     public AbstractMultiworkerIOReactor() throws IOReactorException {
169         this(null, null);
170     }
171 
172     @Deprecated
173     static IOReactorConfig convert(final int workerCount, final HttpParams params) {
174         Args.notNull(params, "HTTP parameters");
175         return IOReactorConfig.custom()
176             .setSelectInterval(params.getLongParameter(NIOReactorPNames.SELECT_INTERVAL, 1000))
177             .setShutdownGracePeriod(params.getLongParameter(NIOReactorPNames.GRACE_PERIOD, 500))
178             .setInterestOpQueued(params.getBooleanParameter(NIOReactorPNames.SELECT_INTERVAL, false))
179             .setIoThreadCount(workerCount)
180             .setSoTimeout(params.getIntParameter(CoreConnectionPNames.SO_TIMEOUT, 0))
181             .setConnectTimeout(params.getIntParameter(CoreConnectionPNames.CONNECTION_TIMEOUT, 0))
182             .setSoTimeout(params.getIntParameter(CoreConnectionPNames.SO_TIMEOUT, 0))
183             .setSoReuseAddress(params.getBooleanParameter(CoreConnectionPNames.SO_REUSEADDR, false))
184             .setSoKeepAlive(params.getBooleanParameter(CoreConnectionPNames.SO_KEEPALIVE, false))
185             .setSoLinger(params.getIntParameter(CoreConnectionPNames.SO_LINGER, -1))
186             .setTcpNoDelay(params.getBooleanParameter(CoreConnectionPNames.TCP_NODELAY, true))
187             .build();
188     }
189 
190     /**
191      * Creates an instance of AbstractMultiworkerIOReactor.
192      *
193      * @param workerCount number of worker I/O reactors.
194      * @param threadFactory the factory to create threads.
195      *   Can be <code>null</code>.
196      * @param params HTTP parameters.
197      * @throws IOReactorException in case if a non-recoverable I/O error.
198      *
199      * @deprecated (4.2) use {@link AbstractMultiworkerIOReactor#AbstractMultiworkerIOReactor(IOReactorConfig, ThreadFactory)}
200      */
201     @Deprecated
202     public AbstractMultiworkerIOReactor(
203             final int workerCount,
204             final ThreadFactory threadFactory,
205             final HttpParams params) throws IOReactorException {
206         this(convert(workerCount, params), threadFactory);
207     }
208 
209     public IOReactorStatus getStatus() {
210         return this.status;
211     }
212 
213     /**
214      * Returns the audit log containing exceptions thrown by the I/O reactor
215      * prior and in the course of the reactor shutdown.
216      *
217      * @return audit log.
218      */
219     public synchronized List<ExceptionEvent> getAuditLog() {
220         if (this.auditLog != null) {
221             return new ArrayList<ExceptionEvent>(this.auditLog);
222         } else {
223             return null;
224         }
225     }
226 
227     /**
228      * Adds the given {@link Throwable} object with the given time stamp
229      * to the audit log.
230      *
231      * @param ex the exception thrown by the I/O reactor.
232      * @param timestamp the time stamp of the exception. Can be
233      * <code>null</code> in which case the current date / time will be used.
234      */
235     protected synchronized void addExceptionEvent(final Throwable ex, final Date timestamp) {
236         if (ex == null) {
237             return;
238         }
239         if (this.auditLog == null) {
240             this.auditLog = new ArrayList<ExceptionEvent>();
241         }
242         this.auditLog.add(new ExceptionEvent(ex, timestamp != null ? timestamp : new Date()));
243     }
244 
245     /**
246      * Adds the given {@link Throwable} object to the audit log.
247      *
248      * @param ex the exception thrown by the I/O reactor.
249      */
250     protected void addExceptionEvent(final Throwable ex) {
251         addExceptionEvent(ex, null);
252     }
253 
254     /**
255      * Sets exception handler for this I/O reactor.
256      *
257      * @param exceptionHandler the exception handler.
258      */
259     public void setExceptionHandler(final IOReactorExceptionHandler exceptionHandler) {
260         this.exceptionHandler = exceptionHandler;
261     }
262 
263     /**
264      * Triggered to process I/O events registered by the main {@link Selector}.
265      * <p>
266      * Super-classes can implement this method to react to the event.
267      *
268      * @param count event count.
269      * @throws IOReactorException in case if a non-recoverable I/O error.
270      */
271     protected abstract void processEvents(int count) throws IOReactorException;
272 
273     /**
274      * Triggered to cancel pending session requests.
275      * <p>
276      * Super-classes can implement this method to react to the event.
277      *
278      * @throws IOReactorException in case if a non-recoverable I/O error.
279      */
280     protected abstract void cancelRequests() throws IOReactorException;
281 
282     /**
283      * Activates the main I/O reactor as well as all worker I/O reactors.
284      * The I/O main reactor will start reacting to I/O events and triggering
285      * notification methods. The worker I/O reactor in their turn will start
286      * reacting to I/O events and dispatch I/O event notifications to the given
287      * {@link IOEventDispatch} interface.
288      * <p>
289      * This method will enter the infinite I/O select loop on
290      * the {@link Selector} instance associated with this I/O reactor and used
291      * to manage creation of new I/O channels. Once a new I/O channel has been
292      * created the processing of I/O events on that channel will be delegated
293      * to one of the worker I/O reactors.
294      * <p>
295      * The method will remain blocked unto the I/O reactor is shut down or the
296      * execution thread is interrupted.
297      *
298      * @see #processEvents(int)
299      * @see #cancelRequests()
300      *
301      * @throws InterruptedIOException if the dispatch thread is interrupted.
302      * @throws IOReactorException in case if a non-recoverable I/O error.
303      */
304     public void execute(
305             final IOEventDispatch eventDispatch) throws InterruptedIOException, IOReactorException {
306         Args.notNull(eventDispatch, "Event dispatcher");
307         synchronized (this.statusLock) {
308             if (this.status.compareTo(IOReactorStatus.SHUTDOWN_REQUEST) >= 0) {
309                 this.status = IOReactorStatus.SHUT_DOWN;
310                 this.statusLock.notifyAll();
311                 return;
312             }
313             Asserts.check(this.status.compareTo(IOReactorStatus.INACTIVE) == 0,
314                     "Illegal state %s", this.status);
315             this.status = IOReactorStatus.ACTIVE;
316             // Start I/O dispatchers
317             for (int i = 0; i < this.dispatchers.length; i++) {
318                 final BaseIOReactor dispatcher = new BaseIOReactor(this.selectTimeout, this.interestOpsQueueing);
319                 dispatcher.setExceptionHandler(exceptionHandler);
320                 this.dispatchers[i] = dispatcher;
321             }
322             for (int i = 0; i < this.workerCount; i++) {
323                 final BaseIOReactor dispatcher = this.dispatchers[i];
324                 this.workers[i] = new Worker(dispatcher, eventDispatch);
325                 this.threads[i] = this.threadFactory.newThread(this.workers[i]);
326             }
327         }
328         try {
329 
330             for (int i = 0; i < this.workerCount; i++) {
331                 if (this.status != IOReactorStatus.ACTIVE) {
332                     return;
333                 }
334                 this.threads[i].start();
335             }
336 
337             for (;;) {
338                 final int readyCount;
339                 try {
340                     readyCount = this.selector.select(this.selectTimeout);
341                 } catch (final InterruptedIOException ex) {
342                     throw ex;
343                 } catch (final IOException ex) {
344                     throw new IOReactorException("Unexpected selector failure", ex);
345                 }
346 
347                 if (this.status.compareTo(IOReactorStatus.ACTIVE) == 0) {
348                     processEvents(readyCount);
349                 }
350 
351                 // Verify I/O dispatchers
352                 for (int i = 0; i < this.workerCount; i++) {
353                     final Worker worker = this.workers[i];
354                     final Exception ex = worker.getException();
355                     if (ex != null) {
356                         throw new IOReactorException(
357                                 "I/O dispatch worker terminated abnormally", ex);
358                     }
359                 }
360 
361                 if (this.status.compareTo(IOReactorStatus.ACTIVE) > 0) {
362                     break;
363                 }
364             }
365 
366         } catch (final ClosedSelectorException ex) {
367             addExceptionEvent(ex);
368         } catch (final IOReactorException ex) {
369             if (ex.getCause() != null) {
370                 addExceptionEvent(ex.getCause());
371             }
372             throw ex;
373         } finally {
374             doShutdown();
375             synchronized (this.statusLock) {
376                 this.status = IOReactorStatus.SHUT_DOWN;
377                 this.statusLock.notifyAll();
378             }
379         }
380     }
381 
382     /**
383      * Activates the shutdown sequence for this reactor. This method will cancel
384      * all pending session requests, close out all active I/O channels,
385      * make an attempt to terminate all worker I/O reactors gracefully,
386      * and finally force-terminate those I/O reactors that failed to
387      * terminate after the specified grace period.
388      *
389      * @throws InterruptedIOException if the shutdown sequence has been
390      *   interrupted.
391      */
392     protected void doShutdown() throws InterruptedIOException {
393         synchronized (this.statusLock) {
394             if (this.status.compareTo(IOReactorStatus.SHUTTING_DOWN) >= 0) {
395                 return;
396             }
397             this.status = IOReactorStatus.SHUTTING_DOWN;
398         }
399         try {
400             cancelRequests();
401         } catch (final IOReactorException ex) {
402             if (ex.getCause() != null) {
403                 addExceptionEvent(ex.getCause());
404             }
405         }
406         this.selector.wakeup();
407 
408         // Close out all channels
409         if (this.selector.isOpen()) {
410             for (final SelectionKey key : this.selector.keys()) {
411                 try {
412                     final Channel channel = key.channel();
413                     if (channel != null) {
414                         channel.close();
415                     }
416                 } catch (final IOException ex) {
417                     addExceptionEvent(ex);
418                 }
419             }
420             // Stop dispatching I/O events
421             try {
422                 this.selector.close();
423             } catch (final IOException ex) {
424                 addExceptionEvent(ex);
425             }
426         }
427 
428         // Attempt to shut down I/O dispatchers gracefully
429         for (int i = 0; i < this.workerCount; i++) {
430             final BaseIOReactor dispatcher = this.dispatchers[i];
431             dispatcher.gracefulShutdown();
432         }
433 
434         final long gracePeriod = this.config.getShutdownGracePeriod();
435 
436         try {
437             // Force shut down I/O dispatchers if they fail to terminate
438             // in time
439             for (int i = 0; i < this.workerCount; i++) {
440                 final BaseIOReactor dispatcher = this.dispatchers[i];
441                 if (dispatcher.getStatus() != IOReactorStatus.INACTIVE) {
442                     dispatcher.awaitShutdown(gracePeriod);
443                 }
444                 if (dispatcher.getStatus() != IOReactorStatus.SHUT_DOWN) {
445                     try {
446                         dispatcher.hardShutdown();
447                     } catch (final IOReactorException ex) {
448                         if (ex.getCause() != null) {
449                             addExceptionEvent(ex.getCause());
450                         }
451                     }
452                 }
453             }
454             // Join worker threads
455             for (int i = 0; i < this.workerCount; i++) {
456                 final Thread t = this.threads[i];
457                 if (t != null) {
458                     t.join(gracePeriod);
459                 }
460             }
461         } catch (final InterruptedException ex) {
462             throw new InterruptedIOException(ex.getMessage());
463         }
464     }
465 
466     /**
467      * Assigns the given channel entry to one of the worker I/O reactors.
468      *
469      * @param entry the channel entry.
470      */
471     protected void addChannel(final ChannelEntry entry) {
472         // Distribute new channels among the workers
473         final int i = Math.abs(this.currentWorker++ % this.workerCount);
474         this.dispatchers[i].addChannel(entry);
475     }
476 
477     /**
478      * Registers the given channel with the main {@link Selector}.
479      *
480      * @param channel the channel.
481      * @param ops interest ops.
482      * @return  selection key.
483      * @throws ClosedChannelException if the channel has been already closed.
484      */
485     protected SelectionKey registerChannel(
486             final SelectableChannel channel, final int ops) throws ClosedChannelException {
487         return channel.register(this.selector, ops);
488     }
489 
490     /**
491      * Prepares the given {@link Socket} by resetting some of its properties.
492      *
493      * @param socket the socket
494      * @throws IOException in case of an I/O error.
495      */
496     protected void prepareSocket(final Socket socket) throws IOException {
497         socket.setTcpNoDelay(this.config.isTcpNoDelay());
498         socket.setKeepAlive(this.config.isSoKeepalive());
499         if (this.config.getSoTimeout() > 0) {
500             socket.setSoTimeout(this.config.getSoTimeout());
501         }
502         if (this.config.getSndBufSize() > 0) {
503             socket.setSendBufferSize(this.config.getSndBufSize());
504         }
505         if (this.config.getRcvBufSize() > 0) {
506             socket.setReceiveBufferSize(this.config.getRcvBufSize());
507         }
508         final int linger = this.config.getSoLinger();
509         if (linger >= 0) {
510             socket.setSoLinger(linger > 0, linger);
511         }
512     }
513 
514     /**
515      * Blocks for the given period of time in milliseconds awaiting
516      * the completion of the reactor shutdown. If the value of
517      * <code>timeout</code> is set to <code>0</code> this method blocks
518      * indefinitely.
519      *
520      * @param timeout the maximum wait time.
521      * @throws InterruptedException if interrupted.
522      */
523     protected void awaitShutdown(final long timeout) throws InterruptedException {
524         synchronized (this.statusLock) {
525             final long deadline = System.currentTimeMillis() + timeout;
526             long remaining = timeout;
527             while (this.status != IOReactorStatus.SHUT_DOWN) {
528                 this.statusLock.wait(remaining);
529                 if (timeout > 0) {
530                     remaining = deadline - System.currentTimeMillis();
531                     if (remaining <= 0) {
532                         break;
533                     }
534                 }
535             }
536         }
537     }
538 
539     public void shutdown() throws IOException {
540         shutdown(2000);
541     }
542 
543     public void shutdown(final long waitMs) throws IOException {
544         synchronized (this.statusLock) {
545             if (this.status.compareTo(IOReactorStatus.ACTIVE) > 0) {
546                 return;
547             }
548             if (this.status.compareTo(IOReactorStatus.INACTIVE) == 0) {
549                 this.status = IOReactorStatus.SHUT_DOWN;
550                 cancelRequests();
551                 return;
552             }
553             this.status = IOReactorStatus.SHUTDOWN_REQUEST;
554         }
555         this.selector.wakeup();
556         try {
557             awaitShutdown(waitMs);
558         } catch (final InterruptedException ignore) {
559         }
560     }
561 
562     static void closeChannel(final Channel channel) {
563         try {
564             channel.close();
565         } catch (final IOException ignore) {
566         }
567     }
568 
569     static class Worker implements Runnable {
570 
571         final BaseIOReactor dispatcher;
572         final IOEventDispatch eventDispatch;
573 
574         private volatile Exception exception;
575 
576         public Worker(final BaseIOReactor dispatcher, final IOEventDispatch eventDispatch) {
577             super();
578             this.dispatcher = dispatcher;
579             this.eventDispatch = eventDispatch;
580         }
581 
582         public void run() {
583             try {
584                 this.dispatcher.execute(this.eventDispatch);
585             } catch (final Exception ex) {
586                 this.exception = ex;
587             }
588         }
589 
590         public Exception getException() {
591             return this.exception;
592         }
593 
594     }
595 
596     static class DefaultThreadFactory implements ThreadFactory {
597 
598         private static volatile int COUNT = 0;
599 
600         public Thread newThread(final Runnable r) {
601             return new Thread(r, "I/O dispatcher " + (++COUNT));
602         }
603 
604     }
605 
606 }