Drain SinkManyEmitterProcessor buffer after cancel #3789
Conversation
| @@ -383,10 +383,12 @@ public Object scanUnsafe(Attr key) { | |||
| } | |||
|
|
|||
| final void drain() { | |||
| if (WIP.getAndIncrement(this) != 0) { | |||
| if (WIP.get(this) != 0) { | |||
There was a problem hiding this comment.
In the cases where more than 1 thread calling the drain function, all the threads are incrementing the WIP counter while only one of them proceeds to clear the queue and decrements the counter. This leaves the WIP in unclean state.
There was a problem hiding this comment.
By separating the atomic operation into two operations you break the exclusive access to the critical section below. Please start with creating JCStress tests if you're considering to spend more time on this. These drain methods are quite a critical piece and ensuring proper lock-free coordination between different actors is essential. Another aspect is performance so that we make as little volatile accesses as we can.
| @@ -398,6 +400,7 @@ final void drain() { | |||
| boolean empty = q == null || q.isEmpty(); | |||
|
|
|||
| if (checkTerminated(d, empty)) { | |||
| WIP.addAndGet(this, -missed); | |||
There was a problem hiding this comment.
Similarly, these returns also leave the WIP in an unclean state.
bajibalu
force-pushed
the
auto-cancel-sink
branch
from
3ad4a0b
to
251d759
Compare
| @@ -383,10 +383,12 @@ public Object scanUnsafe(Attr key) { | |||
| } | |||
|
|
|||
| final void drain() { | |||
| if (WIP.getAndIncrement(this) != 0) { | |||
| if (WIP.get(this) != 0) { | |||
There was a problem hiding this comment.
By separating the atomic operation into two operations you break the exclusive access to the critical section below. Please start with creating JCStress tests if you're considering to spend more time on this. These drain methods are quite a critical piece and ensuring proper lock-free coordination between different actors is essential. Another aspect is performance so that we make as little volatile accesses as we can.
bajibalu
force-pushed
the
auto-cancel-sink
branch
from
251d759
to
d7e3d7d
Compare
|
@chemicL I addressed your comments and also added a JCStress test. But I am not sure whether the test case is correct. |
The internal buffer/queue in SinkManyEmitterProcessor will be drained after all the subscriptions are canceled.
As explained here the queue/buffer in SinkManyEmitterProcessor is not drained properly after the last subscriber canceled the subscription. This was happening due to the WIP marker is left in an unclean state. This PR fixes the issue by updating the WIP marker. I am not sure if this is the ideal approach though.
Fixes #3715