Improved management of parkedWorkersStack in terms of ABA/concurrency;

Also includes documentation, naming and packaging fixes

Author: elizarov

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/LockFreeMPMCQueue.kt

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental.internal
+
+import kotlinx.atomicfu.*
+
+internal open class LockFreeMPMCQueueNode<T> {
+    val next = atomic<T?>(null)
+
+    @PublishedApi internal val nextValue: T? get() = next.value
+}
+
+/*
+ * Michael & Scott lock-free Multi-Producer Multi-Consumer Queue with support for poll with predicate.
+ *
+ * @suppress **This is unstable API and it is subject to change.**
+ */
+internal open class LockFreeMPMCQueue<T : LockFreeMPMCQueueNode<T>> {
+    private val head =
+        @Suppress("UNCHECKED_CAST")
+        atomic(LockFreeMPMCQueueNode<T>() as T) // sentinel
+
+    private val tail = atomic(head.value)
+
+    public fun addLast(node: T): Boolean {
+        tail.loop { curTail ->
+            val curNext = curTail.next.value
+            if (curNext != null) {
+                tail.compareAndSet(curTail, curNext)
+                return@loop // retry
+            }
+            if (curTail.next.compareAndSet(null, node)) {
+                tail.compareAndSet(curTail, node)
+                return true
+            }
+        }
+    }
+
+    public fun removeFistOrNull(): T? {
+        head.loop { curHead ->
+            val next = curHead.next.value ?: return null
+            if (head.compareAndSet(curHead, next)) {
+                return next
+            }
+        }
+    }
+
+    @PublishedApi internal val headValue: T get() = head.value
+    @PublishedApi internal fun headCas(curHead: T, update: T) = head.compareAndSet(curHead, update)
+
+    public inline fun removeFistOrNullIf(predicate: (T) -> Boolean): T? {
+        while (true) {
+            val curHead = headValue
+            val next = curHead.nextValue ?: return null
+            if (!predicate(next)) return null
+            if (headCas(curHead, next)) {
+                return next
+            }
+        }
+    }
+
+    public fun isEmpty(): Boolean = size == 0
+
+    public val size: Int get() = fold(0) { acc, _ -> acc + 1 }
+
+    public inline fun <R> fold(initial: R, operation: (acc: R, T) -> R): R {
+        var acc = initial
+        var cur = headValue
+        while (true) {
+            val next = cur.nextValue ?: break
+            acc = operation(acc, next)
+            cur = next
+        }
+        return acc
+    }
+}

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/CoroutineScheduler.kt

@@ -6,8 +6,14 @@ import java.io.*
 import java.util.concurrent.*
 import kotlin.coroutines.experimental.*
 
+/**
+ * @suppress **This is unstable API and it is subject to change.**
+ */
 // TODO make internal after integration wih Ktor
