Merge pull request AsyncHttpClient#954 from stepancheg/name

AsyncHttpClient `name` option

Author: slandelle

api/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

@@ -34,6 +34,7 @@
 import org.asynchttpclient.filter.ResponseFilter;
 import org.asynchttpclient.proxy.ProxyServer;
 import org.asynchttpclient.proxy.ProxyServerSelector;
+import org.asynchttpclient.util.PrefixIncrementThreadFactory;
 import org.asynchttpclient.util.ProxyUtils;
 
 /**
@@ -66,6 +67,8 @@ public class AsyncHttpClientConfig {
         AHC_VERSION = prop.getProperty("ahc.version", "UNKNOWN");
     }
 
+    protected String name;
+
     protected int connectTimeout;
 
     protected int maxConnections;
@@ -118,7 +121,8 @@ public class AsyncHttpClientConfig {
     protected AsyncHttpClientConfig() {
     }
 
-    private AsyncHttpClientConfig(int connectTimeout,//
+    private AsyncHttpClientConfig(String name,//
+            int connectTimeout,//
             int maxConnections,//
             int maxConnectionsPerHost,//
             int requestTimeout,//
@@ -160,6 +164,7 @@ private AsyncHttpClientConfig(int connectTimeout,//
             boolean keepEncodingHeader,//
             AsyncHttpProviderConfig<?, ?> providerConfig) {
 
+        this.name = name;
         this.connectTimeout = connectTimeout;
         this.maxConnections = maxConnections;
         this.maxConnectionsPerHost = maxConnectionsPerHost;
@@ -178,7 +183,15 @@ private AsyncHttpClientConfig(int connectTimeout,//
         this.proxyServerSelector = proxyServerSelector;
         this.compressionEnforced = compressionEnforced;
         this.userAgent = userAgent;
-        this.applicationThreadPool = applicationThreadPool == null ? Executors.newCachedThreadPool() : applicationThreadPool;
+
+        if (applicationThreadPool != null) {
+            this.applicationThreadPool = applicationThreadPool;
+        } else {
+            PrefixIncrementThreadFactory threadFactory = new PrefixIncrementThreadFactory(
+                    getNameOrDefault() + "-");
+            this.applicationThreadPool = Executors.newCachedThreadPool(threadFactory);
+        }
+
         this.realm = realm;
         this.requestFilters = requestFilters;
         this.responseFilters = responseFilters;
@@ -203,6 +216,32 @@ private AsyncHttpClientConfig(int connectTimeout,//
         this.keepEncodingHeader = keepEncodingHeader;
     }
 
+    /**
+     * Return the name of {@link AsyncHttpClient}, which is used for thread naming
+     * and debugging.
+     *
+     * @return the name.
+     */
+    public String getName() {
+        return name;
+    }
+
+    /**
+     * Return the name of {@link AsyncHttpClient}, or default string if name is null or empty.
+     *
+     * @return the name.
+     */
+    public String getNameOrDefault() {
+        String r = name;
+        if (r == null || r.isEmpty()) {
+            r = defaultName();
+        }
+        if (r == null || r.isEmpty()) {
+            r = "AsyncHttpClient";
+        }
+        return r;
+    }
+
     /**
      * Return the maximum number of connections an {@link AsyncHttpClient} can
      * handle.
@@ -577,6 +616,7 @@ public boolean isKeepEncodingHeader() {
      * Builder for an {@link AsyncHttpClient}
      */
     public static class Builder {
+        private String name = defaultName();
         private int connectTimeout = defaultConnectTimeout();
         private int maxConnections = defaultMaxConnections();
         private int maxConnectionsPerHost = defaultMaxConnectionsPerHost();
@@ -624,6 +664,16 @@ public static class Builder {
         public Builder() {
         }
 
+        /**
+         * Set the name of {@link AsyncHttpClient}. That name is used for thread
+         * naming and can be used for debugging multiple {@link AsyncHttpClient}
+         * instance.
+         */
+        public Builder setName(String name) {
+            this.name = name;
+            return this;
+        }
+
         /**
          * Set the maximum number of connections an {@link AsyncHttpClient} can
          * handle.
@@ -1113,6 +1163,7 @@ public Builder setKeepEncodingHeader(boolean keepEncodingHeader) {
          * @param prototype the configuration to use as a prototype.
          */
         public Builder(AsyncHttpClientConfig prototype) {
+            name = prototype.getName();
             allowPoolingConnections = prototype.isAllowPoolingConnections();
             connectTimeout = prototype.getConnectTimeout();
             pooledConnectionIdleTimeout = prototype.getPooledConnectionIdleTimeout();
@@ -1178,7 +1229,8 @@ public AsyncHttpClientConfig build() {
             if (proxyServerSelector == null)
                 proxyServerSelector = ProxyServerSelector.NO_PROXY_SELECTOR;
 
-            return new AsyncHttpClientConfig(connectTimeout,//
+            return new AsyncHttpClientConfig(name,//
+                    connectTimeout,//
                     maxConnections,//
                     maxConnectionsPerHost,//
                     requestTimeout,//

api/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigBean.java

@@ -51,6 +51,7 @@ void configureFilters() {
     void configureDefaults() {
         maxConnections = defaultMaxConnections();
         maxConnectionsPerHost = defaultMaxConnectionsPerHost();
+        name = defaultName();
         connectTimeout = defaultConnectTimeout();
         webSocketTimeout = defaultWebSocketTimeout();
         pooledConnectionIdleTimeout = defaultPooledConnectionIdleTimeout();
@@ -88,6 +89,11 @@ public Thread newThread(Runnable r) {
         });
     }
 
+    public AsyncHttpClientConfigBean setName(String name) {
+        this.name = name;
+        return this;
+    }
+
     public AsyncHttpClientConfigBean setMaxTotalConnections(int maxConnections) {
         this.maxConnections = maxConnections;
         return this;

api/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java

@@ -19,6 +19,10 @@ private AsyncHttpClientConfigDefaults() {
 
     public static final String ASYNC_CLIENT_CONFIG_ROOT = "org.asynchttpclient.";
 
+    public static String defaultName() {
+        return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getString(ASYNC_CLIENT_CONFIG_ROOT + "name");
+    }
+
     public static int defaultMaxConnections() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getInt(ASYNC_CLIENT_CONFIG_ROOT + "maxConnections");
     }

api/src/main/java/org/asynchttpclient/util/PrefixIncrementThreadFactory.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2015 AsyncHttpClient Project. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ *     http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package org.asynchttpclient.util;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Thread factory that generates thread names by adding incrementing number
+ * to the specified prefix.
+ *
+ * @author Stepan Koltsov
+ */
+public class PrefixIncrementThreadFactory implements ThreadFactory {
+    private final String namePrefix;
+    private final AtomicInteger threadNumber = new AtomicInteger();
+
+    public PrefixIncrementThreadFactory(String namePrefix) {
+        if (namePrefix == null || namePrefix.isEmpty()) {
+            throw new IllegalArgumentException("namePrefix must not be empty");
+        }
+        this.namePrefix = namePrefix;
+    }
+
+    public Thread newThread(Runnable r) {
+        return new Thread(r, namePrefix + threadNumber.incrementAndGet());
+    }
+}

api/src/main/resources/ahc-default.properties

@@ -1,3 +1,4 @@
+org.asynchttpclient.name=AsyncHttpClient
 org.asynchttpclient.maxConnections=-1
 org.asynchttpclient.maxConnectionsPerHost=-1
 org.asynchttpclient.connectTimeout=5000

api/src/test/java/org/asynchttpclient/ThreadNameTest.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2015 AsyncHttpClient Project. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ *     http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package org.asynchttpclient;
+
+import java.util.Arrays;
+import java.util.Random;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+/**
+ * Tests configured client name is used for thread names.
+ *
+ * @author Stepan Koltsov
+ */
+public abstract class ThreadNameTest extends AbstractBasicTest {
+
+    private static Thread[] getThreads() {
+        int count = Thread.activeCount() + 1;
+        for (;;) {
+            Thread[] threads = new Thread[count];
+            int filled = Thread.enumerate(threads);
+            if (filled < threads.length) {
+                return Arrays.copyOf(threads, filled);
+            }
+
+            count *= 2;
+        }
+    }
+
+    @Test(groups = { "standalone", "default_provider" })
+    public void testQueryParameters() throws Exception {
+        String name = "ahc-" + (new Random().nextLong() & 0x7fffffffffffffffL);
+        AsyncHttpClientConfig.Builder config = new AsyncHttpClientConfig.Builder();
+        config.setName(name);
+        try (AsyncHttpClient client = getAsyncHttpClient(config.build())) {
+            Future<Response> f = client.prepareGet("http://127.0.0.1:" + port1 + "/").execute();
+            f.get(3, TimeUnit.SECONDS);
+
+            // We cannot assert that all threads are created with specified name,
+            // so we checking that at least one thread is.
+            boolean found = false;
+            for (Thread thread : getThreads()) {
+                if (thread.getName().startsWith(name)) {
+                    found = true;
+                    break;
+                }
+            }
+
+            Assert.assertTrue(found, "must found threads starting with random string " + name);
+        }
+    }
+}

providers/netty3/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.net.ssl.SSLEngine;
@@ -52,21 +53,26 @@
 import org.asynchttpclient.netty.request.NettyRequestSender;
 import org.asynchttpclient.proxy.ProxyServer;
 import org.asynchttpclient.uri.Uri;
+import org.asynchttpclient.util.PrefixIncrementThreadFactory;
 import org.jboss.netty.bootstrap.ClientBootstrap;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelPipeline;
 import org.jboss.netty.channel.ChannelPipelineFactory;
 import org.jboss.netty.channel.DefaultChannelFuture;
 import org.jboss.netty.channel.group.ChannelGroup;
 import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
+import org.jboss.netty.channel.socket.nio.NioClientBossPool;
 import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
+import org.jboss.netty.channel.socket.nio.NioWorkerPool;
 import org.jboss.netty.handler.codec.http.HttpClientCodec;
 import org.jboss.netty.handler.codec.http.HttpContentDecompressor;
 import org.jboss.netty.handler.codec.http.websocketx.WebSocket08FrameDecoder;
 import org.jboss.netty.handler.codec.http.websocketx.WebSocket08FrameEncoder;
 import org.jboss.netty.handler.codec.http.websocketx.WebSocketFrameAggregator;
 import org.jboss.netty.handler.ssl.SslHandler;
 import org.jboss.netty.handler.stream.ChunkedWriteHandler;
+import org.jboss.netty.util.HashedWheelTimer;
+import org.jboss.netty.util.ThreadNameDeterminer;
 import org.jboss.netty.util.Timer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -181,11 +187,16 @@ public Semaphore apply(Object partitionKey) {
 
         } else {
             ExecutorService e = nettyConfig.getBossExecutorService();
-            if (e == null)
-                e = Executors.newCachedThreadPool();
+            if (e == null) {
+                ThreadFactory threadFactory = new PrefixIncrementThreadFactory(
+                        config.getNameOrDefault() + "-boss-");
+                e = Executors.newCachedThreadPool(threadFactory);
+            }
             int numWorkers = config.getIoThreadMultiplier() * Runtime.getRuntime().availableProcessors();
             LOGGER.trace("Number of application's worker threads is {}", numWorkers);
-            socketChannelFactory = new NioClientSocketChannelFactory(e, config.executorService(), numWorkers);
+            NioClientBossPool nioClientBossPool = new NioClientBossPool(e, 1, new HashedWheelTimer(), ThreadNameDeterminer.CURRENT);
+            NioWorkerPool nioWorkerPool = new NioWorkerPool(config.executorService(), numWorkers, ThreadNameDeterminer.CURRENT);
+            socketChannelFactory = new NioClientSocketChannelFactory(nioClientBossPool, nioWorkerPool);
             allowReleaseSocketChannelFactory = true;
         }
 

providers/netty3/src/test/java/org/asynchttpclient/netty/NettyThreadNameTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright (c) 2015 AsyncHttpClient Project. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ *     http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package org.asynchttpclient.netty;
+
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.RetryRequestTest;
+import org.asynchttpclient.ThreadNameTest;
+
+/**
+ * @author Stepan Koltsov
+ */
+public class NettyThreadNameTest extends ThreadNameTest {
+    @Override
+    public AsyncHttpClient getAsyncHttpClient(AsyncHttpClientConfig config) {
+        return NettyProviderUtil.nettyProvider(config);
+    }
+}

providers/netty4/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -38,6 +38,7 @@
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.util.Timer;
+import io.netty.util.concurrent.DefaultThreadFactory;
 import io.netty.util.internal.chmv8.ConcurrentHashMapV8;
 
 import java.io.IOException;
@@ -172,7 +173,12 @@ public Semaphore apply(Object partitionKey) {
 
         // check if external EventLoopGroup is defined
         allowReleaseEventLoopGroup = nettyConfig.getEventLoopGroup() == null;
-        eventLoopGroup = allowReleaseEventLoopGroup ? new NioEventLoopGroup() : nettyConfig.getEventLoopGroup();
+        if (allowReleaseEventLoopGroup) {
+            DefaultThreadFactory threadFactory = new DefaultThreadFactory(config.getNameOrDefault());
+            eventLoopGroup = new NioEventLoopGroup(0, threadFactory);
+        } else {
+            eventLoopGroup = nettyConfig.getEventLoopGroup();
+        }
         if (eventLoopGroup instanceof OioEventLoopGroup)
             throw new IllegalArgumentException("Oio is not supported");