make sure after resume we just deliver one message at one moment#5525
make sure after resume we just deliver one message at one moment#5525hadadil wants to merge 3 commits intoeclipse-vertx:4.xfrom
Conversation
hadadil
force-pushed
the
do-not-receive-msgs-after-resume
branch
2 times, most recently
from
9d8836e to
bdf86b3
Compare
| throw new NullPointerException(); | ||
| } | ||
| context.emit(msg, handler); | ||
| context.dispatch(msg, handler); |
There was a problem hiding this comment.
this was reverted as it had a side effect:
after resume we were received many messages even we paused again during the first message processing.
| // Check if there are more pending messages in the queue that can be processed next time around | ||
| if (!pending.isEmpty() && demand > 0L) { | ||
| context.nettyEventLoop().execute(() -> { | ||
| if (demand > 0L && !pending.isEmpty()) { |
There was a problem hiding this comment.
I flip here so the demand evaulated first
| if (!pending.isEmpty() && demand > 0L) { | ||
| context.nettyEventLoop().execute(() -> { | ||
| if (demand > 0L && !pending.isEmpty()) { | ||
| context.executor().execute(() -> { |
There was a problem hiding this comment.
here we should schedult the task for the next tick on the context thread (so we have no 2 thread dispatching a message at the same moment).
In the past we had a scenario like:
- received
- pause
- resume
- received + paused
- received ?!
so after we resume we had sometimes got two messages even in the 1st on we said already we don't want to have more right now.
| } | ||
| if (pending.size() < maxBufferedMessages) { | ||
| pending.add(message); | ||
| checkNextTick(); |
There was a problem hiding this comment.
this method simplified a lot and here we always just put it in hte pending (if the buffer not full yet of course) and send a tick (that will decied if we need to do anything right now or not)
as the tick evaulation happen on the contex thread we are safe (single threaded)
hadadil
force-pushed
the
do-not-receive-msgs-after-resume
branch
2 times, most recently
from
98dc54c to
a2860aa
Compare
hadadil
force-pushed
the
do-not-receive-msgs-after-resume
branch
from
a2860aa to
e93ab84
Compare
…tricks could be updatedy
| eb.send(ADDRESS1, "val1"); | ||
| Context ctx = Vertx.currentContext(); | ||
| ctx.runOnContext(v -> { | ||
| consumer.resume(); |
There was a problem hiding this comment.
this was in the past executed on eventloop thread on the context... now this is executed on the context thread (in case we area lready on it it will not schedule new job, in case we are on another thread it will schedule an async job.
in short:
now this is executed here so the logic below was not anymore fine as this call wipe out sync the pending msgs
| if (!pending.isEmpty() && demand > 0L) { | ||
| context.nettyEventLoop().execute(() -> { | ||
| if (demand > 0L && !pending.isEmpty()) { | ||
| context.execute(__ -> { |
There was a problem hiding this comment.
here we do execute the resume/fetch from the context thread so we do not have situations when we call resume and the same time we do receive a msg and we get 2msgs (on on the context thread and one on eventloop thread)
| log.warn("Discarding message as more than " + maxBufferedMessages + " buffered in " + pause + " consumer. address: " + address); | ||
| } | ||
| theHandler = handler; | ||
| return false; |
There was a problem hiding this comment.
this was also not 100% right, when we do discard msg we should return with false to have that in the metrics
1 participant
Motivation:
I had a change recently in 4.5.13 (not a lucky number)
After upgrading to this version we have noticed after resume we do receive more than one message at one moment.
First I extended the test to reproduce the case... and after that I done the prod code change.
Conformance:
You should have signed the Eclipse Contributor Agreement as explained in https://github.com/eclipse/vert.x/blob/master/CONTRIBUTING.md
Please also make sure you adhere to the code style guidelines: https://github.com/vert-x3/wiki/wiki/Vert.x-code-style-guidelines