From 7f668c6fe117f892cd79863fb9805ea5d5a2823c Mon Sep 17 00:00:00 2001 From: Robert Stupp Date: Mon, 12 Dec 2016 20:28:31 +0100 Subject: [PATCH] Thread local pools never cleaned up patch by Robert Stupp; reviewed by Jason Brown for CASSANDRA-13033 --- CHANGES.txt | 1 + .../concurrent/NamedThreadFactory.java | 24 ++++++++++- .../commitlog/AbstractCommitLogService.java | 3 +- .../db/commitlog/CommitLogSegmentManager.java | 3 +- .../cassandra/net/OutboundTcpConnection.java | 41 +++++++++---------- .../cassandra/repair/RepairRunnable.java | 4 +- .../scheduler/RoundRobinScheduler.java | 3 +- .../cassandra/service/StorageService.java | 7 ++-- .../streaming/ConnectionHandler.java | 4 +- .../compress/CompressedInputStream.java | 3 +- 10 files changed, 59 insertions(+), 34 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index bbd47c12ee..5bc30be07c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.11 + * Thread local pools never cleaned up (CASSANDRA-13033) * Set RPC_READY to false when draining or if a node is marked as shutdown (CASSANDRA-12781) * CQL often queries static columns unnecessarily (CASSANDRA-12768) * Make sure sstables only get committed when it's safe to discard commit log records (CASSANDRA-12956) diff --git a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java index 33c80d57bf..22193c4ce4 100644 --- a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java +++ b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java @@ -20,6 +20,9 @@ package org.apache.cassandra.concurrent; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicInteger; +import io.netty.util.concurrent.FastThreadLocal; +import io.netty.util.concurrent.FastThreadLocalThread; + /** * This class is an implementation of the ThreadFactory interface. This * is useful to give Java threads meaningful names which is useful when using @@ -54,12 +57,29 @@ public class NamedThreadFactory implements ThreadFactory public Thread newThread(Runnable runnable) { - String name = id + ":" + n.getAndIncrement(); - Thread thread = new Thread(threadGroup, runnable, name); + String name = id + ':' + n.getAndIncrement(); + Thread thread = new FastThreadLocalThread(threadGroup, threadLocalDeallocator(runnable), name); thread.setPriority(priority); thread.setDaemon(true); if (contextClassLoader != null) thread.setContextClassLoader(contextClassLoader); return thread; } + + /** + * Ensures that {@link FastThreadLocal#remove() FastThreadLocal.remove()} is called when the {@link Runnable#run()} + * method of the given {@link Runnable} instance completes to ensure cleanup of {@link FastThreadLocal} instances. + * This is especially important for direct byte buffers allocated locally for a thread. + */ + public static Runnable threadLocalDeallocator(Runnable r) + { + return () -> + { + try { + r.run(); + } finally { + FastThreadLocal.removeAll(); + } + }; + } } diff --git a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java index 113d1bac56..e5a5887d58 100644 --- a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java +++ b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.db.commitlog; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.concurrent.WaitQueue; import org.slf4j.*; @@ -159,7 +160,7 @@ public abstract class AbstractCommitLogService } }; - thread = new Thread(runnable, name); + thread = new Thread(NamedThreadFactory.threadLocalDeallocator(runnable), name); thread.start(); } diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java index 82cee50892..79dd316588 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java @@ -32,6 +32,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.ColumnFamilyStore; @@ -174,7 +175,7 @@ public class CommitLogSegmentManager run = true; - managerThread = new Thread(runnable, "COMMIT-LOG-ALLOCATOR"); + managerThread = new Thread(NamedThreadFactory.threadLocalDeallocator(runnable), "COMMIT-LOG-ALLOCATOR"); managerThread.start(); } diff --git a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java index f573787fb4..a9dfcdc9c3 100644 --- a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java +++ b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java @@ -45,6 +45,7 @@ import net.jpountz.lz4.LZ4Compressor; import net.jpountz.lz4.LZ4Factory; import net.jpountz.xxhash.XXHashFactory; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.io.util.DataOutputStreamPlus; import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus; import org.apache.cassandra.io.util.WrappedDataOutputStreamPlus; @@ -505,31 +506,27 @@ public class OutboundTcpConnection extends Thread { final AtomicInteger version = new AtomicInteger(NO_VERSION); final CountDownLatch versionLatch = new CountDownLatch(1); - new Thread("HANDSHAKE-" + poolReference.endPoint()) + new Thread(NamedThreadFactory.threadLocalDeallocator(() -> { - @Override - public void run() + try { - try - { - logger.info("Handshaking version with {}", poolReference.endPoint()); - version.set(inputStream.readInt()); - } - catch (IOException ex) - { - final String msg = "Cannot handshake version with " + poolReference.endPoint(); - if (logger.isTraceEnabled()) - logger.trace(msg, ex); - else - logger.info(msg); - } - finally - { - //unblock the waiting thread on either success or fail - versionLatch.countDown(); - } + logger.info("Handshaking version with {}", poolReference.endPoint()); + version.set(inputStream.readInt()); } - }.start(); + catch (IOException ex) + { + final String msg = "Cannot handshake version with " + poolReference.endPoint(); + if (logger.isTraceEnabled()) + logger.trace(msg, ex); + else + logger.info(msg); + } + finally + { + //unblock the waiting thread on either success or fail + versionLatch.countDown(); + } + }),"HANDSHAKE-" + poolReference.endPoint()).start(); try { diff --git a/src/java/org/apache/cassandra/repair/RepairRunnable.java b/src/java/org/apache/cassandra/repair/RepairRunnable.java index 7dd1b31aa6..213e5c5ac8 100644 --- a/src/java/org/apache/cassandra/repair/RepairRunnable.java +++ b/src/java/org/apache/cassandra/repair/RepairRunnable.java @@ -373,7 +373,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti private Thread createQueryThread(final int cmd, final UUID sessionId) { - return new Thread(new WrappedRunnable() + return new Thread(NamedThreadFactory.threadLocalDeallocator(new WrappedRunnable() { // Query events within a time interval that overlaps the last by one second. Ignore duplicates. Ignore local traces. // Wake up upon local trace activity. Query when notified of trace activity with a timeout that doubles every two timeouts. @@ -440,6 +440,6 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti seen[si].clear(); } } - }); + })); } } diff --git a/src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java b/src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java index c98c0fee2d..61dfa501e2 100644 --- a/src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java +++ b/src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java @@ -25,6 +25,7 @@ import java.util.concurrent.TimeoutException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.RequestSchedulerOptions; import org.cliffc.high_scale_lib.NonBlockingHashMap; @@ -69,7 +70,7 @@ public class RoundRobinScheduler implements IRequestScheduler } } }; - Thread scheduler = new Thread(runnable, "REQUEST-SCHEDULER"); + Thread scheduler = new Thread(NamedThreadFactory.threadLocalDeallocator(runnable), "REQUEST-SCHEDULER"); scheduler.start(); logger.info("Started the RoundRobin Request Scheduler"); } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index d70c8dc218..71cbc35820 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -50,6 +50,7 @@ import org.apache.cassandra.auth.AuthMigrationListener; import org.apache.cassandra.batchlog.BatchRemoveVerbHandler; import org.apache.cassandra.batchlog.BatchStoreVerbHandler; import org.apache.cassandra.batchlog.BatchlogManager; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; @@ -613,7 +614,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } // daemon threads, like our executors', continue to run while shutdown hooks are invoked - drainOnShutdown = new Thread(new WrappedRunnable() + drainOnShutdown = new Thread(NamedThreadFactory.threadLocalDeallocator(new WrappedRunnable() { @Override public void runMayThrow() throws InterruptedException, ExecutionException, IOException @@ -628,7 +629,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE logbackHook.setContext((LoggerContext)LoggerFactory.getILoggerFactory()); logbackHook.run(); } - }, "StorageServiceShutdownHook"); + }), "StorageServiceShutdownHook"); Runtime.getRuntime().addShutdownHook(drainOnShutdown); replacing = DatabaseDescriptor.isReplacing(); @@ -3195,7 +3196,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return 0; int cmd = nextRepairCommand.incrementAndGet(); - new Thread(createRepairTask(cmd, keyspace, options, legacy)).start(); + new Thread(NamedThreadFactory.threadLocalDeallocator(createRepairTask(cmd, keyspace, options, legacy))).start(); return cmd; } diff --git a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java index d3d8ed205e..b83c089118 100644 --- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java +++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java @@ -37,6 +37,8 @@ import com.google.common.util.concurrent.SettableFuture; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.io.util.DataOutputStreamPlus; import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus; import org.apache.cassandra.io.util.WrappedDataOutputStreamPlus; @@ -214,7 +216,7 @@ public class ConnectionHandler this.socket = socket; this.protocolVersion = protocolVersion; - new Thread(this, name() + "-" + session.peer).start(); + new Thread(NamedThreadFactory.threadLocalDeallocator(this), name() + "-" + session.peer).start(); } public ListenableFuture close() diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java index d59849f1e7..6577980303 100644 --- a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java +++ b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java @@ -33,6 +33,7 @@ import com.google.common.primitives.Ints; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.io.compress.CompressionMetadata; import org.apache.cassandra.utils.ChecksumType; import org.apache.cassandra.utils.WrappedRunnable; @@ -91,7 +92,7 @@ public class CompressedInputStream extends InputStream this.dataBuffer = new ArrayBlockingQueue<>(Math.min(info.chunks.length, 1024)); this.crcCheckChanceSupplier = crcCheckChanceSupplier; - new Thread(new Reader(source, info, dataBuffer)).start(); + new Thread(NamedThreadFactory.threadLocalDeallocator(new Reader(source, info, dataBuffer))).start(); } public int read() throws IOException