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

add config options for commitlog syncing 

patch by jbellis; reviewed by Jun Rao for CASSANDRA-182

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@798372 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-07-28 02:46:27 +00:00
parent 35ff447ba4
commit fbe6c4a854
6 changed files with 126 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
conf/log4j.properties
View File

@ -18,7 +18,7 @@
# and the pattern to %c instead of %l. (%l is slower.)
# output messages into a rolling log file as well as stdout
log4j.rootLogger=DEBUG,stdout,R
log4j.rootLogger=INFO,stdout,R
# stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender

29
conf/storage-conf.xml
View File

@ -189,6 +189,35 @@
-->
<MemtableObjectCountInMillions>0.01</MemtableObjectCountInMillions>
<!-- Unlike most systems, in Cassandra writes are faster than
reads, so you can afford more of those in parallel.
A good rule of thumb is 2 concurrent reads per processor core.
You especially want more concurrentwrites if you are using
CommitLogSync + CommitLogSyncDelay. -->
<ConcurrentReads>8</ConcurrentReads>
<ConcurrentWrites>32</ConcurrentWrites>
<!-- Turn on CommitLogSync to improve durability.
When enabled, Cassandra won't ack writes until the commit log
has been synced to disk. This is less necessary in Cassandra
than in traditional databases since replication reduces the
odds of losing data from a failure after writing the log
entry but before it actually reaches the disk.
-->
<CommitLogSync>true</CommitLogSync>
<!-- Delay (in microseconds) during which additional commit log
entries may be written before fsync. This will increase
latency slightly, but can vastly improve throughput where
there are many writers. Set to zero to disable
(each entry will be synced individually).
Reasonable values range from a minimal 100 to even 10000
if throughput matters more than latency. (10000us = 10ms
write latency isn't even that bad by traditional db
standards.)
-->
<CommitLogSyncDelay>1000</CommitLogSyncDelay>
<!-- Time to wait before garbage-collection deletion markers.
Set this to a large enough value that you are confident
that the deletion marker will be propagated to all replicas

55
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -63,7 +63,9 @@ public class DatabaseDescriptor
private static String logFileDirectory_;
private static String bootstrapFileDirectory_;
private static boolean rackAware_ = false;
private static int threadsPerPool_ = 4;
private static int consistencyThreads_ = 4; // not configurable
private static int concurrentReaders_ = 8;
private static int concurrentWriters_ = 32;
private static List<String> tables_ = new ArrayList<String>();
private static Set<String> applicationColumnFamilies_ = new HashSet<String>();
@ -114,6 +116,10 @@ public class DatabaseDescriptor
/* initial token in the ring */
private static String initialToken_ = null;
private static boolean commitLogSync_;
private static int commitLogSyncDelay_;
static
{
try
@ -126,6 +132,16 @@ public class DatabaseDescriptor
/* Cluster Name */
clusterName_ = xmlUtils.getNodeValue("/Storage/ClusterName");
String syncRaw = xmlUtils.getNodeValue("/Storage/CommitLogSync");
if (!"false".equals(syncRaw) && !"true".equals(syncRaw))
{
// Bool.valueOf will silently assume false for values it doesn't recognize
throw new ConfigurationException("Unrecognized value for CommitLogSync. Use 'true' or 'false'.");
}
commitLogSync_ = Boolean.valueOf(xmlUtils.getNodeValue("/Storage/CommitLogSync"));
commitLogSyncDelay_ = Integer.valueOf(xmlUtils.getNodeValue("/Storage/CommitLogSyncDelay"));
/* Hashing strategy */
partitionerClass_ = xmlUtils.getNodeValue("/Storage/Partitioner");
try
@ -167,9 +183,16 @@ public class DatabaseDescriptor
rpcTimeoutInMillis_ = Integer.parseInt(rpcTimeoutInMillis);
/* Thread per pool */
String threadsPerPool = xmlUtils.getNodeValue("/Storage/ThreadsPerPool");
if ( threadsPerPool != null )
threadsPerPool_ = Integer.parseInt(threadsPerPool);
String rawReaders = xmlUtils.getNodeValue("/Storage/ConcurrentReads");
if (rawReaders != null)
{
concurrentReaders_ = Integer.parseInt(rawReaders);
}
String rawWriters = xmlUtils.getNodeValue("/Storage/ConcurrentWrites");
if (rawWriters != null)
{
concurrentWriters_ = Integer.parseInt(rawWriters);
}
/* TCP port on which the storage system listens */
String port = xmlUtils.getNodeValue("/Storage/StoragePort");
@ -671,9 +694,19 @@ public class DatabaseDescriptor
return rpcTimeoutInMillis_;
}
public static int getThreadsPerPool()
public static int getConsistencyThreads()
{
return threadsPerPool_;
return consistencyThreads_;
}
public static int getConcurrentReaders()
{
return concurrentReaders_;
}
public static int getConcurrentWriters()
{
return concurrentWriters_;
}
public static String[] getAllDataFileLocations()
@ -821,4 +854,14 @@ public class DatabaseDescriptor
{
return thriftAddress_;
}
public static int getCommitLogSyncDelay()
{
return commitLogSyncDelay_;
}
public static boolean isCommitLogSyncEnabled()
{
return commitLogSync_;
}
}

