Reactor-core: Ignoring the next context swiching with the same thread

[stavalfi]

First, we all need to agree on the following fact:

We use shcedulers with a single thread iff we want to ensure that there is no _race condition_ by ensuring a _mutual exclusion_.
If you disagree, then let's talk about that first before you keep reading because I may not fully understand the purpose of shcedulers with a single thread.

Something is missing

The synchorinize mechanizm in java allow us to obtain the same _synchronization object_ again if
we already obtained it earlier and we didn't free it between the last time we obtained it until the this time. I would like to see the same behavior when using shcedulers with a single thread.

Impact without the suggested mechanizm in reactor lib

We can't _lock_ multiple methods which are invoked by a stream - one by one.
I must implement the _lock_ mechanizm using a semaphore.

Steps to reproduce

Function<Integer, Mono<Integer>> function = realThreadId -> Mono.just(1)
        .publishOn(Schedulers.single())
        .doOnNext(__ -> {
            System.out.println("(1) the thread which is here is: " + realThreadId);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        })
        .publishOn(Schedulers.single())
        .doOnNext(__ -> System.out.println("(2) the thread which is here is: " + realThreadId));

Flux.just(1, 2)
        .flatMap(realThreadId ->
                Mono.just(realThreadId)
                        .subscribeOn(Schedulers.elastic())
                        .flatMap(function),2)
        .subscribe();

Thread.sleep(1000 * 1000);

Expected behavior

(1) the thread which is here is: 1
(1) the thread which is here is: 1
(2) the thread which is here is: 2
(2) the thread which is here is: 2

Actual behavior

(1) the thread which is here is: 1
(1) the thread which is here is: 2
(2) the thread which is here is: 1
(2) the thread which is here is: 2

Reactor Core version

3.2.0.BUILD-SNAPSHOT

JVM version (e.g. java -version)

1.8

[akarnokd]

Schedulers.single() uses a single thread to execute work, however, you used the Schedulers.elastic() which can have many threads to feed into the single-threaded scheduler. The actual behavior is the correct behavior. If you removed the second publishOn, you'd get the expected output.

This flow diagram should better explain what's happening:

[stavalfi]

Hi @akarnokd and thanks for your fast response.

Could you please explain the following:

you used the Schedulers.elastic() which can have many threads to feed into the single-threaded scheduler

Also, due to the fact that I didn't encounter any formal material about how schedulers works I would like to ask you to explain what the image, you provided, describers.

[simonbasle]

@stavalfi I think your original assumption that Schedulers.single() is a mechanism for mutex is wrong. It is more akin to an Executors.newSingleThreadExecutor() where you'd submit one task per value emitted by the upstream Flux. So the second publishOn is not re-entrant.

To explain the diagram, reading from the elastic1 and elastic2 blue boxes:

1. elastic1 publishes realThreadId = 1
2. single prints the first message for 1, and blocks (sleep)
3. elastic2 tries to publish realThreadId = 2 to single, but is blocked
4. at the end of the sleep, single doesn't continue with imperative code but uses a second publishOn. Since it is not re-entrant, that scheduling frees up the thread for the blocked elastic2
5. elastic2 finally manages to publish 2, which prints the first message for 2 and blocks again
6. at the end of that sleep, second publishOn of 2 is scheduled, but that frees up the thread to process the publishOn of 1 that had previously been scheduled
7. second message for 1 is printed
8. second message for 2 is printed

[stavalfi]

Thanks for the description @simonbasle .

Do you mean that we shouldn't use mutex at all? If yes, then I will have a blocking mechanism inside reactor operators which is something we all trying to avoid. If no, then how do I "implement" such a mechanism using reactor operators; is it depends on the particular situation or there is a general way to do it?

[simonbasle]

Mutex is by definition blocking: the excluded task is blocked until the task that has the mutex releases it. So it is a bad fit for a reactive toolkit.

However, if you can represent tasks as a Flux<TaskDescription> of some sort, you'll benefit from the fact that the reactive streams specification forbids concurrent onNext. That way, and by executing each task using a concatMap, you can obtain a guarantee that no 2 task will conflict (provided they are "submitted" in that original sequence).