-class ExperimentalCoroutineDispatcher(corePoolSize: Int = CORE_POOL_SIZE, maxPoolSize: Int = MAX_POOL_SIZE) : CoroutineDispatcher(), Delay, Closeable {
+class ExperimentalCoroutineDispatcher(
+    corePoolSize: Int = CORE_POOL_SIZE,
+    maxPoolSize: Int = MAX_POOL_SIZE
+) : CoroutineDispatcher(), Delay, Closeable {
 
     private val coroutineScheduler = CoroutineScheduler(corePoolSize, maxPoolSize)
 

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/ExperimentalCoroutineDispatcher.kt

@@ -1,10 +1,9 @@
 package kotlinx.coroutines.experimental.scheduling
 
+import kotlinx.coroutines.experimental.internal.*
 import java.util.concurrent.*
 
 
-internal typealias Task = TimedTask
-internal typealias GlobalQueue = TaskQueue
 // TODO most of these fields will be moved to 'object ExperimentalDispatcher'
 
 // 100us as default
@@ -21,12 +20,12 @@ internal val BLOCKING_DEFAULT_PARALLELISM = readFromSystemProperties(
         "kotlinx.coroutines.scheduler.blocking.parallelism", 16)
 
 @JvmField
-internal val CORE_POOL_SIZE = readFromSystemProperties(
-    "kotlinx.coroutines.scheduler.core.pool.size", Runtime.getRuntime().availableProcessors().coerceAtLeast(2))
+internal val CORE_POOL_SIZE = readFromSystemProperties("kotlinx.coroutines.scheduler.core.pool.size",
+    Runtime.getRuntime().availableProcessors().coerceAtLeast(2))
 
 @JvmField
-internal val MAX_POOL_SIZE = readFromSystemProperties(
-    "kotlinx.coroutines.scheduler.max.pool.size", Runtime.getRuntime().availableProcessors() * 128)
+internal val MAX_POOL_SIZE = readFromSystemProperties("kotlinx.coroutines.scheduler.max.pool.size",
+    (Runtime.getRuntime().availableProcessors() * 128).coerceIn(CORE_POOL_SIZE, CoroutineScheduler.MAX_SUPPORTED_POOL_SIZE))
 
 @JvmField
 internal val IDLE_WORKER_KEEP_ALIVE_NS = TimeUnit.SECONDS.toNanos(readFromSystemProperties(
@@ -42,7 +41,18 @@ internal enum class TaskMode {
     PROBABLY_BLOCKING,
 }
 
-internal data class TimedTask(val task: Runnable, val submissionTime: Long, val mode: TaskMode)
+internal data class Task(
+    val block: Runnable,
+    val submissionTime: Long,
+    val mode: TaskMode
+) : LockFreeMPMCQueueNode<Task>()
+
+// Open for tests
+internal open class GlobalQueue : LockFreeMPMCQueue<Task>() {
+    // Open for tests
+    public open fun removeFirstBlockingModeOrNull(): Task? =
+        removeFistOrNullIf { it.mode == TaskMode.PROBABLY_BLOCKING }
+}
 
 internal abstract class TimeSource {
     abstract fun nanoTime(): Long

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/LockFreeQueue.kt

@@ -8,22 +8,25 @@ internal const val BUFFER_CAPACITY = 1 shl BUFFER_CAPACITY_BASE
 internal const val MASK = BUFFER_CAPACITY - 1 // 128 by default
 
 /**
- * Unstable API and subject to change.
  * Tightly coupled with [CoroutineScheduler] queue of pending tasks, but extracted to separate file for simplicity.
- * At any moment queue is used only by [CoroutineScheduler.PoolWorker] threads, has only one producer (worker owning this queue)
+ * At any moment queue is used only by [CoroutineScheduler.Worker] threads, has only one producer (worker owning this queue)
  * and any amount of consumers, other pool workers which are trying to steal work.
  *
- * Fairness
+ * ### Fairness
+ *
  * [WorkQueue] provides semi-FIFO order, but with priority for most recently submitted task assuming
  * that these two (current one and submitted) are communicating and sharing state thus making such communication extremely fast.
  * E.g. submitted jobs [1, 2, 3, 4] will be executed in [4, 1, 2, 3] order.
  *
- * Work offloading
+ * ### Work offloading
+ * 
  * When the queue is full, half of existing tasks are offloaded to global queue which is regularly polled by other pool workers.
  * Offloading occurs in LIFO order for the sake of implementation simplicity: offloads should be extremely rare and occurs only in specific use-cases
  * (e.g. when coroutine starts heavy fork-join-like computation), so fairness is not important.
- * As an alternative, offloading directly to some [CoroutineScheduler.PoolWorker] may be used, but then the strategy of selecting any idle worker
+ * As an alternative, offloading directly to some [CoroutineScheduler.Worker] may be used, but then the strategy of selecting any idle worker
  * should be implemented and implementation should be aware multiple producers.
+ *
+ * @suppress **This is unstable API and it is subject to change.**
  */
 internal class WorkQueue {
 
@@ -54,9 +57,8 @@ internal class WorkQueue {
      * Retrieves and removes task from the head of the queue
      * Invariant: this method is called only by the owner of the queue ([pollExternal] is not)
      */
-    fun poll(): Task? {
-        return lastScheduledTask.getAndSet(null) ?: pollExternal()
-    }
+    fun poll(): Task? =
+        lastScheduledTask.getAndSet(null) ?: pollExternal()
 
     /**
      * Invariant: this method is called only by the owner of the queue
@@ -70,20 +72,18 @@ internal class WorkQueue {
         return addLast(previous, globalQueue)
     }
 
-    // Called only by the owner
+    // Called only by the owner, returns true if no offloading happened, false otherwise
     fun addLast(task: Task, globalQueue: GlobalQueue): Boolean {
-        var addedToGlobalQueue = false
-
+        var noOffloadingHappened = true
         /*
          * We need the loop here because race possible not only on full queue,
          * but also on queue with one element during stealing
          */
         while (!tryAddLast(task)) {
             offloadWork(globalQueue)
-            addedToGlobalQueue = true
+            noOffloadingHappened = false
         }
-
-        return !addedToGlobalQueue
+        return noOffloadingHappened
     }
 
     /**
@@ -135,7 +135,7 @@ internal class WorkQueue {
     private fun offloadWork(target: GlobalQueue) {
         repeat((bufferSize / 2).coerceAtLeast(1)) {
             val task = pollExternal() ?: return
-            target.add(task)
+            target.addLast(task)
         }
     }
 

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/Tasks.kt

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental.internal
+
+import kotlinx.coroutines.experimental.*
+import org.junit.Test
+import kotlin.test.*
+
+class LockFreeMPMCQueueTest : TestBase() {
+    @Test
+    fun testBasic() {
+        val q = LockFreeMPMCQueue<Node>()
+        assertEquals(null, q.removeFistOrNull())
+        assertTrue(q.isEmpty())
+        q.addLast(Node(1))
+        assertEquals(1, q.size)
+        assertEquals(Node(1), q.removeFistOrNull())
+        assertEquals(null, q.removeFistOrNull())
+        assertTrue(q.isEmpty())
+    }
+
+    private data class Node(val v: Int) : LockFreeMPMCQueueNode<Node>()
+}

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/scheduling/WorkQueue.kt

@@ -18,7 +18,7 @@ class CoroutineDispatcherTest : SchedulerTestBase() {
     fun testSingleThread() = runBlocking {
         expect(1)
         withContext(dispatcher) {
-            require(Thread.currentThread() is CoroutineScheduler.PoolWorker)
+            require(Thread.currentThread() is CoroutineScheduler.Worker)
             expect(2)
             val job = async(coroutineContext) {
                 expect(3)

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/internal/LockFreeMPMCQueueTest.kt

@@ -86,8 +86,8 @@ class CoroutineSchedulerTest : TestBase() {
 
     @Test
     fun testRngUniformDistribution() {
-        CoroutineScheduler(1).use { scheduler ->
-            val worker = scheduler.PoolWorker(1)
+        CoroutineScheduler(1, 128).use { scheduler ->
+            val worker = scheduler.Worker(1)
             testUniformDistribution(worker, 2)
             testUniformDistribution(worker, 4)
             testUniformDistribution(worker, 8)
@@ -111,7 +111,7 @@ class CoroutineSchedulerTest : TestBase() {
         ExperimentalCoroutineDispatcher(4, 1)
     }
 
-    private fun testUniformDistribution(worker: CoroutineScheduler.PoolWorker, bound: Int) {
+    private fun testUniformDistribution(worker: CoroutineScheduler.Worker, bound: Int) {
         val result = IntArray(bound)
         val iterations = 10_000_000
         repeat(iterations) {

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/scheduling/CoroutineDispatcherTest.kt

@@ -32,22 +32,22 @@ abstract class SchedulerTestBase : TestBase() {
          * Asserts that any number of pool worker threads in [range] exists at the time of method invocation
          */
         fun checkPoolThreadsExist(range: IntRange) {
-            val threads = Thread.getAllStackTraces().keys.filter { it is CoroutineScheduler.PoolWorker }.count()
+            val threads = Thread.getAllStackTraces().keys.filter { it is CoroutineScheduler.Worker }.count()
             require(threads in range) { "Expected threads in $range interval, but has $threads" }
         }
 
         /**
          * Asserts that [expectedThreadsCount] of pool worker threads exists at the time of method invocation
          */
         fun checkPoolThreadsExist(expectedThreadsCount: Int = CORES_COUNT) {
-            val threads = Thread.getAllStackTraces().keys.filter { it is CoroutineScheduler.PoolWorker }.count()
+            val threads = Thread.getAllStackTraces().keys.filter { it is CoroutineScheduler.Worker }.count()
             require(threads == expectedThreadsCount) { "Expected $expectedThreadsCount threads, but has $threads" }
         }
 
         fun initialPoolSize() = Runtime.getRuntime().availableProcessors().coerceAtMost(2)
 
         private fun maxSequenceNumber(): Int? {
-            return Thread.getAllStackTraces().keys.filter { it is CoroutineScheduler.PoolWorker }
+            return Thread.getAllStackTraces().keys.filter { it is CoroutineScheduler.Worker }
                 .map { sequenceNumber(it.name) }.max()
         }