You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/01/10 01:39:57 UTC
[1/4] storm git commit: STORM-2278: Allow max number of disruptor
queue flusher threads to be configurable
Repository: storm
Updated Branches:
refs/heads/1.x-branch 6b4d1a2f1 -> 9a19b69d3
STORM-2278: Allow max number of disruptor queue flusher threads to be configurable
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/4316fe8a
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/4316fe8a
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/4316fe8a
Branch: refs/heads/1.x-branch
Commit: 4316fe8a7cd7bcf34645b4f3b0b59f06915cdf06
Parents: 6b4d1a2
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Thu Jan 5 14:08:02 2017 -0600
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Tue Jan 10 10:24:29 2017 +0900
----------------------------------------------------------------------
.../jvm/org/apache/storm/utils/DisruptorQueue.java | 14 +++++++++++++-
1 file changed, 13 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/4316fe8a/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java b/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java
index fbae1d1..ddde9cc 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java
@@ -63,6 +63,18 @@ public class DisruptorQueue implements IStatefulObject {
private static final String PREFIX = "disruptor-";
private static final FlusherPool FLUSHER = new FlusherPool();
+ private static int getNumFlusherPoolThreads() {
+ int numThreads = 100;
+ try {
+ String threads = System.getProperty("num_flusher_pool_threads", "100");
+ numThreads = Integer.parseInt(threads);
+ } catch (Exception e) {
+ LOG.warn("Error while parsing number of flusher pool threads", e);
+ }
+ LOG.debug("Reading num_flusher_pool_threads Flusher pool threads: {}", numThreads);
+ return numThreads;
+ }
+
private static class FlusherPool {
private static final String THREAD_PREFIX = "disruptor-flush";
private Timer _timer = new Timer(THREAD_PREFIX + "-trigger", true);
@@ -71,7 +83,7 @@ public class DisruptorQueue implements IStatefulObject {
private HashMap<Long, TimerTask> _tt = new HashMap<>();
public FlusherPool() {
- _exec = new ThreadPoolExecutor(1, 100, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1024), new ThreadPoolExecutor.DiscardPolicy());
+ _exec = new ThreadPoolExecutor(1, getNumFlusherPoolThreads(), 10, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(1024), new ThreadPoolExecutor.DiscardPolicy());
ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setDaemon(true)
.setNameFormat(THREAD_PREFIX + "-task-pool")
[4/4] storm git commit: STORM-2278: CHANGELOG
Posted by ka...@apache.org.
STORM-2278: CHANGELOG
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9a19b69d
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9a19b69d
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9a19b69d
Branch: refs/heads/1.x-branch
Commit: 9a19b69d3c0acb6eeff42a07b90aea524b5f946f
Parents: 58236f6
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Tue Jan 10 10:39:42 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Tue Jan 10 10:39:42 2017 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/9a19b69d/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 84c2d80..daaf121 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.1.0
+ * STORM-2278: Allow max number of disruptor queue flusher threads to be configurable
* STORM-2277: Add shaded jar for Druid connector
* STORM-2274: Support named output streams in Hdfs Spout
* STORM-2204: Adding caching capabilities in HBaseLookupBolt
[2/4] storm git commit: STORM-2278: Addressed review comments
Posted by ka...@apache.org.
STORM-2278: Addressed review comments
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/ce43a24f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/ce43a24f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/ce43a24f
Branch: refs/heads/1.x-branch
Commit: ce43a24f7297074af8927b175fdcecbd6c33223d
Parents: 4316fe8
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Mon Jan 9 15:53:18 2017 -0600
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Tue Jan 10 10:38:52 2017 +0900
----------------------------------------------------------------------
storm-core/src/jvm/org/apache/storm/Config.java | 11 +++++++++++
.../src/jvm/org/apache/storm/utils/DisruptorQueue.java | 11 +++++++++--
2 files changed, 20 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/ce43a24f/storm-core/src/jvm/org/apache/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/Config.java b/storm-core/src/jvm/org/apache/storm/Config.java
index 1b9725e..62fafd3 100644
--- a/storm-core/src/jvm/org/apache/storm/Config.java
+++ b/storm-core/src/jvm/org/apache/storm/Config.java
@@ -2265,6 +2265,17 @@ public class Config extends HashMap<String, Object> {
@isString
public static final Object CLIENT_JAR_TRANSFORMER = "client.jartransformer.class";
+ /**
+ * This is a config that is not likely to be used. Internally the disruptor queue will batch entries written
+ * into the queue. A background thread pool will flush those batches if they get too old. By default that
+ * pool can grow rather large, and sacrifice some CPU time to keep the latency low. In some cases you may
+ * want the queue to be smaller so there is less CPU used, but the latency will increase in some situations.
+ * This configs is on a per cluster bases, if you want to control this on a per topology bases you need to set
+ * the java System property for the worker "num_flusher_pool_threads" to the value you want.
+ */
+ @isInteger
+ public static final String STORM_WORKER_DISRUPTOR_FLUSHER_MAX_POOL_SIZE = "storm.worker.disruptor.flusher.max.pool.size";
+
public static void setClasspath(Map conf, String cp) {
conf.put(Config.TOPOLOGY_CLASSPATH, cp);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/ce43a24f/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java b/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java
index ddde9cc..f2f202b 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/DisruptorQueue.java
@@ -31,6 +31,7 @@ import com.lmax.disruptor.TimeoutException;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;
+import org.apache.storm.Config;
import org.apache.storm.metric.api.IStatefulObject;
import org.apache.storm.metric.internal.RateTracker;
import org.slf4j.Logger;
@@ -62,11 +63,17 @@ public class DisruptorQueue implements IStatefulObject {
private static final Object INTERRUPT = new Object();
private static final String PREFIX = "disruptor-";
private static final FlusherPool FLUSHER = new FlusherPool();
-
+
private static int getNumFlusherPoolThreads() {
int numThreads = 100;
try {
- String threads = System.getProperty("num_flusher_pool_threads", "100");
+ Map<String, Object> conf = Utils.readStormConfig();
+ numThreads = Utils.getInt(conf.get(Config.STORM_WORKER_DISRUPTOR_FLUSHER_MAX_POOL_SIZE), numThreads);
+ } catch (Exception e) {
+ LOG.warn("Error while trying to read system config", e);
+ }
+ try {
+ String threads = System.getProperty("num_flusher_pool_threads", String.valueOf(numThreads));
numThreads = Integer.parseInt(threads);
} catch (Exception e) {
LOG.warn("Error while parsing number of flusher pool threads", e);
[3/4] storm git commit: Merge branch 'STORM-2278-1.x' into 1.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-2278-1.x' into 1.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/58236f61
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/58236f61
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/58236f61
Branch: refs/heads/1.x-branch
Commit: 58236f6166b50f8cc594cc7f0248df3d19e2611b
Parents: 6b4d1a2 ce43a24
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Tue Jan 10 10:38:57 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Tue Jan 10 10:38:57 2017 +0900
----------------------------------------------------------------------
storm-core/src/jvm/org/apache/storm/Config.java | 11 ++++++++++
.../org/apache/storm/utils/DisruptorQueue.java | 21 +++++++++++++++++++-
2 files changed, 31 insertions(+), 1 deletion(-)
----------------------------------------------------------------------