From fbe6c4a854cdf53ec2df092202375bbc2ee81155 Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Tue, 28 Jul 2009 02:46:27 +0000
Subject: [PATCH] add config options for commitlog syncing patch by jbellis;
 reviewed by Jun Rao for CASSANDRA-182

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@798372 13f79535-47bb-0310-9956-ffa450edef68
---
 conf/log4j.properties                         |  2 +-
 conf/storage-conf.xml                         | 29 ++++++++++
 .../cassandra/config/DatabaseDescriptor.java  | 55 +++++++++++++++++--
 .../db/CommitLogExecutorService.java          | 41 ++++++++++++--
 .../cassandra/service/StorageService.java     | 18 +++---
 test/conf/storage-conf.xml                    |  2 +
 6 files changed, 126 insertions(+), 21 deletions(-)

diff --git a/conf/log4j.properties b/conf/log4j.properties
index 914414d895..307dc69273 100644
--- a/conf/log4j.properties
+++ b/conf/log4j.properties
@@ -18,7 +18,7 @@
 # and the pattern to %c instead of %l.  (%l is slower.)
 
 # output messages into a rolling log file as well as stdout
-log4j.rootLogger=DEBUG,stdout,R
+log4j.rootLogger=INFO,stdout,R
 
 # stdout
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
diff --git a/conf/storage-conf.xml b/conf/storage-conf.xml
index e8afa96b20..34e3accf7b 100644
--- a/conf/storage-conf.xml
+++ b/conf/storage-conf.xml
@@ -189,6 +189,35 @@
     -->
     <MemtableObjectCountInMillions>0.01</MemtableObjectCountInMillions>
 
+    <!-- Unlike most systems, in Cassandra writes are faster than
+         reads, so you can afford more of those in parallel.
+	 A good rule of thumb is 2 concurrent reads per processor core.
+         You especially want more concurrentwrites if you are using
+         CommitLogSync + CommitLogSyncDelay. -->
+    <ConcurrentReads>8</ConcurrentReads>
+    <ConcurrentWrites>32</ConcurrentWrites>
+
+    <!-- Turn on CommitLogSync to improve durability.
+         When enabled, Cassandra won't ack writes until the commit log
+         has been synced to disk.  This is less necessary in Cassandra
+         than in traditional databases since replication reduces the
+         odds of losing data from a failure after writing the log
+         entry but before it actually reaches the disk.
+    -->
+    <CommitLogSync>true</CommitLogSync>
+    <!-- Delay (in microseconds) during which additional commit log
+         entries may be written before fsync.  This will increase
+         latency slightly, but can vastly improve throughput where
+         there are many writers.  Set to zero to disable
+         (each entry will be synced individually).
+	 Reasonable values range from a minimal 100 to even 10000
+         if throughput matters more than latency.  (10000us = 10ms
+         write latency isn't even that bad by traditional db
+         standards.)
+    -->
+    <CommitLogSyncDelay>1000</CommitLogSyncDelay>
+         
+
     <!-- Time to wait before garbage-collection deletion markers.
          Set this to a large enough value that you are confident
          that the deletion marker will be propagated to all replicas
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index ff117d5532..fe485b6edd 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -63,7 +63,9 @@ public class DatabaseDescriptor
     private static String logFileDirectory_;
     private static String bootstrapFileDirectory_;
     private static boolean rackAware_ = false;
-    private static int threadsPerPool_ = 4;
+    private static int consistencyThreads_ = 4; // not configurable
+    private static int concurrentReaders_ = 8;
+    private static int concurrentWriters_ = 32;
     private static List<String> tables_ = new ArrayList<String>();
     private static Set<String> applicationColumnFamilies_ = new HashSet<String>();
 
@@ -114,6 +116,10 @@ public class DatabaseDescriptor
     /* initial token in the ring */
     private static String initialToken_ = null;
 
