Hardening automatic context propagation

[chemicL]

Summary

This change addresses Flux, Mono, and Publisher implementations that are implemented outside of reactor-core for the purposes of automatic restoration of ThreadLocal state from the Context.
Original issue: #3478

Motivation

Arbitrary Flux, Mono, or raw Publisher implementations can deliver signals from any Thread they desire, with no need to use Scheduler-controlled Threads. The implementations in reactor-core, whenever they switch Threads or when they receive signals out-of-band (e.g. request(n)) restore the ThreadLocal state. Recently, factories for Flux and Mono creation from external sources, such as a raw Publisher, or Future implementations, have added guards via a dedicated ThreadLocal-restoring Subscriber to deal with signals from the sources that are not performing the restoration. However, custom Flux, Mono, and Publisher sources not wrapped via from() method can deliver signals without the ThreadLocal state restored. This is the case with reactor-netty's Flux implementation delivering signals from Netty's event loop. Also, while wrapping the sources obtained via Flux.from and Mono.from factories works fine, plenty of operators, such as flatMap, accept a raw Publisher and don't provide any guard against changing Thread, which means these operators need to be addressed.

Core ideas

This PR applies wrapping of either the Producer or Subscriber. It is driven by a Scannable flag on the Publisher. We do it in these scenarios:

1. Wrap the Publisher when we assemble an operator at runtime of another operator (flatMap for example).
2. Wrap the Subscriber at operators that allow the user to directly emit signals from a different Thread (Sinks for example).
3. Wrap the Subscriber when subscribe(Subscriber) from RS Publisher is called and the source is not protected.

Why wrap the Publisher and not always wrap the Subscriber? The reason is that even when we wrap the Subscriber, the user can still call the base method, which would again wrap the Subscriber. We could also introduce a flag on the Subscriber type to determine that or do plenty of instanceof checks. The latter would be too expensive, while the former could potentially be a very intrusive and long effort.

Changes

• Added a new Scannable attribute: INTERNAL_PRODUCER, which allows determining if a Flux or Mono is a reactor-core implementation.
• Reviewed Scannable implementations to report INTERNAL_PRODUCER properly for internal implementations.
• Introduced new methods in Operators class, which allow wrapping the Subscriber with a context restoration guard whenever the Publisher is not an INTERNAL_PRODUCER and allow wrapping the Publisher directly by applying an operator that restores ThreadLocals when signals travel downstream.
• In Flux, Mono, InternalFluxPublisher, InternalMonoPublisher, FluxFromMonoOperator, MonoFromFluxOperator, MonoFromPublisher, the subscribe(Subscriber) method from Publisher uses Subscriber wrapping as a last-resort chance to restore ThreadLocals. In case subscribe(CoreSubscriber) is used, we have no way to intercept it, as it is abstract, so we need other means of wrapping too.
• Flux.from, Mono.from, Mono.fromDirect and their respective wrap() methods also apply the wrapping.
• In operators which receive a Publisher from the user during runtime (for instance flatMap receives it as a return value of a lambda provided by the user), either Operators.toFluxOrMono is used or direct type wrapping when it is known what is expected.
• Added plenty of tests that depict the issue for the identified cases that are able to accept a Publisher that can switch a Thread when delivering signals

Considered alternatives

A workaround would be to recommend users to wrap any non-reactor implementation with .contextWrite(Function.identity()), however that seems impractical and can be a source of errors when a library unaware of such a recommendation doesn't do so for its sources. Even if a library would do so, the user could apply another such guard leading to undesired performance impact. Therefore, it seems like the solution needs to land in reactor-core.

A different way to implement the barrier that restores ThreadLocal values would be to introduce a parent class for Flux and for Mono that is on the subscribe path which wraps every Subscriber when necessary. This would not prevent against raw Publisher implementations though. [TBC]

TODO

• Review ParallelFlux implementations.
• Review SourceProducer implementations.
• Review InnerProducer implementations.
• Review ConnectableFlux implementations and hot producers.
• Review GroupedFlux implementations.
• Review Sinks and Flux and Mono creators that have generative sinks.
• Review Processors implementations. (DEPRECATED, skip)
• Deliver onComplete asynchronously by the test Flux and Mono and validate the ThreadLocal state.
• Validate onError cases.
• Review tests and extract common patterns (methods accepting a Supplier of the chain to validate signals on)
• Consider a from(Publisher) method that returns a Publisher that is either a Flux or Mono, which has appropriate guards (Mono for a non-INTERNAL_PRODUCER Mono, Flux for anything else). Implemented as Operators.toFluxOrMono.

[OlegDokuka]

the minor outcome is that super scan will go through the same attr one more time (hey, potential small perf penalty). So... may be it worth just adding attr explicitly as it was before

[chemicL]

Actually, the individual implementations should not duplicate what the parent does, just enhancements and changes should be reported.

[OlegDokuka]

That is for common cases. Here, in low-level library, that was likely done performance wise. Even it was not important before, now we would have to be sure that by doing common OOP stuff we don't impact performance on the hot path (since now that code is part of the hot path and any redundancy will add significant impact at scale).
Such thing may invisible for latency of a single request, but at scale, especially in the case when just few Netty threads are doing all the work such small Impact in 0,00001% will have %1 overall perf degradation for 100_000 queued requests which would have to be served by the same number of threads. Thus such unthoughtful change may impact someones production.

[chemicL]

The reasoning you give assumes that 1 thread is handling the queue. If you keep maintaining the queue filled at 100k pending requests constantly it means you're in much bigger trouble. To put things into perspective assume mean time to handle a request is 1ms. That means you'd wait 100s for a request to start being processed. That 0,00001% impact, or 1% overall impact is honestly meaningless to you at that point. Your request is already waiting for 1 minute 40 seconds. It doesn't matter you'd wait an additional second, having the end result being 1 minute 41 seconds.

Putting that discussion aside, as you suggest, I will see whether inlining kicks in, as I'd expect it should with such short methods, but if this is a tricky situation to the JIT, we'll see that in JMH tests, which I'm also going to perform.

[OlegDokuka]

one main suggestion for now is to eliminate all super.scanUnsafe(key) and replace it with explicit if (key == InternalProducerAttr.INSTANCE) return true;.

[OlegDokuka]

LG overall with minor comments

[chemicL]

one main suggestion for now is to eliminate all super.scanUnsafe(key) and replace it with explicit if (key == InternalProducerAttr.INSTANCE) return true;.

I'd suggest the opposite. Removing duplicates in children if the parent does some common job.

[OlegDokuka]

one main suggestion for now is to eliminate all super.scanUnsafe(key) and replace it with explicit if (key == InternalProducerAttr.INSTANCE) return true;.

I'd suggest the opposite. Removing duplicates in children if the parent does some common job.

Calling super should increase calls depth hence decrease inlining probability. I would do JMH perf test on that to see diff.

Anyways, this should not stop us on merging that effort. I suggest revising perf aspect separately. FYI, in reactor we prefer duplication and copying over DRY rule where possible. That may sound ridiculous but it worth it for hot path which this scan operation becomes with this PR