From 92bef67ddda8ea67d7bbb1cc07a676cf5b8e7aeb Mon Sep 17 00:00:00 2001 From: "Ilia.Shulgin" Date: Fri, 17 Oct 2025 16:19:06 +0200 Subject: [PATCH] [vcs] Fix race in SingleTaskRunner GitOrigin-RevId: df99f8eef9fa5a697549817cdcc75154842a2a41 --- .../CommitChangesViewWithToolbarPanel.kt | 17 ++++--- .../vcs/impl/shared/SingleTaskRunner.kt | 47 +++++++---------- .../intellij/vcsUtil/SingleTaskRunnerTest.kt | 50 ++++--------------- .../git4idea/repo/GitUntrackedFilesHolder.kt | 2 +- 4 files changed, 39 insertions(+), 77 deletions(-) diff --git a/platform/vcs-impl/shared/src/com/intellij/openapi/vcs/changes/CommitChangesViewWithToolbarPanel.kt b/platform/vcs-impl/shared/src/com/intellij/openapi/vcs/changes/CommitChangesViewWithToolbarPanel.kt index 172865016025..4213b1bf9b2a 100644 --- a/platform/vcs-impl/shared/src/com/intellij/openapi/vcs/changes/CommitChangesViewWithToolbarPanel.kt +++ b/platform/vcs-impl/shared/src/com/intellij/openapi/vcs/changes/CommitChangesViewWithToolbarPanel.kt @@ -34,6 +34,8 @@ import org.jetbrains.annotations.CalledInAny import java.lang.Runnable import kotlin.time.Duration.Companion.milliseconds +private val REFRESH_DELAY = 100.milliseconds + @ApiStatus.Internal class CommitChangesViewWithToolbarPanel( changesView: ChangesListView, @@ -42,7 +44,7 @@ class CommitChangesViewWithToolbarPanel( val project: Project get() = changesView.project private val settings get() = ChangesViewSettings.getInstance(project) - private val refresher = SingleTaskRunner(cs, 100.milliseconds) { + private val refresher = SingleTaskRunner(cs) { refreshView() } @@ -57,8 +59,8 @@ class CommitChangesViewWithToolbarPanel( init { refresher.start() cs.launch(Dispatchers.UI) { - refresher.getPendingTasksFlow().collect { - isBusy -> changesView.setPaintBusy(isBusy) + refresher.getIdleFlow().collect { idle -> + changesView.setPaintBusy(!idle) } } } @@ -119,13 +121,14 @@ class CommitChangesViewWithToolbarPanel( @CalledInAny private fun scheduleRefresh(withDelay: Boolean, @RequiresBackgroundThread callback: Runnable? = null) { - if (withDelay) { + if (!withDelay && callback == null) { refresher.request() + return } - else { - refresher.requestNow() - } + cs.launch { + if (withDelay) delay(REFRESH_DELAY) + refresher.request() refresher.awaitNotBusy() callback?.run() } diff --git a/platform/vcs-impl/shared/src/com/intellij/platform/vcs/impl/shared/SingleTaskRunner.kt b/platform/vcs-impl/shared/src/com/intellij/platform/vcs/impl/shared/SingleTaskRunner.kt index 057bb8296fe3..326a614c45ce 100644 --- a/platform/vcs-impl/shared/src/com/intellij/platform/vcs/impl/shared/SingleTaskRunner.kt +++ b/platform/vcs-impl/shared/src/com/intellij/platform/vcs/impl/shared/SingleTaskRunner.kt @@ -6,11 +6,7 @@ import com.intellij.openapi.diagnostic.getOrHandleException import com.intellij.openapi.diagnostic.logger import com.intellij.openapi.progress.checkCanceled import kotlinx.coroutines.* -import kotlinx.coroutines.channels.Channel -import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.MutableStateFlow -import kotlinx.coroutines.flow.combine -import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.* import org.jetbrains.annotations.ApiStatus import kotlin.time.Duration @@ -22,49 +18,39 @@ import kotlin.time.Duration @ApiStatus.Internal class SingleTaskRunner( cs: CoroutineScope, - private val delay: Duration = Duration.Companion.ZERO, private val task: suspend () -> Unit, ) { - private val requested = MutableStateFlow(false) - private val busy = MutableStateFlow(false) + /** + * Represents the state of the runner: + * 0 - idle, + * 1 - a single task is either queued or running, + * 2 - has one running task and one queued task. + */ + private val requestCounter = MutableStateFlow(0) - private val runNow = Channel(capacity = Channel.CONFLATED) - - fun getPendingTasksFlow(): Flow = requested.combine(busy) { requested, busy -> !requested && !busy } + fun getIdleFlow(): Flow = requestCounter.map { it == 0 } private val processorJob = cs.launch(Dispatchers.Default, CoroutineStart.LAZY) { try { while (true) { checkCanceled() - requested.first { it } - if (delay.isPositive()) { - withTimeoutOrNull(delay) { runNow.receive() } - } - busy.value = true - requested.value = false + requestCounter.first { it > 0 } checkCanceled() runCatching { task() }.getOrHandleException { LOG.error("Task failed", it) } - busy.value = false + requestCounter.update { it.dec() } } } finally { - requested.value = false - busy.value = false + requestCounter.value = 0 LOG.debug { "Task processing finished" } } } fun request() { if (processorJob.isCancelled) return - requested.value = true - } - - fun requestNow() { - if (processorJob.isCancelled) return - runNow.trySend(Unit) - requested.value = true + requestCounter.update { it.inc().coerceAtMost(2) } } fun start() { @@ -76,10 +62,15 @@ class SingleTaskRunner( * Await the state where the task is not executed and there are no requests to do so. */ suspend fun awaitNotBusy() { - getPendingTasksFlow().first { it } + getIdleFlow().first { it } } companion object { private val LOG = logger() + + fun delayedTaskRunner(cs: CoroutineScope, delay: Duration, task: suspend () -> Unit): SingleTaskRunner = SingleTaskRunner(cs) { + delay(delay) + task() + } } } \ No newline at end of file diff --git a/platform/vcs-impl/testSrc/com/intellij/vcsUtil/SingleTaskRunnerTest.kt b/platform/vcs-impl/testSrc/com/intellij/vcsUtil/SingleTaskRunnerTest.kt index 4ccd0c69c7d0..543b38099827 100644 --- a/platform/vcs-impl/testSrc/com/intellij/vcsUtil/SingleTaskRunnerTest.kt +++ b/platform/vcs-impl/testSrc/com/intellij/vcsUtil/SingleTaskRunnerTest.kt @@ -5,18 +5,13 @@ import com.intellij.platform.util.coroutines.childScope import com.intellij.platform.vcs.impl.shared.SingleTaskRunner import com.intellij.testFramework.LoggedErrorProcessor import com.intellij.testFramework.common.timeoutRunBlocking -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.TimeoutCancellationException -import kotlinx.coroutines.cancel +import kotlinx.coroutines.* import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first -import kotlinx.coroutines.withTimeout import org.junit.jupiter.api.Assertions import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows -import kotlin.time.Duration -import kotlin.time.Duration.Companion internal class SingleTaskRunnerTest { @Test @@ -43,41 +38,13 @@ internal class SingleTaskRunnerTest { } } - @Test - fun `test instant execution`() = timeoutRunBlocking { - var counter = 0 - withRunner({ counter++ }, delay = Duration.INFINITE) { - start() - request() - requestNow() - awaitNotBusy() - assertEquals(1, counter) - } - } - - @Test - fun `test delayed execution after instant`() = timeoutRunBlocking { - var counter = 0 - withRunner({ counter++ }, delay = Duration.INFINITE) { - start() - request() - requestNow() - awaitNotBusy() - request() - assertThrows { - withTimeout(100) { - awaitNotBusy() - } - } - assertEquals(1, counter) - } - } - @Test fun `test execution debounced`() = timeoutRunBlocking { var counter = 0 + val taskStarted = CompletableDeferred() val runAllowed = MutableStateFlow(false) withRunner({ + taskStarted.complete(Unit) runAllowed.first { it } counter++ }) { @@ -85,11 +52,13 @@ internal class SingleTaskRunnerTest { repeat(10) { request() + taskStarted.await() } runAllowed.value = true awaitNotBusy() - assertEquals(1, counter) + assertEquals(2, counter) + counter = 0 runAllowed.value = false request() assertThrows { @@ -97,10 +66,10 @@ internal class SingleTaskRunnerTest { awaitNotBusy() } } - assertEquals(1, counter) + assertEquals(0, counter) runAllowed.value = true awaitNotBusy() - assertEquals(2, counter) + assertEquals(1, counter) } } @@ -131,11 +100,10 @@ internal class SingleTaskRunnerTest { private inline fun CoroutineScope.withRunner( noinline task: suspend () -> Unit, - delay: Duration = Companion.ZERO, consumer: SingleTaskRunner.() -> Unit, ) { val cs = childScope("BG runner") - SingleTaskRunner(cs, delay, task).also(consumer) + SingleTaskRunner(cs, task).also(consumer) cs.cancel() } } \ No newline at end of file diff --git a/plugins/git4idea/src/git4idea/repo/GitUntrackedFilesHolder.kt b/plugins/git4idea/src/git4idea/repo/GitUntrackedFilesHolder.kt index a6763757ca15..6f64c7997162 100644 --- a/plugins/git4idea/src/git4idea/repo/GitUntrackedFilesHolder.kt +++ b/plugins/git4idea/src/git4idea/repo/GitUntrackedFilesHolder.kt @@ -69,7 +69,7 @@ class GitUntrackedFilesHolder internal constructor( get() = untrackedFiles.initialized init { - updateRunner = SingleTaskRunner(cs, 500.milliseconds, ::update) + updateRunner = SingleTaskRunner.delayedTaskRunner(cs, 500.milliseconds, ::update) cs.launch(start = CoroutineStart.UNDISPATCHED) { try { project.serviceAsync().awaitInitialVfsRefreshFinished()