merge from 0.7

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1081925 13f79535-47bb-0310-9956-ffa450edef68

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

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

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<Runnable>(), new NamedThreadFactory(threadPoolName, priority));
+        this(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory(threadPoolName, priority));
     }
 
-    public DebuggableThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory)
+    public DebuggableThreadPoolExecutor(int corePoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> 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<Runnable> queue = executor.getQueue();
+                while (true)
                 {
-                    BlockingQueue<Runnable> 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);
         }
     }
 }

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<Runnable> workQueue, 
                                              NamedThreadFactory threadFactory,
                                              String jmxPath)
     {
-        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, jmxPath);
+        super(corePoolSize, keepAliveTime, unit, workQueue, threadFactory, jmxPath);
     }
     
 }

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

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<Runnable>(), new NamedThreadFactory(threadPoolName), "internal");
+        this(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory(threadPoolName), "internal");
     }
 
     public JMXEnabledThreadPoolExecutor(String threadPoolName, String jmxPath)
     {
-        this(1, 1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory(threadPoolName), jmxPath);
+        this(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory(threadPoolName), jmxPath);
     }
 
     public JMXEnabledThreadPoolExecutor(String threadPoolName, int priority)
     {
-        this(1, 1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory(threadPoolName, priority), "internal");
+        this(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory(threadPoolName, priority), "internal");
     }
 
     public JMXEnabledThreadPoolExecutor(int corePoolSize,
-                                        int maximumPoolSize,
                                         long keepAliveTime,
                                         TimeUnit unit,
                                         BlockingQueue<Runnable> workQueue,
                                         NamedThreadFactory threadFactory,
                                         String jmxPath)
     {
-        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
+        super(corePoolSize, keepAliveTime, unit, workQueue, threadFactory);
         super.prestartAllCoreThreads();
 
         MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();

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<Runnable>(),
@@ -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<Runnable>(),

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,

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

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<Runnable>(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<Runnable>(),
+                                               new LinkedBlockingQueue<Runnable>(DatabaseDescriptor.getFlushQueueSize()),
                                                new NamedThreadFactory("FlushWriter"),
                                                "internal");
     public static final ExecutorService postFlushExecutor = new JMXEnabledThreadPoolExecutor("MemtablePostFlusher");

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

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

test/unit/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutorTest.java

@@ -35,7 +35,6 @@ public class DebuggableThreadPoolExecutorTest
     {
         LinkedBlockingQueue<Runnable> q = new LinkedBlockingQueue<Runnable>(1);
         DebuggableThreadPoolExecutor executor = new DebuggableThreadPoolExecutor(1,
-                                                                                 1,
                                                                                  Integer.MAX_VALUE,
                                                                                  TimeUnit.MILLISECONDS,
                                                                                  q,