Coroutine execution always performed with dispatch if the dispatcher is bound to a DuplicatedContext

[tsegismont]

Regression introduced by #234 and reported by @bbyk in https://github.com/vert-x3/vertx-lang-kotlin/issues/234#issuecomment-1756706247

[tsegismont]

Fixed by #244

[bbyk]

Thanks @tsegismont ! Hopefully the last thing along these lines: I think you might be missing the following override in the new class ContextCoroutineDispatcher.

  override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle {
    return (delegate as Delay).invokeOnTimeout(timeMillis, block, context)
  }

With the override the timeout functionality will run with io.vertx.kotlin.coroutines.VertxCoroutineExecutor#schedule(java.lang.Runnable, long, java.util.concurrent.TimeUnit)

While without the override, it will rely on DefaultDelay.invokeOnTimeout(timeMillis, block, context) which is a different executor which gets involved.

For example, with the override and the code like this

      try {
        withTimeout(100) {
          delay(1000)
        }
      } catch (e: TimeoutCancellationException) {

you will get

while without the override it's

The issue doesn't change correctness, to be sure, but users might expect the functionality to stay exclusively with the Vert.x threads

[tsegismont]

@bbyk Thanks for you thorough review

I've tried to reproduce but the test passes:

@Test
  fun `test withTimeout execution dispatched on Vertx coroutine executor`(testContext: TestContext) {
    val latch = testContext.async()
    val context = (vertx as VertxInternal).getOrCreateContext()
    val duplicatedContext = context.duplicate()
    val promise = Promise.promise<Any>()
    duplicatedContext.runOnContext {
      GlobalScope.launch(vertx.dispatcher()) {
        try {
          withTimeout(100) {
            promise.future().await()
          }
        } catch (e: TimeoutCancellationException) {
          testContext.assertEquals(ContextInternal.current(), duplicatedContext)
          latch.complete()
        }
      }
    }
  }

Can you share your reproducer?

[bbyk]

@tsegismont I am sharing it below for a different concurrency primitive. But first I'd like to mention again that the missing override doesn't change correctness of withTimeout, it only leads to spawning a non-Vert.x thread for running timeouts which, I assume, is less desirable. You can put a breakpoint in kotlinx.coroutines.TimeoutCoroutine#run to see what I see on the screenshot.

You can have a more contrived case, though, that actually fails in a test. E.g. the following test fails ( a kludge: when you paste the sketched code, please remove private access modifier from VertxCoroutineExecutor for the sake of the test run - I didn't feel like re-implementing it in-place ). If you un-comment the code in invokeOnTimeout the test will start to pass.

  @InternalCoroutinesApi
  @Test
  fun `test select-onTimeout execution dispatched on Vertx coroutine executor`(testContext: TestContext) {
    val latch = testContext.async()
    val context = (vertx as VertxInternal).getOrCreateContext()
    val duplicatedContext = context.duplicate()
    val invokeTimeoutOnDuplicatedContextDispatcher = object : CoroutineDispatcher(), Delay {
      override fun dispatch(context: CoroutineContext, block: Runnable) {
        testContext.fail()
      }

      override fun isDispatchNeeded(context: CoroutineContext): Boolean {
        return false
      }

      override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
        testContext.fail()
      }

      override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle {
        return super.invokeOnTimeout(timeMillis, block, context)
//        val delay = VertxCoroutineExecutor(duplicatedContext).asCoroutineDispatcher() as Delay
//        return delay.invokeOnTimeout(timeMillis, block, context)
      }
    }
    val promise = Promise.promise<Any>()
    duplicatedContext.runOnContext {
      GlobalScope.launch(vertx.dispatcher()) {
        val neverEndingJob = launch {
          promise.future().await()
        }
        withContext(invokeTimeoutOnDuplicatedContextDispatcher) {
          // still on Vert.x
          testContext.assertEquals(duplicatedContext, ContextInternal.current())
          select<Unit> {
            neverEndingJob.onJoin {
            }
            onTimeout(100) {
              testContext.assertEquals(duplicatedContext, ContextInternal.current())
              neverEndingJob.cancel()
              latch.complete()
            }
          }
        }
      }
    }
  }

[tsegismont]

I'd like to mention again that the missing override doesn't change correctness of withTimeout, it only leads to spawning a non-Vert.x thread for running timeouts which, I assume, is less desirable.

Got you. This will be fixed by https://github.com/vert-x3/vertx-lang-kotlin/pull/246