Simplify semaphore logic in ChannelManager (AsyncHttpClient#1372)

* avoid using `java.util.concurrent.Semaphore` which is blocking,
  and its blocking part is not used, which is confusing.
* Use `NonBlockingSemaphoreInfinite` to avoid nulls and ifs
  when connection count is not limited.

Author: stepancheg

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -45,7 +45,6 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -109,9 +108,9 @@ public class ChannelManager {
     private final ChannelPool channelPool;
     private final ChannelGroup openChannels;
     private final boolean maxTotalConnectionsEnabled;
-    private final Semaphore freeChannels;
+    private final NonBlockingSemaphoreLike freeChannels;
     private final boolean maxConnectionsPerHostEnabled;
-    private final ConcurrentHashMap<Object, Semaphore> freeChannelsPerHost = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<Object, NonBlockingSemaphore> freeChannelsPerHost = new ConcurrentHashMap<>();
 
     private AsyncHttpClientHandler wsHandler;
 
@@ -141,18 +140,21 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) {
         maxTotalConnectionsEnabled = config.getMaxConnections() > 0;
         maxConnectionsPerHostEnabled = config.getMaxConnectionsPerHost() > 0;
 
+        freeChannels = maxTotalConnectionsEnabled ?
+                new NonBlockingSemaphore(config.getMaxConnections()) :
+                NonBlockingSemaphoreInfinite.INSTANCE;
+
         if (maxTotalConnectionsEnabled || maxConnectionsPerHostEnabled) {
             openChannels = new DefaultChannelGroup("asyncHttpClient", GlobalEventExecutor.INSTANCE) {
                 @Override
                 public boolean remove(Object o) {
                     boolean removed = super.remove(o);
                     if (removed) {
-                        if (maxTotalConnectionsEnabled)
-                            freeChannels.release();
+                        freeChannels.release();
                         if (maxConnectionsPerHostEnabled) {
                             Object partitionKey = Channel.class.cast(o).attr(partitionKeyAttr).getAndSet(null);
                             if (partitionKey != null) {
-                                Semaphore hostFreeChannels = freeChannelsPerHost.get(partitionKey);
+                                NonBlockingSemaphore hostFreeChannels = freeChannelsPerHost.get(partitionKey);
                                 if (hostFreeChannels != null)
                                     hostFreeChannels.release();
                             }
@@ -161,10 +163,8 @@ public boolean remove(Object o) {
                     return removed;
                 }
             };
-            freeChannels = new Semaphore(config.getMaxConnections());
         } else {
             openChannels = new DefaultChannelGroup("asyncHttpClient", GlobalEventExecutor.INSTANCE);
-            freeChannels = null;
         }
 
         handshakeTimeout = config.getHandshakeTimeout();
@@ -338,15 +338,17 @@ public boolean removeAll(Channel connection) {
     }
 
     private boolean tryAcquireGlobal() {
-        return !maxTotalConnectionsEnabled || freeChannels.tryAcquire();
+        return freeChannels.tryAcquire();
     }
 
-    private Semaphore getFreeConnectionsForHost(Object partitionKey) {
-        return freeChannelsPerHost.computeIfAbsent(partitionKey, pk -> new Semaphore(config.getMaxConnectionsPerHost()));
+    private NonBlockingSemaphoreLike getFreeConnectionsForHost(Object partitionKey) {
+        return maxConnectionsPerHostEnabled ?
+                freeChannelsPerHost.computeIfAbsent(partitionKey, pk -> new NonBlockingSemaphore(config.getMaxConnectionsPerHost())) :
+                NonBlockingSemaphoreInfinite.INSTANCE;
     }
 
     private boolean tryAcquirePerHost(Object partitionKey) {
-        return !maxConnectionsPerHostEnabled || getFreeConnectionsForHost(partitionKey).tryAcquire();
+        return getFreeConnectionsForHost(partitionKey).tryAcquire();
     }
 
     public void acquireChannelLock(Object partitionKey) throws IOException {
@@ -355,8 +357,7 @@ public void acquireChannelLock(Object partitionKey) throws IOException {
         if (!tryAcquireGlobal())
             throw tooManyConnections;
         if (!tryAcquirePerHost(partitionKey)) {
-            if (maxTotalConnectionsEnabled)
-                freeChannels.release();
+            freeChannels.release();
 
             throw tooManyConnectionsPerHost;
         }
@@ -383,10 +384,8 @@ public void closeChannel(Channel channel) {
     }
 
     public void releaseChannelLock(Object partitionKey) {
-        if (maxTotalConnectionsEnabled)
-            freeChannels.release();
-        if (maxConnectionsPerHostEnabled)
-            getFreeConnectionsForHost(partitionKey).release();
+        freeChannels.release();
+        getFreeConnectionsForHost(partitionKey).release();
     }
 
     public void registerOpenChannel(Channel channel, Object partitionKey) {

client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphore.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2017 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.channel;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Semaphore-like API, but without blocking.
+ *
+ * @author Stepan Koltsov
+ */
+class NonBlockingSemaphore implements NonBlockingSemaphoreLike {
+
+    private final AtomicInteger permits;
+
+    public NonBlockingSemaphore(int permits) {
+        this.permits = new AtomicInteger(permits);
+    }
+
+    @Override
+    public void release() {
+        permits.incrementAndGet();
+    }
+
+    @Override
+    public boolean tryAcquire() {
+        for (;;) {
+            int count = permits.get();
+            if (count <= 0) {
+                return false;
+            }
+            if (permits.compareAndSet(count, count - 1)) {
+                return true;
+            }
+        }
+    }
+}

client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreInfinite.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2017 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.channel;
+
+/**
+ * Non-blocking semaphore-like object with infinite permits.
+ *
+ * So try-acquire always succeeds.
+ *
+ * @author Stepan Koltsov
+ */
+enum NonBlockingSemaphoreInfinite implements NonBlockingSemaphoreLike {
+    INSTANCE;
+
+    @Override
+    public void release() {
+    }
+
+    @Override
+    public boolean tryAcquire() {
+        return true;
+    }
+}

client/src/main/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreLike.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2017 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.channel;
+
+/**
+ * Non-blocking semaphore API.
+ *
+ * @author Stepan Koltsov
+ */
+interface NonBlockingSemaphoreLike {
+    void release();
+
+    boolean tryAcquire();
+}

client/src/test/java/org/asynchttpclient/netty/channel/NonBlockingSemaphoreTest.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright (c) 2017 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.channel;
+
+import java.util.concurrent.Semaphore;
+
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+/**
+ * @author Stepan Koltsov
+ */
+public class NonBlockingSemaphoreTest {
+
+    private static class Mirror {
+        private final Semaphore real;
+        private final NonBlockingSemaphore nonBlocking;
+
+        public Mirror(int permits) {
+            real = new Semaphore(permits);
+            nonBlocking = new NonBlockingSemaphore(permits);
+        }
+
+        public boolean tryAcquire() {
+            boolean a = real.tryAcquire();
+            boolean b = nonBlocking.tryAcquire();
+            assertEquals(a, b);
+            return a;
+        }
+
+        public void release() {
+            real.release();
+            nonBlocking.release();
+        }
+    }
+
+    @Test
+    public void test0() {
+        Mirror mirror = new Mirror(0);
+        assertFalse(mirror.tryAcquire());
+    }
+
+    @Test
+    public void three() {
+        Mirror mirror = new Mirror(3);
+        for (int i = 0; i < 3; ++i) {
+            assertTrue(mirror.tryAcquire());
+        }
+        assertFalse(mirror.tryAcquire());
+        mirror.release();
+        assertTrue(mirror.tryAcquire());
+    }
+
+    @Test
+    public void negative() {
+        Mirror mirror = new Mirror(-1);
+        assertFalse(mirror.tryAcquire());
+        mirror.release();
+        assertFalse(mirror.tryAcquire());
+        mirror.release();
+        assertTrue(mirror.tryAcquire());
+    }
+
+}