BlockHound: ReentrantLock.lock in Channels

[fluidsonic]

I use send in a channelFlow which is collected on Dispatchers.Default.
That in turn calls jdk.internal.misc.Unsafe#park and BlockHound complains about that.

Are suspending Channel functions supposed to be used on the IO dispatcher or is this merely BlockHound overreacting?

reactor.blockhound.BlockingOperationError: Blocking call! jdk.internal.misc.Unsafe#park
    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 kotlinx.coroutines.channels.ArrayChannel.offerInternal(ArrayChannel.kt:313)
    at kotlinx.coroutines.channels.AbstractSendChannel.send(AbstractChannel.kt:135)
    at kotlinx.coroutines.channels.ChannelCoroutine.send$suspendImpl(ChannelCoroutine.kt)
    at kotlinx.coroutines.channels.ChannelCoroutine.send(ChannelCoroutine.kt)
    at eve.FlowKt$combineLatest$2$1$1$invokeSuspend$$inlined$forEachIndexed$lambda$1$1.invokeSuspend(Flow.kt:27)
    at eve.FlowKt$combineLatest$2$1$1$invokeSuspend$$inlined$forEachIndexed$lambda$1$1.invoke(Flow.kt)
    at kotlinx.coroutines.flow.FlowKt__MergeKt$mapLatest$1.invokeSuspend(Merge.kt:214)
    at kotlinx.coroutines.flow.FlowKt__MergeKt$mapLatest$1.invoke(Merge.kt)
    at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3$invokeSuspend$$inlined$collect$1$lambda$1.invokeSuspend(Merge.kt:34)
    at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3$invokeSuspend$$inlined$collect$1$lambda$1.invoke(Merge.kt)
    at kotlinx.coroutines.intrinsics.UndispatchedKt.startCoroutineUndispatched(Undispatched.kt:55)
    at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:111)
    at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:158)
    at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch(Builders.common.kt:56)
    at kotlinx.coroutines.BuildersKt.launch(Unknown Source)
    at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch$default(Builders.common.kt:49)
    at kotlinx.coroutines.BuildersKt.launch$default(Unknown Source)
    at kotlinx.coroutines.flow.internal.ChannelFlowTransformLatest$flowCollect$3$invokeSuspend$$inlined$collect$1.emit(Collect.kt:138)
    at eve.EveFlowKt$asEveFlow$1$$special$$inlined$map$1$2.emit(Collect.kt:135)
    at kotlinx.coroutines.flow.SharedFlowImpl.collect(SharedFlow.kt:318)
    at kotlinx.coroutines.flow.SharedFlowImpl$collect$1.invokeSuspend(SharedFlow.kt)
    at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
    at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
    at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:571)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:738)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:678)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:665)

reactor.blockhound.BlockingOperationError: Blocking call! jdk.internal.misc.Unsafe#park
	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 kotlinx.coroutines.channels.ArrayChannel.pollInternal(ArrayChannel.kt:317)
	at kotlinx.coroutines.channels.AbstractChannel.receiveOrClosed-ZYPwvRU(AbstractChannel.kt:629)
	at kotlinx.coroutines.channels.ChannelCoroutine.receiveOrClosed-ZYPwvRU$suspendImpl(ChannelCoroutine.kt)
	at kotlinx.coroutines.channels.ChannelCoroutine.receiveOrClosed-ZYPwvRU(ChannelCoroutine.kt)
	at kotlinx.coroutines.flow.FlowKt__ChannelsKt.emitAllImpl$FlowKt__ChannelsKt(Channels.kt:50)
	at kotlinx.coroutines.flow.FlowKt__ChannelsKt$emitAllImpl$1.invokeSuspend(Channels.kt)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:571)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:738)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:678)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:665)

reactor.blockhound.BlockingOperationError: Blocking call! jdk.internal.misc.Unsafe#park
	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 kotlinx.coroutines.channels.ArrayChannel.enqueueReceiveInternal(ArrayChannel.kt:319)
	at kotlinx.coroutines.channels.AbstractChannel.enqueueReceive(AbstractChannel.kt:604)
	at kotlinx.coroutines.channels.AbstractChannel.access$enqueueReceive(AbstractChannel.kt:496)
	at kotlinx.coroutines.channels.AbstractChannel.receiveSuspend(AbstractChannel.kt:583)
	at kotlinx.coroutines.channels.AbstractChannel.receiveOrClosed-ZYPwvRU(AbstractChannel.kt:632)
	at kotlinx.coroutines.channels.ChannelCoroutine.receiveOrClosed-ZYPwvRU$suspendImpl(ChannelCoroutine.kt)
	at kotlinx.coroutines.channels.ChannelCoroutine.receiveOrClosed-ZYPwvRU(ChannelCoroutine.kt)
	at kotlinx.coroutines.flow.FlowKt__ChannelsKt.emitAllImpl$FlowKt__ChannelsKt(Channels.kt:50)
	at kotlinx.coroutines.flow.FlowKt__ChannelsKt$emitAllImpl$1.invokeSuspend(Channels.kt)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:571)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:738)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:678)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:665)

[dkhalanskyjb]

Thank you! Yes, this is a false positive. Using channels in dispatchers other than IO is okay, locking there is just an implementation detail and doesn't impact performance. Channels (and some other stuff) are added to the list of allowed blocking calls via #2331.