Fix deadlock in FrameLimiter (#211)

We run FrameLimiter in another thread in SoftwareRedrawer. So FrameLimiter should be thread-safe. It is not thread safe now.

We can run it from the Swing thread though, but probably it will be used in Compose directly, where can be multiple threads (in Application frame clock, for example)

Deadlock:
[couroutine 2] onRequest.trySend(Unit)
[couroutine 1] onRequest.receive()
[couroutine 1] onResult.getAndSet(CompletableDeferred()).complete(value)
[couroutine 2] onResult.get().await() // wait
[couroutine 1] onRequest.receive()  // wait

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

@@ -18,6 +18,8 @@ private const val NanosecondsPerMillisecond = 1_000_000L
  *      frameJob.cancelAndJoin()
  *  }
  * ```
+ *
+ * Can be accessed from multiple threads.
  */
 @OptIn(ExperimentalTime::class)
 @Suppress("UNUSED_PARAMETER")

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

 package org.jetbrains.skiko
 
-import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.channels.Channel
-import java.util.concurrent.atomic.AtomicReference
+import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.resume
 
 /**
  * Behaves as Channel<Unit>(Channel.RENDEZVOUS), but with ability to send value to all current consumers
- * (which await on `receive` method)
+ * (which await on `receive` method).
  */
 internal class RendezvousBroadcastChannel<T> {
     private val onRequest = Channel<Unit>(Channel.CONFLATED)
-    private val onResult = AtomicReference(CompletableDeferred<T>())
+    private val receivers = mutableListOf<Continuation<T>>()
 
     /**
-     * Send value to all current consumers which await value on `receive` method, or await for the first one
+     * Send value to all current consumers which await value on `receive` method, or await for the first one.
+     *
+     * Can't be called concurrently from multiple threads.
      */
     suspend fun sendAll(value: T) {
         onRequest.receive()
-        onResult.getAndSet(CompletableDeferred()).complete(value)
+        val receiversCopy = synchronized(receivers) {
+            mutableListOf<Continuation<T>>().apply {
+                addAll(receivers)
+                receivers.clear()
+            }
+        }
+        for (receiver in receiversCopy) {
+            receiver.resume(value)
+        }
     }
 
     /**
      * Wait when the producer will send a value and return it.
+     *
+     * Can be called concurrently from multiple threads.
      */
-    suspend fun receive(): T {
+    suspend fun receive(): T = suspendCancellableCoroutine { continuation ->
+        synchronized(receivers) {
+            receivers.add(continuation)
+        }
         onRequest.trySend(Unit)
-        return onResult.get().await()
     }
 }
\ No newline at end of file

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

@@ -180,6 +180,41 @@ class FrameLimiterTest {
         }
     }
 
+    @Test(timeout = 30000)
+    fun `multithreaded awaiter`() {
+        val scope = CoroutineScope(Dispatchers.IO)
+        val frameLimiter = FrameLimiter(scope, { 0 }, nanoTime = System::nanoTime)
+
+        runBlocking(Dispatchers.IO) {
+            repeat(50000) {
+                frameLimiter.awaitNextFrame()
+            }
+        }
+
+        scope.cancel()
+    }
+
+    @Test(timeout = 30000)
+    fun `multiple multithreaded awaiters`() {
+        val scope = CoroutineScope(Dispatchers.IO)
+        val frameLimiter = FrameLimiter(scope, { 0 }, nanoTime = System::nanoTime)
+
+        runBlocking(Dispatchers.IO) {
+            repeat(3) {
+                launch {
+                    repeat(50000) {
+                        frameLimiter.awaitNextFrame()
+                        yield()
+                        yield()
+                        yield()
+                    }
+                }
+            }
+        }
+
+        scope.cancel()
+    }
+
     private inline fun <reified T : Throwable> assertThrow(body: () -> Unit) {
         var actualE: Throwable? = null
         try {

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

 package org.jetbrains.skiko
 
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.TimeoutCancellationException
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.*
 import kotlinx.coroutines.test.runBlockingTest
-import kotlinx.coroutines.withTimeout
 import org.junit.Assert.assertEquals
 import org.junit.Test
 import kotlin.random.Random
@@ -71,6 +66,30 @@ class RendezvousBroadcastChannelTest {
         assertEquals(listOf(1, 1, 1, 1, 1), actualValues)
     }
 
+    @Test(timeout = 30000)
+    fun `multithreading sending and receiving should not cause deadlock`() {
+        val channel = RendezvousBroadcastChannel<Int>()
+
+        runBlocking {
+            val receiverJobs = (1..10).map {
+                launch(Dispatchers.IO) {
+                    repeat(1000) {
+                        channel.receive()
+                    }
+                }
+            }
+
+            val sendingJob = launch(Dispatchers.IO) {
+                while (true) {
+                    channel.sendAll(1)
+                }
+            }
+
+            receiverJobs.joinAll()
+            sendingJob.cancel()
+        }
+    }
+
     @Test
     fun `first send should not end if there is no received value`() {
         var isExceptionThrown = false