new PR for STORM-2306 : Messaging subsystem redesign - #2502
Conversation
|
@revans2 , UTs for the storm-client module were fixed (including JCQueueTest). They run cleanly for me. Are you seeing them fail in your local setup ? Rebasing: Due to the length of the review cycle this PR needed and the need for a stable point for perf runs, I avoided keeping up to date with latest master, so far. Planning to address all review comments first. There are around 300 commits on master to catch up on right now. Will take me a week I think to rebase. |
|
@roshannaik I may have checked it out wrong, I will try again... |
|
Yes I checked it out wrong... My bad. |
|
I am still seeing test failures when running
I'll see if there is more that I can come up with, and I'll try to spend more time looking at the actual code changes. |
revans2
left a comment
There was a problem hiding this comment.
I finished a pass through all of the code. I didn't get as in depth as I had hoped, especially with threading.
I am nervous that some of the routing code that was originally called from the thread that pulled off of the send queue may not be thread safe, and is not being called from arbitrary user threads during an emit.
But overall it looks fairly good.
| @@ -0,0 +1,132 @@ | |||
| --- | |||
There was a problem hiding this comment.
Great Documentation, but can we have some of the other docs link to it?
There was a problem hiding this comment.
I am putting a link to this doc from Concepts.md. If you have better place in mind please let me know.
| * @param func the function to run | ||
| */ | ||
| public void scheduleRecurringMs(long delayMs, final long recurMs, final Runnable func) { | ||
| scheduleMs(delayMs, new Runnable() { |
There was a problem hiding this comment.
nit could we use a java 8 lambda here instead?
scheduleMs(delayMs, () -> {
func.run();
// This avoids a race condition with cancel-timer.
scheduleMs(recurMs, this, true, 0);
});
There was a problem hiding this comment.
Doesn't seem feasible due to the 'this' reference inside the lambda.
| public static final String STORMS_ROOT = "storms"; | ||
| public static final String SUPERVISORS_ROOT = "supervisors"; | ||
| public static final String WORKERBEATS_ROOT = "workerbeats"; | ||
| public static final String BACKPRESSURE_ROOT = "backpressure"; |
There was a problem hiding this comment.
In order to support running older topology versions under a newer 2.x nimbus/etc we should still keep around the basic setup and cleanup of the backpressure nodes in zookeeper at least until a 3.x release.
There was a problem hiding this comment.
@revans2 can u you clarify ? 2.x will require topo jars to be rebuilt for it.
| metricMap.put("skipped-max-spout-ms", skippedMaxSpoutMs); | ||
| metricMap.put("skipped-throttle-ms", skippedThrottleMs); | ||
| metricMap.put("skipped-inactive-ms", skippedInactiveMs); | ||
| metricMap.put("skipped-backpressure-ms", skippedBackPressureMs); |
There was a problem hiding this comment.
./docs/Metrics.md describes these. It should be updated to remove skipped-throttle-ms and replaced with skipped-backpressure-ms (and preferably mention that in older versions of storm skipped-throttle-ms would have been similar)
| if (entry.getValue().isEmptyOverflow()) { | ||
| // move task from bpTasks to noBpTasks | ||
| nonBpTasks.add(entry.getKey()); | ||
| itr.remove(); |
There was a problem hiding this comment.
It looks like there is a race condition here with recordBackpressure. If nonBpTasks.add(entry.getKey()) finishes and then we get a context switch and recordBackpressure completes fully for the same taskId then itr.remove happens we might have an inconsistent state.
There was a problem hiding this comment.
Next update will simplify the logic by using a single map and fix this issue as well.
| oldS = newS; | ||
| } | ||
| if (++count == printFreq) { | ||
| System.err.printf(" ***> %s - %,.2f\n", name, mean()); |
There was a problem hiding this comment.
Can we log this instead of printing it? Also could you file a follow on JIRA so that when we go to a better metrics implementation that we use that instead of printing/logging?
|
|
||
| public static final Object INTERRUPT = new Object(); | ||
|
|
||
| private final ThroughputMeter emptyMeter = new ThroughputMeter("EmptyBatch"); |
There was a problem hiding this comment.
This is never reported anywhere... Do we want to just delete it and move ThroughputMeter to perf?
| if (s==null) // then stop running it | ||
| break; | ||
| if (s>0) | ||
| Thread.sleep(s); |
There was a problem hiding this comment.
This needs to be Time.sleep if we want simulated time to work properly....
| new HashMap<>(), | ||
| new AtomicBoolean(false)); | ||
| return new TupleImpl(context, values, 1, stream); | ||
| return new TupleImpl(context, values, "component", 1, stream); |
There was a problem hiding this comment.
This I think is the cause of some of the test failures....
| protected final long lowMemoryThresholdMB; | ||
| protected final long mediumMemoryThresholdMb; | ||
| protected final long mediumMemoryGracePeriodMs; | ||
| private static int port = 5006; // TODO: Roshan: remove this after stabilization |
There was a problem hiding this comment.
Yes please make sure this is removed, as leaving it in place is a security vulnerability.
| /** | ||
| * Send sampled data to the eventlogger if the global or component level debug flag is set (via nimbus api). | ||
| */ | ||
| public void sendToEventLogger(Executor executor, List values, |
There was a problem hiding this comment.
@HeartSaVioR you pointed out some optimizations are possible to this .. that we can tackle in another jira ... can you elaborate or capture your thoughts into a jira ?
There was a problem hiding this comment.
I reread the code and found sampling percentage can be changed. I was thinking about reducing random.nextDouble(), but in this case we may not be able to do that. Please ignore my previous comment.
HeartSaVioR
left a comment
There was a problem hiding this comment.
I can see only one comment is yet to address. Thanks for addressing review comments.
| ((BoltExecutorStats) executor.getStats()).boltFailedTuple( | ||
| input.getSourceComponent(), input.getSourceStreamId(), delta); | ||
| boltFailInfo.applyOn(task.getUserContext()); | ||
| if (delta != 0) { |
There was a problem hiding this comment.
Looks like missed spot : this should be delta >= 0.
https://github.com/apache/storm/pull/2241/files?diff=split#r158213916
(when (<= 0 delta)
(stats/bolt-failed-tuple! executor-stats
(.getSourceComponent tuple)
(.getSourceStreamId tuple)
delta))))
d6ed79d to
d68720e
Compare
|
Update:
The cause of the last two failures is unclear to me as they are passing on my local machine and I cant see anything interesting in the Travis logs. @revans2 : Have that open question for you wrt that race condition you brought up wrt ExecutorTransfer::remotesBatchSz. Rest everything has been addressed. |
|
Looks like storm-core error is related to worker crash while shutting down. https://travis-ci.org/apache/storm/jobs/336372156 |
d68720e to
658d4af
Compare
|
This looks really really good. I ran through the unit tests and some performance tests and everything looks great. Now I feel like I can get excited about this going in. I am going to spend tomorrow going through the code, but I am really hopeful that we can get this merged in next week. |
8619035 to
9ce6e59
Compare
|
@revans2 and @HeartSaVioR I am hoping we can revisit this topic of how to allow concurrent emits without making it internally thread safe, after this PR is merged. |
9ce6e59 to
5a8d732
Compare
revans2
left a comment
There was a problem hiding this comment.
I made it through the code and it looks good. Most things are nits.
I would like to see the link to the perf docs fixed.
I would like to understand why the acker code changed
and most of all I want to make sure that the backpressure serialization works when java serialization is disabled.
But once those happen I am +1 on the change, and the others nits can be wither cleaned up now or in a follow on JIRA.
|
|
||
| ### Performance Tuning | ||
|
|
||
| Refer to [performance tuning guide](docs/CONTRIBUTING.md) |
There was a problem hiding this comment.
I don't think CONTRIBUTING.md is the performance tuning guide, also the path is relative, and since Performance.md is in the same directory as this we don't need the docs/ in the link
| package org.apache.storm.flux.multilang; | ||
|
|
||
|
|
||
| import org.junit.Ignore; |
There was a problem hiding this comment.
nit: I don't think this is used in here, so can we remove it?
| <snakeyaml.version>1.11</snakeyaml.version> | ||
| <httpclient.version>4.3.3</httpclient.version> | ||
| <clojure.tools.cli.version>0.2.4</clojure.tools.cli.version> | ||
| <jctools.version>2.0.1</jctools.version> |
There was a problem hiding this comment.
Why does flux need jctools? Shouldn't it come with storm-client?
There was a problem hiding this comment.
I dont see any references to jctools in any flux related pom. i do see it in the storm-client's pom. Can you point to the specific offending location ?
There was a problem hiding this comment.
I guess I misread things, please ignore this comment.
| * vs. CPU usage | ||
| * Selects the Bolt's Wait Strategy to use when there are no incoming msgs. Used to trade off latency vs CPU usage. | ||
| */ | ||
| @isString |
There was a problem hiding this comment.
Can we check if this is an instance of the proper parent interface?
| * 2. The spout has hit maxSpoutPending and can't emit any more tuples | ||
| */ | ||
| @isString | ||
| public static final String TOPOLOGY_BACKPRESSURE_WAIT_STRATEGY="topology.backpressure.wait.strategy"; |
There was a problem hiding this comment.
Here too if this is a class name it would be good to verify that it is an instance of a given class early on.
| private IWaitStrategy backPressureWaitStrategy; | ||
|
|
||
| JCQueue transferQueue; // [remoteTaskId] -> JCQueue. Some entries maybe null (if no emits to those tasksIds from this worker) | ||
| AtomicBoolean[] remoteBackPressureStatus; // [[remoteTaskId] -> true/false : indicates if remote task is under BP. |
There was a problem hiding this comment.
Same here for package private.
| if ((!Acker.ACKER_COMPONENT_ID.equals(componentId) && Utils.isSystemId(componentId)) | ||
| || (!enableMessageTimeout && isSpout)) { | ||
| LOG.info("Timeouts disabled for executor {}:{}", componentId, executorId); | ||
| LOG.info("Timeouts disabled for executor " + componentId + ":" + executorId); |
There was a problem hiding this comment.
nit: why did we go back to String concatenation?
| private final Boolean isEventLoggers; | ||
| private final Boolean isDebug; | ||
| private final RotatingMap<Long, TupleInfo> pending; | ||
| private TupleInfo globalTupleInfo = new TupleInfo(); // thread safety: assumes Collector.emit*() calls are externally synchronized (if needed). |
There was a problem hiding this comment.
This feels like a good assumption for a spout, but I would like to understand the cost of making this thread safe (thread local instance etc), and at least document it if that cost is high, or preferably find a low cost solution to throw an exception if it does happen.
There was a problem hiding this comment.
Accessing the Thread Local (TL) instance via ThreadLocal.get() typically involves a map lookup behind the scenes.
Related Note: I reluctantly used TL for latching on to JCQueue.BatchInserter instance for the producers to JCQueue. Reluctant since I noticed perf hit when doing some targeted microbenchmarking. I used it anyway because it was a perf improvement over the ConcurrentHashMap employed in Disruptor, and eliminating TL needed a bigger change to the interface and producers. I think it is possible to achieve TL free JCQueue and gain some perf.. perhaps in a follow up jira.
Although many decisions were measured, due to scope it was not feasible to measure each one. So, in the critical path, I have taken this general approach of :
- avoid locks & synchronization ... and try using lock/wait-free approaches where synchronization is unavoidable.
- avoid map lookups and object creation
This was a case of avoiding synchronization, (TL) map lookups & object allocation.
There was a problem hiding this comment.
Can we do a sanity check for the fast path + documentation? Check if the thread id is the same as the id of the main thread we expect emits to come from. If so we go with the fast path, if not we have a thread local or do some kind of locking + documentation about why you never want the spout to emit from a background thread.
There was a problem hiding this comment.
To get the id of the current thread involves a call to Thread.currentThread() which is quite expensive... so not good to use in fast path.
I am introducing that check if topology.debug is enabled as a compromise. This mode could
be used mode to do any checks in dev mode that are unnecessary or expensive to do repeatedly in production.
I have opened: STORM-2945 to nail down and document background emits support.. we can document both spout & bolt support semantics together in the same document.
| import java.util.concurrent.atomic.AtomicLong; | ||
|
|
||
| // Instances of this type are sent from NettyWorker to upstream WorkerTransfer to indicate BackPressure situation | ||
| public class BackPressureStatus implements java.io.Serializable { |
There was a problem hiding this comment.
Why does this need to be serializable if we are explicitly using kryo for all of the serialization? Can we just register the class with kryo instead so this will also work when java serialization is disabled?
There was a problem hiding this comment.
I was not aware of that. Do we have an example of what i should do instead ?
There was a problem hiding this comment.
There was a problem hiding this comment.
You should be able to get away with just registering the class and not providing a serializer.
There was a problem hiding this comment.
Hmm interesting.
Wondering why ControlMessage, MessageBatch & SaslMessageToken dont follow the same pattern of registering ?
Can you confirm that you are suggesting the following steps ?
- Remove the inheritance from Serializable interface
- Register BackPressureStatus.class with kryo
- Remove the BackPressureStatus.buffer() and BackPressureStatus.read() methods ?
There was a problem hiding this comment.
ControlMessage, MessageBatch and SaslMessageToken are handled explicitly by MessageDecoder and MessageEncoder using the buffer and read methods.
When ControlMessage, SaslMessageToken, or Message Batch writes themselves out to a buffer they do not use kryo to do it.
It is a hand coded protocol (for good or bad).
The messages inside MessageBatch have already been serialized using kryo elsewhere in the pipeline.
For the load messages we hid them as a tuple inside a MessageBatch. We did this so we could do a rolling upgrade with it, but it is an ugly hack.
BackPressureStatus is serialized/deserialized using MessageEncoder/MessageDecoder, but it uses kryo internally to do it.
So please remove Serializable from BackPressureStatus. Register it with kryo. Do not remove BackPressureStatus.buffer() nor BackPressureStatus.read(). I think the changes to KryoTupleSerializer to send a BackPressureState can be removed assuming no one is calling them directly.
The simplest way to test this is to run a topology with multiple workers and topology.fall.back.on.java.serialization=false. This is a config that makes it so kryo does not fall back to java serialization when trying to use kryo.
For normal tuples/end users you can set topology.testing.always.try.serialize=true and every tuple emitted will be serialized, even in local mode. This is a way to unit test that you have setup kryo appropriately, but this BackPressureStatus is a special message so we need to do it differently.
There was a problem hiding this comment.
Thanks for the detailed comment. Made the changes and tested them as well.
|
|
||
| @Override | ||
| public void flush() { | ||
| //NOOP //TODO: Roshan: validate if this is OK |
There was a problem hiding this comment.
TODO needs to go away. Is this OK to not have a flush?
f92a206 to
dc53b2a
Compare
|
I think I have addressed all the major and minor issues as well. |
|
|
||
| public final String workerId; | ||
| public String workerId; | ||
| public final long id; // monotonically increasing id |
There was a problem hiding this comment.
Just wondering: the characteristic of monotonically increasing guarantee id to be unique in a worker, but not among workers, and also reset to 0 after worker crash and restart. Does it hurt the backpressure logic at any chance?
| private static AtomicLong bpCount = new AtomicLong(0); | ||
|
|
||
| public String workerId; | ||
| public final long id; // monotonically increasing id |
There was a problem hiding this comment.
Sorry to leave two duplicated comments: I commented to old commit.
Just wondering: the characteristic of monotonically increasing guarantees id to be unique in a worker, but not among workers, and also reset to 0 after worker crash and restart. Does it hurt the backpressure logic at any chance?
There was a problem hiding this comment.
Its only for debugging purposes.. so that we can co-relate sent & recvd msgs. I have used it to measure latency involved in transmission of BackPressureStatus msgs.
There was a problem hiding this comment.
OK thanks for clarification. I thought it as a kind of guarantee we should ensure. Maybe better to clear out that that's not a requirement and only for debugging purpose.
revans2
left a comment
There was a problem hiding this comment.
+1 this looks good to me. I know this is big so if there are others looking at this patch please say something soon. Otherwise @roshannaik could you squash your commits tomorrow sometime and then I will be happy to merge this in.
| import java.util.List; | ||
| import java.util.Random; | ||
|
|
||
| // Methods are not thread safe. Each thread expected to have a separate instance, or else synchronize externally |
There was a problem hiding this comment.
nit: better to make it as javadoc so that it can be exposed to more ways.
As @revans2 stated, I also think this is a good assumption for a spout, but even better to update the restriction if we have documented any. That's just a 2 cents, not blocker.
There was a problem hiding this comment.
To nail down and document the concurrent emits semantics I had opened STORM-2945
There was a problem hiding this comment.
I thought STORM-2945 was filed to find the way to support background emit without external synchronization, so likely having the chance to keep it unresolved in 2.x. If you intended to document how to enable background emit with current state in STORM-2945, please ignore the comment here.
I still think it is good to change class line comment to be a javadoc format, but it is just a nit, not a thing to block.
There was a problem hiding this comment.
It is to figure out what will have for Storm 2.0... since we cannot make any breaking changes even if we like to thereafter until 3.0. will change this to javadoc.
There was a problem hiding this comment.
OK. Please let me know if you plan to figure out in time frame of Storm 2.0.0. I'll add it in epic of releasing Storm 2.0.0. Thanks!
There was a problem hiding this comment.
yes please thats the intent.
dc53b2a to
5b54809
Compare
|
I'm running performance test (TVL) to see there's any regression here. As @revans2 already did performance test, I wouldn't spend the time going through too deeply (just couple of tests). I would provide +1 after things are going well. |
|
Here's my test result: TVL with rate 85000 and max spout pending 5000. spouts/splitters/counters are set to same as worker count. Again, this is just to see if there's clear performance regression from TVL which we have been using. Full analysis like google doc in STORM-2306 would require so much efforts and resources (dedicated machines). If we could update the numbers with latest master vs latest patch of STORM-2306 it would be really great, but we should be OK if we postpone measuring numbers just before releasing Storm 2.0.0 comparing with latest Storm 1.x release.
In overall, this patch shows half of mean latency, whereas 99%ile/99.9%ile of latencies show less difference (not a kind of linear). It also shows this patch consumes more CPU. The gap is fairly small for single-worker but becomes somewhat huge on 4 workers. I might be wrong about testing so it would be really nice if someone also runs the multi-workers test and shares the result. Even my test result is not wrong I'm +1, because this patch introduces better design of backpressure which should have been enabled by default but we disabled by default since it affected performance. We could file some follow-up issues (with priority to blocker if needed) if there're something we should address. |
HeartSaVioR
left a comment
There was a problem hiding this comment.
+1 Thanks for the amazing contribution and also your patience.
Will quash the commits and refresh this PR so that it can be committed. |
|
Yes I only did the performance test for single node. Script for running TVL:
You may not want to rely on latency on the UI, since it includes metrics while starting up which we would want to discard before numbers are stabilized. If you would want to pick up the numbers from UI, please read it at least after 15 mins later so that metrics regarding first 5 mins would be out of window for last 10 mins. I'm also planning to share some scripts regarding verifying release, running daemons in a node via tmux/tmuxinator. |
5b54809 to
c313bfb
Compare
c313bfb to
bc4c480
Compare
|
Commits Quashed. Thanks. Will use follow up jiras to address any issues discovered. |
|
I want to clarify that my performance tests on the latest code were fairly simple. I did more exhaustive performance tests on an older version of the code that looked good to me, but I didn't save any of the numbers. I am planning to merge this in shortly once I verify all of the unit tests still pass, but I think we should see if we can reproduce the increased CPU utilization that @HeartSaVioR saw in a follow on JIRA. |
…to STORM-2306 STORM-2306: Messaging subsystem redesign This closes #2502
|
Nice feeling for me to see this merged in and I will look into the CPU usage issue reported by @HeartSaVioR. Just Wanted to specifically call out and thank the following folks who have been very helpful in making this possible: |
|
Congrats @roshannaik great effort and perseverance to get this in and thanks to @revans2 for reviewing in great detail. |
|
@HeartSaVioR The below cmd was run. All runs were similar so just posting results from one run for each. Mac Book Runs:Master without 2306 (#ab7b4ca) [Latency from UI =240ms]
Master with 2306 (#09e0123) [Latency from UI =89ms]
Linux Server Runs:Master without 2306 (#ab7b4ca) [Latency from UI =15ms]
Master with 2306 (#09e0123) [Latency from UI =7.8ms]
Summary:On Linux Both CPU and mem usage was significantly better for 2306. On Macbook: Both CPU & mem usage were relatively close but slightly favoring 2306. Again, the actual latency was much better for 2306. |
|
@roshannaik FYI my desktop spec and OS here: OS: Ubuntu 17.10, 4.13.0-32-generic #35-Ubuntu SMP Thu Jan 25 09:13:46 UTC 2018 x86_64 x86_64 x86_64 |
|
I suspect this issue won’t show up on the setups I am using and will need to be triaged on your setup itself. Will work with you offline on that. BTW what version of Java are you using ? |
|
Hi, thanks for the great patch. But I ran into some problems. I ran on ResourceAwareScheduler and it's not working properly. It looks like the __acker-executor was not able to receive messages from spouts and bolts. And spouts and bolts continued to retry sending messages to acker. It then led to another problem: I tried to run on the storm right before this merge and it works properly. I then tried to run it on the storm right after this merge and this issue appears. Could you please verify? Thanks! |
|
@Ethanlm should be able to take a look in a couple days. can you share any settings you are using outside in config file other than the one you noted on the cmd line (toplogy.debug=true) ? esp RAS related. |
|
@roshannaik Thanks for looking into this. I used default configs on secure cluster with ResourceAwareScheduler. |
|
Not very familiar with RAS and how it works ..... it seems to be default enabled ... Can you elaborate what setup was needed for RAS (if any) in your case ? |
|
Yes sure. My storm.yaml is as follows: |
|
@Ethanlm can you please open a jira for this issue ? |
|
@roshannaik Yes sure. https://issues.apache.org/jira/browse/STORM-2983 This blocks my current work and I would really appreciate it if it can be solved soon. Thanks |
Since the old PR page had become unusable (due to lots of old comments). Creating this new PR as a replacement for the old PR.
Results for comparative performance runs between master branch (#aaebc3b) and 2306.
https://docs.google.com/document/d/1A5k41UjVFY8jZg01BHc1fFxmI0AcFz5jRiKhNjhOj7I/edit?usp=sharing