activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cshan...@apache.org
Subject activemq git commit: https://issues.apache.org/jira/browse/AMQ-6535
Date Mon, 12 Dec 2016 16:26:40 GMT
Repository: activemq
Updated Branches:
  refs/heads/activemq-5.14.x 80b5fdcf2 -> ae30662fb


https://issues.apache.org/jira/browse/AMQ-6535

Fixing the auto+nio+ssl transport so that the protocol detection task
will properly terminate on timeout and not continue to run. Also
lowered the default detection timeout to 15 seconds instead of 30
seconds to match the InactivityMonitor default.

(cherry picked from commit 27238b2dd790cea38d95da46b23ed141f5847782)


Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/ae30662f
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/ae30662f
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/ae30662f

Branch: refs/heads/activemq-5.14.x
Commit: ae30662fbdc2bc263e7c7c2e3cda3f8bbb063907
Parents: 80b5fdc
Author: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Authored: Mon Dec 12 11:15:43 2016 -0500
Committer: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Committed: Mon Dec 12 11:26:26 2016 -0500

----------------------------------------------------------------------
 .../transport/auto/AutoTcpTransportServer.java  | 86 ++++++++++++++++----
 .../auto/nio/AutoNIOSSLTransportServer.java     | 33 ++++++--
 .../transport/nio/AutoInitNioSSLTransport.java  | 19 +----
 .../activemq/transport/nio/NIOSSLTransport.java | 18 ----
 4 files changed, 97 insertions(+), 59 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/ae30662f/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportServer.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportServer.java
