Merge pull request #741 from Kotlin/array-queue

Use array-based queue for Scheduler.GlobalQueue & optimize Task

Author: qwwdfsad

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

@@ -18,8 +18,8 @@ private const val RESUMED = 2
  */
 internal abstract class AbstractContinuation<in T>(
     public final override val delegate: Continuation<T>,
-    public final override val resumeMode: Int
-) : Continuation<T>, DispatchedTask<T> {
+    resumeMode: Int
+) : DispatchedTask<T>(resumeMode), Continuation<T> {
 
     /*
      * Implementation notes

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

@@ -81,11 +81,10 @@ internal object UndispatchedEventLoop {
 internal class DispatchedContinuation<in T>(
     @JvmField val dispatcher: CoroutineDispatcher,
     @JvmField val continuation: Continuation<T>
-) : Continuation<T> by continuation, DispatchedTask<T> {
+) : DispatchedTask<T>(MODE_ATOMIC_DEFAULT), Continuation<T> by continuation {
     @JvmField
     @Suppress("PropertyName")
     internal var _state: Any? = UNDEFINED
-    public override var resumeMode: Int = 0
     @JvmField // pre-cached value to avoid ctx.fold on every resumption
     internal val countOrElement = threadContextElements(context)
 
@@ -204,20 +203,22 @@ internal fun <T> Continuation<T>.resumeDirectWithException(exception: Throwable)
     else -> resumeWithException(exception)
 }
 
-internal interface DispatchedTask<in T> : Runnable {
-    public val delegate: Continuation<T>
-    public val resumeMode: Int get() = MODE_CANCELLABLE
+internal abstract class DispatchedTask<in T>(
+    @JvmField var resumeMode: Int
+) : SchedulerTask() {
+    public abstract val delegate: Continuation<T>
 
-    public fun takeState(): Any?
+    public abstract fun takeState(): Any?
 
     @Suppress("UNCHECKED_CAST")
-    public fun <T> getSuccessfulResult(state: Any?): T =
+    public open fun <T> getSuccessfulResult(state: Any?): T =
         state as T
 
     public fun getExceptionalResult(state: Any?): Throwable? =
         (state as? CompletedExceptionally)?.cause
 
-    public override fun run() {
+    public final override fun run() {
+        val taskContext = this.taskContext
         try {
             val delegate = delegate as DispatchedContinuation<T>
             val continuation = delegate.continuation
@@ -237,6 +238,8 @@ internal interface DispatchedTask<in T> : Runnable {
             }
         } catch (e: Throwable) {
             throw DispatchException("Unexpected exception running $this", e)
+        } finally {
+            taskContext.afterTask()
         }
     }
 }

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

@@ -0,0 +1,13 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+internal expect abstract class SchedulerTask() : Runnable
+
+internal expect interface SchedulerTaskContext
+
+internal expect val SchedulerTask.taskContext: SchedulerTaskContext
+
+internal expect inline fun SchedulerTaskContext.afterTask()

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

@@ -51,7 +51,7 @@ internal fun delayNanosToMillis(timeNanos: Long): Long =
 @Suppress("PrivatePropertyName")
 private val CLOSED_EMPTY = Symbol("CLOSED_EMPTY")
 
-private typealias Queue<T> = LockFreeMPSCQueueCore<T>
+private typealias Queue<T> = LockFreeTaskQueueCore<T>
 
 internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
     // null | CLOSED_EMPTY | task | Queue<Runnable>
@@ -150,7 +150,7 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
                     queue === CLOSED_EMPTY -> return false
                     else -> {
                         // update to full-blown queue to add one more
-                        val newQueue = Queue<Runnable>(Queue.INITIAL_CAPACITY)
+                        val newQueue = Queue<Runnable>(Queue.INITIAL_CAPACITY, singleConsumer = true)
                         newQueue.addLast(queue as Runnable)
                         newQueue.addLast(task)
                         if (_queue.compareAndSet(queue, newQueue)) return true
@@ -191,7 +191,7 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
                     queue === CLOSED_EMPTY -> return
                     else -> {
                         // update to full-blown queue to close
-                        val newQueue = Queue<Runnable>(Queue.INITIAL_CAPACITY)
+                        val newQueue = Queue<Runnable>(Queue.INITIAL_CAPACITY, singleConsumer = true)
                         newQueue.addLast(queue as Runnable)
                         if (_queue.compareAndSet(queue, newQueue)) return
                     }

core/kotlinx-coroutines-core/src/SchedulerTask.kt

@@ -0,0 +1,18 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlinx.coroutines.scheduling.*
+
+internal actual typealias SchedulerTask = Task
+
+internal actual typealias SchedulerTaskContext = TaskContext
+
+@Suppress("EXTENSION_SHADOWED_BY_MEMBER")
+internal actual val SchedulerTask.taskContext: SchedulerTaskContext get() = taskContext
+
+@Suppress("NOTHING_TO_INLINE", "EXTENSION_SHADOWED_BY_MEMBER")
+internal actual inline fun SchedulerTaskContext.afterTask() =
+    afterTask()