Basic Channel interfaces and RendezvousChannel implementation

Author: elizarov

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

@@ -20,15 +20,37 @@ public interface CancellableContinuation<in T> : Continuation<T>, Job {
      * Returns `true` if this continuation was cancelled. It implies that [isActive] is `false`.
      */
     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.
+     */
+    fun tryResume(value: T, onSuccess: ((Any?) -> Unit)? = null): Boolean
+
+    /**
+     * Makes this continuation cancellable. Use it with `holdCancellability` optional parameter to
+     * [suspendCancellableCoroutine] function. It throws [IllegalStateException] if invoked more than once.
+     */
+    fun initCancellability()
 }
 
 /**
  * Suspend coroutine similar to [suspendCoroutine], but provide an implementation of [CancellableContinuation] to
  * the [block]. This function throws [CancellationException] if the coroutine is cancelled while suspended.
+ *
+ * If [holdCancellability] optional parameter is `true`, then the coroutine is suspended, but it is not
+ * cancellable until [CancellableContinuation.initCancellability] is invoked.
  */
-public inline suspend fun <T> suspendCancellableCoroutine(crossinline block: (CancellableContinuation<T>) -> Unit): T =
+public inline suspend fun <T> suspendCancellableCoroutine(
+    holdCancellability: Boolean = false,
+    crossinline block: (CancellableContinuation<T>) -> Unit
+): T =
     suspendCoroutineOrReturn { cont ->
         val safe = SafeCancellableContinuation(cont, getParentJobOrAbort(cont))
+        if (!holdCancellability) safe.initCancellability()
         block(safe)
         safe.getResult()
     }
@@ -63,7 +85,9 @@ internal class SafeCancellableContinuation<in T>(
         const val YIELD = 3 // used by cancellable "yield"
     }
 
