Repository: netty/netty
Issue: SimpleChannelPool/FixedChannelPool: Don't put unhealthy channel back to the pool on release
Labels: feature

Currently it is _required_ to release channel to the channel pool for `FixedChannelPool`. 
Failure to release a channel may cause that `FixedChannelPool.acquiredChannelCount` becomes equal or bigger then `FixedChannelPool.maxConnections`. So basically it means that we have to release back all the channels even those ones that are not _healthy_, and when channel is released it simply put back into a `CimpleChannelPool.deque` regardless if it's healthy or not.
Now, on acquire SimpelChannelPool will check if channel is healthy before returning it. If it turns out to be unhealthy then it will discard it and poll the next one from the pool(if there any left in pool) and will continue doing so until it gets healthy channel or until the `deque` is empty(in which case it will just create a new channel). Though this sort of lazy-discard approach works, I think there is a room for improvement, such as we can filter out unhealthy channels at the time when we release them. 

Here is example where such improvement may help:
1) Client application requested bunch of channels
2) Before releasing them back to pool all channels become closed cause remote tier closed them
3) We end up having channelPool full of unhealthy channels.
4) Next time we will try to acquire new channel we will have to cleanup/discard all unhealthy channels first which may take some time depending on number of unhealthy channels.

It also make sense to filter our unhealthy channels on release because in most cases when someone releases a channel he normally doesn't care about release performance. The more critical in my opinion is acquire performance.


Issue comments:
@normanmaurer - I recall discussing this when we originally reviewed the pool.  I think your rational was since we have to check the health anyways when we pull out of the pool, and the health check may be expensive (may include RTT), we wanted to minimize the number of health checks done. WDYT?

@Scottmitch yep that was my argument, I may been wrong here and we should kind of make this configurable. WDYT ?

@normanmaurer yes. As this issue describes I think there is value in having the option to check up front too.

Fixed by https://github.com/netty/netty/pull/4078

@Scottmitch Please make sure not to introduce a build breakage when you cherry-pick into 4.0 and 4.1

@trustin - I'm sorry. Will be more careful in the future. Thanks for cleaning up after my mess.

@Scottmitch No worries. I make the same mistake over and over. :-)


Pull request review comments:
As discussed in https://github.com/netty/netty/issues/4077#issuecomment-130461684. Perhaps we should make this check configurable (some boolean flag to indicate we should check health when putting back into pool).

@Scottmitch sure, will do. What do you think the default behavior should be? Check health on release by default or do not check by default?

My vote is default should check the health. If this hurts performance in specific use cases it can be switched off. Our default health checker `ChannelHealthChecker.ACTIVE` shouldn't involve any network traffic so should be OK. @normanmaurer - WDYT?

I'm not sure we need to verify / require this text is in the message.

Also do we need the `ExpectedException` `@Rule`?  Is this not something we can catch in normal code flow? I think we typically catch expected exceptions and `fail()` if we don't get the exception.

@Scottmitch, so that I know for sure that expected exception(not any random) had been thrown. 

@ioanbsu - Understood, but I don't think we need to couple the tests so tightly to the implementation that the message in the exception is verified. I think just verifying the correct type of exception is thrown at the correct time should be sufficient.

@Scottmitch, got it. removed message validation.

You also need to empty the stack trace as done for "FULL_EXCEPTION" below.

`true` -> `{@code true}`.

`true` -> `{@code true}`.

Is it necessary to have `offerHealthyOnly` as an argument? Seems like we are just passing in the member variable when ever this is called.

Consider renaming to `releaseHeathCheck`....and also the same for the publicly visible in constructors.

remove `final`.

should this just be `<p>`, or do you want to wrap the above in `<p>...</p>`?

You're right. will fix.

Why?

This is not part of my changes actually. This part got auto-reformatted by IDEA based on the rules that are used in netty project. I can certainly change it to <p>

just a nit, but it is not needed.

Why this change?

This is general code style practice that I try to follow in my projects. It helps to prevent possible bugs. If there's no need to change the value of a variable and someone does, the IDE and/or compiler will provide a warning.
It may also be a helping hint to compiler to perform some optimizations for the code.

I don't mind removing it if in netty projects you don't have such rules.

Lets revert for this PR. I understand the rational, and do not necessarily disagree, but lets keep the set of changes focused on what is required to resolve the issue.  If need be we can open a followup PR focused on using `final` more consistently.

Sounds good.

Done.


