Repository: cassandra Updated Branches: refs/heads/cassandra-1.2 2b002b469 -> f72738125 refs/heads/cassandra-2.0 c54958a9d -> 0dbb4e028
Fix race in Thrift server that can exhaust the pool of worker threads patch by Christian Rolf and jbellis for CASSANDRA-6788 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f7273812 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f7273812 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f7273812 Branch: refs/heads/cassandra-1.2 Commit: f72738125339c21a6981ecd2bfe86035244e8257 Parents: 2b002b4 Author: Jonathan Ellis <[email protected]> Authored: Sat Jun 7 08:21:33 2014 -0500 Committer: Jonathan Ellis <[email protected]> Committed: Sat Jun 7 08:21:33 2014 -0500 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../thrift/CustomTThreadPoolServer.java | 25 ++++++++++---------- 2 files changed, 14 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/f7273812/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 6bd3aba..11ded69 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -31,6 +31,7 @@ * Fix availability validation for LOCAL_ONE CL (CASSANDRA-7319) * Hint streaming can cause decommission to fail (CASSANDRA-7219) * RepairTask didn't send a correct message on IllegalArgumentException (CASSANDRA-7336) + * Fix race in Thrift server that can exhaust the pool of worker threads (CASSANDRA-6788) 1.2.16 http://git-wip-us.apache.org/repos/asf/cassandra/blob/f7273812/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java b/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java index 7014443..6913fba 100644 --- a/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java +++ b/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java @@ -21,6 +21,7 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; import java.net.SocketTimeoutException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -97,7 +98,7 @@ public class CustomTThreadPoolServer extends TServer { try { - Thread.sleep(100); + Thread.sleep(10); } catch (InterruptedException e) { @@ -122,6 +123,12 @@ public class CustomTThreadPoolServer extends TServer logger.warn("Transport error occurred during acceptance of message.", ttx); } } + catch (RejectedExecutionException e) + { + // worker thread decremented activeClients but hadn't finished exiting + logger.debug("Dropping client connection because our limit of {} has been reached", args.maxWorkerThreads); + continue; + } if (activeClients.get() >= args.maxWorkerThreads) logger.warn("Maximum number of clients " + args.maxWorkerThreads + " reached"); @@ -218,19 +225,13 @@ public class CustomTThreadPoolServer extends TServer } finally { - activeClients.decrementAndGet(); if (socket != null) ThriftSessionManager.instance.connectionComplete(socket); - } - - if (inputTransport != null) - { - inputTransport.close(); - } - - if (outputTransport != null) - { - outputTransport.close(); + if (inputTransport != null) + inputTransport.close(); + if (outputTransport != null) + outputTransport.close(); + activeClients.decrementAndGet(); } } }
