parallelize post-recovery flushes.

patch by jbellis; reviewed by Brandon Williams for CASSANDRA-539

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

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -1004,11 +1004,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
         return isSuper_;
     }
 
-    public void flushMemtableOnRecovery() throws IOException
-    {
-        getMemtableThreadSafe().flushOnRecovery();
-    }
-
     public int getMemtableColumnsCount()
     {
         return getMemtableThreadSafe().getCurrentObjectCount();

src/java/org/apache/cassandra/db/CommitLog.java

@@ -32,6 +32,7 @@ import java.util.*;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -283,8 +284,9 @@ public class CommitLog
 
     void recover(File[] clogs) throws IOException
     {
-        DataInputBuffer bufIn = new DataInputBuffer();
+        Set<Table> tablesRecovered = new HashSet<Table>();
 
+        DataInputBuffer bufIn = new DataInputBuffer();
         for (File file : clogs)
         {
             int bufferSize = (int)Math.min(file.length(), 32 * 1024 * 1024);
@@ -299,8 +301,6 @@ public class CommitLog
             if (logger_.isDebugEnabled())
                 logger_.debug("Replaying " + file + " starting at " + lowPos);
 
-            Set<Table> tablesRecovered = new HashSet<Table>();
-
             /* read the logs populate RowMutation and apply */
             while (!reader.isEOF())
             {
@@ -348,10 +348,24 @@ public class CommitLog
                 }
             }
             reader.close();
-            /* apply the rows read -- success will result in the CL file being discarded */
-            for (Table table : tablesRecovered)
+        }
+
+        // flush replayed tables, allowing commitlog segments to be removed
+        List<Future<?>> futures = new ArrayList<Future<?>>();
+        for (Table table : tablesRecovered)
+        {
+            futures.addAll(table.flush());
+        }
+        // wait for flushes to finish before continuing with startup
+        for (Future<?> future : futures)
+        {
+            try
             {
-                table.flush(true);
+                future.get();
+            }
+            catch (Exception e)
+            {
+                throw new RuntimeException(e);
             }
         }
     }

src/java/org/apache/cassandra/db/Memtable.java

@@ -166,15 +166,6 @@ public class Memtable implements Comparable<Memtable>, IFlushable<DecoratedKey>
         }
     }
 
-    /** flush synchronously (in the current thread, not on the executors).
-     *  only the recover code should call this. */
-    void flushOnRecovery() throws IOException {
-        if (!isClean())
-        {
-            writeSortedContents(getSortedKeys());
-        }
-    }
-
     // for debugging
     public String contents()
     {

src/java/org/apache/cassandra/db/Table.java

@@ -24,6 +24,7 @@ import java.io.File;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.concurrent.Future;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.dht.Range;
@@ -487,19 +488,16 @@ public class Table
         }
     }
 
-    public void flush(boolean fRecovery) throws IOException
+    public List<Future<?>> flush() throws IOException
     {
+        List<Future<?>> futures = new ArrayList<Future<?>>();
         for (String cfName : columnFamilyStores_.keySet())
         {
-            if (fRecovery)
-            {
-                columnFamilyStores_.get(cfName).flushMemtableOnRecovery();
-            }
-            else
-            {
-                columnFamilyStores_.get(cfName).forceFlush();
-            }
+            Future<?> future = columnFamilyStores_.get(cfName).forceFlush();
+            if (future != null)
+                futures.add(future);
         }
+        return futures;
     }
 
     // for binary load path.  skips commitlog.

src/java/org/apache/cassandra/io/Streaming.java

@@ -52,7 +52,7 @@ public class Streaming
                 Table table = Table.open(tName);
                 if (logger.isDebugEnabled())
                   logger.debug("Flushing memtables ...");
-                table.flush(false);
+                table.flush();
                 if (logger.isDebugEnabled())
                   logger.debug("Performing anticompaction ...");
                 /* Get the list of files that need to be streamed */