RSocketRequester: unsafe usage of streamId across all interactions

[mostroverkhov]

PR #807 introduced RequestOperator, whose callbacks hookOnFirstRequest(), hookOnRemainingRequests() are not signalling events in serial way. streamId plain reference is unsafely accessed from within above, and also within termination callbacks hookOnTerminal() - called on onComplete & onError signals - that are allowed to race with upstream signals by the RS.

[OlegDokuka]

PR #807 introduced RequestOperator, whose callbacks hookOnFirstRequest(), hookOnRemainingRequests() are not signalling events in serial way

This is no true and it contradicts reactive streams rule 2.7
https://github.com/reactive-streams/reactive-streams-jvm/tree/v1.0.3#2.7

request(N) and request(N) can not overlap each other, so having hookOnFirstRequest racing with hookOnRemainingRequests() is impossible due to spec and implementation in the project reactor

streamId plain reference is unsafely accessed from within above, and also within termination callbacks hookOnTerminal()

Not true as well since hookOnTerminal may be called only by the remote responder frame which happens right after the first request has happened (because in hookOnFirstRequest we initiate stream, so termination can not happen before the stream is initiated). So all the subsequent read should observe the write (that happens in hookOnFirstRequest).

The only possible case is RSocket termination, but it is a known issue which will be addressed properly

[mostroverkhov]

This is no true and it contradicts reactive streams rule 2.7

I mention rule 2.7 specifically -

The intent of this rule is to permit the calling of the request and cancel methods (including from *multiple threads*)

Are you saying Subscriber (as implemented by user of reactor) must establish happen-before between separate subscription.request(n) if It wants to call them from different threads, guaranteeing there is no race between calls? It seems reactor subscribers and sources are prepared to handle different thread request(N):
https://github.com/reactor/reactor-core/blob/54ab847862462c8990ee3bcc14a524a9e1118031/reactor-core/src/main/java/reactor/core/publisher/StrictSubscriber.java#L129
https://github.com/reactor/reactor-core/blob/54ab847862462c8990ee3bcc14a524a9e1118031/reactor-core/src/main/java/reactor/core/publisher/FluxCreate.java#L498

Btw, RequestOperator callbacks are called by at least 2 different threads: "serializer" thread on request (requestN - upstream signal), and netty transport thread on termination (onError/onComplete - downstream signal). Upstream and downstream signals are not serialized, and its responsibility of operator impl to establish happens-before relationship - otherwise streamId at least should be volatile.

RequestOperator doing atomics dance between request(n) and cancel(), which also should not race according to said rule 2.7 -

rsocket-java/rsocket-core/src/main/java/io/rsocket/core/RequestOperator.java

Line 81 in d8da87a

if (WIP.getAndIncrement(this) != 0) {

rsocket-java/rsocket-core/src/main/java/io/rsocket/core/RequestOperator.java

Line 116 in d8da87a

if (WIP.getAndIncrement(this) != 0) {

can you elaborate why doing so for request(n)/cancel, but not subsequent request(n)/request(n)?

[mostroverkhov]

Let me reopen issue as I am weighting whether to switch to 1.0.0 release from my own fork - I will close this by end of the weekend

[OlegDokuka]

Part 1. Rule 2.7

This rule certainly says that Subscriber MUST ensure Serial access.

Must means that the Subscriber mandated to take care of the following:

1. In the context of a Signal, non-overlapping.

Explicitly says -> no simultaneous calls on request

2. In the context of the JVM, calls to methods on an object are serial if and only if there is a happens-before relationship between those calls (implying also that the calls do not overlap).

That said that if there is racing on the Subscriber level on the given subscription instance, the Subscriber MUST ensure that racing is serialized.

3. When the calls are performed asynchronously, coordination to establish the happens-before relationship is to be implemented using techniques such as, but not limited to, atomics, monitors, or locks.

That is how you can implement that

@viktorklang correct me if I'm wrong

Part 2. Strict Subscriber

This guard stands specifically for the case where the upstream subscription is not there, but Subscriber does the request in advance (implemented specifically for reactor internal usage where StrictSubscriber is sent as subscription and then subscribed to the upstream on the late phase. See the following code).

It means that there is racing between onSubscribe(s) from the upstream and request(n) from the downstream. Hence, proper synchronization is required in that specific case.

Part 3. Flux Create

Publisher, I would say, a unique case (and please @simonbasle @viktorklang correct me if I'm wrong) where request may overlap with another request on the Publisher Subscription. Since in order to achieve asynchronous demand from the downstream, while the emitting is in process, FluxCreate, as well as all Publishers implementations in Project Reactor / RxJava does this guarding.

Part 4. Stream ID value observability

Btw, RequestOperator callbacks are called by at least 2 different threads: "serializer

That is true.
However, the initial issue is that

streamId is unsafelly accessed.

If we refer to JMM. Every volatile write makes accessible all changes that happen before. So, every volatile read makes visible those (non-volatile changes)

It means, that we write streamId, -> then we put generated frame to the queue -> queue is drained by the netty pool (at this point we have volatile write since task is executed by Netty EventLoop) -> frame is delivered to the Responder RSocket -> Responder RSocket sends CompleteFrame -> frame is handled by Netty Even Loop (hooray we have volatile read on the same EvenLoop and right, channel is bound to the same event loop which guarantees serial reading) -> onTerminal is handled so having that sequence streamId should be visible due to happens-before relationship

[OlegDokuka]

Feel free to reopen this issue if you have a valid test that reproduces this particular scenario with Project Reactor / RxJava / any valid reactive-stream implementation

[OlegDokuka]

Well. I can see the same guard in the original source of reactive-streams ref exaples. (https://github.com/reactive-streams/reactive-streams-jvm/blob/master/examples/src/main/java/org/reactivestreams/example/unicast/RangePublisher.java#L116)

But I'm still not sure this is applicable for intermediate operators. Thus input from @viktorklang side is really appreciated

[OlegDokuka]

Alright. After some thoughts I ended up there there is no violation of the spec. there is no possibility to call requests simultaneously for intermediate operators.
Publisher subscription is called serially.

E.g let's take into account the FluxPublishOn case where the first request happens on the publisher thread -> new onNext calls are scheduled on the worker, and the worker does subscription.request(N) on that specific worker's thread in a serial manner.

The same valid to FluxSubscribeOn which schedules all subscription.request(n) on the same Worker so there is proper serialization as well.

It means that there is a possibility that upstream subscription is being called from a different thread (specifically PublishOn case) while actual.onXXX is being called from the, e.g. RangeSubscription, but this execution still has racing between request calls and has this happens before the sequence

onSubscribe(RangeSubscription) -> s.request(n) -> onNext -> scheduling -> subscription.request(n)

This sequence says that the first subscription.requestN started data sending, the element was delivered on the different thread and the next request happened from the different thread, at the same time, it means that the previous request is technically completed and the next call from the rescheduled onNext is not overlapping concurrently with the first one. This means that all work done in the first request will be available and visible.

The guard is needed to ensure that if elements are still being sent and the new requests come in, we have to make sure that modification of reuqested holder is no modified concurrently