-    init { initParentJob(parentJob) }
+    override fun initCancellability() {
+        initParentJob(parentJob)
+    }
 
     fun getResult(): Any? {
         val decision = this.decision // volatile read
@@ -80,6 +104,16 @@ internal class SafeCancellableContinuation<in T>(
     override val isCancelled: Boolean
         get() = getState() is Cancelled
 
+    override fun tryResume(value: T, onSuccess: ((Any?) -> Unit)?): Boolean {
+        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
+            }
+        }
+    }
+
     @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,14 +213,16 @@ internal open class JobSupport : AbstractCoroutineContextElement(Job), Job {
     /**
      * Tries to update current [state][getState] of this job.
      */
-    fun updateState(expect: Any, update: Any?): Boolean {
+    fun updateState(expect: Any, update: Any?, onSuccess: ((Any?) -> Unit)? = null): 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. Invoke completion handlers
+        // #3. Additional (optional) callback
+        onSuccess?.invoke(update)
+        // #4. Invoke completion handlers
         val reason = (update as? CompletedExceptionally)?.cancelReason
         var completionException: Throwable? = null
         when (expect) {
@@ -242,7 +244,7 @@ internal open class JobSupport : AbstractCoroutineContextElement(Job), Job {
             // otherwise -- do nothing (Empty)
             else -> check(expect == Empty)
         }
-        // #4. Do other (overridable) processing after completion handlers
+        // #5. Do other (overridable) processing after completion handlers
         completionException?.let { handleCompletionException(it) }
         afterCompletion(update)
         return true

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Channel.kt

@@ -0,0 +1,167 @@
+package kotlinx.coroutines.experimental.channels
+
+import kotlinx.coroutines.experimental.CancellationException
+import kotlinx.coroutines.experimental.CoroutineScope
+import kotlinx.coroutines.experimental.Job
+import kotlinx.coroutines.experimental.yield
+import java.util.*
+
+/**
+ * Sender's interface to [Channel].
+ */
+public interface SendChannel<in E> {
+    /**
+     * Returns `true` if this channel was closed by invocation of [close] and thus
+     * the [send] attempt will throw [ClosedSendChannelException].
+     */
+    public val isClosedForSend: Boolean
+
+    /**
+     * Returns `true` if the channel is full (out of capacity) and the [send] attempt will suspend.
+     * This function returns `false` for [isClosedForSend] channel.
+     */
+    public val isFull: Boolean
+
+    /**
+     * Adds [element] into to this queue, suspending the caller while this queue [isFull],
+     * or throws [ClosedSendChannelException] if the channel [isClosedForSend].
+     *
+     * This suspending function is cancellable. If the [Job] of the current coroutine is completed while this
+     * function is suspended, this function immediately resumes with [CancellationException].
+     * Cancellation of suspended send is *atomic* -- when this function
+     * throws [CancellationException] it means that the [element] was not sent to this channel.
+     *
+     * Note, that this function does not check for cancellation when it is not suspended.
+     * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed.
+     */
+    public suspend fun send(element: E)
+
+    /**
+     * Adds [element] into this queue if it is possible to do so immediately without violating capacity restrictions
+     * and returns `true`. Otherwise, it returns `false` immediately
+     * or throws [ClosedSendChannelException] if the channel [isClosedForSend].
+     */
+    public fun offer(element: E): Boolean
+
+    /**
+     * Closes this channel. This is an idempotent operation -- repeated invocations of this function have no effect.
+     * Conceptually, its sends a special close token of this channel. Immediately after invocation of this function
+     * [isClosedForSend] starts returning `true`. However, [isClosedForReceive][ReceiveChannel.isClosedForReceive]
+     * on the side of [ReceiveChannel] starts returning `true` only after all previously sent elements
+     * are received.
+     */
+    public fun close()
+}
+
+/**
+ * Receiver's interface to [Channel].
+ */
+public interface ReceiveChannel<out E> {
+    /**
+     * Returns `true` if this channel was closed by invocation of [close][SendChannel.close] on the [SendChannel]
+     * side and all previously sent items were already received, so that the [receive] attempt will
+     * throw [ClosedReceiveChannelException].
+     */
+    public val isClosedForReceive: Boolean
+
+    /**
+     * Returns `true` if the channel is empty (contains no elements) and the [receive] attempt will suspend.
+     * This function returns `false` for [isClosedForReceive] channel.
+     */
+    public val isEmpty: Boolean
+
+    /**
+     * Retrieves and removes the element from this channel suspending the caller while this channel [isEmpty]
+     * or throws [ClosedReceiveChannelException] if the channel [isClosedForReceive].
+     *
+     * This suspending function is cancellable. If the [Job] of the current coroutine is completed while this
+     * function is suspended, this function immediately resumes with [CancellationException].
+     * Cancellation of suspended receive is *atomic* -- when this function
+     * throws [CancellationException] it means that the element was not retrieved from this channel.
+     *
+     * Note, that this function does not check for cancellation when it is not suspended.
+     * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed.
+     */
+    public suspend fun receive(): E
+
+    /**
+     * Retrieves and removes the element from this channel suspending the caller while this channel [isEmpty]
+     * or returns `null` if the channel [isClosedForReceive].
+     *
+     * This suspending function is cancellable. If the [Job] of the current coroutine is completed while this
+     * function is suspended, this function immediately resumes with [CancellationException].
+     * Cancellation of suspended receive is *atomic* -- when this function
+     * throws [CancellationException] it means that the element was not retrieved from this channel.
+     *
+     * Note, that this function does not check for cancellation when it is not suspended.
+     * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed.
+     */
+    public suspend fun receiveOrNull(): E?
+
+    /**
+     * Retrieves and removes the head of this queue, or returns `null` if this queue [isEmpty]
+     * or [isClosedForReceive].
+     */
+    public fun pool(): E?
+
+    /**
+     * Returns new iterator to receive elements from this channels using `for` loop.
+     */
+    public operator fun iterator(): ChannelIterator<E>
+}
+
+/**
+ * Iterator for [ReceiveChannel]. Instances of this interface are *not thread-safe* and shall not be used
+ * from concurrent coroutines.
+ */
+public interface ChannelIterator<out E> {
+    /**
+     * Returns `true` if the channel has more elements suspending the caller while this channel
+     * [isEmpty][ReceiveChannel.isEmpty] or `false` [ClosedReceiveChannelException] if the channel
+     * [isClosedForReceive][ReceiveChannel.isClosedForReceive].
+     * This function retrieves and removes the element from this channel for the subsequent invocation
+     * of [next].
+     *
+     * This suspending function is cancellable. If the [Job] of the current coroutine is completed while this
+     * function is suspended, this function immediately resumes with [CancellationException].
+     * Cancellation of suspended receive is *atomic* -- when this function
+     * throws [CancellationException] it means that the element was not retrieved from this channel.
+     *
+     * Note, that this function does not check for cancellation when it is not suspended.
+     * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed.
+     */
+    public suspend operator fun hasNext(): Boolean
+
+    /**
+     * Retrieves and removes the element from this channel suspending the caller while this channel
+     * [isEmpty][ReceiveChannel.isEmpty] or throws [ClosedReceiveChannelException] if the channel
+     * [isClosedForReceive][ReceiveChannel.isClosedForReceive].
+     *
+     * This suspending function is cancellable. If the [Job] of the current coroutine is completed while this
+     * function is suspended, this function immediately resumes with [CancellationException].
+     * Cancellation of suspended receive is *atomic* -- when this function
+     * throws [CancellationException] it means that the element was not retrieved from this channel.
+     *
+     * Note, that this function does not check for cancellation when it is not suspended.
+     * Use [yield] or [CoroutineScope.isActive] to periodically check for cancellation in tight loops if needed.
+     */
+    public suspend operator fun next(): E
+}
+
+/**
+ * Channel is a non-blocking primitive for communication between sender using [SendChannel] and receiver using [ReceiveChannel].
+ * Conceptually, a channel is similar to [BlockingQueue][java.util.concurrent.BlockingQueue],
+ * but it has suspending operations instead of blocking ones and it can be closed.
+ */
+public interface Channel<E> : SendChannel<E>, ReceiveChannel<E>
+
+/**
+ * Indicates attempt to [send][SendChannel.send] on [isClosedForSend][SendChannel.isClosedForSend] channel.
+ */
+public class ClosedSendChannelException : IllegalStateException()
+
+/**
+ * Indicates attempt to [receive][ReceiveChannel.receive] on [isClosedForReceive][ReceiveChannel.isClosedForReceive]
+ * channel.
+ */
+public class ClosedReceiveChannelException : NoSuchElementException()