~tmp

Author: qwwdfsad

build.gradle

@@ -128,6 +128,7 @@ allprojects {
         google()
         mavenCentral()
         maven { url "https://maven.pkg.jetbrains.space/kotlin/p/kotlin/dev" }
+        mavenLocal()
     }
 }
 

buildSrc/settings.gradle.kts

@@ -10,6 +10,7 @@ pluginManagement {
             maven("https://cache-redirector.jetbrains.com/plugins.gradle.org/m2")
         } else {
             maven("https://plugins.gradle.org/m2")
+            mavenLocal()
         }
         if (build_snapshot_train?.toBoolean() == true) {
             mavenLocal()

gradle.properties

@@ -10,7 +10,7 @@ kotlin_version=1.7.10
 # Dependencies
 junit_version=4.12
 junit5_version=5.7.0
-atomicfu_version=0.18.2
+atomicfu_version=0.18.2-conditions
 knit_version=0.4.0
 html_version=0.7.2
 lincheck_version=2.14.1

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

@@ -4,7 +4,9 @@
 
 package kotlinx.coroutines
 
+import kotlinx.atomicfu.*
 import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 import kotlin.native.concurrent.*
 
@@ -63,28 +65,62 @@ internal class WorkerDispatcher(name: String) : CloseableCoroutineDispatcher(),
     }
 }
 
