From 0eb658654c39db3daa9b8ba15c53be5e8e78eeae Mon Sep 17 00:00:00 2001 From: Eldar Abusalimov Date: Thu, 13 Sep 2018 00:33:00 +0300 Subject: [PATCH] AsyncExecutionSupport: Consider expired constraint satisfied This makes the whole chain of constraints to be checked and rescheduled if necessary. Otherwise, an expired constraint would ruin dispatching of the rest constraints by giving up to using the fallback dispatcher. Also make the tests work solely on the EDT. This fixes them being flaky due to a race between the EDT and a pooled thread used for running test coroutines. Follow-up: IDEA-CR-36112 AsyncExecutionSupport: Adapt to launch/async/withContext API, add tests --- .../application/impl/AsyncExecutionSupport.kt | 66 +++++++--------- .../application/impl/AppUIExecutorTest.kt | 78 ++++++++++++------- 2 files changed, 79 insertions(+), 65 deletions(-) diff --git a/platform/platform-impl/src/com/intellij/openapi/application/impl/AsyncExecutionSupport.kt b/platform/platform-impl/src/com/intellij/openapi/application/impl/AsyncExecutionSupport.kt index a01339eeb473..548c78b0b54f 100644 --- a/platform/platform-impl/src/com/intellij/openapi/application/impl/AsyncExecutionSupport.kt +++ b/platform/platform-impl/src/com/intellij/openapi/application/impl/AsyncExecutionSupport.kt @@ -101,33 +101,25 @@ internal abstract class AsyncExecutionSupport> : AsyncExec private fun dispatchChain(context: CoroutineContext, block: Runnable) { for (dispatcher in myChain) { if (dispatcher.isScheduleNeeded(context)) { - dispatcher.doSchedule(context) { isRetryDispatchNeeded -> - if (isRetryDispatchNeeded) { - LOG.assertTrue(!dispatcher.isScheduleNeeded(context), this) - this.retryDispatch(context, block, causeDispatcher = dispatcher) - } - else { - block.run() - } - } - return + return dispatcher.doSchedule(context, Runnable { + LOG.assertTrue(!dispatcher.isScheduleNeeded(context), this) + this.retryDispatch(context, block, causeDispatcher = dispatcher) + }) } } block.run() } - protected open fun retryDispatch(context: CoroutineContext, - block: Runnable, + protected open fun retryDispatch(context: CoroutineContext, block: Runnable, causeDispatcher: ChainedDispatcher) = this.dispatch(context, block) protected abstract fun isScheduleNeeded(context: CoroutineContext): Boolean - protected abstract fun doSchedule(context: CoroutineContext, - block: (isRetryDispatchNeeded: Boolean) -> Unit) + protected abstract fun doSchedule(context: CoroutineContext, retryDispatchRunnable: Runnable) } /** A DelegateDispatcher backed by a ContextConstraint. */ internal abstract class ChainedConstraintDispatcher(delegate: CoroutineDispatcher, - protected val constraint: ContextConstraint) : ChainedDispatcher(delegate) { + protected open val constraint: ContextConstraint) : ChainedDispatcher(delegate) { override fun isScheduleNeeded(context: CoroutineContext): Boolean = !constraint.isCorrectContext override fun toString() = constraint.toString() } @@ -135,19 +127,15 @@ internal abstract class AsyncExecutionSupport> : AsyncExec /** @see SimpleContextConstraint */ internal class SimpleConstraintDispatcher(delegate: CoroutineDispatcher, constraint: SimpleContextConstraint) : ChainedConstraintDispatcher(delegate, constraint) { - - - override fun doSchedule(context: CoroutineContext, - block: (isRetryDispatchNeeded: Boolean) -> Unit) { - (constraint as SimpleContextConstraint).schedule(Runnable { - block(true) - }) + override fun doSchedule(context: CoroutineContext, retryDispatchRunnable: Runnable) { + (constraint as SimpleContextConstraint).schedule(retryDispatchRunnable) } } /** @see ExpirableContextConstraint */ internal class ExpirableConstraintDispatcher(delegate: CoroutineDispatcher, constraint: ExpirableContextConstraint) : ChainedConstraintDispatcher(delegate, constraint) { + override val constraint get() = super.constraint as ExpirableContextConstraint private val myInvokeOnDisposal = object : THashSet(ContainerUtil.identityStrategy()) { var isDisposed: Boolean = false @@ -174,28 +162,30 @@ internal abstract class AsyncExecutionSupport> : AsyncExec override fun initializeJob(job: Job) { super.initializeJob(job) - (constraint as ExpirableContextConstraint).expirable.cancelJobOnDisposal(job) { + constraint.expirable.cancelJobOnDisposal(job) { myInvokeOnDisposal.disposeAndGet().forEach(Runnable::run) } } - override fun isScheduleNeeded(context: CoroutineContext): Boolean = myInvokeOnDisposal.isDisposed || !constraint.isCorrectContext + override fun isScheduleNeeded(context: CoroutineContext): Boolean = !(myInvokeOnDisposal.isDisposed || constraint.isCorrectContext) - override fun doSchedule(context: CoroutineContext, - block: (isRetryDispatchNeeded: Boolean) -> Unit) { + override fun doSchedule(context: CoroutineContext, retryDispatchRunnable: Runnable) { val runOnce = RunOnce() - val fallbackRunnable = runOnce.runnable { + val unscheduledRunnable = runOnce.runnable { + LOG.assertTrue(myInvokeOnDisposal.isDisposed, "Must only be called on disposal of ${constraint.expirable}") LOG.assertTrue(context[Job]!!.isCancelled, "Job should have been cancelled through initializeJob()") - fallbackDispatch(context) { block(false) } + // Although it is tempting to invoke the retryDispatchRunnable directly, + // it is better to avoid executing arbitrary code from inside the disposal handler. + fallbackDispatch(context, retryDispatchRunnable) // invokeLater, basically } - if (!myInvokeOnDisposal.register(fallbackRunnable)) { - fallbackRunnable.run() + if (!myInvokeOnDisposal.register(unscheduledRunnable)) { + unscheduledRunnable.run() } else { - (constraint as ExpirableContextConstraint).scheduleExpirable(runOnce.runnable { - myInvokeOnDisposal.unregister(fallbackRunnable) - block(true) + constraint.scheduleExpirable(runOnce.runnable { + myInvokeOnDisposal.unregister(unscheduledRunnable) + retryDispatchRunnable.run() }) } } @@ -211,7 +201,7 @@ internal abstract class AsyncExecutionSupport> : AsyncExec override fun isScheduleNeeded(context: CoroutineContext): Boolean = false override fun doSchedule(context: CoroutineContext, - block: (isRetryDispatchNeeded: Boolean) -> Unit) = throw UnsupportedOperationException() + retryDispatchRunnable: Runnable) = throw UnsupportedOperationException() override fun dispatch(context: CoroutineContext, block: Runnable) { resetAttemptCount() @@ -230,7 +220,7 @@ internal abstract class AsyncExecutionSupport> : AsyncExec // The continuation block MUST be invoked at some point in order to give the coroutine a chance // to handle the cancellation exception and exit gracefully. // At this point we can only provide a guarantee to resume it on EDT with a proper modality state. - fallbackDispatch(context) { block.run() } + fallbackDispatch(context, block) } } @@ -287,17 +277,17 @@ internal abstract class AsyncExecutionSupport> : AsyncExec } } - internal fun CoroutineDispatcher.fallbackDispatch(context: CoroutineContext, block: () -> Unit) { + internal fun CoroutineDispatcher.fallbackDispatch(context: CoroutineContext, block: Runnable) { val primaryDispatcher = context[ContinuationInterceptor] as? CoroutineDispatcher val fallbackDispatcher = primaryDispatcher?.chainFallbackDispatcher ?: this.chainFallbackDispatcher // fallbackDispatcher.dispatchIfNeededOrInvoke(context, block) if (fallbackDispatcher !== Unconfined) { // Invoke later unconditionally to avoid running arbitrary code from inside dispose() handler. - fallbackDispatcher.dispatch(context, Runnable { block() }) + fallbackDispatcher.dispatch(context, block) } else { - block() + block.run() } } diff --git a/platform/platform-tests/testSrc/com/intellij/openapi/application/impl/AppUIExecutorTest.kt b/platform/platform-tests/testSrc/com/intellij/openapi/application/impl/AppUIExecutorTest.kt index 0bc96701caac..fda89cf01617 100644 --- a/platform/platform-tests/testSrc/com/intellij/openapi/application/impl/AppUIExecutorTest.kt +++ b/platform/platform-tests/testSrc/com/intellij/openapi/application/impl/AppUIExecutorTest.kt @@ -16,6 +16,7 @@ import kotlinx.coroutines.experimental.* import kotlinx.coroutines.experimental.channels.Channel import java.util.concurrent.LinkedBlockingQueue import javax.swing.SwingUtilities +import kotlin.coroutines.experimental.CoroutineContext /** * @author eldar @@ -28,8 +29,12 @@ class AppUIExecutorTest : LightPlatformTestCase() { UIUtil.dispatchAllInvocationEvents() } + private object SwingDispatcher : CoroutineDispatcher() { + override fun dispatch(context: CoroutineContext, block: Runnable) = SwingUtilities.invokeLater(block) + } + private fun Deferred.joinNonBlocking() { - var countDown = 300 + var countDown = 5 object : Runnable { override fun run() { if (isCompleted) { @@ -59,7 +64,7 @@ class AppUIExecutorTest : LightPlatformTestCase() { val executor = AppUIExecutor.onUiThread(ModalityState.any()) .expireWith(disposable) - async(Unconfined) { + async(SwingDispatcher) { queue.add("start") launch((executor as AppUIExecutorEx).createJobContext(coroutineContext)) { @@ -106,7 +111,7 @@ class AppUIExecutorTest : LightPlatformTestCase() { .inTransaction(getProject()) .inWriteAction() - async(Unconfined) { + async(SwingDispatcher) { val pdm = PsiDocumentManager.getInstance(getProject()) val commitChannel = Channel() val job = launch((executor as AppUIExecutorEx).createJobContext(coroutineContext)) { @@ -197,7 +202,7 @@ class AppUIExecutorTest : LightPlatformTestCase() { queue.add("start") - async(Unconfined) { + async(SwingDispatcher) { launch(executor.createJobContext()) { queue.add("coroutine start") assertTrue(scheduled) @@ -266,7 +271,7 @@ class AppUIExecutorTest : LightPlatformTestCase() { override fun toString() = "test" }) - async(Unconfined) { + async(SwingDispatcher) { queue.add("start") val channel = Channel() @@ -321,9 +326,16 @@ class AppUIExecutorTest : LightPlatformTestCase() { fun `test custom expirable constraints disposed during dispatching`() { val queue = LinkedBlockingQueue() - var scheduled = false + var outerScheduled = false + var innerScheduled = false var shouldDisposeOnDispatch = false + fun emit(s: String) { + fun Boolean.toFlagString(name: String) = if (this) name else "!$name" + val contextState = "context: ${outerScheduled.toFlagString("outer")} + ${innerScheduled.toFlagString("inner")}" + queue += "[$contextState] $s" + } + val disposable = object : Disposable.Parent { override fun beforeTreeDispose() { queue.add("disposable.beforeTreeDispose()") @@ -340,7 +352,7 @@ class AppUIExecutorTest : LightPlatformTestCase() { override val expirable = disposable override val isCorrectContext: Boolean - get() = scheduled + get() = outerScheduled override fun scheduleExpirable(runnable: Runnable) { if (shouldDisposeOnDispatch) { @@ -351,36 +363,48 @@ class AppUIExecutorTest : LightPlatformTestCase() { queue.add("refuse to run already disposed") return } - scheduled = true + outerScheduled = true runnable.run() - scheduled = false + outerScheduled = false } - override fun toString() = "test" + override fun toString() = "test outer" + }) + .withConstraint(object : AsyncExecution.SimpleContextConstraint { + override val isCorrectContext: Boolean + get() = innerScheduled + + override fun schedule(runnable: Runnable) { + innerScheduled = true + runnable.run() + innerScheduled = false + } + + override fun toString() = "test inner" }) - async(Unconfined) { - queue.add("start") + async(SwingDispatcher) { + emit("start") val channel = Channel() val job = launch(executor.createJobContext(coroutineContext)) { - queue.add("coroutine start") - assertTrue(scheduled) + emit("coroutine start") + assertTrue(outerScheduled) channel.receive() - assertTrue(scheduled) + assertTrue(outerScheduled) try { shouldDisposeOnDispatch = true - queue.add("before receive") + emit("before receive") channel.receive() - queue.add("after receive disposed") + emit("after receive disposed") yield() - queue.add("after yield disposed") + emit("after yield disposed") } catch (e: Throwable) { - queue.add("coroutine yield caught ${e.javaClass.simpleName} because of ${e.cause?.javaClass?.simpleName}") + emit("coroutine yield caught ${e.javaClass.simpleName} because of ${e.cause?.javaClass?.simpleName}") throw e } - queue.add("coroutine end") + emit("coroutine end") channel.close() } @@ -395,19 +419,19 @@ class AppUIExecutorTest : LightPlatformTestCase() { } } job.join() - queue.add("end") + emit("end") assertOrderedEquals(queue, - "start", - "coroutine start", - "before receive", + "[context: !outer + !inner] start", + "[context: outer + inner] coroutine start", + "[context: outer + inner] before receive", "disposing", "disposable.beforeTreeDispose()", "disposable.dispose()", "refuse to run already disposed", - "after receive disposed", // channel.receive() is atomic - "coroutine yield caught JobCancellationException because of DisposedException", - "end") + "[context: !outer + inner] after receive disposed", // channel.receive() is atomic + "[context: !outer + inner] coroutine yield caught JobCancellationException because of DisposedException", + "[context: !outer + !inner] end") }.joinNonBlocking() } } \ No newline at end of file