diff --git a/CHANGES.txt b/CHANGES.txt index 46f594dc42..fda6719e77 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -18,8 +18,10 @@ * fix tombstone handling in repair and sstable2json (CASSANDRA-2279) * clear Built flag in system table when dropping an index (CASSANDRA-2320) * validate index names (CASSANDRA-1761) + * add memtable_flush_queue_size defaulting to 4 (CASSANDRA-2333) * allow job configuration to set the CL used in Hadoop jobs (CASSANDRA-2331) * queue secondary indexes for flush before the parent (CASSANDRA-2330) + * shut down server for OOM on a Thrift thread (CASSANDRA-2269) >>>>>>> .merge-right.r1081840 diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 98524aa54c..4b02273bae 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -148,6 +148,11 @@ concurrent_writes: 32 # By default this will be set to the amount of data directories defined. #memtable_flush_writers: 1 +# the number of full memtables to allow pending flush, that is, +# waiting for a writer thread. At a minimum, this should be set to +# the maximum number of secondary indexes created on a single CF. +memtable_flush_queue_size: 4 + # Buffer size to use when performing contiguous column slices. # Increase this to the size of the column slices you typically perform sliced_buffer_size_in_kb: 64 diff --git a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java index 78dde47af9..b8d3778b74 100644 --- a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java @@ -26,59 +26,74 @@ import java.util.concurrent.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +/** + * This class encorporates some Executor best practices for Cassandra. Most of the executors in the system + * should use or extend this. There are two main improvements over a vanilla TPE: + * + * - If a task throws an exception, the default uncaught exception handler will be invoked; if there is + * no such handler, the exception will be logged. + * - MaximumPoolSize is not supported. Here is what that means (quoting TPE javadoc): + * + * If fewer than corePoolSize threads are running, the Executor always prefers adding a new thread rather than queuing. + * If corePoolSize or more threads are running, the Executor always prefers queuing a request rather than adding a new thread. + * If a request cannot be queued, a new thread is created unless this would exceed maximumPoolSize, in which case, the task will be rejected. + * + * We don't want this last stage of creating new threads if the queue is full; it makes it needlessly difficult to + * reason about the system's behavior. In other words, if DebuggableTPE has allocated our maximum number of (core) + * threads and the queue is full, we want the enqueuer to block. But to allow the number of threads to drop if a + * stage is less busy, core thread timeout is enabled. + */ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor { protected static Logger logger = LoggerFactory.getLogger(DebuggableThreadPoolExecutor.class); public DebuggableThreadPoolExecutor(String threadPoolName, int priority) { - this(1, 1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName, priority)); + this(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName, priority)); } - public DebuggableThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory) + public DebuggableThreadPoolExecutor(int corePoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory) { - super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory); + super(corePoolSize, corePoolSize, keepAliveTime, unit, workQueue, threadFactory); + allowCoreThreadTimeOut(true); - if (maximumPoolSize > 1) + // preserve task serialization. this is more complicated than it needs to be, + // since TPE rejects if queue.offer reports a full queue. we'll just + // override this with a handler that retries until it gets in. ugly, but effective. + // (there is an extensive analysis of the options here at + // http://today.java.net/pub/a/today/2008/10/23/creating-a-notifying-blocking-thread-pool-executor.html) + this.setRejectedExecutionHandler(new RejectedExecutionHandler() { - // clearly strict serialization is not a requirement. just make the calling thread execute. - this.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); - } - else - { - // preserve task serialization. this is more complicated than it needs to be, - // since TPE rejects if queue.offer reports a full queue. we'll just - // override this with a handler that retries until it gets in. ugly, but effective. - // (there is an extensive analysis of the options here at - // http://today.java.net/pub/a/today/2008/10/23/creating-a-notifying-blocking-thread-pool-executor.html) - this.setRejectedExecutionHandler(new RejectedExecutionHandler() + public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) { - public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) + BlockingQueue queue = executor.getQueue(); + while (true) { - BlockingQueue queue = executor.getQueue(); - while (true) + if (executor.isShutdown()) + throw new RejectedExecutionException("ThreadPoolExecutor has shut down"); + try { - if (executor.isShutdown()) - throw new RejectedExecutionException("ThreadPoolExecutor has shut down"); - try - { - if (queue.offer(task, 1000, TimeUnit.MILLISECONDS)) - break; - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } + if (queue.offer(task, 1000, TimeUnit.MILLISECONDS)) + break; + } + catch (InterruptedException e) + { + throw new AssertionError(e); } } - }); - } + } + }); } + @Override public void afterExecute(Runnable r, Throwable t) { super.afterExecute(r,t); + logExceptionsAfterExecute(r, t); + } + public static void logExceptionsAfterExecute(Runnable r, Throwable t) + { // exceptions wrapped by FutureTask if (r instanceof FutureTask) { @@ -92,7 +107,9 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor } catch (ExecutionException e) { - if (Thread.getDefaultUncaughtExceptionHandler() != null) + if (Thread.getDefaultUncaughtExceptionHandler() == null) + logger.error("Error in ThreadPoolExecutor", e.getCause()); + else Thread.getDefaultUncaughtExceptionHandler().uncaughtException(Thread.currentThread(), e.getCause()); } } @@ -100,7 +117,10 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor // exceptions for non-FutureTask runnables [i.e., added via execute() instead of submit()] if (t != null) { - logger.error("Error in ThreadPoolExecutor", t); + if (Thread.getDefaultUncaughtExceptionHandler() == null) + logger.error("Error in ThreadPoolExecutor", t); + else + Thread.getDefaultUncaughtExceptionHandler().uncaughtException(Thread.currentThread(), t); } } } diff --git a/src/java/org/apache/cassandra/concurrent/JMXConfigurableThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/JMXConfigurableThreadPoolExecutor.java index aa5d433874..dbcd641146 100644 --- a/src/java/org/apache/cassandra/concurrent/JMXConfigurableThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/JMXConfigurableThreadPoolExecutor.java @@ -25,14 +25,13 @@ public class JMXConfigurableThreadPoolExecutor extends JMXEnabledThreadPoolExecu { public JMXConfigurableThreadPoolExecutor(int corePoolSize, - int maximumPoolSize, - long keepAliveTime, + long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, NamedThreadFactory threadFactory, String jmxPath) { - super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, jmxPath); + super(corePoolSize, keepAliveTime, unit, workQueue, threadFactory, jmxPath); } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/concurrent/JMXConfigurableThreadPoolExecutorMBean.java b/src/java/org/apache/cassandra/concurrent/JMXConfigurableThreadPoolExecutorMBean.java index 7b8066b377..054a999b1c 100644 --- a/src/java/org/apache/cassandra/concurrent/JMXConfigurableThreadPoolExecutorMBean.java +++ b/src/java/org/apache/cassandra/concurrent/JMXConfigurableThreadPoolExecutorMBean.java @@ -20,9 +20,7 @@ package org.apache.cassandra.concurrent; public interface JMXConfigurableThreadPoolExecutorMBean extends JMXEnabledThreadPoolExecutorMBean { - void setCorePoolSize(int n); int getCorePoolSize(); - } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java index 3ea9b20baf..6a9112dc6d 100644 --- a/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/JMXEnabledThreadPoolExecutor.java @@ -38,28 +38,27 @@ public class JMXEnabledThreadPoolExecutor extends DebuggableThreadPoolExecutor i public JMXEnabledThreadPoolExecutor(String threadPoolName) { - this(1, 1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName), "internal"); + this(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName), "internal"); } public JMXEnabledThreadPoolExecutor(String threadPoolName, String jmxPath) { - this(1, 1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName), jmxPath); + this(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName), jmxPath); } public JMXEnabledThreadPoolExecutor(String threadPoolName, int priority) { - this(1, 1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName, priority), "internal"); + this(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName, priority), "internal"); } public JMXEnabledThreadPoolExecutor(int corePoolSize, - int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, NamedThreadFactory threadFactory, String jmxPath) { - super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory); + super(corePoolSize, keepAliveTime, unit, workQueue, threadFactory); super.prestartAllCoreThreads(); MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java index 91b548b953..b96c1b3d25 100644 --- a/src/java/org/apache/cassandra/concurrent/StageManager.java +++ b/src/java/org/apache/cassandra/concurrent/StageManager.java @@ -41,8 +41,8 @@ public class StageManager { stages.put(Stage.MUTATION, multiThreadedConfigurableStage(Stage.MUTATION, getConcurrentWriters())); stages.put(Stage.READ, multiThreadedConfigurableStage(Stage.READ, getConcurrentReaders())); - stages.put(Stage.REQUEST_RESPONSE, multiThreadedStage(Stage.REQUEST_RESPONSE, Math.max(2, Runtime.getRuntime().availableProcessors()))); - stages.put(Stage.INTERNAL_RESPONSE, multiThreadedStage(Stage.INTERNAL_RESPONSE, Math.max(2, Runtime.getRuntime().availableProcessors()))); + stages.put(Stage.REQUEST_RESPONSE, multiThreadedStage(Stage.REQUEST_RESPONSE, Runtime.getRuntime().availableProcessors())); + stages.put(Stage.INTERNAL_RESPONSE, multiThreadedStage(Stage.INTERNAL_RESPONSE, Runtime.getRuntime().availableProcessors())); stages.put(Stage.REPLICATE_ON_WRITE, multiThreadedConfigurableStage(Stage.REPLICATE_ON_WRITE, getConcurrentReplicators())); // the rest are all single-threaded stages.put(Stage.STREAM, new JMXEnabledThreadPoolExecutor(Stage.STREAM)); @@ -50,17 +50,12 @@ public class StageManager stages.put(Stage.ANTI_ENTROPY, new JMXEnabledThreadPoolExecutor(Stage.ANTI_ENTROPY)); stages.put(Stage.MIGRATION, new JMXEnabledThreadPoolExecutor(Stage.MIGRATION)); stages.put(Stage.MISC, new JMXEnabledThreadPoolExecutor(Stage.MISC)); - stages.put(Stage.READ_REPAIR, multiThreadedStage(Stage.READ_REPAIR, Math.max(2, Runtime.getRuntime().availableProcessors()))); + stages.put(Stage.READ_REPAIR, multiThreadedStage(Stage.READ_REPAIR, Runtime.getRuntime().availableProcessors())); } private static ThreadPoolExecutor multiThreadedStage(Stage stage, int numThreads) { - // avoid running afoul of requirement in DebuggableThreadPoolExecutor that single-threaded executors - // must have unbounded queues - assert numThreads > 1 : "multi-threaded stages must have at least 2 threads"; - return new JMXEnabledThreadPoolExecutor(numThreads, - numThreads, KEEPALIVE, TimeUnit.SECONDS, new LinkedBlockingQueue(), @@ -70,10 +65,7 @@ public class StageManager private static ThreadPoolExecutor multiThreadedConfigurableStage(Stage stage, int numThreads) { - assert numThreads > 1 : "multi-threaded stages must have at least 2 threads"; - return new JMXConfigurableThreadPoolExecutor(numThreads, - numThreads, KEEPALIVE, TimeUnit.SECONDS, new LinkedBlockingQueue(), diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index d979a2852d..5bbd89c56a 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -115,6 +115,7 @@ public class Config public boolean compaction_preheat_key_cache = true; public boolean incremental_backups = false; + public int memtable_flush_queue_size = 4; public static enum CommitLogSync { periodic, diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 6b3ccb491d..b7f6e54827 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1248,4 +1248,9 @@ public class DatabaseDescriptor { return conf.incremental_backups; } + + public static int getFlushQueueSize() + { + return conf.memtable_flush_queue_size; + } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 62f48f2f21..270821c98d 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -79,19 +79,17 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * called, all data up to the given context has been persisted to SSTables. */ private static final ExecutorService flushSorter - = new JMXEnabledThreadPoolExecutor(1, - Runtime.getRuntime().availableProcessors(), + = new JMXEnabledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), StageManager.KEEPALIVE, TimeUnit.SECONDS, new LinkedBlockingQueue(Runtime.getRuntime().availableProcessors()), new NamedThreadFactory("FlushSorter"), "internal"); private static final ExecutorService flushWriter - = new JMXEnabledThreadPoolExecutor(1, - DatabaseDescriptor.getFlushWriters(), + = new JMXEnabledThreadPoolExecutor(DatabaseDescriptor.getFlushWriters(), StageManager.KEEPALIVE, TimeUnit.SECONDS, - new SynchronousQueue(), + new LinkedBlockingQueue(DatabaseDescriptor.getFlushQueueSize()), new NamedThreadFactory("FlushWriter"), "internal"); public static final ExecutorService postFlushExecutor = new JMXEnabledThreadPoolExecutor("MemtablePostFlusher"); diff --git a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java index 489b3ece5c..00a10c6e39 100644 --- a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java @@ -33,6 +33,7 @@ import org.apache.log4j.PropertyConfigurator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.config.DatabaseDescriptor; @@ -363,6 +364,7 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon protected void afterExecute(Runnable r, Throwable t) { super.afterExecute(r, t); + DebuggableThreadPoolExecutor.logExceptionsAfterExecute(r, t); state.get().logout(); } diff --git a/src/java/org/apache/cassandra/service/WriteResponseHandler.java b/src/java/org/apache/cassandra/service/WriteResponseHandler.java index 4e33056320..cf8be4cbe5 100644 --- a/src/java/org/apache/cassandra/service/WriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/WriteResponseHandler.java @@ -74,36 +74,23 @@ public class WriteResponseHandler extends AbstractWriteResponseHandler protected int determineBlockFor(String table) { - int blockFor = 0; switch (consistencyLevel) { case ONE: - blockFor = 1; - break; + return 1; case ANY: - blockFor = 1; - break; + return 1; case TWO: - blockFor = 2; - break; + return 2; case THREE: - blockFor = 3; - break; + return 3; case QUORUM: - blockFor = (writeEndpoints.size() / 2) + 1; - break; + return (writeEndpoints.size() / 2) + 1; case ALL: - blockFor = writeEndpoints.size(); - break; + return writeEndpoints.size(); default: throw new UnsupportedOperationException("invalid consistency level: " + consistencyLevel.toString()); } - // at most one node per range can bootstrap at a time, and these will be added to the write until - // bootstrap finishes (at which point we no longer need to write to the old ones). - assert 1 <= blockFor && blockFor <= 2 * Table.open(table).getReplicationStrategy().getReplicationFactor() - : String.format("invalid response count %d for replication factor %d", - blockFor, Table.open(table).getReplicationStrategy().getReplicationFactor()); - return blockFor; } public void assureSufficientLiveNodes() throws UnavailableException diff --git a/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java b/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java index 4f8993ac22..54f0c2ede3 100644 --- a/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java +++ b/test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java @@ -35,7 +35,6 @@ public class DebuggableThreadPoolExecutorTest { LinkedBlockingQueue q = new LinkedBlockingQueue(1); DebuggableThreadPoolExecutor executor = new DebuggableThreadPoolExecutor(1, - 1, Integer.MAX_VALUE, TimeUnit.MILLISECONDS, q,