+    private static boolean commitLogSync_;
+
+    private static int commitLogSyncDelay_;
+
     static
     {
         try
@@ -126,6 +132,16 @@ public class DatabaseDescriptor
             /* Cluster Name */
             clusterName_ = xmlUtils.getNodeValue("/Storage/ClusterName");
 
+            String syncRaw = xmlUtils.getNodeValue("/Storage/CommitLogSync");
+            if (!"false".equals(syncRaw) && !"true".equals(syncRaw))
+            {
+                // Bool.valueOf will silently assume false for values it doesn't recognize
+                throw new ConfigurationException("Unrecognized value for CommitLogSync.  Use 'true' or 'false'.");
+            }
+            commitLogSync_ = Boolean.valueOf(xmlUtils.getNodeValue("/Storage/CommitLogSync"));
+
+            commitLogSyncDelay_ = Integer.valueOf(xmlUtils.getNodeValue("/Storage/CommitLogSyncDelay"));
+
             /* Hashing strategy */
             partitionerClass_ = xmlUtils.getNodeValue("/Storage/Partitioner");
             try
@@ -167,9 +183,16 @@ public class DatabaseDescriptor
                 rpcTimeoutInMillis_ = Integer.parseInt(rpcTimeoutInMillis);
 
             /* Thread per pool */
-            String threadsPerPool = xmlUtils.getNodeValue("/Storage/ThreadsPerPool");
-            if ( threadsPerPool != null )
-                threadsPerPool_ = Integer.parseInt(threadsPerPool);
+            String rawReaders = xmlUtils.getNodeValue("/Storage/ConcurrentReads");
+            if (rawReaders != null)
+            {
+                concurrentReaders_ = Integer.parseInt(rawReaders);
+            }
+            String rawWriters = xmlUtils.getNodeValue("/Storage/ConcurrentWrites");
+            if (rawWriters != null)
+            {
+                concurrentWriters_ = Integer.parseInt(rawWriters);
+            }
 
             /* TCP port on which the storage system listens */
             String port = xmlUtils.getNodeValue("/Storage/StoragePort");
@@ -671,9 +694,19 @@ public class DatabaseDescriptor
         return rpcTimeoutInMillis_;
     }
 
-    public static int getThreadsPerPool()
+    public static int getConsistencyThreads()
     {
-        return threadsPerPool_;
+        return consistencyThreads_;
+    }
+
+    public static int getConcurrentReaders()
+    {
+        return concurrentReaders_;
+    }
+
+    public static int getConcurrentWriters()
+    {
+        return concurrentWriters_;
     }
 
     public static String[] getAllDataFileLocations()
@@ -821,4 +854,14 @@ public class DatabaseDescriptor
     {
         return thriftAddress_;
     }
+
+    public static int getCommitLogSyncDelay()
+    {
+        return commitLogSyncDelay_;
+    }
+
+    public static boolean isCommitLogSyncEnabled()
+    {
+        return commitLogSync_;
+    }
 }
diff --git a/src/java/org/apache/cassandra/db/CommitLogExecutorService.java b/src/java/org/apache/cassandra/db/CommitLogExecutorService.java
index 238ad5a363..b79fda8724 100644
--- a/src/java/org/apache/cassandra/db/CommitLogExecutorService.java
+++ b/src/java/org/apache/cassandra/db/CommitLogExecutorService.java
@@ -6,6 +6,8 @@ import java.util.Queue;
 import java.util.ArrayList;
 import java.io.IOException;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
+
 public class CommitLogExecutorService extends AbstractExecutorService
 {
     Queue<CheaterFutureTask> queue;
@@ -17,18 +19,48 @@ public class CommitLogExecutorService extends AbstractExecutorService
         {
             public void run()
             {
-                while (true)
+                if (DatabaseDescriptor.isCommitLogSyncEnabled())
                 {
-                    process();
+                    while (true)
+                    {
+                        processWithSyncDelay();
+                    }
+                }
+                else
+                {
+                    while (true)
+                    {
+                        process();
+                    }
                 }
             }
         };
         new Thread(runnable).start();
     }
 
