Release channel lock when channel is not offered to pool, see AsyncHttpClient#1377

Author: slandelle

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

@@ -133,9 +133,7 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) {
         maxTotalConnectionsEnabled = config.getMaxConnections() > 0;
         maxConnectionsPerHostEnabled = config.getMaxConnectionsPerHost() > 0;
 
-        freeChannels = maxTotalConnectionsEnabled ?
-                new NonBlockingSemaphore(config.getMaxConnections()) :
-                NonBlockingSemaphoreInfinite.INSTANCE;
+        freeChannels = maxTotalConnectionsEnabled ? new NonBlockingSemaphore(config.getMaxConnections()) : NonBlockingSemaphoreInfinite.INSTANCE;
 
         if (maxTotalConnectionsEnabled || maxConnectionsPerHostEnabled) {
             openChannels = new DefaultChannelGroup("asyncHttpClient", GlobalEventExecutor.INSTANCE) {
@@ -309,16 +307,15 @@ public final void tryToOfferChannelToPool(Channel channel, AsyncHandler<?> async
             if (asyncHandler instanceof AsyncHandlerExtensions)
                 AsyncHandlerExtensions.class.cast(asyncHandler).onConnectionOffer(channel);
             if (channelPool.offer(channel, partitionKey)) {
-                if (maxConnectionsPerHostEnabled)
+                if (maxConnectionsPerHostEnabled) {
                     channel.attr(partitionKeyAttr).setIfAbsent(partitionKey);
-            } else {
-                // rejected by pool
-                closeChannel(channel);
+                }
+                return;
             }
-        } else {
-            // not offered
-            closeChannel(channel);
         }
+        // rejected by pool or not offered
+        releaseChannelLock(partitionKey);
+        closeChannel(channel);
     }
 
     public Channel poll(Uri uri, String virtualHost, ProxyServer proxy, ChannelPoolPartitioning connectionPoolPartitioning) {
@@ -335,9 +332,8 @@ private boolean tryAcquireGlobal() {
     }
 
     private NonBlockingSemaphoreLike getFreeConnectionsForHost(Object partitionKey) {
-        return maxConnectionsPerHostEnabled ?
-                freeChannelsPerHost.computeIfAbsent(partitionKey, pk -> new NonBlockingSemaphore(config.getMaxConnectionsPerHost())) :
-                NonBlockingSemaphoreInfinite.INSTANCE;
+        return maxConnectionsPerHostEnabled ? freeChannelsPerHost.computeIfAbsent(partitionKey, pk -> new NonBlockingSemaphore(config.getMaxConnectionsPerHost()))
+                : NonBlockingSemaphoreInfinite.INSTANCE;
     }
 
     private boolean tryAcquirePerHost(Object partitionKey) {

client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java

@@ -49,10 +49,11 @@ private void finishUpdate(final NettyResponseFuture<?> future, Channel channel,
         future.cancelTimeouts();
 
         boolean keepAlive = future.isKeepAlive();
-        if (expectOtherChunks && keepAlive)
+        if (expectOtherChunks && keepAlive) {
             channelManager.drainChannelAndOffer(channel, future);
-        else
+        } else {
             channelManager.tryToOfferChannelToPool(channel, future.getAsyncHandler(), keepAlive, future.getPartitionKey());
+        }
 
         try {
             future.done();