From 6f00a24296179b0abc9673c94f76d1e551ebdeab Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Dec 2009 05:02:08 +0000 Subject: [PATCH] respect memtable thresholds when replaying commit log patch by jbellis; reviewed by Jun Rao for CASSANDRA-609 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@888701 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + .../cassandra/db/ColumnFamilyStore.java | 33 ++++++++++--------- .../org/apache/cassandra/db/CommitLog.java | 4 +-- .../org/apache/cassandra/db/RowMutation.java | 9 +++-- src/java/org/apache/cassandra/db/Table.java | 17 +++------- 5 files changed, 32 insertions(+), 32 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 60c3a28863..a90baf64bb 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -18,6 +18,7 @@ * avoid GCing tombstones except on major compaction (CASSANDRA-604) * increase failure conviction threshold, resulting in less nodes incorrectly (and temporarily) marked as down (CASSANDRA-610) + * respect memtable thresholds during log replay (CASSANDRA-609) 0.5.0 beta diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 5dbc5ac369..00a6b14c2c 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -372,7 +372,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean columnFamily_, SSTable.TEMPFILE_MARKER, fileIndexGenerator_.incrementAndGet()); } - Future switchMemtable(Memtable oldMemtable) throws IOException + Future switchMemtable(Memtable oldMemtable, final boolean writeCommitLog) throws IOException { /** * If we can get the writelock, that means no new updates can come in and @@ -382,7 +382,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean Table.flusherLock_.writeLock().lock(); try { - final CommitLog.CommitLogContext ctx = CommitLog.open().getContext(); + final CommitLog.CommitLogContext ctx = CommitLog.open().getContext(); // this is harmless if !writeCommitLog if (oldMemtable.isFrozen()) { @@ -401,7 +401,12 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean try { condition.await(); - onMemtableFlush(ctx); + if (writeCommitLog) + { + // if we're not writing to the commit log, we are replaying the log, so marking + // the log header with "you can discard anything written before the context" is not valid + onMemtableFlush(ctx); + } } catch (Exception e) { @@ -438,7 +443,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean if (memtable_.isClean()) return null; - return switchMemtable(memtable_); + return switchMemtable(memtable_, true); } void forceBlockingFlush() throws IOException, ExecutionException, InterruptedException @@ -560,17 +565,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean } } - /* - * This version is used only on start up when we are recovering from logs. - * Hence no locking is required since we process logs on the main thread. In - * the future we may want to parellelize the log processing for a table by - * having a thread per log file present for recovery. Re-visit at that time. - */ - void applyNow(String key, ColumnFamily columnFamily) throws IOException - { - getMemtableThreadSafe().put(key, columnFamily); - } - /* * This method is called when the Memtable is frozen and ready to be flushed * to disk. This method informs the CommitLog that a particular ColumnFamily @@ -1005,6 +999,15 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean return memtables; } + /** + * submits flush sort on the flushSorter executor, which will in turn submit to flushWriter when sorted. + * TODO because our executors use CallerRunsPolicy, when flushSorter fills up, no writes will proceed + * because the next flush will start executing on the caller, mutation-stage thread that has the + * flush write lock held. (writes aquire this as a read lock before proceeding.) + * This is good, because it backpressures flushes, but bad, because we can't write until that last + * flushing thread finishes sorting, which will almost always be longer than any of the flushSorter threads proper + * (since, by definition, it started last). + */ Condition submitFlush(final IFlushable flushable) { logger_.info("Enqueuing flush of " + flushable); diff --git a/src/java/org/apache/cassandra/db/CommitLog.java b/src/java/org/apache/cassandra/db/CommitLog.java index 5f97207d18..fb9c982bc9 100644 --- a/src/java/org/apache/cassandra/db/CommitLog.java +++ b/src/java/org/apache/cassandra/db/CommitLog.java @@ -352,11 +352,11 @@ public class CommitLog { try { - table.applyNow(rm); + rm.apply(false); } catch (IOException e) { - throw new RuntimeException(e); + throw new IOError(e); } } } diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index 76d692e4be..bec3322101 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -199,8 +199,13 @@ public class RowMutation implements Serializable * to the table that is obtained by calling Table.open(). */ public void apply() throws IOException - { - Table.open(table_).apply(this, this.getSerializedBuffer()); + { + apply(true); + } + + public void apply(boolean writeCommitLog) throws IOException + { + Table.open(table_).apply(this, this.getSerializedBuffer(), writeCommitLog); } /* diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 30d4dc2dfc..2b6f39a5bc 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -428,14 +428,15 @@ public class Table * Once this happens the data associated with the individual column families * is also written to the column family store's memtable. */ - void apply(RowMutation mutation, DataOutputBuffer serializedMutation) throws IOException + void apply(RowMutation mutation, DataOutputBuffer serializedMutation, boolean writeCommitLog) throws IOException { HashMap memtablesToFlush = new HashMap(2); flusherLock_.readLock().lock(); try { - CommitLog.open().add(mutation, serializedMutation); + if (writeCommitLog) + CommitLog.open().add(mutation, serializedMutation); for (ColumnFamily columnFamily : mutation.getColumnFamilies()) { @@ -452,17 +453,7 @@ public class Table // usually mTF will be empty and this will be a no-op for (Map.Entry entry : memtablesToFlush.entrySet()) - entry.getKey().switchMemtable(entry.getValue()); - } - - void applyNow(RowMutation row) throws IOException - { - String key = row.key(); - for (ColumnFamily columnFamily : row.getColumnFamilies()) - { - ColumnFamilyStore cfStore = columnFamilyStores_.get(columnFamily.name()); - cfStore.applyNow( key, columnFamily ); - } + entry.getKey().switchMemtable(entry.getValue(), writeCommitLog); } public List> flush() throws IOException