Patch that resolved the issue:
diff --git a/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java b/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
index bd89bd64612..1f57b7d640d 100644
--- a/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
+++ b/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java
@@ -121,7 +121,34 @@ public FixedChannelPool(Bootstrap bootstrap,
                             ChannelHealthChecker healthCheck, AcquireTimeoutAction action,
                             final long acquireTimeoutMillis,
                             int maxConnections, int maxPendingAcquires) {
-        super(bootstrap, handler, healthCheck);
+        this(bootstrap, handler, healthCheck, action, acquireTimeoutMillis, maxConnections, maxPendingAcquires, true);
+    }
+
+    /**
+     * Creates a new instance.
+     *
+     * @param bootstrap             the {@link Bootstrap} that is used for connections
+     * @param handler               the {@link ChannelPoolHandler} that will be notified for the different pool actions
+     * @param healthCheck           the {@link ChannelHealthChecker} that will be used to check if a {@link Channel} is
+     *                              still healty when obtain from the {@link ChannelPool}
+     * @param action                the {@link AcquireTimeoutAction} to use or {@code null} if non should be used.
+     *                              In this case {@param acquireTimeoutMillis} must be {@code -1}.
+     * @param acquireTimeoutMillis  the time (in milliseconds) after which an pending acquire must complete or
+     *                              the {@link AcquireTimeoutAction} takes place.
+     * @param maxConnections        the numnber of maximal active connections, once this is reached new tries to
+     *                              acquire a {@link Channel} will be delayed until a connection is returned to the
+     *                              pool again.
+     * @param maxPendingAcquires    the maximum number of pending acquires. Once this is exceed acquire tries will
+     *                              be failed.
+     * @param releaseHealthCheck    will check channel health before offering back if this parameter set to
+     *                              {@code true}.
+     */
+    public FixedChannelPool(Bootstrap bootstrap,
+                            ChannelPoolHandler handler,
+                            ChannelHealthChecker healthCheck, AcquireTimeoutAction action,
+                            final long acquireTimeoutMillis,
+                            int maxConnections, int maxPendingAcquires, final boolean releaseHealthCheck) {
+        super(bootstrap, handler, healthCheck, releaseHealthCheck);
         if (maxConnections < 1) {
             throw new IllegalArgumentException("maxConnections: " + maxConnections + " (expected: >= 1)");
         }
diff --git a/transport/src/main/java/io/netty/channel/pool/SimpleChannelPool.java b/transport/src/main/java/io/netty/channel/pool/SimpleChannelPool.java
index 311279a2292..72cd7cf9b89 100644
--- a/transport/src/main/java/io/netty/channel/pool/SimpleChannelPool.java
+++ b/transport/src/main/java/io/netty/channel/pool/SimpleChannelPool.java
@@ -43,13 +43,17 @@
 public class SimpleChannelPool implements ChannelPool {
     private static final AttributeKey<SimpleChannelPool> POOL_KEY = AttributeKey.newInstance("channelPool");
     private static final IllegalStateException FULL_EXCEPTION = new IllegalStateException("ChannelPool full");
+    private static final IllegalStateException UNHEALTHY_NON_OFFERED_TO_POOL =
+            new IllegalStateException("Channel is unhealthy not offering it back to pool");
     static {
         FULL_EXCEPTION.setStackTrace(EmptyArrays.EMPTY_STACK_TRACE);
+        UNHEALTHY_NON_OFFERED_TO_POOL.setStackTrace(EmptyArrays.EMPTY_STACK_TRACE);
     }
     private final Deque<Channel> deque = PlatformDependent.newConcurrentDeque();
     private final ChannelPoolHandler handler;
     private final ChannelHealthChecker healthCheck;
     private final Bootstrap bootstrap;
+    private final boolean releaseHealthCheck;
 
     /**
      * Creates a new instance using the {@link ChannelHealthChecker#ACTIVE}.
@@ -67,11 +71,27 @@ public SimpleChannelPool(Bootstrap bootstrap, final ChannelPoolHandler handler)
      * @param bootstrap         the {@link Bootstrap} that is used for connections
      * @param handler           the {@link ChannelPoolHandler} that will be notified for the different pool actions
      * @param healthCheck       the {@link ChannelHealthChecker} that will be used to check if a {@link Channel} is
-     *                          still healty when obtain from the {@link ChannelPool}
+     *                          still healthy when obtain from the {@link ChannelPool}
      */
     public SimpleChannelPool(Bootstrap bootstrap, final ChannelPoolHandler handler, ChannelHealthChecker healthCheck) {
+        this(bootstrap, handler, healthCheck, true);
+    }
+
+    /**
+     * Creates a new instance.
+     *
+     * @param bootstrap          the {@link Bootstrap} that is used for connections
+     * @param handler            the {@link ChannelPoolHandler} that will be notified for the different pool actions
+     * @param healthCheck        the {@link ChannelHealthChecker} that will be used to check if a {@link Channel} is
+     *                           still healthy when obtain from the {@link ChannelPool}
+     * @param releaseHealthCheck will offercheck channel health before offering back if this parameter set to
+     *                           {@code true}.
+     */
+    public SimpleChannelPool(Bootstrap bootstrap, final ChannelPoolHandler handler, ChannelHealthChecker healthCheck,
+                             boolean releaseHealthCheck) {
         this.handler = checkNotNull(handler, "handler");
         this.healthCheck = checkNotNull(healthCheck, "healthCheck");
+        this.releaseHealthCheck = releaseHealthCheck;
         // Clone the original Bootstrap as we want to set our own handler
         this.bootstrap = checkNotNull(bootstrap, "bootstrap").clone();
         this.bootstrap.handler(new ChannelInitializer<Channel>() {
@@ -183,9 +203,9 @@ private void notifyHealthCheck(Future<Boolean> future, Channel ch, Promise<Chann
     }
 
     /**
-     * Bootstrap a new {@link Channel}. The default implementation uses {@link Bootstrap#connect()},
-     * sub-classes may override this.
-     *
+     * Bootstrap a new {@link Channel}. The default implementation uses {@link Bootstrap#connect()}, sub-classes may
+     * override this.
+     * <p>
      * The {@link Bootstrap} that is passed in here is cloned via {@link Bootstrap#clone()}, so it is safe to modify.
      */
     protected ChannelFuture connectChannel(Bootstrap bs) {
@@ -230,11 +250,10 @@ private void doReleaseChannel(Channel channel, Promise<Void> promise) {
                          promise);
         } else {
             try {
-                if (offerChannel(channel)) {
-                    handler.channelReleased(channel);
-                    promise.setSuccess(null);
+                if (releaseHealthCheck) {
+                    doHealthCheckOnRelease(channel, promise);
                 } else {
-                    closeAndFail(channel, FULL_EXCEPTION, promise);
+                    releaseAndOffer(channel, promise);
                 }
             } catch (Throwable cause) {
                 closeAndFail(channel, cause, promise);
@@ -242,6 +261,46 @@ private void doReleaseChannel(Channel channel, Promise<Void> promise) {
         }
     }
 
+    private void doHealthCheckOnRelease(final Channel channel, final Promise<Void> promise) throws Exception {
+        final Future<Boolean> f = healthCheck.isHealthy(channel);
+        if (f.isDone()) {
+            releaseAndOfferIfHealthy(channel, promise, f);
+        } else {
+            f.addListener(new FutureListener<Boolean>() {
+                @Override
+                public void operationComplete(Future<Boolean> future) throws Exception {
+                    releaseAndOfferIfHealthy(channel, promise, f);
+                }
+            });
+        }
+    }
+
+    /**
+     * Adds the channel back to the pool only if the channel is healty.
+     * @param channel the channel to put back to the pool
+     * @param promise offer operation promise.
+     * @param future the future that contains information fif channel is healthy or not.
+     * @throws Exception in case when failed to notify handler about release operation.
+     */
+    private void releaseAndOfferIfHealthy(Channel channel, Promise<Void> promise, Future<Boolean> future)
+            throws Exception {
+        if (future.getNow()) { //channel turns out to be healthy, offering and releasing it.
+            releaseAndOffer(channel, promise);
+        } else { //channel ont healthy, just releasing it.
+            handler.channelReleased(channel);
+            closeAndFail(channel, UNHEALTHY_NON_OFFERED_TO_POOL, promise);
+        }
+    }
+
+    private void releaseAndOffer(Channel channel, Promise<Void> promise) throws Exception {
+        if (offerChannel(channel)) {
+            handler.channelReleased(channel);
+            promise.setSuccess(null);
+        } else {
+            closeAndFail(channel, FULL_EXCEPTION, promise);
+        }
+    }
+
     private static void closeChannel(Channel channel) {
         channel.attr(POOL_KEY).getAndSet(null);
         channel.close();


Test patch:
diff --git a/transport/src/test/java/io/netty/channel/pool/SimpleChannelPoolTest.java b/transport/src/test/java/io/netty/channel/pool/SimpleChannelPoolTest.java
index 20d024ddd88..20ca46b7837 100644
--- a/transport/src/test/java/io/netty/channel/pool/SimpleChannelPoolTest.java
+++ b/transport/src/test/java/io/netty/channel/pool/SimpleChannelPoolTest.java
@@ -25,7 +25,11 @@
 import io.netty.channel.local.LocalAddress;
 import io.netty.channel.local.LocalChannel;
 import io.netty.channel.local.LocalServerChannel;
+import io.netty.util.concurrent.Future;
+import org.hamcrest.CoreMatchers;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import java.util.Queue;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -35,6 +39,9 @@
 public class SimpleChannelPoolTest {
     private static final String LOCAL_ADDR_ID = "test.id";
 
+    @Rule
+    public ExpectedException expectedException = ExpectedException.none();
+
     @Test
     public void testAcquire() throws Exception {
         EventLoopGroup group = new DefaultEventLoopGroup();
@@ -142,4 +149,94 @@ protected boolean offerChannel(Channel ch) {
         channel2.close().sync();
         group.shutdownGracefully();
     }
+
+    /**
+     * Tests that if channel was unhealthy it is not offered back to the pool.
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testUnhealthyChannelIsNotOffered() throws Exception {
+        EventLoopGroup group = new DefaultEventLoopGroup();
+        LocalAddress addr = new LocalAddress(LOCAL_ADDR_ID);
+        Bootstrap cb = new Bootstrap();
+        cb.remoteAddress(addr);
+        cb.group(group)
+          .channel(LocalChannel.class);
+
+        ServerBootstrap sb = new ServerBootstrap();
+        sb.group(group)
+          .channel(LocalServerChannel.class)
+          .childHandler(new ChannelInitializer<LocalChannel>() {
+              @Override
+              public void initChannel(LocalChannel ch) throws Exception {
+                  ch.pipeline().addLast(new ChannelHandlerAdapter());
+              }
+          });
+
+        // Start server
+        Channel sc = sb.bind(addr).syncUninterruptibly().channel();
+        ChannelPoolHandler handler = new CountingChannelPoolHandler();
+        ChannelPool pool = new SimpleChannelPool(cb, handler);
+        Channel channel1 = pool.acquire().syncUninterruptibly().getNow();
+        pool.release(channel1).syncUninterruptibly();
+        Channel channel2 = pool.acquire().syncUninterruptibly().getNow();
+        //first check that when returned healthy then it actually offered back to the pool.
+        assertSame(channel1, channel2);
+
+        expectedException.expect(IllegalStateException.class);
+        channel1.close().syncUninterruptibly();
+        try {
+            pool.release(channel1).syncUninterruptibly();
+        } catch (Exception e) {
+            throw e;
+        } finally {
+            sc.close().syncUninterruptibly();
+            channel2.close().syncUninterruptibly();
+            group.shutdownGracefully();
+        }
+    }
+
+    /**
+     * Tests that if channel was unhealthy it is was offered back to the pool because
+     * it was requested not to validate channel health on release.
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testUnhealthyChannelIsOfferedWhenNoHealthCheckRequested() throws Exception {
+        EventLoopGroup group = new DefaultEventLoopGroup();
+        LocalAddress addr = new LocalAddress(LOCAL_ADDR_ID);
+        Bootstrap cb = new Bootstrap();
+        cb.remoteAddress(addr);
+        cb.group(group)
+          .channel(LocalChannel.class);
+
+        ServerBootstrap sb = new ServerBootstrap();
+        sb.group(group)
+          .channel(LocalServerChannel.class)
+          .childHandler(new ChannelInitializer<LocalChannel>() {
+              @Override
+              public void initChannel(LocalChannel ch) throws Exception {
+                  ch.pipeline().addLast(new ChannelHandlerAdapter());
+              }
+          });
+
+        // Start server
+        Channel sc = sb.bind(addr).syncUninterruptibly().channel();
+        ChannelPoolHandler handler = new CountingChannelPoolHandler();
+        ChannelPool pool = new SimpleChannelPool(cb, handler, ChannelHealthChecker.ACTIVE, false);
+        Channel channel1 = pool.acquire().syncUninterruptibly().getNow();
+        channel1.close().syncUninterruptibly();
+        Future<Void> releaseFuture =
+                pool.release(channel1, channel1.eventLoop().<Void>newPromise()).syncUninterruptibly();
+        assertThat(releaseFuture.isSuccess(), CoreMatchers.is(true));
+
+        Channel channel2 = pool.acquire().syncUninterruptibly().getNow();
+        //verifying that in fact the channel2 is different that means is not pulled from the pool
+        assertNotSame(channel1, channel2);
+        sc.close().syncUninterruptibly();
+        channel2.close().syncUninterruptibly();
+        group.shutdownGracefully();
+    }
 }