Mono.fromCallable Threading Behaviour

Question

I am trying to understand the behaviour of blocking functions in Reactor, but something else has completely thrown me off my study. Here is the code:

  public static void main(String[] args) throws InterruptedException {
    Flux.range(1, 100_000)
        .doOnNext(a -> System.out.println(a + ", thread: " + Thread.currentThread().getName()))
        .flatMap(a -> Mono.fromCallable(() -> blockingMethod(a)).subscribeOn(Schedulers.elastic()))
        .subscribe();
    System.out.println("Here");
    Thread.sleep(Integer.MAX_VALUE);
  }

  private static int blockingMethod(int s) {
    try {
      Thread.sleep(100_000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return s;
  }

Here's a summary of what happens AFAIK:

Subscription happens on the main thread.
main becomes free inside the flatMap to bring the next element from upstream. Therefore, doOnNext should always print main.
After processing 100_000 elements, main would become free and print here.

Instead, this is what happens:

The first 256 elements are printed on main (in doOnNext) as expected.
After around 1 second, the next 256, then the next and so on. Elements from the second batch onwards are printed on elastic threads.

Here are my questions:

Why are elements being processed in batch of 256? Schedulers.elastic() should create threads on demand, so ideally there should always be a thread available to take the request from main (ignoring JVM restrictions on the number of threads that I can create).

Why are elements in the second 'batch' (and beyond) being printed on elastic threads? I expect them to be published on main. In fact, this is what happens when you remove the blocking call as

   public static void main(String[] args) throws InterruptedException {
       Flux.range(1, 100_000)
           .doOnNext(a -> System.out.println(a + ", thread: " + Thread.currentThread().getName()))
           .flatMap(a -> Mono.just(a).subscribeOn(Schedulers.elastic()))
           .subscribe();
      System.out.println("Here");
      Thread.sleep(Integer.MAX_VALUE);
  }

Here, all elements print main in doOnNext and here is printed only when the stream finishes (freeing the main thread).

Am I missing something?

By default, max concurrence of flatMap operator is 256. So it can merge only 256 upstream concurrently. — Mister_Jesus
And subscribeOn affects to all chain of operators (No matter where was the call - inside other nested operator or before subscribe). You can see that the first bundle of doOnNext callbacks was performed in main thread because subscribeOn was not called yet. If you will print values in subscriber, we will see that its all expected and all call will be in elastic. That why you shouldn't use doOn* operators with heavy operations. — Mister_Jesus
@Mister_Jesus thanks for the answer. Why does the example without the blocking call emit all items on the main thread then? (in doOnNext)? — Prashant Pandey
@Mister_Jesus not sure how it answers my question. In the example without the blocking call (the bottom-most chunk of code), all items are printed on the main thread. This means subscribeOn didn't change the source of the top-level stream right? — Prashant Pandey

Balajee Balajee · Accepted Answer · 2020-12-03T18:51:47

You need to understand that the entire reactor is built on the reactive streams spec. Every operator above is thus a publisher and subscriber combo.

In scenario 2

   public static void main(String[] args) throws InterruptedException {
       Flux.range(1, 100_000)
           .doOnNext(a -> System.out.println(a + ", thread: " + Thread.currentThread().getName()))
           .flatMap(a -> Mono.just(a).subscribeOn(Schedulers.elastic()))
           .subscribe();
      System.out.println("Here");
      Thread.sleep(Integer.MAX_VALUE);
  }

The publisher starts on the main thread (Flux.Range) the subscriber (doOnNext) is subscribing (invoking the onNext call on the main Thread) and it is delegated to an elastic thread which is consuming faster than the subscriber can publish. So the entire Flux.Range gets invoked on the main thread and gets distributed to the elastic thread pool to process

In scenario 1

  public static void main(String[] args) throws InterruptedException {
    Flux.range(1, 100_000)
        .doOnNext(a -> System.out.println(a + ", Publisher: " + Thread.currentThread().getName()))
        .flatMap(a -> Mono.fromCallable(() -> blockingMethod(a)).subscribeOn(Schedulers.elastic())).doOnNext(a -> System.out.println(a + ", Subscriber: " + Thread.currentThread().getName()))
        .subscribe();
    System.out.println("Here");
    Thread.sleep(Integer.MAX_VALUE);
  }

  private static int blockingMethod(int s) {
    try {
      Thread.sleep(100_000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return s;
  }

The first 256 calls of Flux.Range happen on the main thread. Please remember that reactive streams is a back pressure driven programming and since the flat map can only merge 256 upstreams, the flat map operator (which is a subscriber) does not invoke the onNext call after 256 events, since they are all blocked/waiting to complete.

The main thread now is freed up, since it finished it responsibility of taking part in the reactive pipeline and steps out of the conveyer belt. i.e basically goes on to execute the next line of code. Which is why you see "Here" printed after the first 256 elements. Then the main thread goes on to get blocked indefinitely as per the code.

Once one of the blocking threads complete the job, they would trigger the subscriber to invoke the onNext on the thread that completed the earlier request, which is the elastic thread. The elastic thread would now delegate the call to another elastic thread in the inner reactor pipeline.

Which means for the same item i emitted, the thread that prints "Publisher" & the thread that prints "Subscriber" is not the same, even though they are elastic thread. Feel free to verify the below code

Mono.fromCallable Threading Behaviour

2 Answers