natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.X' into trunk

This commit is contained in:
Robert Stupp 2016-12-12 20:39:58 +01:00
parent a92ca61548 e303863258
commit 2d21cbda22
9 changed files with 58 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

22
src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java
View File

@ -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();
}
};
}
}

3
src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java
View File

@ -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

3
src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
View File

@ -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();
}

3
src/java/org/apache/cassandra/index/sasi/TermIterator.java
View File

@ -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<Long, Token>
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); }};
}
});
}

42
src/java/org/apache/cassandra/net/OutboundTcpConnection.java
View File

@ -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
{

4
src/java/org/apache/cassandra/repair/RepairRunnable.java
View File

@ -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();
}
}
});
}));
}
}

12
src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java
View File

@ -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<String, WeightedQueue>();
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");
}

7
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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;
}