subscribeOn(Schedulers.parallel()) doesn't execute in parallel when located right after a just()

[antonio-marrero]

I've noticed a different behaviour when you place a subscribeOn straigh after a just from when you put another operator in-between.
For example:

CountDownLatch latch = new CountDownLatch(1);

Flux<Integer> flux = Flux.range(0, 30);
flux.flatMap(count -> Flux.just(count)
     .subscribeOn(Schedulers.parallel())
     .map(i -> highCpuProcess(i)))
     .doAfterTerminate(latch::countDown)
     .subscribe( m -> System.out.println("Subscriber received - " + m + " on thread: " + Thread.currentThread().getName()));

latch.await();

This always creates four threads but only uses one and never runs the processes in parallel.

On the contrary, when you put an operator in the middle the four threads are used, I use log() in my example but it can be any operator:

CountDownLatch latch = new CountDownLatch(1);

Flux<Integer> flux = Flux.range(0, 30);
flux.flatMap(count -> Flux.just(count)
     .log()
     .subscribeOn(Schedulers.parallel())
     .map(i -> highCpuProcess(i)))
     .doAfterTerminate(latch::countDown)
     .subscribe( m -> System.out.println("Subscriber received - " + m + " on thread: " + Thread.currentThread().getName()));

latch.await();

In this last case, the four threads are used:

[antonio-marrero]

NOTE: Using Mono instead of Flux works fine

CountDownLatch latch = new CountDownLatch(1);
		
Flux<Integer> flux = Flux.range(0, 30);
flux.flatMap(count -> Mono.just(count)
		.subscribeOn(Schedulers.parallel())
		.map(i -> highCpuProcess(i)))
.doAfterTerminate(latch::countDown)
.subscribe( m -> System.out.println("Subscriber received - " + m + " on thread: " +   Thread.currentThread().getName()));
latch.await();

[akarnokd]

This is a cross-boundary fusion problem with flatMap and an async-fuseable source that schedules emissions internally, like the fused just+subscribeOn and observeOn in general.

I don't know yet how would be possible to keep such fusion, therefore, I suggest disabling fusion on these classes:

• FluxSubscribeOnValue
• FluxPublishOn

You can simply return NONE or strip the classes from QueueSubscription

[smaldini]

Yes for FluxSubscribeOnValue.
I wonder if we can keep the fusion for FluxPublishOn if its not backfused tho or parent is not fuseable ?