+    private void process()
+    {
+        while (queue.isEmpty())
+        {
+            try
+            {
+                Thread.sleep(1);
+            }
+            catch (InterruptedException e)
+            {
+                throw new RuntimeException(e);
+            }
+        }
+
+        while (!queue.isEmpty())
+        {
+            queue.remove().run();
+        }
+    }
+
     private ArrayList<CheaterFutureTask> incompleteTasks = new ArrayList<CheaterFutureTask>();
     private ArrayList taskValues = new ArrayList(); // TODO not sure how to generify this
-    void process()
+    private void processWithSyncDelay()
     {
         while (queue.isEmpty())
         {
@@ -45,9 +77,10 @@ public class CommitLogExecutorService extends AbstractExecutorService
         // attempt to do a bunch of LogRecordAdder ops before syncing
         incompleteTasks.clear();
         taskValues.clear();
+        long end = System.nanoTime() + 1000 * DatabaseDescriptor.getCommitLogSyncDelay();
         while (!queue.isEmpty()
                && queue.peek().getRawCallable() instanceof CommitLog.LogRecordAdder
-               && incompleteTasks.size() < 20)
+               && (incompleteTasks.isEmpty() || System.nanoTime() < end))
         {
             CheaterFutureTask task = queue.remove();
             incompleteTasks.add(task);
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 52b466d328..d74f6ca82a 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -61,7 +61,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
     /* All stage identifiers */
     public final static String mutationStage_ = "ROW-MUTATION-STAGE";
     public final static String readStage_ = "ROW-READ-STAGE";
-    public final static String mrStage_ = "MAP-REDUCE-STAGE";
     
     /* All verb handler identifiers */
     public final static String mutationVerbHandler_ = "ROW-MUTATION-VERB-HANDLER";
@@ -223,16 +222,15 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
         MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.rangeVerbHandler_, new RangeVerbHandler());
         
         /* register the stage for the mutations */
-        int threadCount = DatabaseDescriptor.getThreadsPerPool();
-        consistencyManager_ = new DebuggableThreadPoolExecutor(threadCount,
-        		threadCount,
-                Integer.MAX_VALUE, TimeUnit.SECONDS,
-                new LinkedBlockingQueue<Runnable>(), new ThreadFactoryImpl(
-                        "CONSISTENCY-MANAGER"));
+        consistencyManager_ = new DebuggableThreadPoolExecutor(DatabaseDescriptor.getConsistencyThreads(),
+                                                               DatabaseDescriptor.getConsistencyThreads(),
+                                                               Integer.MAX_VALUE, TimeUnit.SECONDS,
+                                                               new LinkedBlockingQueue<Runnable>(), new ThreadFactoryImpl("CONSISTENCY-MANAGER"));
         
-        StageManager.registerStage(StorageService.mutationStage_, new MultiThreadedStage(StorageService.mutationStage_, threadCount));
-        StageManager.registerStage(StorageService.readStage_, new MultiThreadedStage(StorageService.readStage_, 2*threadCount));        
-        StageManager.registerStage(StorageService.mrStage_, new MultiThreadedStage(StorageService.mrStage_, threadCount));
+        StageManager.registerStage(StorageService.mutationStage_,
+                                   new MultiThreadedStage(StorageService.mutationStage_, DatabaseDescriptor.getConcurrentWriters()));
+        StageManager.registerStage(StorageService.readStage_,
+                                   new MultiThreadedStage(StorageService.readStage_, DatabaseDescriptor.getConcurrentReaders()));
 
         if ( DatabaseDescriptor.isRackAware() )
             nodePicker_ = new RackAwareStrategy(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort());
diff --git a/test/conf/storage-conf.xml b/test/conf/storage-conf.xml
index a8bae24e6c..7b3acab189 100644
--- a/test/conf/storage-conf.xml
+++ b/test/conf/storage-conf.xml
@@ -18,6 +18,8 @@
  -->
 <Storage>
    <ClusterName>Test Cluster</ClusterName>
+   <CommitLogSync>true</CommitLogSync>
+   <CommitLogSyncDelay>100</CommitLogSyncDelay>
    <Partitioner>org.apache.cassandra.dht.OrderPreservingPartitioner</Partitioner>
    <RackAware>false</RackAware>
    <ReplicationFactor>1</ReplicationFactor>