ArrayChannel implementation and tests

Author: elizarov

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/CancellableContinuation.kt

@@ -22,13 +22,16 @@ public interface CancellableContinuation<in T> : Continuation<T>, Job {
     val isCancelled: Boolean
 
     /**
-     * Tries to resume this continuation with a given value and returns `true` if it was successful,
-     * or `false` otherwise (it was already resumed or cancelled).
-     *
-     * An optional [onSuccess] callback is invoked with [value] as its parameter after the state of this continuation
-     * is updated (so that is cannot be cancelled anymore), but before it is actually resumed.
+     * Tries to resume this continuation with a given value and returns non-null object token if it was successful,
+     * or `null` otherwise (it was already resumed or cancelled). When non-null object was returned,
+     * [completeResume] must be invoked with it.
      */
-    fun tryResume(value: T, onSuccess: ((Any?) -> Unit)? = null): Boolean
+    fun tryResume(value: T): Any?
+
+    /**
+     * Completes the execution of [tryResume] on its non-null result.
+     */
+    fun completeResume(token: Any)
 
     /**
      * Makes this continuation cancellable. Use it with `holdCancellability` optional parameter to
@@ -104,16 +107,20 @@ internal class SafeCancellableContinuation<in T>(
     override val isCancelled: Boolean
         get() = getState() is Cancelled
 
-    override fun tryResume(value: T, onSuccess: ((Any?) -> Unit)?): Boolean {
+    override fun tryResume(value: T): Any? {
         while (true) { // lock-free loop on state
             val state = getState() // atomic read
             when (state) {
-                is Active -> if (updateState(state, value, onSuccess)) return true
-                else -> return false // cannot resume -- not active anymore
+                is Active -> if (tryUpdateState(state, value)) return state
+                else -> return null // cannot resume -- not active anymore
             }
         }
     }
 
+    override fun completeResume(token: Any) {
+        completeUpdateState(token, getState())
+    }
+
     @Suppress("UNCHECKED_CAST")
     override fun afterCompletion(state: Any?) {
         val decision = this.decision // volatile read

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Job.kt

@@ -213,16 +213,24 @@ internal open class JobSupport : AbstractCoroutineContextElement(Job), Job {
     /**
      * Tries to update current [state][getState] of this job.
      */
-    fun updateState(expect: Any, update: Any?, onSuccess: ((Any?) -> Unit)? = null): Boolean {
+    fun updateState(expect: Any, update: Any?): Boolean {
+        if (!tryUpdateState(expect, update)) return false
+        completeUpdateState(expect, update)
+        return true
+    }
+
+    fun tryUpdateState(expect: Any, update: Any?): Boolean  {
         require(expect is Active && update !is Active) // only active -> inactive transition is allowed
         if (!STATE.compareAndSet(this, expect, update)) return false
         // #1. Update linked state before invoking completion handlers
         onStateUpdate(update)
         // #2. Unregister from parent job
         registration?.unregister() // volatile read registration _after_ state was updated
-        // #3. Additional (optional) callback
-        onSuccess?.invoke(update)
-        // #4. Invoke completion handlers
+        return true // continues in completeUpdateState
+    }
+
+    fun completeUpdateState(expect: Any, update: Any?) {
+        // #3. Invoke completion handlers
         val reason = (update as? CompletedExceptionally)?.cancelReason
         var completionException: Throwable? = null
         when (expect) {
@@ -244,10 +252,9 @@ internal open class JobSupport : AbstractCoroutineContextElement(Job), Job {
             // otherwise -- do nothing (Empty)
             else -> check(expect == Empty)
         }
-        // #5. Do other (overridable) processing after completion handlers
+        // #4. Do other (overridable) processing after completion handlers
         completionException?.let { handleCompletionException(it) }
         afterCompletion(update)
-        return true
     }
 
     final override val isActive: Boolean get() = state is Active