From 94bfaeb2e4c538d1173e4bef8a0d566e8e3fef70 Mon Sep 17 00:00:00 2001 From: saran Date: Fri, 7 Aug 2026 09:30:17 +0530 Subject: [PATCH] Fix a crash caused by renaming a live worker thread from another thread Fixes #2234 --- .../jvm/src/scheduling/CoroutineScheduler.kt | 20 +++++- .../jvm/test/scheduling/SchedulerTestBase.kt | 14 +--- .../jvm/test/scheduling/WorkerRenameTest.kt | 71 +++++++++++++++++++ 3 files changed, 92 insertions(+), 13 deletions(-) create mode 100644 kotlinx-coroutines-core/jvm/test/scheduling/WorkerRenameTest.kt diff --git a/kotlinx-coroutines-core/jvm/src/scheduling/CoroutineScheduler.kt b/kotlinx-coroutines-core/jvm/src/scheduling/CoroutineScheduler.kt index 1885579051..102c5a44eb 100644 --- a/kotlinx-coroutines-core/jvm/src/scheduling/CoroutineScheduler.kt +++ b/kotlinx-coroutines-core/jvm/src/scheduling/CoroutineScheduler.kt @@ -608,8 +608,17 @@ internal class CoroutineScheduler( @Volatile // volatile for push/pop operation into parkedWorkersStack var indexInArray = 0 set(index) { - name = "$schedulerName-worker-${if (index == 0) "TERMINATED" else index.toString()}" field = index + // Renaming a live thread from another thread can require the runtime to + // suspend it first (see #2234), which can time out and crash the process + // if that thread doesn't reach a safepoint quickly enough. Only rename + // synchronously here if it's safe to do so: either this thread hasn't + // started yet (no live peer to suspend), or we ARE this thread. Otherwise, + // defer to runWorker()'s self-heal check, which is always a same-thread + // (and therefore always safe) rename. + if (!isAlive || Thread.currentThread() === this) { + name = "$schedulerName-worker-${if (index == 0) "TERMINATED" else index.toString()}" + } } constructor(index: Int) : this() { @@ -706,9 +715,18 @@ internal class CoroutineScheduler( @JvmField var mayHaveLocalTasks = false + // Tracks the index this worker's name was last set to by itself, so a rename + // deferred by indexInArray's setter (because it was requested by another + // thread) gets applied here instead, on this worker's own thread. + private var appliedNameIndex = indexInArray + private fun runWorker() { var rescanned = false while (!isTerminated && state != WorkerState.TERMINATED) { + if (appliedNameIndex != indexInArray) { + appliedNameIndex = indexInArray + name = "$schedulerName-worker-${if (appliedNameIndex == 0) "TERMINATED" else appliedNameIndex.toString()}" + } val task = findTask(mayHaveLocalTasks) // Task found. Execute and repeat if (task != null) { diff --git a/kotlinx-coroutines-core/jvm/test/scheduling/SchedulerTestBase.kt b/kotlinx-coroutines-core/jvm/test/scheduling/SchedulerTestBase.kt index 3ca6bb7e19..b051bd65a9 100644 --- a/kotlinx-coroutines-core/jvm/test/scheduling/SchedulerTestBase.kt +++ b/kotlinx-coroutines-core/jvm/test/scheduling/SchedulerTestBase.kt @@ -36,18 +36,8 @@ abstract class SchedulerTestBase : TestBase() { } private fun maxSequenceNumber(): Int? { - return Thread.getAllStackTraces().keys.asSequence().filter { it is CoroutineScheduler.Worker } - .map { sequenceNumber(it.name) }.maxOrNull() - } - - private fun sequenceNumber(threadName: String): Int { - val suffix = threadName.substring(threadName.lastIndexOf("-") + 1) - val separatorIndex = suffix.indexOf(' ') - if (separatorIndex == -1) { - return suffix.toInt() - } - - return suffix.substring(0, separatorIndex).toInt() + return Thread.getAllStackTraces().keys.asSequence() + .filterIsInstance().maxOfOrNull { it.indexInArray } } suspend fun Iterable.joinAll() = forEach { it.join() } diff --git a/kotlinx-coroutines-core/jvm/test/scheduling/WorkerRenameTest.kt b/kotlinx-coroutines-core/jvm/test/scheduling/WorkerRenameTest.kt new file mode 100644 index 0000000000..915741b9d3 --- /dev/null +++ b/kotlinx-coroutines-core/jvm/test/scheduling/WorkerRenameTest.kt @@ -0,0 +1,71 @@ +package kotlinx.coroutines.scheduling + +import kotlinx.coroutines.testing.* +import org.junit.Test +import java.lang.Runnable +import java.util.concurrent.* +import java.util.concurrent.atomic.* +import kotlin.test.* + +/** + * Regression test for #2234: renaming a live worker thread from a different + * thread can require the runtime to suspend it first, which can time out and + * crash the process (see CoroutineScheduler.Worker.indexInArray). Verifies + * that a cross-thread indexInArray change never renames the thread directly, + * and that the worker corrects its own name once it next runs. + */ +class WorkerRenameTest : TestBase() { + + @Test + fun testCrossThreadIndexChangeDoesNotRenameLiveWorker() { + CoroutineScheduler(1, 2, schedulerName = "WorkerRenameTest").use { scheduler -> + val workerRef = AtomicReference() + val started = CountDownLatch(1) + val release = CountDownLatch(1) + + scheduler.dispatch(Runnable { + workerRef.set(Thread.currentThread() as CoroutineScheduler.Worker) + started.countDown() + release.await() + }) + started.await() + val worker = workerRef.get() + + try { + val nameBefore = worker.name + val otherIndex = worker.indexInArray + 1000 + worker.indexInArray = otherIndex // cross-thread: this is the test thread, not `worker` + assertEquals(nameBefore, worker.name, "cross-thread rename must not happen synchronously") + } finally { + release.countDown() + } + + val expectedName = "WorkerRenameTest-worker-${worker.indexInArray}" + val deadline = System.currentTimeMillis() + 5_000 + while (worker.name != expectedName && System.currentTimeMillis() < deadline) { + Thread.sleep(10) + } + assertEquals(expectedName, worker.name, "worker should self-heal its own name once it runs again") + } + } + + @Test + fun testSelfRenameHappensSynchronously() { + CoroutineScheduler(1, 2, schedulerName = "WorkerRenameTest").use { scheduler -> + val done = CountDownLatch(1) + var expectedName: String? = null + var nameAfterSelfRename: String? = null + + scheduler.dispatch(Runnable { + val self = Thread.currentThread() as CoroutineScheduler.Worker + self.indexInArray = self.indexInArray + 1000 // same-thread: self-rename + expectedName = "WorkerRenameTest-worker-${self.indexInArray}" + nameAfterSelfRename = self.name // must already reflect the change, synchronously + done.countDown() + }) + + done.await() + assertEquals(expectedName, nameAfterSelfRename, "self-rename must apply synchronously, not deferred") + } + } +}