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 2012/12/05 21:51:02 UTC
[8/12] git commit: Allow overriding available processors with
-Dcassandra.available_processors Patch by brandonwilliams,
reviewed by iamaleksey for CASSANDRA-4790
Allow overriding available processors with
-Dcassandra.available_processors
Patch by brandonwilliams, reviewed by iamaleksey for
CASSANDRA-4790
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/2f3b5bfb
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/2f3b5bfb
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/2f3b5bfb
Branch: refs/heads/trunk
Commit: 2f3b5bfb2da968b408062d56b9d41ab420b95ba2
Parents: 99c8338
Author: Brandon Williams <br...@apache.org>
Authored: Wed Dec 5 14:26:13 2012 -0600
Committer: Brandon Williams <br...@apache.org>
Committed: Wed Dec 5 14:26:13 2012 -0600
----------------------------------------------------------------------
.../apache/cassandra/concurrent/StageManager.java | 9 +++++----
src/java/org/apache/cassandra/config/Config.java | 3 ++-
.../cassandra/config/DatabaseDescriptor.java | 2 +-
.../PeriodicCommitLogExecutorService.java | 3 ++-
.../db/compaction/ParallelCompactionIterable.java | 2 +-
.../cassandra/hadoop/ColumnFamilyRecordWriter.java | 3 ++-
.../apache/cassandra/io/sstable/SSTableReader.java | 2 +-
.../org/apache/cassandra/service/StorageProxy.java | 2 +-
.../apache/cassandra/thrift/CustomTHsHaServer.java | 3 ++-
.../org/apache/cassandra/utils/FBUtilities.java | 8 ++++++++
10 files changed, 25 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/concurrent/StageManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java
index 7ca45f4..287b19e 100644
--- a/src/java/org/apache/cassandra/concurrent/StageManager.java
+++ b/src/java/org/apache/cassandra/concurrent/StageManager.java
@@ -24,6 +24,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.DatabaseDescriptor.*;
@@ -41,21 +42,21 @@ public class StageManager
public static final long KEEPALIVE = 60; // seconds to keep "extra" threads alive for when idle
- public static final int MAX_REPLICATE_ON_WRITE_TASKS = 1024 * Runtime.getRuntime().availableProcessors();
+ public static final int MAX_REPLICATE_ON_WRITE_TASKS = 1024 * FBUtilities.getAvailableProcessors();
static
{
stages.put(Stage.MUTATION, multiThreadedConfigurableStage(Stage.MUTATION, getConcurrentWriters()));
stages.put(Stage.READ, multiThreadedConfigurableStage(Stage.READ, getConcurrentReaders()));
- stages.put(Stage.REQUEST_RESPONSE, multiThreadedStage(Stage.REQUEST_RESPONSE, Runtime.getRuntime().availableProcessors()));
- stages.put(Stage.INTERNAL_RESPONSE, multiThreadedStage(Stage.INTERNAL_RESPONSE, Runtime.getRuntime().availableProcessors()));
+ stages.put(Stage.REQUEST_RESPONSE, multiThreadedStage(Stage.REQUEST_RESPONSE, FBUtilities.getAvailableProcessors()));
+ stages.put(Stage.INTERNAL_RESPONSE, multiThreadedStage(Stage.INTERNAL_RESPONSE, FBUtilities.getAvailableProcessors()));
stages.put(Stage.REPLICATE_ON_WRITE, multiThreadedConfigurableStage(Stage.REPLICATE_ON_WRITE, getConcurrentReplicators(), MAX_REPLICATE_ON_WRITE_TASKS));
// the rest are all single-threaded
stages.put(Stage.GOSSIP, new JMXEnabledThreadPoolExecutor(Stage.GOSSIP));
stages.put(Stage.ANTI_ENTROPY, new JMXEnabledThreadPoolExecutor(Stage.ANTI_ENTROPY));
stages.put(Stage.MIGRATION, new JMXEnabledThreadPoolExecutor(Stage.MIGRATION));
stages.put(Stage.MISC, new JMXEnabledThreadPoolExecutor(Stage.MISC));
- stages.put(Stage.READ_REPAIR, multiThreadedStage(Stage.READ_REPAIR, Runtime.getRuntime().availableProcessors()));
+ stages.put(Stage.READ_REPAIR, multiThreadedStage(Stage.READ_REPAIR, FBUtilities.getAvailableProcessors()));
stages.put(Stage.TRACING, tracingExecutor());
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/config/Config.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index fa8f273..492bb7a 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.config;
import org.apache.cassandra.cache.SerializingCacheProvider;
import org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions;
import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
+import org.apache.cassandra.utils.FBUtilities;
/**
* A class that contains configuration properties for the cassandra node it runs within.
@@ -100,7 +101,7 @@ public class Config
/* if the size of columns or super-columns are more than this, indexing will kick in */
public Integer column_index_size_in_kb = 64;
public Integer in_memory_compaction_limit_in_mb = 64;
- public Integer concurrent_compactors = Runtime.getRuntime().availableProcessors();
+ public Integer concurrent_compactors = FBUtilities.getAvailableProcessors();
public volatile Integer compaction_throughput_mb_per_sec = 16;
public Boolean multithreaded_compaction = false;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 641ac64..f6095af 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -387,7 +387,7 @@ public class DatabaseDescriptor
}
if (conf.concurrent_compactors == null)
- conf.concurrent_compactors = Runtime.getRuntime().availableProcessors();
+ conf.concurrent_compactors = FBUtilities.getAvailableProcessors();
if (conf.concurrent_compactors <= 0)
throw new ConfigurationException("concurrent_compactors should be strictly greater than 0");
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogExecutorService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogExecutorService.java b/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogExecutorService.java
index 39978ef..94f593e 100644
--- a/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogExecutorService.java
+++ b/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogExecutorService.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.util.concurrent.*;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.WrappedRunnable;
class PeriodicCommitLogExecutorService implements ICommitLogExecutorService
@@ -32,7 +33,7 @@ class PeriodicCommitLogExecutorService implements ICommitLogExecutorService
public PeriodicCommitLogExecutorService(final CommitLog commitLog)
{
- queue = new LinkedBlockingQueue<Runnable>(1024 * Runtime.getRuntime().availableProcessors());
+ queue = new LinkedBlockingQueue<Runnable>(1024 * FBUtilities.getAvailableProcessors());
Runnable runnable = new WrappedRunnable()
{
public void runMayThrow() throws Exception
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java b/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
index 56fce20..b19395c 100644
--- a/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
+++ b/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
@@ -135,7 +135,7 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable
private final List<RowContainer> rows = new ArrayList<RowContainer>();
private int row = 0;
- private final ThreadPoolExecutor executor = new DebuggableThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
+ private final ThreadPoolExecutor executor = new DebuggableThreadPoolExecutor(FBUtilities.getAvailableProcessors(),
Integer.MAX_VALUE,
TimeUnit.MILLISECONDS,
new SynchronousQueue<Runnable>(),
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
index 3b66976..909c291 100644
--- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
+++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
@@ -30,6 +30,7 @@ import org.apache.cassandra.client.RingCache;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.thrift.*;
+import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.RecordWriter;
@@ -101,7 +102,7 @@ implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<Mutation>>
{
this.conf = conf;
this.ringCache = new RingCache(conf);
- this.queueSize = conf.getInt(ColumnFamilyOutputFormat.QUEUE_SIZE, 32 * Runtime.getRuntime().availableProcessors());
+ this.queueSize = conf.getInt(ColumnFamilyOutputFormat.QUEUE_SIZE, 32 * FBUtilities.getAvailableProcessors());
this.clients = new HashMap<Range,RangeClient>();
batchThreshold = conf.getLong(ColumnFamilyOutputFormat.BATCH_THRESHOLD, 32);
consistencyLevel = ConsistencyLevel.valueOf(ConfigHelper.getWriteConsistencyLevel(conf));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
index 8fe820a..1af1b11 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -225,7 +225,7 @@ public class SSTableReader extends SSTable
{
final Collection<SSTableReader> sstables = new LinkedBlockingQueue<SSTableReader>();
- ExecutorService executor = DebuggableThreadPoolExecutor.createWithFixedPoolSize("SSTableBatchOpen", Runtime.getRuntime().availableProcessors());
+ ExecutorService executor = DebuggableThreadPoolExecutor.createWithFixedPoolSize("SSTableBatchOpen", FBUtilities.getAvailableProcessors());
for (final Map.Entry<Descriptor, Set<Component>> entry : entries)
{
Runnable runnable = new Runnable()
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 0c3eae9..1884132 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -79,7 +79,7 @@ public class StorageProxy implements StorageProxyMBean
public static final StorageProxy instance = new StorageProxy();
- private static volatile int maxHintsInProgress = 1024 * Runtime.getRuntime().availableProcessors();
+ private static volatile int maxHintsInProgress = 1024 * FBUtilities.getAvailableProcessors();
private static final AtomicInteger totalHintsInProgress = new AtomicInteger();
private static final Map<InetAddress, AtomicInteger> hintsInProgress = new MapMaker().concurrencyLevel(1).makeComputingMap(new Function<InetAddress, AtomicInteger>()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java b/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java
index 250a549..a199a27 100644
--- a/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java
+++ b/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java
@@ -30,6 +30,7 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
+import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -378,7 +379,7 @@ public class CustomTHsHaServer extends TNonblockingServer
.outputProtocolFactory(args.tProtocolFactory)
.processor(args.processor);
// Check for available processors in the system which will be equal to the IO Threads.
- return new CustomTHsHaServer(serverArgs, executorService, Runtime.getRuntime().availableProcessors());
+ return new CustomTHsHaServer(serverArgs, executorService, FBUtilities.getAvailableProcessors());
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2f3b5bfb/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index 2f59ca1..d894036 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -68,6 +68,14 @@ public class FBUtilities
private static volatile InetAddress localInetAddress;
private static volatile InetAddress broadcastInetAddress;
+ public static int getAvailableProcessors()
+ {
+ if (System.getProperty("cassandra.available_processors") != null)
+ return Integer.parseInt(System.getProperty("cassandra.available_processors"));
+ else
+ return FBUtilities.getAvailableProcessors();
+ }
+
private static final ThreadLocal<MessageDigest> localMD5Digest = new ThreadLocal<MessageDigest>()
{
@Override