natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

multithread recovery 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@880889 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-11-16 18:22:14 +00:00
parent f1279e9618
commit 6691b2c295
13 changed files with 86 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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

2
src/java/org/apache/cassandra/concurrent/IStage.java
View File

@ -116,4 +116,6 @@ public interface IStage
* @return task count.
*/
public long getPendingTasks();
public long getCompletedTasks();
}

5
src/java/org/apache/cassandra/concurrent/MultiThreadedStage.java
View File

@ -87,4 +87,9 @@ public class MultiThreadedStage implements IStage
public long getPendingTasks(){
return executorService_.getPendingTasks();
}
public long getCompletedTasks()
{
return executorService_.getCompletedTasks();
}
}

5
src/java/org/apache/cassandra/concurrent/SingleThreadedStage.java
View File

@ -96,4 +96,9 @@ public class SingleThreadedStage implements IStage
public long getPendingTasks(){
return executorService_.getPendingTasks();
}
public long getCompletedTasks()
{
return executorService_.getCompletedTasks();
}
}

14
src/java/org/apache/cassandra/concurrent/StageManager.java
View File

@ -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<String, IStage > stageQueues_ = new HashMap<String, IStage>();
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

61
src/java/org/apache/cassandra/db/CommitLog.java
View File

@ -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<Table> tablesRecovered = new HashSet<Table>();
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<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>(rm.getColumnFamilies());
/* remove column families that have already been flushed */
for (ColumnFamily columnFamily : columnFamilies)
final Collection<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>(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<Future<?>> futures = new ArrayList<Future<?>>();
for (Table table : tablesRecovered)

3
src/java/org/apache/cassandra/db/RangeCommand.java
View File

@ -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()));
}

3
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -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;

4
src/java/org/apache/cassandra/db/RowMutation.java
View File

@ -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<String, List<ColumnOrSuperColumn>> cfmap)

3
src/java/org/apache/cassandra/db/RowMutationMessage.java
View File

@ -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")

3
src/java/org/apache/cassandra/dht/BootstrapMetadataMessage.java
View File

@ -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];

2
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -465,7 +465,7 @@ public class StorageProxy implements StorageProxyMBean
for (ReadCommand command: commands)
{
Callable<Object> callable = new weakReadLocalCallable(command);
futures.add(StageManager.getStage(StorageService.readStage_).execute(callable));
futures.add(StageManager.getStage(StageManager.readStage_).execute(callable));
}
for (Future<Object> future : futures)
{

13
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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<AbstractReplicationStrategy> 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()
{