Set kotlinx.coroutines to 1.7.3; Use limited parallelism instead of `Dispatchers.IO` (#798)

* set `kotlinx.coroutines` to `1.7.3`

* update tests after coroutines.tests update

* use `limitedParallelism` instead of bare `Dispatchers.IO`

rename

* use virtual time aware dispatchers in tests that use `runTest`

* better naming

* remove usages of experimental coroutines api

* remove usages of experimental time api

* use dispatcher from `coroutineScope` in `FrameLimiter`

`FrameLimiter` must cancel waiting when its scope canceled

* extract the `dispatcherToBlockOn` into separate file

* use dispatcher from `coroutineScope` in `FrameLimiter`

* Fix FrameLimiterTest

---------
Co-authored-by: Igor Demin <igor.demin@jetbrains.com>

skiko/build.gradle.kts

@@ -16,7 +16,7 @@ plugins {
     id("de.undercouch.download") version "5.4.0"
 }
 
-val coroutinesVersion = "1.5.2"
+val coroutinesVersion = "1.7.3"
 
 fun targetSuffix(os: OS, arch: Arch): String {
     return "${os.id}_${arch.id}"

skiko/src/awtMain/kotlin/org/jetbrains/skiko/HardwareLayer.kt

@@ -2,6 +2,7 @@ package org.jetbrains.skiko
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
+import org.jetbrains.skiko.redrawer.dispatcherToBlockOn
 import java.awt.Canvas
 import java.awt.Component
 import java.awt.Graphics
@@ -130,7 +131,7 @@ internal fun layerFrameLimiter(
     }
 
     return FrameLimiter(
-        scope + Dispatchers.IO,
+        scope + dispatcherToBlockOn,
         frameMillis = {
             frames.trySend(Unit)
             (1000 / state.frameLimit).toLong()

skiko/src/awtMain/kotlin/org/jetbrains/skiko/redrawer/Direct3DRedrawer.kt

 package org.jetbrains.skiko.redrawer
 
-import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.withContext
 import org.jetbrains.skia.DirectContext
 import org.jetbrains.skia.Surface
@@ -15,6 +15,7 @@ internal class Direct3DRedrawer(
     analytics: SkiaLayerAnalytics,
     private val properties: SkiaLayerProperties
 ) : AWTRedrawer(layer, analytics, GraphicsApi.DIRECT3D) {
+
     private val contextHandler = Direct3DContextHandler(layer)
     override val renderInfo: String get() = contextHandler.rendererInfo()
 
@@ -77,7 +78,7 @@ internal class Direct3DRedrawer(
 
     private suspend fun draw() {
         inDrawScope {
-            withContext(Dispatchers.IO) {
+            withContext(dispatcherToBlockOn) {
                 drawAndSwap(withVsync = properties.isVsyncEnabled)
             }
         }

skiko/src/awtMain/kotlin/org/jetbrains/skiko/redrawer/DispatcherToBlockOn.kt

+package org.jetbrains.skiko.redrawer
+
+import kotlinx.coroutines.asCoroutineDispatcher
+import java.util.concurrent.Executors
+
+
+/**
+ * Dispatcher intended for use in coroutines that blocks (not suspends) for indefinite amount of time.
+ * Now we use it e.g. for waiting for VSYNC.
+ * We can't use `Dispatchers.IO` here because it's limited by 64 threads and under heavy IO workload all of them might be occupied
+ * which leads to skipped frames
+ */
+internal val dispatcherToBlockOn = Executors.newCachedThreadPool().asCoroutineDispatcher()
\ No newline at end of file

skiko/src/awtMain/kotlin/org/jetbrains/skiko/redrawer/MetalRedrawer.kt

 package org.jetbrains.skiko.redrawer
 
-import kotlinx.coroutines.CancellationException
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.withContext
+import kotlinx.coroutines.*
 import org.jetbrains.skiko.*
 import org.jetbrains.skiko.context.MetalContextHandler
+import java.util.concurrent.Executors
 import javax.swing.SwingUtilities.*
 
 /**
@@ -132,7 +130,7 @@ internal class MetalRedrawer(
         // Executors.newSingleThreadExecutor().asCoroutineDispatcher(): 50 FPS, 150% CPU
         // Dispatchers.IO: 50 FPS, 200% CPU
         inDrawScope {
-            withContext(Dispatchers.IO) {
+            withContext(dispatcherToBlockOn) {
                 performDraw()
             }
         }

skiko/src/awtMain/kotlin/org/jetbrains/skiko/redrawer/WindowsOpenGLRedrawer.kt

 package org.jetbrains.skiko.redrawer
 
-import kotlinx.coroutines.CancellationException
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.withContext
+import kotlinx.coroutines.*
 import org.jetbrains.skiko.*
 import org.jetbrains.skiko.context.OpenGLContextHandler
+import java.util.concurrent.Executors
 
 internal class WindowsOpenGLRedrawer(
     private val layer: SkiaLayer,
@@ -110,7 +109,7 @@ internal class WindowsOpenGLRedrawer(
 
             val isVsyncEnabled = toRedrawVisible.all { it.properties.isVsyncEnabled }
             if (isVsyncEnabled) {
-                withContext(Dispatchers.IO) {
+                withContext(dispatcherToBlockOn) {
                     dwmFlush() // wait for vsync
                 }
             }

skiko/src/jvmMain/kotlin/org/jetbrains/skiko/FrameLimiter.kt

 package org.jetbrains.skiko
 
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.withContext
-import kotlin.time.ExperimentalTime
-
-private const val NanosecondsPerMillisecond = 1_000_000L
+import kotlinx.coroutines.*
+import java.util.concurrent.Executors
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.nanoseconds
+import kotlin.time.TimeSource
 
 /**
  * Limit the duration of the frames (to avoid high CPU usage) to [frameMillis].
@@ -17,7 +16,8 @@ private const val NanosecondsPerMillisecond = 1_000_000L
 class FrameLimiter(
     private val coroutineScope: CoroutineScope,
     private val frameMillis: () -> Long,
-    private val nanoTime: () -> Long = System::nanoTime
+    private val impreciseDelay: suspend (Long) -> Unit = ::delay,
+    private val currentTime: () -> Duration = { System.nanoTime().nanoseconds }
 ) {
     private val channel = RendezvousBroadcastChannel<Unit>()
 
@@ -31,15 +31,15 @@ class FrameLimiter(
     }
 
     private suspend fun preciseDelay(millis: Long) {
-        val start = nanoTime()
+        val start = currentTime()
         // delay aren't precise, so we should measure what is the actual precision of delay is,
         // so we don't wait longer than we need
-        var actual1msDelay = 1L
+        var actual1msDelay = 1.milliseconds
 
-        while (nanoTime() - start <= millis * NanosecondsPerMillisecond - actual1msDelay) {
-            val beforeDelay = nanoTime()
-            delay(1) // TODO do multiple delays instead of the single one consume more energy? Test it
-            actual1msDelay = maxOf(actual1msDelay, nanoTime() - beforeDelay)
+        while (currentTime() - start <= millis.milliseconds - actual1msDelay) {
+            val beforeDelay = currentTime()
+            impreciseDelay(1) // TODO do multiple delays instead of the single one consume more energy? Test it
+            actual1msDelay = maxOf(actual1msDelay, currentTime() - beforeDelay)
         }
     }
 

skiko/src/jvmTest/kotlin/org/jetbrains/skiko/FrameDispatcherTest.kt

 package org.jetbrains.skiko
 
 import kotlinx.coroutines.*
-import kotlinx.coroutines.test.TestCoroutineScope
-import kotlinx.coroutines.test.runBlockingTest
+import kotlinx.coroutines.test.*
 import org.junit.Assert.assertEquals
 import org.junit.Test
 import java.util.concurrent.Executors
 
-@OptIn(ExperimentalCoroutinesApi::class)
 class FrameDispatcherTest {
     private var frameCount = 0
 
     @Test
-    fun `shouldn't call onFrame after the creating`() = test {
-        FrameDispatcher(scope = this) {
+    fun `shouldn't call onFrame after the creating`() = runTest {
+        val frameDispatcher = FrameDispatcher(scope = this) {
             frameCount++
         }
 
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
 
         assertEquals(0, frameCount)
+        frameDispatcher.cancel()
     }
 
     @Test
-    fun `scheduleFrame after creating`() = test {
+    fun `scheduleFrame after creating`() = runTest {
         val frameDispatcher = FrameDispatcher(scope = this) {
             frameCount++
         }
 
         frameDispatcher.scheduleFrame()
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
 
         assertEquals(1, frameCount)
+        frameDispatcher.cancel()
     }
 
     @Test
-    fun `scheduleFrame multiple times after creating`() = test {
+    fun `scheduleFrame multiple times after creating`() = runTest {
         val frameDispatcher = FrameDispatcher(scope = this) {
             frameCount++
         }
@@ -44,42 +44,45 @@ class FrameDispatcherTest {
         frameDispatcher.scheduleFrame()
         frameDispatcher.scheduleFrame()
         frameDispatcher.scheduleFrame()
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
 
         assertEquals(1, frameCount)
+        frameDispatcher.cancel()
     }
 
     @Test
-    fun `scheduleFrame second time after first onFrame`() = test {
+    fun `scheduleFrame second time after first onFrame`() = runTest {
         val frameDispatcher = FrameDispatcher(scope = this) {
             frameCount++
         }
 
         frameDispatcher.scheduleFrame()
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
 
         frameDispatcher.scheduleFrame()
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
         assertEquals(2, frameCount)
+        frameDispatcher.cancel()
     }
 
     @Test
-    fun `scheduleFrame second time twice after first onFrame`() = test {
+    fun `scheduleFrame second time twice after first onFrame`() = runTest {
         val frameDispatcher = FrameDispatcher(scope = this) {
             frameCount++
         }
 
         frameDispatcher.scheduleFrame()
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
 
         frameDispatcher.scheduleFrame()
         frameDispatcher.scheduleFrame()
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
         assertEquals(2, frameCount)
+        frameDispatcher.cancel()
     }
 
     @Test
-    fun `scheduleFrame during onFrame`() = test {
+    fun `scheduleFrame during onFrame`() = runTest {
         lateinit var frameDispatcher: FrameDispatcher
         frameDispatcher = FrameDispatcher(scope = this) {
             frameCount++
@@ -98,10 +101,11 @@ class FrameDispatcherTest {
 
         yield()
         assertEquals(4, frameCount)
+        frameDispatcher.cancel()
     }
 
     @Test
-    fun `scheduleFrame multiple times during onFrame`() = test {
+    fun `scheduleFrame multiple times during onFrame`() = runTest {
         lateinit var frameDispatcher: FrameDispatcher
         frameDispatcher = FrameDispatcher(scope = this) {
             frameCount++
@@ -122,13 +126,13 @@ class FrameDispatcherTest {
 
         yield()
         assertEquals(4, frameCount)
+        frameDispatcher.cancel()
     }
 
     @Test
-    fun `cancel coroutine scope`() = test {
-        val scope = CoroutineScope(coroutineContext)
-
+    fun `cancel coroutine scope`() = runTest {
         lateinit var frameDispatcher: FrameDispatcher
+        val scope = CoroutineScope(coroutineContext + Job())
         frameDispatcher = FrameDispatcher(scope) {
             frameCount++
             frameDispatcher.scheduleFrame()
@@ -148,7 +152,7 @@ class FrameDispatcherTest {
         yield()
         assertEquals(3, frameCount)
 
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
         assertEquals(3, frameCount)
     }
 
@@ -184,13 +188,4 @@ class FrameDispatcherTest {
 
         assertEquals(listOf("frame0", "task", "frame1"), history)
     }
-
-    private fun test(
-        block: suspend TestCoroutineScope.() -> Unit
-    ) = runBlockingTest {
-        pauseDispatcher()
-        val job = Job()
-        TestCoroutineScope(coroutineContext + job).block()
-        job.cancel()
-    }
 }
\ No newline at end of file

skiko/src/jvmTest/kotlin/org/jetbrains/skiko/FrameLimiterTest.kt

 package org.jetbrains.skiko
 
 import kotlinx.coroutines.*
-import kotlinx.coroutines.test.DelayController
-import kotlinx.coroutines.test.TestCoroutineDispatcher
-import kotlinx.coroutines.test.TestCoroutineScope
-import kotlinx.coroutines.test.runBlockingTest
+import kotlinx.coroutines.test.*
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
 import org.junit.Test
-import kotlin.coroutines.CoroutineContext
 import kotlin.math.ceil
+import kotlin.time.Duration.Companion.milliseconds
 
 @OptIn(ExperimentalCoroutinesApi::class)
 class FrameLimiterTest {
@@ -151,11 +148,37 @@ class FrameLimiterTest {
         assertEquals(frames.map { it * 10 }, ticks3)
     }
 
+    private fun frameLimiterTest(
+        frameLimitMillis: Long,
+        delayPrecisionMillis: Long,
+        block: suspend TestScope.(FrameLimiter) -> Unit
+    ) =
+        runFrameTest {
+            val limiter = FrameLimiter(
+                backgroundScope,
+                frameMillis = { frameLimitMillis },
+                currentTime = { testScheduler.currentTime.milliseconds },
+                impreciseDelay = { timeMillis ->
+                    val ms = ceil(timeMillis.toDouble() / delayPrecisionMillis).toInt() * delayPrecisionMillis
+                    delay(ms)
+                }
+            )
+            block(limiter)
+        }
+
+
+    private fun runFrameTest(block: suspend TestScope.() -> Unit) = runTest {
+        block()
+        testScheduler.advanceUntilIdle()
+    }
+
     @Test
-    fun `cancel scope before awaitNextFrame`() = runBlockingTest {
-        pauseDispatcher()
+    fun `cancel scope before awaitNextFrame`() = runTest {
         val scope = CoroutineScope(coroutineContext + Job())
-        val frameLimiter = FrameLimiter(scope, { 10 }, nanoTime = { currentTime * 1_000_000 })
+        val frameLimiter = FrameLimiter(
+            scope,
+            frameMillis = { 10 },
+            currentTime = { testScheduler.currentTime.milliseconds })
 
         scope.cancel()
 
@@ -165,11 +188,13 @@ class FrameLimiterTest {
     }
 
     @Test
-    fun `cancel scope after awaitNextFrame`() = runBlockingTest {
-        pauseDispatcher()
+    fun `cancel scope after awaitNextFrame`() = runTest {
         val scope = CoroutineScope(coroutineContext + Job())
 
-        val frameLimiter = FrameLimiter(scope, { 10 }, nanoTime = { currentTime * 1_000_000 })
+        val frameLimiter = FrameLimiter(
+            scope,
+            frameMillis = { 10 },
+            currentTime = { testScheduler.currentTime.milliseconds })
 
         launch {
             scope.cancel()
@@ -180,10 +205,20 @@ class FrameLimiterTest {
         }
     }
 
+    private inline fun <reified T : Throwable> assertThrow(body: () -> Unit) {
+        var actualE: Throwable? = null
+        try {
+            body()
+        } catch (e: Throwable) {
+            actualE = e
+        }
+        assertTrue("Actual ${actualE?.javaClass}, expected ${T::class.java}", actualE is T)
+    }
+
     @Test(timeout = 30000)
     fun `multithreaded awaiter`() {
         val scope = CoroutineScope(Dispatchers.IO)
-        val frameLimiter = FrameLimiter(scope, { 0 }, nanoTime = System::nanoTime)
+        val frameLimiter = FrameLimiter(scope, { 0 })
 
         runBlocking(Dispatchers.IO) {
             repeat(50000) {
@@ -197,7 +232,7 @@ class FrameLimiterTest {
     @Test(timeout = 30000)
     fun `multiple multithreaded awaiters`() {
         val scope = CoroutineScope(Dispatchers.IO)
-        val frameLimiter = FrameLimiter(scope, { 0 }, nanoTime = System::nanoTime)
+        val frameLimiter = FrameLimiter(scope, { 0 })
 
         runBlocking(Dispatchers.IO) {
             repeat(3) {
@@ -214,61 +249,4 @@ class FrameLimiterTest {
 
         scope.cancel()
     }
-
-    private inline fun <reified T : Throwable> assertThrow(body: () -> Unit) {
-        var actualE: Throwable? = null
-        try {
-            body()
-        } catch (e: Throwable) {
-            actualE = e
-        }
-        assertTrue("Actual ${actualE?.javaClass}, expected ${T::class.java}", actualE is T)
-    }
-
-    private fun frameLimiterTest(
-        frameLimitMillis: Long,
-        delayPrecisionMillis: Long,
-        block: suspend TestCoroutineScope.(FrameLimiter) -> Unit
-    ) {
-        runFrameTest(
-            delayPrecisionMillis = delayPrecisionMillis
-        ) {
-            val scope = CoroutineScope(coroutineContext + Job())
-            val limiter = FrameLimiter(
-                this,
-                frameMillis = { frameLimitMillis },
-                nanoTime = { currentTime * 1_000_000 }
-            )
-            block(limiter)
-            scope.cancel()
-        }
-    }
-
-    private fun runFrameTest(
-        delayPrecisionMillis: Long,
-        block: suspend TestCoroutineScope.() -> Unit
-    ) = runBlockingTest{
-        val dispatcher = NonpreciseTestCoroutineDispatcher(delayPrecisionMillis)
-        dispatcher.pauseDispatcher()
-        val scope = TestCoroutineScope(dispatcher)
-        scope.launch {
-            scope.block()
-        }
-        dispatcher.advanceUntilIdle()
-    }
-
-    @OptIn(InternalCoroutinesApi::class)
-    private class NonpreciseTestCoroutineDispatcher(
-        private val delayPrecisionMillis: Long,
-        private val original: TestCoroutineDispatcher = TestCoroutineDispatcher()
-    ) : CoroutineDispatcher(), Delay, DelayController by original {
-        override fun dispatch(context: CoroutineContext, block: Runnable) {
-            original.dispatch(context, block)
-        }
-
-        override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
-            val delay = ceil(timeMillis.toDouble() / delayPrecisionMillis).toInt() * delayPrecisionMillis
-            original.scheduleResumeAfterDelay(delay, continuation)
-        }
-    }
 }
\ No newline at end of file

skiko/src/jvmTest/kotlin/org/jetbrains/skiko/RendezvousBroadcastChannelTest.kt

 package org.jetbrains.skiko
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.pauseDispatcher
 import kotlinx.coroutines.test.runBlockingTest
+import kotlinx.coroutines.test.runTest
 import org.junit.Assert.assertEquals
 import org.junit.Test
 import kotlin.random.Random
@@ -172,10 +175,8 @@ class RendezvousBroadcastChannelTest {
         assertEquals(true, isExceptionThrown)
     }
 
-    @OptIn(ExperimentalCoroutinesApi::class)
     @Test
-    fun `produce values, consume from multiple coroutines`() = runBlockingTest {
-        pauseDispatcher()
+    fun `produce values, consume from multiple coroutines`() = runTest {
 
         val frames1 = mutableListOf<Int>()
         val frames2 = mutableListOf<Int>()
@@ -211,7 +212,7 @@ class RendezvousBroadcastChannelTest {
             }
         }
 
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
         produceJob.cancel()
 
         assertEquals((0 until 1000).toList(), frames1)

skiko/src/jvmTest/kotlin/org/jetbrains/skiko/TaskTest.kt

@@ -2,14 +2,12 @@ package org.jetbrains.skiko
 
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
-import kotlinx.coroutines.test.TestCoroutineScope
-import kotlinx.coroutines.test.runBlockingTest
+import kotlinx.coroutines.test.*
 import kotlinx.coroutines.yield
 import org.junit.Assert.assertFalse
 import org.junit.Assert.assertTrue
@@ -19,38 +17,38 @@ import java.util.concurrent.Executors.newSingleThreadExecutor
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.random.Random
 
-@OptIn(ExperimentalCoroutinesApi::class)
 internal class TaskTest {
     @Test
-    fun `runAndAwait with finish`() = test {
+    fun `runAndAwait with finish`() = runTest {
         val task = Task()
 
         val job = launch {
             task.runAndAwait {}
         }
 
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
 
         task.finish()
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
 
         assertTrue(job.isCompleted)
     }
 
     @Test
-    fun `runAndAwait without finish`() = test {
+    fun `runAndAwait without finish`() = runTest {
         val task = Task()
 
         val job = launch {
             task.runAndAwait {}
         }
 
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
         assertFalse(job.isCompleted)
+        job.cancel()
     }
 
     @Test
-    fun `finish inside runAndAwait`() = test {
+    fun `finish inside runAndAwait`() = runTest {
         val task = Task()
 
         val job = launch {
@@ -59,12 +57,12 @@ internal class TaskTest {
             }
         }
 
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
         assertTrue(job.isCompleted)
     }
 
     @Test
-    fun `finish before runAndAwait`() = test {
+    fun `finish before runAndAwait`() = runTest {
         val task = Task()
 
         val job = launch {
@@ -72,8 +70,9 @@ internal class TaskTest {
             task.runAndAwait {}
         }
 
-        advanceUntilIdle()
+        testScheduler.advanceUntilIdle()
         assertFalse(job.isCompleted)
+        job.cancel()
     }
 
     @Test(timeout = 5000)
@@ -193,13 +192,4 @@ internal class TaskTest {
             needsDisplay.set(true)
         }
     }
-
-    private fun test(
-        block: suspend TestCoroutineScope.() -> Unit
-    ) = runBlockingTest {
-        pauseDispatcher()
-        val job = Job()
-        TestCoroutineScope(coroutineContext + job).block()
-        job.cancel()
-    }
 }
\ No newline at end of file