-private class MultiWorkerDispatcher(name: String, workersCount: Int) : CloseableCoroutineDispatcher() {
-    private val tasksQueue = Channel<Runnable>(Channel.UNLIMITED)
-    private val workers = Array(workersCount) { Worker.start(name = "$name-$it") }
+public class MultiWorkerDispatcher(
+    private val name: String, private val workersCount: Int
+) : CloseableCoroutineDispatcher() {
+    private val runningWorkers = atomic(0)
+    private val queue = DispatcherQueue()
+    private val workers = atomicArrayOfNulls<Worker>(workersCount)
+    private val isTerminated = atomic(false)
 
-    init {
-        workers.forEach { w -> w.executeAfter(0L) { workerRunLoop() } }
+    override fun dispatch(context: CoroutineContext, block: Runnable) {
+        if (runningWorkers.value != workersCount) {
+            tryAddWorker()
+        }
+        queue.put(block)
     }
 
-    private fun workerRunLoop() = runBlocking {
-        for (task in tasksQueue) {
-            // TODO error handling
-            task.run()
+    private fun tryAddWorker() {
+        runningWorkers.loop {
+            if (it == workersCount) return
+            if (runningWorkers.compareAndSet(it, it + 1)) {
+                addWorker(it)
+                return
+            }
         }
     }
 
-    override fun dispatch(context: CoroutineContext, block: Runnable) {
-        // TODO handle rejections
-        tasksQueue.trySend(block)
+    private fun addWorker(sequenceNumber: Int) {
+        val worker = Worker.start(name = "$name-#$sequenceNumber")
+        workers[sequenceNumber].value = worker
+        worker.executeAfter(0L) {
+            workerLoop()
+        }
+    }
+
+    private fun workerLoop() {
+        while (!isTerminated.value) {
+            val runnable = queue.take()
+            runnable.run()
+        }
     }
 
     override fun close() {
-        tasksQueue.close()
-        workers.forEach { it.requestTermination().result }
+        // TODO it races with worker creation
+        if (!isTerminated.compareAndSet(false, true)) return
+        repeat(workersCount) {
+            queue.put(Runnable {}) // Empty poison pill to wakeup workers and make them check isTerminated
+        }
+
+        val requests = ArrayList<Future<Unit>>()
+        for (i in 0 until workers.size) {
+            val worker = workers[i].value ?: continue
+            requests += worker.requestTermination(false)
+        }
+        for (request in requests) {
+            request.result // Wait for workers termination
+        }
+
+//        queue.close()
     }
 }

kotlinx-coroutines-core/native/src/internal/BlockingQueue.kt

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.internal
+
+/**
+ * A domain-specific bounded array-backed blocking queue used
+ * in multithreaded dispatchers.
+ *
+ * It is written in the most straightforward way and is not expected to be a bottleneck.
+ *
+ * NB: this implementation used POSIX mutexes and should be explicitly [closed][close]
+ * in order to release underlying POSIX-specific primitives.
+ */
+internal class BlockingQueue<T: Any>(capacity: Int) {
+
+    init {
+        // Prior to all resource allocations
+        require(capacity >= 1) { "Capacity should be positive, but had $capacity" }
+    }
+
+    private val lock = SpinLockWithCondition()
+    private val notEmptyCondition = lock.Condition()
+    private val notFullCondition = lock.Condition()
+
+    private val elements = arrayOfNulls<Any?>(capacity)
+    private var tailIndex = 0
+    private var headIndex = 0
+    private var size = 0
+
+    fun put(value: T) = lock.withLock {
+        while (size == elements.size) {
+            notFullCondition.wait()
+        }
+        elements[headIndex] = value
+        if (++headIndex == elements.size) headIndex = 0
+        ++size
+        notEmptyCondition.notifyOne()
+    }
+
+    fun take(): T = lock.withLock {
+        while (size == 0) {
+            notEmptyCondition.wait()
+        }
+        val result = elements[tailIndex]
+        elements[tailIndex] = null
+        if (++tailIndex == elements.size) tailIndex = 0
+        --size
+        notFullCondition.notifyOne()
+        @Suppress("UNCHECKED_CAST")
+        return result as T
+    }
+
+    fun close() {
+        // TODO
+    }
+}

kotlinx-coroutines-core/native/src/internal/DispatcherQueue.kt

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.internal
+
+import kotlinx.atomicfu.*
+import kotlinx.coroutines.*
+
+internal class DispatcherQueue {
+
+    private val queue = LockFreeTaskQueue<Runnable>(false)
+    private val lock = SpinLockWithCondition()
+    private val nonEmpty = lock.Condition()
+    private val nonEmptyWaiters = atomic(0)
+
+    fun put(value: Runnable) {
+        queue.addLast(value)
+        notifyNonEmpty()
+    }
+
+    private fun notifyNonEmpty() {
+        if (nonEmptyWaiters.value == 0) return
+        lock.withLock {
+            nonEmpty.notifyOne()
+        }
+    }
+
+    fun take(): Runnable {
+        takeFastPath()?.let { return it }
+        return takeSlowPath()
+    }
+
+    private fun takeFastPath(): Runnable? {
+        val fastPathResult = queue.removeFirstOrNull()
+        if (fastPathResult !== null) {
+            return fastPathResult
+        }
+        return null
+    }
+
+    private fun takeSlowPath(): Runnable = lock.withLock {
+        nonEmptyWaiters.incrementAndGet()
+        while (true) {
+            // Check if `put` missed our increment
+            val result = takeFastPath()
+            if (result !== null) {
+                nonEmptyWaiters.decrementAndGet()
+                return result
+            }
+            nonEmpty.wait()
+        }
+        TODO() // Hack for some reason required for this code to compile
+    }
+}

kotlinx-coroutines-core/native/src/internal/FastLock.kt

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.internal
+
+import kotlinx.atomicfu.*
+import kotlin.native.concurrent.*
+import kotlin.test.*
+
+private val UNPARK_TASK = {}
+private const val DEFAULT = 0
+private const val PARKED = 1
+private const val UNPARKED = 2
+
+
+internal class SpinLockWithCondition {
+
+    private val isLocked = atomic(false)
+
+    private class ParkSupport(private val worker: Worker) {
+        private val parkingState = atomic(0)
+
+        fun park() {
+            parkingState.loop { state ->
+                when (state) {
+                    DEFAULT -> {
+                        if (parkingState.compareAndSet(DEFAULT, PARKED)) {
+                            worker.park(-1)
+                            return
+                        }
+                    }
+                    UNPARKED -> {
+                        return
+                    }
+                    PARKED -> {
+                        fail("Unexpected state: worker is already parked")
+                    }
+                }
+            }
+        }
+
+        fun unpark() {
+            parkingState.loop { state ->
+                when (state) {
+                    DEFAULT -> {
+                        if (parkingState.compareAndSet(DEFAULT, UNPARKED)) {
+                            return
+                        }
+                    }
+                    UNPARKED -> {
+                        fail("Unexpected state: worker is already unparked")
+                    }
+                    PARKED -> {
+                        parkingState.compareAndSet(DEFAULT, UNPARKED)
+                        worker.executeAfter(0L, UNPARK_TASK)
+                        return
+                    }
+                }
+            }
+        }
+    }
+
+    inner class Condition {
+        private val waitingWorkers = ArrayList<ParkSupport>()
+
+        fun wait() {
+            assertLocked()
+            val worker = ParkSupport(Worker.current)
+            waitingWorkers += worker
+            unlock() // Release before parking
+            worker.park()
+            lock() // Immediately lock
+        }
+
+        fun notifyOne() {
+            assertLocked()
+            waitingWorkers.removeLastOrNull()?.unpark()
+        }
+    }
+
+    public fun lock() {
+        isLocked.loop { locked ->
+            if (locked) return@loop
+            else if (isLocked.compareAndSet(false, true)) return
+        }
+    }
+
+
+    public fun unlock() {
+        require(isLocked.compareAndSet(true, false)) { "Lock release should always succeed" }
+    }
+
+    private fun assertLocked() {
+        require(isLocked.value) { "Lock should be locked" }
+    }
+}
+
+internal inline fun <T> SpinLockWithCondition.withLock(block: () -> T): T {
+    try {
+        lock()
+        return block()
+    } finally {
+        unlock()
+    }
+}