AsyncHttpClient name option

Fixes AsyncHttpClient#952

Name option is useful for two purposes:

## Thread name

Typical application with several netty and/or async-http-client
instances may have hundreds of threads like "New I/O worker 120",
which makes it hard to understand which thread belongs to which
part of the application.

Specified name is used in AsyncHttpClient to name threads created
by AsyncHttpClient and netty.

If name is not specified, `AsyncHttpClient` string is used.

## Debugging

`name` is stored in `AsyncHttpClientConfig`, and available in
debugger debugging when application has multiple client instances.
It can be useful to distinguish between instance, for example, when
breakpoint is set inside of AsyncHttpClient.

Author: stepancheg

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");