mirror of
https://gitflic.ru/project/openide/openide.git
synced 2026-09-27 10:03:11 +07:00
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
This commit is contained in:
+28
-38
@@ -101,33 +101,25 @@ internal abstract class AsyncExecutionSupport<E : AsyncExecution<E>> : 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<E : AsyncExecution<E>> : 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<Runnable>(ContainerUtil.identityStrategy()) {
|
||||
var isDisposed: Boolean = false
|
||||
@@ -174,28 +162,30 @@ internal abstract class AsyncExecutionSupport<E : AsyncExecution<E>> : 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<E : AsyncExecution<E>> : 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<E : AsyncExecution<E>> : 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<E : AsyncExecution<E>> : 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()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+51
-27
@@ -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<Unit>.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<Unit>()
|
||||
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<Unit>()
|
||||
@@ -321,9 +326,16 @@ class AppUIExecutorTest : LightPlatformTestCase() {
|
||||
fun `test custom expirable constraints disposed during dispatching`() {
|
||||
val queue = LinkedBlockingQueue<String>()
|
||||
|
||||
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<Unit>()
|
||||
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()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user