You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by br...@apache.org on 2010/12/10 18:35:14 UTC
svn commit: r1044450 - in /cassandra/branches/cassandra-0.6: CHANGES.txt
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
src/java/org/apache/cassandra/service/StorageService.java
Author: brandonwilliams
Date: Fri Dec 10 17:35:14 2010
New Revision: 1044450
URL: http://svn.apache.org/viewvc?rev=1044450&view=rev
Log:
correct ordering of drain operations so CL.recover is no longer necessary. Patch by jbellis and brandonwilliams, reviewed by jbellis for CASSANDRA-1408
Modified:
cassandra/branches/cassandra-0.6/CHANGES.txt
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/StorageService.java
Modified: cassandra/branches/cassandra-0.6/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.6/CHANGES.txt?rev=1044450&r1=1044449&r2=1044450&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.6/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.6/CHANGES.txt Fri Dec 10 17:35:14 2010
@@ -16,6 +16,8 @@
* reduce fat client timeout (CASSANDRA-1730)
* cleanup smallest CFs first to increase free temp space for larger ones
(CASSANDRA-1811)
+ * correct ordering of drain operations so CL.recover is no longer necessary
+ (CASSANDRA-1408)
0.6.8
Modified: cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/db/ColumnFamilyStore.java?rev=1044450&r1=1044449&r2=1044450&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/db/ColumnFamilyStore.java (original)
+++ cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/db/ColumnFamilyStore.java Fri Dec 10 17:35:14 2010
@@ -71,27 +71,27 @@ public class ColumnFamilyStore implement
* For BinaryMemtable that's about all that happens. For live Memtables there are two other things
* that switchMemtable does (which should be the only caller of submitFlush in this case).
* First, it puts the Memtable into memtablesPendingFlush, where it stays until the flush is complete
- * and it's been added as an SSTableReader to ssTables_. Second, it adds an entry to commitLogUpdater
+ * and it's been added as an SSTableReader to ssTables_. Second, it adds an entry to postFlushExecutor
* that waits for the flush to complete, then calls onMemtableFlush. This allows multiple flushes
* to happen simultaneously on multicore systems, while still calling onMF in the correct order,
* which is necessary for replay in case of a restart since CommitLog assumes that when onMF is
* called, all data up to the given context has been persisted to SSTables.
*/
- private static ExecutorService flushSorter_
+ private static final ExecutorService flushSorter
= new JMXEnabledThreadPoolExecutor(1,
Runtime.getRuntime().availableProcessors(),
Integer.MAX_VALUE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(Runtime.getRuntime().availableProcessors()),
new NamedThreadFactory("FLUSH-SORTER-POOL"));
- private static ExecutorService flushWriter_
+ private static final ExecutorService flushWriter
= new JMXEnabledThreadPoolExecutor(1,
DatabaseDescriptor.getAllDataFileLocations().length,
Integer.MAX_VALUE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(DatabaseDescriptor.getAllDataFileLocations().length),
new NamedThreadFactory("FLUSH-WRITER-POOL"));
- private static ExecutorService commitLogUpdater_ = new JMXEnabledThreadPoolExecutor("MEMTABLE-POST-FLUSHER");
+ public static final ExecutorService postFlushExecutor = new JMXEnabledThreadPoolExecutor("MEMTABLE-POST-FLUSHER");
private static final int KEY_RANGE_FILE_BUFFER_SIZE = 256 * 1024;
@@ -480,7 +480,7 @@ public class ColumnFamilyStore implement
memtable_ = new Memtable(this);
// a second executor that makes sure the onMemtableFlushes get called in the right order,
// while keeping the wait-for-flush (future.get) out of anything latency-sensitive.
- return commitLogUpdater_.submit(new WrappedRunnable()
+ return postFlushExecutor.submit(new WrappedRunnable()
{
public void runMayThrow() throws InterruptedException, IOException
{
@@ -747,7 +747,7 @@ public class ColumnFamilyStore implement
{
logger_.info("Enqueuing flush of " + flushable);
final Condition condition = new SimpleCondition();
- flushable.flushAndSignal(condition, flushSorter_, flushWriter_);
+ flushable.flushAndSignal(condition, flushSorter, flushWriter);
return condition;
}
Modified: cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/StorageService.java?rev=1044450&r1=1044449&r2=1044450&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/StorageService.java Fri Dec 10 17:35:14 2010
@@ -1606,6 +1606,10 @@ public class StorageService implements I
MessagingService.shutdown();
setMode("Draining: emptying MessageService pools", false);
MessagingService.waitFor();
+
+ setMode("Draining: clearing mutation stage", false);
+ mutationStage.shutdown();
+ mutationStage.awaitTermination(3600, TimeUnit.SECONDS);
// lets flush.
setMode("Draining: flushing column families", false);
@@ -1614,18 +1618,11 @@ public class StorageService implements I
f.get();
- setMode("Draining: replaying commit log", false);
+ ColumnFamilyStore.postFlushExecutor.shutdown();
+ ColumnFamilyStore.postFlushExecutor.awaitTermination(60, TimeUnit.SECONDS);
CommitLog.instance().forceNewSegment();
// want to make sure that any segments deleted as a result of flushing are gone.
DeletionService.waitFor();
- CommitLog.recover();
-
- // commit log recovery just sends work to the mutation stage. (there could have already been work there anyway.
- // Either way, we need to let this one drain naturally, and then we're finished.
- setMode("Draining: clearing mutation stage", false);
- mutationStage.shutdown();
- while (!mutationStage.isTerminated())
- mutationStage.awaitTermination(5, TimeUnit.SECONDS);
setMode("Node is drained", true);
}