From 6691b2c295ffdc324cf460c73016b319c36e11cb Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 16 Nov 2009 18:22:14 +0000 Subject: [PATCH] multithread recovery git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@880889 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + .../apache/cassandra/concurrent/IStage.java | 2 + .../concurrent/MultiThreadedStage.java | 5 ++ .../concurrent/SingleThreadedStage.java | 5 ++ .../cassandra/concurrent/StageManager.java | 14 +++++ .../org/apache/cassandra/db/CommitLog.java | 61 ++++++++++++++----- .../org/apache/cassandra/db/RangeCommand.java | 3 +- .../org/apache/cassandra/db/ReadCommand.java | 3 +- .../org/apache/cassandra/db/RowMutation.java | 4 +- .../cassandra/db/RowMutationMessage.java | 3 +- .../dht/BootstrapMetadataMessage.java | 3 +- .../cassandra/service/StorageProxy.java | 2 +- .../cassandra/service/StorageService.java | 13 +--- 13 files changed, 86 insertions(+), 33 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index f350479779..201863a8b0 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -41,6 +41,7 @@ interfaces (CASSANDRA-546) * stress.py benchmarking tool improvements (several tickets) * optimized replica placement code (CASSANDRA-525) + * faster log replay on restart (CASSANDRA-539, -540) 0.4.2 diff --git a/src/java/org/apache/cassandra/concurrent/IStage.java b/src/java/org/apache/cassandra/concurrent/IStage.java index ef615907e1..69460d8b8d 100644 --- a/src/java/org/apache/cassandra/concurrent/IStage.java +++ b/src/java/org/apache/cassandra/concurrent/IStage.java @@ -116,4 +116,6 @@ public interface IStage * @return task count. */ public long getPendingTasks(); + + public long getCompletedTasks(); } diff --git a/src/java/org/apache/cassandra/concurrent/MultiThreadedStage.java b/src/java/org/apache/cassandra/concurrent/MultiThreadedStage.java index e9cff8272f..868e4b2b4d 100644 --- a/src/java/org/apache/cassandra/concurrent/MultiThreadedStage.java +++ b/src/java/org/apache/cassandra/concurrent/MultiThreadedStage.java @@ -87,4 +87,9 @@ public class MultiThreadedStage implements IStage public long getPendingTasks(){ return executorService_.getPendingTasks(); } + + public long getCompletedTasks() + { + return executorService_.getCompletedTasks(); + } } diff --git a/src/java/org/apache/cassandra/concurrent/SingleThreadedStage.java b/src/java/org/apache/cassandra/concurrent/SingleThreadedStage.java index 1eb30b330d..71415512dc 100644 --- a/src/java/org/apache/cassandra/concurrent/SingleThreadedStage.java +++ b/src/java/org/apache/cassandra/concurrent/SingleThreadedStage.java @@ -96,4 +96,9 @@ public class SingleThreadedStage implements IStage public long getPendingTasks(){ return executorService_.getPendingTasks(); } + + public long getCompletedTasks() + { + return executorService_.getCompletedTasks(); + } } diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java index 65a6bdb390..f1500f3528 100644 --- a/src/java/org/apache/cassandra/concurrent/StageManager.java +++ b/src/java/org/apache/cassandra/concurrent/StageManager.java @@ -23,6 +23,9 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutorService; +import static org.apache.cassandra.config.DatabaseDescriptor.getConcurrentWriters; +import static org.apache.cassandra.config.DatabaseDescriptor.getConcurrentReaders; + /** * This class manages all stages that exist within a process. The application registers @@ -33,6 +36,17 @@ import java.util.concurrent.ExecutorService; public class StageManager { private static Map stageQueues_ = new HashMap(); + + public final static String readStage_ = "ROW-READ-STAGE"; + public final static String mutationStage_ = "ROW-MUTATION-STAGE"; + public final static String streamStage_ = "STREAM-STAGE"; + + static + { + StageManager.registerStage(mutationStage_, new MultiThreadedStage(mutationStage_, getConcurrentWriters())); + StageManager.registerStage(readStage_, new MultiThreadedStage(readStage_, getConcurrentReaders())); + StageManager.registerStage(streamStage_, new SingleThreadedStage(streamStage_)); + } /** * Register a stage with the StageManager diff --git a/src/java/org/apache/cassandra/db/CommitLog.java b/src/java/org/apache/cassandra/db/CommitLog.java index 7b7d7d24e7..057d19bfe7 100644 --- a/src/java/org/apache/cassandra/db/CommitLog.java +++ b/src/java/org/apache/cassandra/db/CommitLog.java @@ -24,6 +24,8 @@ import org.apache.cassandra.io.DataInputBuffer; import org.apache.cassandra.io.DataOutputBuffer; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FileUtils; +import org.apache.cassandra.concurrent.StageManager; + import org.apache.commons.lang.StringUtils; import org.apache.log4j.Logger; @@ -285,13 +287,15 @@ public class CommitLog void recover(File[] clogs) throws IOException { Set tablesRecovered = new HashSet
(); + assert StageManager.getStage(StageManager.mutationStage_).getCompletedTasks() == 0; + int rows = 0; DataInputBuffer bufIn = new DataInputBuffer(); for (File file : clogs) { int bufferSize = (int)Math.min(file.length(), 32 * 1024 * 1024); BufferedRandomAccessFile reader = new BufferedRandomAccessFile(file.getAbsolutePath(), "r", bufferSize); - CommitLogHeader clHeader = readCommitLogHeader(reader); + final CommitLogHeader clHeader = readCommitLogHeader(reader); /* seek to the lowest position where any CF has non-flushed data */ int lowPos = CommitLogHeader.getLowestPosition(clHeader); if (lowPos == 0) @@ -324,32 +328,61 @@ public class CommitLog bufIn.reset(bytes, bytes.length); /* read the commit log entry */ - RowMutation rm = RowMutation.serializer().deserialize(bufIn); + final RowMutation rm = RowMutation.serializer().deserialize(bufIn); if (logger_.isDebugEnabled()) logger_.debug(String.format("replaying mutation for %s.%s: %s", rm.getTable(), rm.key(), "{" + StringUtils.join(rm.getColumnFamilies(), ", ") + "}")); - Table table = Table.open(rm.getTable()); + final Table table = Table.open(rm.getTable()); tablesRecovered.add(table); - Collection columnFamilies = new ArrayList(rm.getColumnFamilies()); - /* remove column families that have already been flushed */ - for (ColumnFamily columnFamily : columnFamilies) + final Collection columnFamilies = new ArrayList(rm.getColumnFamilies()); + final long entryLocation = reader.getFilePointer(); + Runnable runnable = new Runnable() { - int id = table.getColumnFamilyId(columnFamily.name()); - if (!clHeader.isDirty(id) || reader.getFilePointer() < clHeader.getPosition(id)) + public void run() { - rm.removeColumnFamily(columnFamily); + /* remove column families that have already been flushed before applying the rest */ + for (ColumnFamily columnFamily : columnFamilies) + { + int id = table.getColumnFamilyId(columnFamily.name()); + if (!clHeader.isDirty(id) || entryLocation < clHeader.getPosition(id)) + { + rm.removeColumnFamily(columnFamily); + } + } + if (!rm.isEmpty()) + { + try + { + table.applyNow(rm); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } } - } - if (!rm.isEmpty()) - { - table.applyNow(rm); - } + }; + StageManager.getStage(StageManager.mutationStage_).execute(runnable); + rows++; } reader.close(); } + // wait for all the writes to finish on the mutation stage + while (StageManager.getStage(StageManager.mutationStage_).getCompletedTasks() < rows) + { + try + { + Thread.sleep(10); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + } + // flush replayed tables, allowing commitlog segments to be removed List> futures = new ArrayList>(); for (Table table : tablesRecovered) diff --git a/src/java/org/apache/cassandra/db/RangeCommand.java b/src/java/org/apache/cassandra/db/RangeCommand.java index 8f0fc3ced3..863868827f 100644 --- a/src/java/org/apache/cassandra/db/RangeCommand.java +++ b/src/java/org/apache/cassandra/db/RangeCommand.java @@ -29,6 +29,7 @@ import org.apache.cassandra.io.DataInputBuffer; import org.apache.cassandra.net.Message; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.concurrent.StageManager; public class RangeCommand { @@ -54,7 +55,7 @@ public class RangeCommand DataOutputBuffer dob = new DataOutputBuffer(); serializer.serialize(this, dob); return new Message(FBUtilities.getLocalAddress(), - StorageService.readStage_, + StageManager.readStage_, StorageService.rangeVerbHandler_, Arrays.copyOf(dob.getData(), dob.getLength())); } diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 1f88a88ba5..68b18b3065 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -31,6 +31,7 @@ import org.apache.cassandra.service.StorageService; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.concurrent.StageManager; public abstract class ReadCommand @@ -53,7 +54,7 @@ public abstract class ReadCommand ByteArrayOutputStream bos = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream(bos); ReadCommand.serializer().serialize(this, dos); - return new Message(FBUtilities.getLocalAddress(), StorageService.readStage_, StorageService.readVerbHandler_, bos.toByteArray()); + return new Message(FBUtilities.getLocalAddress(), StageManager.readStage_, StorageService.readVerbHandler_, bos.toByteArray()); } public final QueryPath queryPath; diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index d6d9d654ac..8a97e80bec 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -36,12 +36,12 @@ import org.apache.commons.lang.StringUtils; import org.apache.cassandra.io.DataOutputBuffer; import org.apache.cassandra.io.ICompactSerializer; -import java.net.InetAddress; import org.apache.cassandra.net.Message; import org.apache.cassandra.service.*; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.concurrent.StageManager; public class RowMutation implements Serializable { @@ -224,7 +224,7 @@ public class RowMutation implements Serializable ByteArrayOutputStream bos = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream(bos); serializer().serialize(this, dos); - return new Message(FBUtilities.getLocalAddress(), StorageService.mutationStage_, verbHandlerName, bos.toByteArray()); + return new Message(FBUtilities.getLocalAddress(), StageManager.mutationStage_, verbHandlerName, bos.toByteArray()); } public static RowMutation getRowMutation(String table, String key, Map> cfmap) diff --git a/src/java/org/apache/cassandra/db/RowMutationMessage.java b/src/java/org/apache/cassandra/db/RowMutationMessage.java index fc3e6b4bbd..1f1cb740aa 100644 --- a/src/java/org/apache/cassandra/db/RowMutationMessage.java +++ b/src/java/org/apache/cassandra/db/RowMutationMessage.java @@ -30,6 +30,7 @@ import org.apache.cassandra.io.ICompactSerializer; import org.apache.cassandra.net.Message; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.concurrent.StageManager; public class RowMutationMessage implements Serializable { @@ -51,7 +52,7 @@ public class RowMutationMessage implements Serializable ByteArrayOutputStream bos = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream( bos ); RowMutationMessage.serializer().serialize(this, dos); - return new Message(FBUtilities.getLocalAddress(), StorageService.mutationStage_, verbHandlerName, bos.toByteArray()); + return new Message(FBUtilities.getLocalAddress(), StageManager.mutationStage_, verbHandlerName, bos.toByteArray()); } @XmlElement(name="RowMutation") diff --git a/src/java/org/apache/cassandra/dht/BootstrapMetadataMessage.java b/src/java/org/apache/cassandra/dht/BootstrapMetadataMessage.java index 635fd9672f..719ca50cef 100644 --- a/src/java/org/apache/cassandra/dht/BootstrapMetadataMessage.java +++ b/src/java/org/apache/cassandra/dht/BootstrapMetadataMessage.java @@ -24,6 +24,7 @@ import org.apache.cassandra.io.ICompactSerializer; import org.apache.cassandra.net.Message; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.concurrent.StageManager; /** @@ -57,7 +58,7 @@ class BootstrapMetadataMessage { throw new IOError(e); } - return new Message(FBUtilities.getLocalAddress(), StorageService.streamStage_, StorageService.bootstrapMetadataVerbHandler_, bos.toByteArray() ); + return new Message(FBUtilities.getLocalAddress(), StageManager.streamStage_, StorageService.bootstrapMetadataVerbHandler_, bos.toByteArray() ); } protected BootstrapMetadata[] bsMetadata_ = new BootstrapMetadata[0]; diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index c8e1365680..66ad34ae1d 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -465,7 +465,7 @@ public class StorageProxy implements StorageProxyMBean for (ReadCommand command: commands) { Callable callable = new weakReadLocalCallable(command); - futures.add(StageManager.getStage(StorageService.readStage_).execute(callable)); + futures.add(StageManager.getStage(StageManager.readStage_).execute(callable)); } for (Future future : futures) { diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index c74233ac9d..bc23170682 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -65,11 +65,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public final static String STATE_LEAVING = "LEAVING"; public final static String STATE_LEFT = "LEFT"; - /* All stage identifiers */ - public final static String mutationStage_ = "ROW-MUTATION-STAGE"; - public final static String readStage_ = "ROW-READ-STAGE"; - public final static String streamStage_ = "STREAM-STAGE"; - /* All verb handler identifiers */ public final static String mutationVerbHandler_ = "ROW-MUTATION-VERB-HANDLER"; public final static String binaryVerbHandler_ = "BINARY-VERB-HANDLER"; @@ -235,12 +230,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto MessagingService.instance().registerVerbHandlers(streamInitiateDoneVerbHandler_, new Streaming.StreamInitiateDoneVerbHandler()); MessagingService.instance().registerVerbHandlers(streamFinishedVerbHandler_, new Streaming.StreamFinishedVerbHandler()); - StageManager.registerStage(StorageService.mutationStage_, - new MultiThreadedStage(StorageService.mutationStage_, DatabaseDescriptor.getConcurrentWriters())); - StageManager.registerStage(StorageService.readStage_, - new MultiThreadedStage(StorageService.readStage_, DatabaseDescriptor.getConcurrentReaders())); - StageManager.registerStage(StorageService.streamStage_, new SingleThreadedStage(StorageService.streamStage_)); - Class cls = DatabaseDescriptor.getReplicaPlacementStrategyClass(); Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class}; try @@ -998,7 +987,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } } }; - StageManager.getStage(streamStage_).execute(new Runnable() + StageManager.getStage(StageManager.streamStage_).execute(new Runnable() { public void run() {