Fix DefaultExecutor having a separate thread and being used for cleanup work and running Dispatchers.Unconfined #4277

dkhalanskyjb · 2024-11-21T12:46:48Z

qwwdfsad · 2025-03-11T11:36:20Z

kotlinx-coroutines-core/native/src/EventLoop.kt

-        val delayTimeMillis = delayNanosToMillis(delayedTask.nanoTime - now)
-        DefaultExecutor.invokeOnTimeout(delayTimeMillis, delayedTask, EmptyCoroutineContext)
+        startThreadOrObtainSleepingThread()?.let {
+            it.executeAfter(0L, {})


Did the comment get lost? send an empty task to unpark the waiting event loop this one

qwwdfsad · 2025-04-08T09:32:19Z

kotlinx-coroutines-core/common/src/Dispatchers.common.kt

+ *
+ * `close` needs to run somewhere, but it can't run on the closed dispatcher.
+ *
+ * On the JVM and Native, we reschedule to the thread pool backing `Dispatchers.IO`,


[Reading as it goes] From the contract standpoint: here we change the behaviour -- previously, all such cleanups have been executed synchronously (because DefaultExecutor is single-threaded); with this change, it's no longer the case.

Not sure if it will actually affect anyone, rather pointing it out

qwwdfsad · 2025-04-08T09:40:58Z

kotlinx-coroutines-core/concurrent/src/Dispatchers.concurrent.kt

+     *   If we were to create a separate view of [Dispatchers.IO] for cleanup, this number would be exceeded, which
+     *   is a regression.
+     */
+    Dispatchers.IO.dispatch(Dispatchers.IO, task)


Thanks for the writeup!

qwwdfsad · 2025-04-08T09:44:05Z

kotlinx-coroutines-core/jvm/src/DefaultDelay.kt

+            unTrackTask(this) /** see the comment in [startThreadOrObtainSleepingThread] */
+            try {
+                while (true) {
+                    Thread.interrupted() // just reset interruption flag


Please account #4399 when rebasing on develop

Could you elaborate? What's the relation between that PR and this piece of code?

qwwdfsad · 2025-04-08T09:45:06Z

kotlinx-coroutines-core/jvm/src/DefaultDelay.kt

+    override fun startThreadOrObtainSleepingThread(): Thread? {
+        // Check if the thread is already running
+        _thread.value?.let { return it }
+        /* Now we know that at the moment of this call the thread was not initially running.


Could you please use a consistent style of comments here?

The rules I use for writing comments are roughly this:

If I have to use code references, I use /** */.

If not, then I use // if it's a short phrase that fits on one line of code.

For multiline comments or comments that are at risk of becoming multiline, /* */.

I could just always use /** */ everywhere, but why not use the other forms if the language provides them?

qwwdfsad · 2025-04-08T09:49:17Z

kotlinx-coroutines-core/jvm/src/scheduling/CoroutineScheduler.kt

-    fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false) {
-        trackTask() // this is needed for virtual time support
+    fun dispatch(
+        block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false, track: Boolean = true


Suggested change

block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false, track: Boolean = true

block: Runnable, taskContext: TaskContext = NonBlockingContext,

fair: Boolean = false, track: Boolean = true

But why? It's 117 characters, so it fits the 120-character limit.

qwwdfsad · 2025-04-08T09:52:12Z

kotlinx-coroutines-core/jvm/src/scheduling/CoroutineScheduler.kt

        val task = createTask(block, taskContext)
+        if (track) trackTask(task) // this is needed for virtual time support


An idea: you can simplify the code here and avoid extra checks with the following trick:

Untrackable tasks (which are extremely rare) are wrapped into NonTrackableRunnable : Runnable

The only time tracker is the test one -- it can always check if obj is an instance of non-trackable runnable

qwwdfsad · 2025-04-08T10:49:00Z

kotlinx-coroutines-core/native/src/DefaultDelay.kt

+/** A view separate from [Dispatchers.IO].
+ * [Int.MAX_VALUE] instead of `1` to avoid needlessly using the [LimitedDispatcher] machinery. */
+private val ioView by lazy { Dispatchers.IO.limitedParallelism(Int.MAX_VALUE) }
+// Without `lazy`, there is a cycle between `ioView` and `DefaultDelay` initialization, leading to a segfault.


Probably should be above the declaration line.
Could you please also drop a clarification here?
I.e.:
there is a cycle between ioViewandDefaultDelay initialization
->
there is a cycle between ioViewandDefaultDelay initialization (limitedParallelism -> LimitedDispatcher -> DefaultDelay dependency in constructor -> this file -> ioView)

Before this change, DefaultExecutor was occasionally used for executing the work of dispatchers that no longer function. This is no longer the case: instead, Dispatchers.IO is used for that on our multithreaded targets.

…created

Tests segfault on my machine with this stacktrace: 0 kfun:kotlinx.coroutines.CoroutineDispatcher#limitedParallelism(kotlin.Int;kotlin.String?){}kotlinx.coroutines.CoroutineDispatcher-trampoline () at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/common/src/CoroutineDispatcher.kt:176 1 0x00000000003bed59 in kfun:kotlinx.coroutines.CoroutineDispatcher#limitedParallelism$default(kotlin.Int;kotlin.String?;kotlin.Int){}kotlinx.coroutines.CoroutineDispatcher (_this=0x0, parallelism=2147483647, name=0x0, $mask0=2) at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/common/src/CoroutineDispatcher.kt:176 2 0x00000000004bfa97 in kfun:kotlinx.coroutines.$init_global#internal.18 () at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/native/src/DefaultDelay.kt:101 3 0x0000000000cac194 in CallInitGlobalPossiblyLock () 4 0x00000000004bfb60 in kfun:kotlinx.coroutines#<get-DefaultDelay>(){}kotlinx.coroutines.Delay () at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/native/src/DefaultDelay.kt:1 5 0x0000000000496840 in kfun:kotlinx.coroutines.internal.LimitedDispatcher#<init>(kotlinx.coroutines.CoroutineDispatcher;kotlin.Int;kotlin.String?){} ($this=0x7ffff64a0668, dispatcher=0x7ffff64a0620, parallelism=64, name=0x0) at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/common/src/internal/LimitedDispatcher.kt:26 6 0x00000000003bec1d in kfun:kotlinx.coroutines.CoroutineDispatcher#limitedParallelism(kotlin.Int;kotlin.String?){}kotlinx.coroutines.CoroutineDispatcher (_this=0x7ffff64a0620, parallelism=64, name=0x0) at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/common/src/CoroutineDispatcher.kt:178 7 0x00000000004c53ad in kfun:kotlinx.coroutines.MultiWorkerDispatcher.limitedParallelism#internal (_this=0x7ffff64a0620, parallelism=64, name=0x0) at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/native/src/MultithreadedDispatchers.kt:151 8 0x0000000000b3de7a in kfun:kotlinx.coroutines.CoroutineDispatcher#limitedParallelism(kotlin.Int;kotlin.String?){}kotlinx.coroutines.CoroutineDispatcher-trampoline () at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/common/src/CoroutineDispatcher.kt:176 9 0x00000000003bed59 in kfun:kotlinx.coroutines.CoroutineDispatcher#limitedParallelism$default(kotlin.Int;kotlin.String?;kotlin.Int){}kotlinx.coroutines.CoroutineDispatcher (_this=0x7ffff64a0620, parallelism=64, name=0x0, $mask0=2) at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/common/src/CoroutineDispatcher.kt:176 10 0x00000000004c13c0 in kfun:kotlinx.coroutines.DefaultIoScheduler.<init>#internal ($this=0x7ffff6630700) at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/native/src/Dispatchers.kt:27 11 0x00000000004c127f in kfun:kotlinx.coroutines.DefaultIoScheduler.$init_global#internal () at /home/dkhalansky/IdeaProjects/kotlinx.coroutines/kotlinx-coroutines-core/native/src/Dispatchers.kt:1

dkhalanskyjb force-pushed the dk-remove-defaultexecutor branch 3 times, most recently from 17af696 to 660ee49 Compare December 4, 2024 14:13

dkhalanskyjb mentioned this pull request Jan 14, 2025

Reach target parallelism in .limitedParallelism of failing dispatchers #4330

Merged

dkhalanskyjb force-pushed the dk-remove-defaultexecutor branch from 389e9eb to 448ecaa Compare January 28, 2025 11:19

dkhalanskyjb force-pushed the dk-remove-defaultexecutor branch from a1ef099 to 4cb7adc Compare February 24, 2025 15:22

dkhalanskyjb requested a review from qwwdfsad February 25, 2025 14:10

dkhalanskyjb marked this pull request as ready for review February 25, 2025 14:10

qwwdfsad reviewed Apr 8, 2025

View reviewed changes

dkhalanskyjb added 15 commits April 9, 2025 11:50

Do not use DefaultExecutor for cleanup work

ed8c45b

Before this change, DefaultExecutor was occasionally used for executing the work of dispatchers that no longer function. This is no longer the case: instead, Dispatchers.IO is used for that on our multithreaded targets.

Add some internal documentation

a97acdd

Use a Dispatchers.IO thread for DefaultDelay on the JVM

e09543d

Cleanup

0577a29

WIP

6d442f3

Fix VirtualTimeSource not being able to skip delays

5399006

Fix VirtualTimeSource not noticing that DefaultDelay is about to get …

65ccc36

…created

Fix a test

d2492e3

Fix the segfault

22c5391

Implement some tests for the new functionality

80ef17e

Fix warnings

045e52f

Work around some JS miscompilation

d0aa7fc

Update the documentation of time-based operators

54a155c

Fix a broken guide example

0857fee

dkhalanskyjb force-pushed the dk-remove-defaultexecutor branch from 8ed3f71 to 0857fee Compare April 9, 2025 12:31

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix DefaultExecutor having a separate thread and being used for cleanup work and running Dispatchers.Unconfined #4277

Fix DefaultExecutor having a separate thread and being used for cleanup work and running Dispatchers.Unconfined #4277

dkhalanskyjb commented Nov 21, 2024

qwwdfsad Mar 11, 2025

qwwdfsad Apr 8, 2025

qwwdfsad Apr 8, 2025

qwwdfsad Apr 8, 2025

dkhalanskyjb Apr 8, 2025

qwwdfsad Apr 8, 2025

dkhalanskyjb Apr 8, 2025

qwwdfsad Apr 8, 2025

dkhalanskyjb Apr 8, 2025

qwwdfsad Apr 8, 2025

qwwdfsad Apr 8, 2025

	block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false, track: Boolean = true
	block: Runnable, taskContext: TaskContext = NonBlockingContext,
	fair: Boolean = false, track: Boolean = true

		val task = createTask(block, taskContext)
		if (track) trackTask(task) // this is needed for virtual time support

Fix DefaultExecutor having a separate thread and being used for cleanup work and running Dispatchers.Unconfined #4277

Are you sure you want to change the base?

Fix DefaultExecutor having a separate thread and being used for cleanup work and running Dispatchers.Unconfined #4277

Conversation

dkhalanskyjb commented Nov 21, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment