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