Reactor-core: Access to the current scheduler

[vy]

It is not possible to get a handle to the currently running scheduler. Though there are certain occasions where such a thing would be desired. Consider the following example:

Flux.range(1, 2, 3)
    .flatMap(id -> {
        return Mono
            .fromCallable(() -> repository.findById(id))
            .subscribeOn(repositoryScheduler);
    })
    .map(record -> process(record))
    .subscribe(System.out::println);

Given, repository.findById() is blocking, it makes perfect sense to confine these calls to a dedicated thread pool, which is repositoryScheduler in this example. That being said, you would not want to occupy repositoryScheduler for consecutive process(record) calls. There you better switch to another threading context, preferably to the call initiating one. Though that turns out to be not possible:

Flux.range(1, 2, 3)
    .flatMap(id -> {
        @Nullable Scheduler currentScheduler = /* What to type here? */;
                Scheduler effectiveScheduler = currentScheduler == null
                        ? /* some other scheduler */
                        : currentScheduler;
        return Mono
            .fromCallable(() -> repository.findById(id))
            .subscribeOn(repositoryScheduler)
            .publishOn(effectiveScheduler);
    })
    .map(record -> process(record))
    .subscribe(System.out::println);

Hence I ask for a simple aid like Schedulers.current() to use in cases similar to the above one. What if there are no schedulers involved? Schedulers.current() can just return null.

[akarnokd]

Have a specific single-threaded theadpool via Schedulers.fromExecutorService and you can return to it.

Schedulers.current() can just return null

publishOn would then fail with NullPointerException so this approach doesn't work.

[vy]

Have a specific single-threaded theadpool via Schedulers.fromExecutorService and you can return to it.

But how would flatMap block know about the caller?

publishOn would then fail with NullPointerException so this approach doesn't work.

No, this approach does work, though might not be the best way to implement it. My snippet was just an example. If you are smart enough to use schedulers, you would better be smart enough to check a @Nullable method return value. But since you have a point about NPE, I will fix it in the snippet.

[akarnokd]

Why would it need to know about it? You can use backpressure and limited concurrency with a dedicated scheduler to achieve sequentiality.

[vy]

I am lost. Would you mind sharing a concrete example where the caller and callee of the flatMap block have no assumptions about each other?

[akarnokd]

Scheduler shared = Schedulers.fromExecutorService(Executors.newSingleThreadedExecutor());

Flux.range(1, 2, 3)
        .subscribeOn(shared)
    .flatMap(id -> {
        return Mono
            .fromCallable(() -> repository.findById(id))
            .subscribeOn(repositoryScheduler)
    })
        .publishOn(shared)
    .map(record -> process(record))
    .subscribe(System.out::println);

[OlegDokuka]

It impossible to unify that approach because of Java limitations. Thread class is plain enough and does not support scheduling task on it. As an example - Main Thread of any standard Java Application. Once it has left there is no way to schedule something on it again.

Anyway. David's example depict the way of how it might be done correctly.

[vy]

Thanks so much for your prompt replies @akarnokd. Though I still think I am not able to communicate my intent properly. In your example, subscriber has the following two assumptions:

1. The source is single-threaded.
2. flatMap block uses a separate scheduler.

These are the assumptions I strive to avoid making and besides not even able to do in my case. Let me give a more detailed example:

private Flux<Integer> read() { /* ... */ }

private Mono<Integer> enrich(Integer id) {
    return Mono
            .fromCallable(() -> repository.findById(id))
            .subscribeOn(repositoryScheduler);
}

private Mono<Integer> process(Integer id) { /* ... */ }

public void main() {
    read()
            .flatMap(this::enrich)
            .map(this::process)
            .subscribe(System.out::println);
}

Here I want to wire read(), enrich(), and process() such that main() needs to have zero knowledge of what kind of scheduler or threading context the callees entail. I cannot touch read(), process() and I do not have any knowledge of their internals. Reflecting this back to your recommendations: main() cannot wrap enrich() with a single-threaded scheduler, because it does not know whether read() is single-threaded or not. Even if it did, I still believe it is enrich()'s responsibility to manage threading context due to a particular preference of its own design. That is why I strive to confine the solution within enrich() body, not anywhere else.

I hope this makes my case more clear -- unless I totally get you wrong.

[vy]

@OlegDokuka, I do not agree. You are right that I cannot (re)use a Thread, though I don't want to either. I just want to have a handle to the Scheduler. Maybe (probably?) there is a better way to implement this, but something I have in mind: In schedulers, having a ThreadLocal pointing to the Scheduler instance. That being said, I did not work out this idea in detail. Maybe there is another technical obstacle.

[akarnokd]

read could be a blocking generator in which case returning to its Scheduler will hang your flow. Also flows are sequential in nature and you can route them onto different threads with publishOn. Reactor is a composition library which also means you don't have to make too many assumptions about your sources or consumers beyond that they conform to the Reactive Streams protocols. If you need to ensure more than that, such as running steps on particular threads, use publishOn. When in doubt, use publishOn.

[vy]

read could be a blocking generator in which case returning to its Scheduler will hang your flow.

Right. But isn't this always the case when you're playing around with threading contexts? You can still shoot yourself in the foot without using Schedulers.current() I am requesting.

Also flows are sequential in nature and you can route them onto different threads with publishOn. Reactor is a composition library which also means you don't have to make too many assumptions about your sources or consumers beyond that they conform to the Reactive Streams protocols. If you need to ensure more than that, such as running steps on particular threads, use publishOn. When in doubt, use publishOn.

Agreed. I also want to use publishOn, but I don't know which scheduler to pick.

@akarnokd, I really appreciated your kind answers and I don't want to waste the precious time of developers that could have been invested into the project. My request is really simple: I need an implementation for Schedulers.current(). Yes, there are many caveats around using that, but so does the rest of the rx toolbox. Is it technically possible to provide Schedulers.current() in Reactor? If so, in this ticket I am requesting for that.

[smaldini]

It might be possible to do that using a property in a Thread extension (I need to check about reference leaking tho)