diff --git a/CHANGES.txt b/CHANGES.txt index f59f89f374..d1135bbc2b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -133,6 +133,7 @@ * Restore resumable hints delivery (CASSANDRA-11960) * Properly record CAS contention (CASSANDRA-12626) Merged from 3.0: + * 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 85edf742d9..22193c4ce4 100644 --- a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java +++ b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java @@ -20,6 +20,7 @@ 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; /** @@ -56,12 +57,29 @@ public class NamedThreadFactory implements ThreadFactory public Thread newThread(Runnable runnable) { - String name = id + ":" + n.getAndIncrement(); - Thread thread = new FastThreadLocalThread(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/AbstractCommitLogSegmentManager.java b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java index 00ddf44843..eff35f4086 100644 --- a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java +++ b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java @@ -30,6 +30,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import net.nicoulaj.compilecommand.annotations.DontInline; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.*; @@ -146,7 +147,7 @@ public abstract class AbstractCommitLogSegmentManager }; shutdown = false; - managerThread = new Thread(runnable, "COMMIT-LOG-ALLOCATOR"); + managerThread = new Thread(NamedThreadFactory.threadLocalDeallocator(runnable), "COMMIT-LOG-ALLOCATOR"); managerThread.start(); // for simplicity, ensure the first segment is allocated before continuing diff --git a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java index 7b56da3c34..834aa0d556 100644 --- a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java +++ b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java @@ -26,6 +26,7 @@ import org.slf4j.LoggerFactory; import com.codahale.metrics.Timer.Context; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.db.commitlog.CommitLogSegment.Allocation; import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.concurrent.WaitQueue; @@ -149,7 +150,7 @@ public abstract class AbstractCommitLogService }; shutdown = false; - thread = new Thread(runnable, name); + thread = new Thread(NamedThreadFactory.threadLocalDeallocator(runnable), name); thread.start(); } diff --git a/src/java/org/apache/cassandra/index/sasi/TermIterator.java b/src/java/org/apache/cassandra/index/sasi/TermIterator.java index 5b08a5605e..1ddfcb9582 100644 --- a/src/java/org/apache/cassandra/index/sasi/TermIterator.java +++ b/src/java/org/apache/cassandra/index/sasi/TermIterator.java @@ -23,6 +23,7 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.index.sasi.disk.OnDiskIndexBuilder; import org.apache.cassandra.index.sasi.disk.Token; @@ -58,7 +59,7 @@ public class TermIterator extends RangeIterator public Thread newThread(Runnable task) { - return new Thread(task, currentThread + "-SEARCH-" + count.incrementAndGet()) {{ setDaemon(true); }}; + return new Thread(NamedThreadFactory.threadLocalDeallocator(task), currentThread + "-SEARCH-" + count.incrementAndGet()) {{ setDaemon(true); }}; } }); } diff --git a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java index ceb0fa0d1e..1a3dd1de03 100644 --- a/src/java/org/apache/cassandra/net/OutboundTcpConnection.java +++ b/src/java/org/apache/cassandra/net/OutboundTcpConnection.java @@ -46,6 +46,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; @@ -495,31 +496,28 @@ public class OutboundTcpConnection extends FastThreadLocalThread { final AtomicInteger version = new AtomicInteger(NO_VERSION); final CountDownLatch versionLatch = new CountDownLatch(1); - new Thread("HANDSHAKE-" + poolReference.endPoint()) + Runnable r = () -> { - @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(); + } + }; + new Thread(NamedThreadFactory.threadLocalDeallocator(r), "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 efa82346da..013dcec073 100644 --- a/src/java/org/apache/cassandra/repair/RepairRunnable.java +++ b/src/java/org/apache/cassandra/repair/RepairRunnable.java @@ -376,7 +376,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. @@ -443,6 +443,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..904deb3aa9 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; @@ -59,17 +60,14 @@ public class RoundRobinScheduler implements IRequestScheduler taskCount = new Semaphore(options.throttle_limit - 1); queues = new NonBlockingHashMap(); - Runnable runnable = new Runnable() + Runnable runnable = () -> { - public void run() + while (true) { - while (true) - { - schedule(); - } + schedule(); } }; - 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 4cff40943b..8719c6c5ec 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -56,6 +56,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; @@ -634,7 +635,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 @@ -649,7 +650,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE logbackHook.setContext((LoggerContext)LoggerFactory.getILoggerFactory()); logbackHook.run(); } - }, "StorageServiceShutdownHook"); + }), "StorageServiceShutdownHook"); Runtime.getRuntime().addShutdownHook(drainOnShutdown); replacing = isReplacing(); @@ -3532,7 +3533,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; }