Reactor-core: BlockHound detects blocking call on cancel Timeout task

[robotmrv]

BlockHound sometimes detects blocking call on cancel timeout task.

Expected Behavior

BlockHound should not detect blocking calls in reactor-core

Actual Behavior

BlockHound detects blocking calls in reactor-core

Steps to Reproduce

Here is code which could reproduce issue

@BeforeAll
static void setUp() {
    BlockHound.install();
}

@Test
void cancelTimeoutBlocks() {
    CopyOnWriteArrayList<Throwable> errors = new CopyOnWriteArrayList<>();
    Schedulers.onHandleError((thread, throwable) -> {
        errors.add(throwable);
    });
    Scheduler sc = Schedulers.newParallel("my-sc", 10, true);
    Flux<Long> test = Flux.interval(Duration.ZERO, Duration.ofMillis(5))
            .flatMap(it -> {
                return Mono.just(it).hide()
                        .timeout(Duration.ofMillis(10_000))
                        .subscribeOn(sc);
            }, 20)
            .subscribeOn(Schedulers.parallel())
            .take(Duration.ofMillis(5_000));
    test.blockLast();
    Assertions.assertThat(errors).isEmpty();
}

Run it and check for logs

13:36:58.222 [my-sc-12] ERROR reactor.core.scheduler.Schedulers - Scheduler worker in group main failed with an uncaught exception
java.lang.Error: Blocking call! jdk.internal.misc.Unsafe#park
    at reactor.blockhound.BlockHound$Builder.lambda$new$0(BlockHound.java:196)
    at reactor.blockhound.BlockHound$Builder.lambda$install$6(BlockHound.java:318)
    at reactor.blockhound.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:46)
    at java.base/jdk.internal.misc.Unsafe.park(Unsafe.java)
    at java.base/java.util.concurrent.locks.LockSupport.park(LockSupport.java:194)
    at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:885)
    at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:917)
    at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1240)
    at java.base/java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:267)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.remove(ScheduledThreadPoolExecutor.java:1046)
    at java.base/java.util.concurrent.ThreadPoolExecutor.remove(ThreadPoolExecutor.java:1751)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.cancel(ScheduledThreadPoolExecutor.java:293)
    at reactor.core.scheduler.SchedulerTask.dispose(SchedulerTask.java:126)
    at reactor.core.publisher.MonoDelay$MonoDelayRunnable.cancel(MonoDelay.java:135)
    at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.drainLoop(Operators.java:2051)
    at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.drain(Operators.java:2020)
    at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.cancel(Operators.java:1832)
    at reactor.core.publisher.Operators.terminate(Operators.java:1108)
    at reactor.core.publisher.StrictSubscriber.cancel(StrictSubscriber.java:155)
    at reactor.core.publisher.FluxTimeout$TimeoutTimeoutSubscriber.cancel(FluxTimeout.java:409)
    at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.onNext(FluxTimeout.java:159)
    at reactor.core.publisher.FluxHide$HideSubscriber.onNext(FluxHide.java:74)
    at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2186)
    at reactor.core.publisher.FluxHide$HideSubscriber.request(FluxHide.java:58)
    at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1994)
    at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.onSubscribe(FluxTimeout.java:148)
    at reactor.core.publisher.FluxHide$HideSubscriber.onSubscribe(FluxHide.java:69)
    at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
    at reactor.core.publisher.Mono.subscribe(Mono.java:4105)
    at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:124)
    at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
    at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)

Possible Solution

Probably extend allowed blocking methods or document it

Your Environment

• Reactor version(s) used:
  reactor-core:3.3.1.RELEASE,
  blockhound:1.0.1.RELEASE
• Other relevant libraries versions (eg. netty, ...):
• JVM version (javar -version):
  openjdk version "11" 2018-09-25
  OpenJDK Runtime Environment 18.9 (build 11+28)
  OpenJDK 64-Bit Server VM 18.9 (build 11+28, mixed mode)
• OS and version (eg uname -a):
  Win10

[bsideup]

Hi @robotmrv,

The code you shared does not fail for me. Do you see it failing all the time or occasionally?

[robotmrv]

Hi @bsideup ,
Initial Test "passes". But There are several logs like this.
It reproduces all the time on my PC.
But before making it reproducible, I played around with sc properties and delays.
I've updated TC so it will fail test.

[robotmrv]

Updated dependencies to

reactor-core:3.3.2.RELEASE,
blockhound:1.0.2.RELEASE

and it reproduces

[bsideup]

@robotmrv I was unable to reproduce it with your code, but managed to find a more isolated reproducer :)

See #2046.

[robotmrv]

@bsideup I've checked with your changes. Now there is no Blocking Call reports

[bsideup]

@robotmrv thanks a lot for checking! 👍