b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportServer.java
index 8eeb6ac..8b9a73f 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportServer.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/AutoTcpTransportServer.java
@@ -27,8 +27,6 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -77,8 +75,10 @@ public class AutoTcpTransportServer extends TcpTransportServer {
 
     protected BrokerService brokerService;
 
+    protected final ThreadPoolExecutor newConnectionExecutor;
+    protected final ThreadPoolExecutor protocolDetectionExecutor;
     protected int maxConnectionThreadPoolSize = Integer.MAX_VALUE;
-    protected int protocolDetectionTimeOut = 30000;
+    protected int protocolDetectionTimeOut = 15000;
 
     private static final FactoryFinder TRANSPORT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/");
     private final ConcurrentMap<String, TransportFactory> transportFactories = new
ConcurrentHashMap<String, TransportFactory>();
@@ -157,12 +157,21 @@ public class AutoTcpTransportServer extends TcpTransportServer {
 
         //Use an executor service here to handle new connections.  Setting the max number
         //of threads to the maximum number of connections the thread count isn't unbounded
-        service = new ThreadPoolExecutor(maxConnectionThreadPoolSize,
+        newConnectionExecutor = new ThreadPoolExecutor(maxConnectionThreadPoolSize,
                 maxConnectionThreadPoolSize,
                 30L, TimeUnit.SECONDS,
                 new LinkedBlockingQueue<Runnable>());
         //allow the thread pool to shrink if the max number of threads isn't needed
-        service.allowCoreThreadTimeOut(true);
+        //and the pool can grow and shrink as needed if contention is high
+        newConnectionExecutor.allowCoreThreadTimeOut(true);
+
+        //Executor for waiting for bytes to detection of protocol
+        protocolDetectionExecutor = new ThreadPoolExecutor(maxConnectionThreadPoolSize,
+                maxConnectionThreadPoolSize,
+                30L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<Runnable>());
+        //allow the thread pool to shrink if the max number of threads isn't needed
+        protocolDetectionExecutor.allowCoreThreadTimeOut(true);
 
         this.brokerService = brokerService;
         this.enabledProtocols = enabledProtocols;
@@ -173,10 +182,32 @@ public class AutoTcpTransportServer extends TcpTransportServer {
         return maxConnectionThreadPoolSize;
     }
 
+    /**
+     * Set the number of threads to be used for processing connections.  Defaults
+     * to Integer.MAX_SIZE.  Set this value to be lower to reduce the
+     * number of simultaneous connection attempts.  If not set then the maximum number of
+     * threads will generally be controlled by the transport maxConnections setting:
+     * {@link TcpTransportServer#setMaximumConnections(int)}.
+     *<p>
+     * Note that this setter controls two thread pools because connection attempts
+     * require 1 thread to start processing the connection and another thread to read from
the
+     * socket and to detect the protocol. Two threads are needed because some transports
+     * block on socket read so the first thread needs to be able to abort the second thread
on timeout.
+     * Therefore this setting will set each thread pool to the size passed in essentially
giving
+     * 2 times as many potential threads as the value set.
+     *<p>
+     * Both thread pools will close idle threads after a period of time
+     * essentially allowing the thread pools to grow and shrink dynamically based on load.
+     *
+     * @see {@link TcpTransportServer#setMaximumConnections(int)}.
+     * @param maxConnectionThreadPoolSize
+     */
     public void setMaxConnectionThreadPoolSize(int maxConnectionThreadPoolSize) {
         this.maxConnectionThreadPoolSize = maxConnectionThreadPoolSize;
-        service.setCorePoolSize(maxConnectionThreadPoolSize);
-        service.setMaximumPoolSize(maxConnectionThreadPoolSize);
+        newConnectionExecutor.setCorePoolSize(maxConnectionThreadPoolSize);
+        newConnectionExecutor.setMaximumPoolSize(maxConnectionThreadPoolSize);
+        protocolDetectionExecutor.setCorePoolSize(maxConnectionThreadPoolSize);
+        protocolDetectionExecutor.setMaximumPoolSize(maxConnectionThreadPoolSize);
     }
 
     public void setProtocolDetectionTimeOut(int protocolDetectionTimeOut) {
@@ -219,16 +250,13 @@ public class AutoTcpTransportServer extends TcpTransportServer {
         return enabledProtocols == null || enabledProtocols.isEmpty();
     }
 
-
-    protected final ThreadPoolExecutor service;
-
     @Override
     protected void handleSocket(final Socket socket) {
         final AutoTcpTransportServer server = this;
         //This needs to be done in a new thread because
         //the socket might be waiting on the client to send bytes
         //doHandleSocket can't complete until the protocol can be detected
-        service.submit(new Runnable() {
+        newConnectionExecutor.submit(new Runnable() {
             @Override
             public void run() {
                 server.doHandleSocket(socket);
@@ -239,30 +267,37 @@ public class AutoTcpTransportServer extends TcpTransportServer {
     @Override
     protected TransportInfo configureTransport(final TcpTransportServer server, final Socket
socket) throws Exception {
         final InputStream is = socket.getInputStream();
-        ExecutorService executor = Executors.newSingleThreadExecutor();
-
         final AtomicInteger readBytes = new AtomicInteger(0);
         final ByteBuffer data = ByteBuffer.allocate(8);
+
         // We need to peak at the first 8 bytes of the buffer to detect the protocol
-        Future<?> future = executor.submit(new Runnable() {
+        Future<?> future = protocolDetectionExecutor.submit(new Runnable() {
             @Override
             public void run() {
                 try {
                     do {
+                        //will block until enough bytes or read or a timeout
+                        //and the socket is closed
                         int read = is.read();
                         if (read == -1) {
                             throw new IOException("Connection failed, stream is closed.");
                         }
                         data.put((byte) read);
                         readBytes.incrementAndGet();
-                    } while (readBytes.get() < 8);
+                    } while (readBytes.get() < 8 && !Thread.interrupted());
                 } catch (Exception e) {
                     throw new IllegalStateException(e);
                 }
             }
         });
 
-        waitForProtocolDetectionFinish(future, readBytes);
+        try {
+            //If this fails and throws an exception and the socket will be closed
+            waitForProtocolDetectionFinish(future, readBytes);
+        } finally {
+            //call cancel in case task didn't complete
+            future.cancel(true);
+        }
         data.flip();
         ProtocolInfo protocolInfo = detectProtocol(data.array());
 
@@ -320,8 +355,23 @@ public class AutoTcpTransportServer extends TcpTransportServer {
     }
     @Override
     protected void doStop(ServiceStopper stopper) throws Exception {
-        if (service != null) {
-            service.shutdown();
+        if (newConnectionExecutor != null) {
+            newConnectionExecutor.shutdownNow();
+            try {
+                if (!newConnectionExecutor.awaitTermination(3, TimeUnit.SECONDS)) {
+                    LOG.warn("Auto Transport newConnectionExecutor didn't shutdown cleanly");
+                }
+            } catch (InterruptedException e) {
+            }
+        }
+        if (protocolDetectionExecutor != null) {
+            protocolDetectionExecutor.shutdownNow();
+            try {
+                if (!protocolDetectionExecutor.awaitTermination(3, TimeUnit.SECONDS)) {
+                    LOG.warn("Auto Transport protocolDetectionExecutor didn't shutdown cleanly");
+                }
+            } catch (InterruptedException e) {
+            }
         }
         super.doStop(stopper);
     }

http://git-wip-us.apache.org/repos/asf/activemq/blob/ae30662f/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNIOSSLTransportServer.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNIOSSLTransportServer.java
b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNIOSSLTransportServer.java
index cb38d7e..572352e 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNIOSSLTransportServer.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/transport/auto/nio/AutoNIOSSLTransportServer.java
@@ -7,8 +7,6 @@ import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 import javax.net.ServerSocketFactory;
@@ -101,8 +99,6 @@ public class AutoNIOSSLTransportServer extends AutoTcpTransportServer {
 
     @Override
     protected TransportInfo configureTransport(final TcpTransportServer server, final Socket
socket) throws Exception {
-        ExecutorService executor = Executors.newSingleThreadExecutor();
-
         //The SSLEngine needs to be initialized and handshake done to get the first command
and detect the format
         //The wireformat doesn't need properties set here because we aren't using this format
during the SSL handshake
         final AutoInitNioSSLTransport in = new AutoInitNioSSLTransport(wireFormatFactory.createWireFormat(),
socket);
@@ -117,17 +113,38 @@ public class AutoNIOSSLTransportServer extends AutoTcpTransportServer
{
         in.start();
         SSLEngine engine = in.getSslSession();
 
-        Future<?> future = executor.submit(new Runnable() {
+        //Attempt to read enough bytes to detect the protocol until the timeout period
+        //is reached
+        Future<?> future = protocolDetectionExecutor.submit(new Runnable() {
             @Override
             public void run() {
-                //Wait for handshake to finish initializing
+                int attempts = 0;
                 do {
+                    if(attempts > 0) {
+                        try {
+                            //increase sleep period each attempt to prevent high cpu usage
+                            //if the client is hung and not sending bytes
+                            int sleep = attempts >= 1024 ? 1024 : 4 * attempts;
+                            Thread.sleep(sleep);
+                        } catch (InterruptedException e) {
+                            break;
+                        }
+                    }
+                    //In the future it might be better to register a nonblocking selector
+                    //to be told when bytes are ready
                     in.serviceRead();
-                } while(in.getReadSize().get() < 8);
+                    attempts++;
+                } while(in.getReadSize().get() < 8 && !Thread.interrupted());
             }
         });
 
-        waitForProtocolDetectionFinish(future, in.getReadSize());
+        try {
+            //If this fails and throws an exception and the socket will be closed
+            waitForProtocolDetectionFinish(future, in.getReadSize());
+        } finally {
+            //call cancel in case task didn't complete which will interrupt the task
+            future.cancel(true);
+        }
         in.stop();
 
         InitBuffer initBuffer = new InitBuffer(in.getReadSize().get(), ByteBuffer.allocate(in.getReadData().length));

http://git-wip-us.apache.org/repos/asf/activemq/blob/ae30662f/activemq-broker/src/main/java/org/apache/activemq/transport/nio/AutoInitNioSSLTransport.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/nio/AutoInitNioSSLTransport.java
b/activemq-broker/src/main/java/org/apache/activemq/transport/nio/AutoInitNioSSLTransport.java
index 0bf4145..449c7ae 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/transport/nio/AutoInitNioSSLTransport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/transport/nio/AutoInitNioSSLTransport.java
@@ -135,21 +135,6 @@ public class AutoInitNioSSLTransport extends NIOSSLTransport {
 
     }
 
-
-    @Override
-    protected void finishHandshake() throws Exception {
-        if (handshakeInProgress) {
-            handshakeInProgress = false;
-            nextFrameSize = -1;
-
-            // Once handshake completes we need to ask for the now real sslSession
-            // otherwise the session would return 'SSL_NULL_WITH_NULL_NULL' for the
-            // cipher suite.
-            sslSession = sslEngine.getSession();
-
-        }
-    }
-
     public SSLEngine getSslSession() {
         return this.sslEngine;
     }
@@ -180,6 +165,10 @@ public class AutoInitNioSSLTransport extends NIOSSLTransport {
                 if (!plain.hasRemaining()) {
                     int readCount = secureRead(plain);
 
+                    if (readCount == 0) {
+                        break;
+                    }
+
                     // channel is closed, cleanup
                     if (readCount == -1) {
                         onException(new EOFException());

http://git-wip-us.apache.org/repos/asf/activemq/blob/ae30662f/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
index 3bceb0e..64e96be 100644
--- a/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
+++ b/activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
@@ -156,7 +156,6 @@ public class NIOSSLTransport extends NIOTransport {
                 doHandshake();
             }
 
-           // if (hasSslEngine) {
             selection = SelectorManager.getInstance().register(channel, new SelectorManager.Listener()
{
                 @Override
                 public void onSelect(SelectorSelection selection) {
@@ -233,23 +232,6 @@ public class NIOSSLTransport extends NIOTransport {
             // otherwise the session would return 'SSL_NULL_WITH_NULL_NULL' for the
             // cipher suite.
             sslSession = sslEngine.getSession();
-
-            // listen for events telling us when the socket is readable.
-            selection = SelectorManager.getInstance().register(channel, new SelectorManager.Listener()
{
-                @Override
-                public void onSelect(SelectorSelection selection) {
-                    serviceRead();
-                }
-
-                @Override
-                public void onError(SelectorSelection selection, Throwable error) {
-                    if (error instanceof IOException) {
-                        onException((IOException) error);
-                    } else {
-                        onException(IOExceptionSupport.create(error));
-                    }
-                }
-            });
         }
     }
 


Mime
View raw message