41
src/java/org/apache/cassandra/db/CommitLogExecutorService.java
View File

@ -6,6 +6,8 @@ import java.util.Queue;
import java.util.ArrayList;
import java.io.IOException;
import org.apache.cassandra.config.DatabaseDescriptor;
public class CommitLogExecutorService extends AbstractExecutorService
{
Queue<CheaterFutureTask> queue;
@ -17,18 +19,48 @@ public class CommitLogExecutorService extends AbstractExecutorService
{
public void run()
{
while (true)
if (DatabaseDescriptor.isCommitLogSyncEnabled())
{
process();
while (true)
{
processWithSyncDelay();
}
}
else
{
while (true)
{
process();
}
}
}
};
new Thread(runnable).start();
}
private void process()
{
while (queue.isEmpty())
{
try
{
Thread.sleep(1);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
}
while (!queue.isEmpty())
{
queue.remove().run();
}
}
private ArrayList<CheaterFutureTask> incompleteTasks = new ArrayList<CheaterFutureTask>();
private ArrayList taskValues = new ArrayList(); // TODO not sure how to generify this
void process()
private void processWithSyncDelay()
{
while (queue.isEmpty())
{
@ -45,9 +77,10 @@ public class CommitLogExecutorService extends AbstractExecutorService
// attempt to do a bunch of LogRecordAdder ops before syncing
incompleteTasks.clear();
taskValues.clear();
long end = System.nanoTime() + 1000 * DatabaseDescriptor.getCommitLogSyncDelay();
while (!queue.isEmpty()
&& queue.peek().getRawCallable() instanceof CommitLog.LogRecordAdder
&& incompleteTasks.size() < 20)
&& (incompleteTasks.isEmpty() || System.nanoTime() < end))
{
CheaterFutureTask task = queue.remove();
incompleteTasks.add(task);

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

@ -61,7 +61,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
/* All stage identifiers */
public final static String mutationStage_ = "ROW-MUTATION-STAGE";
public final static String readStage_ = "ROW-READ-STAGE";
public final static String mrStage_ = "MAP-REDUCE-STAGE";
/* All verb handler identifiers */
public final static String mutationVerbHandler_ = "ROW-MUTATION-VERB-HANDLER";
@ -223,16 +222,15 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.rangeVerbHandler_, new RangeVerbHandler());
/* register the stage for the mutations */
int threadCount = DatabaseDescriptor.getThreadsPerPool();
consistencyManager_ = new DebuggableThreadPoolExecutor(threadCount,
threadCount,
Integer.MAX_VALUE, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(), new ThreadFactoryImpl(
"CONSISTENCY-MANAGER"));
consistencyManager_ = new DebuggableThreadPoolExecutor(DatabaseDescriptor.getConsistencyThreads(),
DatabaseDescriptor.getConsistencyThreads(),
Integer.MAX_VALUE, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(), new ThreadFactoryImpl("CONSISTENCY-MANAGER"));
StageManager.registerStage(StorageService.mutationStage_, new MultiThreadedStage(StorageService.mutationStage_, threadCount));
StageManager.registerStage(StorageService.readStage_, new MultiThreadedStage(StorageService.readStage_, 2*threadCount));
StageManager.registerStage(StorageService.mrStage_, new MultiThreadedStage(StorageService.mrStage_, threadCount));
StageManager.registerStage(StorageService.mutationStage_,
new MultiThreadedStage(StorageService.mutationStage_, DatabaseDescriptor.getConcurrentWriters()));
StageManager.registerStage(StorageService.readStage_,
new MultiThreadedStage(StorageService.readStage_, DatabaseDescriptor.getConcurrentReaders()));
if ( DatabaseDescriptor.isRackAware() )
nodePicker_ = new RackAwareStrategy(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort());

2
test/conf/storage-conf.xml
View File

@ -18,6 +18,8 @@
-->
<Storage>
<ClusterName>Test Cluster</ClusterName>
<CommitLogSync>true</CommitLogSync>
<CommitLogSyncDelay>100</CommitLogSyncDelay>
<Partitioner>org.apache.cassandra.dht.OrderPreservingPartitioner</Partitioner>
<RackAware>false</RackAware>
<ReplicationFactor>1</ReplicationFactor>