Implemented "onLock" clause for Mutex.lock and added optional "owner" parameter to all Mutex funs

Author: elizarov

kotlinx-coroutines-core/README.md

@@ -47,6 +47,7 @@ Top-level suspending functions:
 | [SendChannel][kotlinx.coroutines.experimental.channels.SendChannel]    | [send][kotlinx.coroutines.experimental.channels.SendChannel.send]                      | [onSend][kotlinx.coroutines.experimental.selects.SelectBuilder.onSend]                   | [offer][kotlinx.coroutines.experimental.channels.SendChannel.offer]
 | [ReceiveChannel][kotlinx.coroutines.experimental.channels.ReceiveChannel] | [receive][kotlinx.coroutines.experimental.channels.ReceiveChannel.receive]             | [onReceive][kotlinx.coroutines.experimental.selects.SelectBuilder.onReceive]             | [poll][kotlinx.coroutines.experimental.channels.ReceiveChannel.poll]
 | [ReceiveChannel][kotlinx.coroutines.experimental.channels.ReceiveChannel] | [receiveOrNull][kotlinx.coroutines.experimental.channels.ReceiveChannel.receiveOrNull] | [onReceiveOrNull][kotlinx.coroutines.experimental.selects.SelectBuilder.onReceiveOrNull] | [poll][kotlinx.coroutines.experimental.channels.ReceiveChannel.poll]
+| [Mutex][kotlinx.coroutines.experimental.sync.Mutex]          | [lock][kotlinx.coroutines.experimental.sync.Mutex.lock]                            | [onLock][kotlinx.coroutines.experimental.selects.SelectBuilder.onLock]                   | [tryLock][kotlinx.coroutines.experimental.sync.Mutex.tryLock]
 
 Cancellation support for user-defined suspending functions is available with [suspendCancellableCoroutine]
 helper function. [NonCancellable] job object is provided to suppress cancellation with 
@@ -100,6 +101,7 @@ Select expression to perform multiple suspending operations simultaneously until
 <!--- INDEX kotlinx.coroutines.experimental.sync -->
 [kotlinx.coroutines.experimental.sync.Mutex]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.sync/-mutex/index.html
 [kotlinx.coroutines.experimental.sync.Mutex.lock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.sync/lock.html
+[kotlinx.coroutines.experimental.sync.Mutex.tryLock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.sync/try-lock.html
 <!--- INDEX kotlinx.coroutines.experimental.channels -->
 [kotlinx.coroutines.experimental.channels.produce]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/produce.html
 [kotlinx.coroutines.experimental.channels.ProducerJob]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-producer-job/index.html
@@ -122,4 +124,5 @@ Select expression to perform multiple suspending operations simultaneously until
 [kotlinx.coroutines.experimental.selects.SelectBuilder.onSend]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/on-send.html
 [kotlinx.coroutines.experimental.selects.SelectBuilder.onReceive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/on-receive.html
 [kotlinx.coroutines.experimental.selects.SelectBuilder.onReceiveOrNull]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/on-receive-or-null.html
+[kotlinx.coroutines.experimental.selects.SelectBuilder.onLock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/on-lock.html
 <!--- END -->

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

@@ -764,14 +764,6 @@ internal class UnregisterOnCompletion(
     override fun toString(): String = "UnregisterOnCompletion[$registration]"
 }
 
-private class CancelOnCompletion(
-    parentJob: Job,
-    @JvmField val subordinateJob: Job
-) : JobNode<Job>(parentJob) {
-    override fun invoke(reason: Throwable?) { subordinateJob.cancel(reason) }
-    override fun toString(): String = "CancelOnCompletion[$subordinateJob]"
-}
-
 private class ParentOnCompletion(
     parentJob: Job,
     @JvmField val subordinateJob: JobSupport
@@ -804,8 +796,6 @@ private class SelectJoinOnCompletion<R>(
     override fun toString(): String = "SelectJoinOnCompletion[$select]"
 }
 
-
-
 private class JobImpl(parent: Job? = null) : JobSupport(true) {
     init { initParentJob(parent) }
 }

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

@@ -192,7 +192,7 @@ public abstract class AbstractChannel<E> : Channel<E> {
         element: E,
         select: SelectInstance<R>,
         block: suspend () -> R
-    ) : AddLastDesc(queue, SendSelect(element, select, block)) {
+    ) : AddLastDesc<SendSelect<R>>(queue, SendSelect(element, select, block)) {
         override fun failure(affected: LockFreeLinkedListNode, next: Any): Any? {
             if (affected is ReceiveOrClosed<*>) {
                 return affected as? Closed<*> ?: ENQUEUE_FAILED
@@ -370,11 +370,11 @@ public abstract class AbstractChannel<E> : Channel<E> {
         }
     }
 
-    private inner class TryEnqueueReceiveDesc<E, R>(
+    private inner class TryEnqueueReceiveDesc<in E, R>(
         select: SelectInstance<R>,
         block: suspend (E?) -> R,
         nullOnClose: Boolean
-    ) : AddLastDesc(queue, ReceiveSelect(select, block, nullOnClose)) {
+    ) : AddLastDesc<ReceiveSelect<R, E>>(queue, ReceiveSelect(select, block, nullOnClose)) {
         override fun failure(affected: LockFreeLinkedListNode, next: Any): Any? {
             if (affected is Send) return ENQUEUE_FAILED
             return null

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

@@ -48,7 +48,7 @@ public typealias RemoveFirstDesc<T> = LockFreeLinkedListNode.RemoveFirstDesc<T>
 /**
  * @suppress **This is unstable API and it is subject to change.**
  */
-public typealias AddLastDesc = LockFreeLinkedListNode.AddLastDesc
+public typealias AddLastDesc<T> = LockFreeLinkedListNode.AddLastDesc<T>
 
 /**
  * Doubly-linked concurrent list node with remove support.
@@ -160,7 +160,7 @@ public open class LockFreeLinkedListNode {
         }
     }
 
-    public fun describeAddLast(node: Node): AddLastDesc = AddLastDesc(this, node)
+    public fun <T : Node> describeAddLast(node: T): AddLastDesc<T> = AddLastDesc(this, node)
 
     /**
      * Adds last item to this list atomically if the [condition] is true.
@@ -293,7 +293,10 @@ public open class LockFreeLinkedListNode {
 
     // ------ multi-word atomic operations helpers ------
 
-    public open class AddLastDesc(val queue: Node, val node: Node) : AbstractAtomicDesc() {
+    public open class AddLastDesc<out T : Node>(
+        @JvmField val queue: Node,
+        @JvmField val node: T
+    ) : AbstractAtomicDesc() {
         init {
             // require freshly allocated node here
             check(node._next === node && node._prev === node)
@@ -337,7 +340,9 @@ public open class LockFreeLinkedListNode {
         }
     }
 
-    public open class RemoveFirstDesc<T>(val queue: Node) : AbstractAtomicDesc() {
+    public open class RemoveFirstDesc<T>(
+        @JvmField val queue: Node
+    ) : AbstractAtomicDesc() {
         @Suppress("UNCHECKED_CAST")
         public val result: T get() = affectedNode!! as T
 

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/Select.kt

@@ -22,6 +22,8 @@ import kotlinx.coroutines.experimental.channels.ClosedSendChannelException
 import kotlinx.coroutines.experimental.channels.ReceiveChannel
 import kotlinx.coroutines.experimental.channels.SendChannel
 import kotlinx.coroutines.experimental.internal.AtomicDesc
+import kotlinx.coroutines.experimental.internal.LockFreeLinkedListNode
+import kotlinx.coroutines.experimental.sync.Mutex
 import kotlin.coroutines.experimental.Continuation
 import kotlin.coroutines.experimental.CoroutineContext
 import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
@@ -66,6 +68,14 @@ public interface SelectBuilder<in R> {
      * the original [close][SendChannel.close] cause exception if the channel has _failed_.
      */
     public fun <E> ReceiveChannel<E>.onReceiveOrNull(block: suspend (E?) -> R)
+
+    /**
+     * Clause for [Mutex.lock] suspending function that selects the given [block] when the mutex is locked.
+     *
+     * @param owner Optional owner token for debugging. When `owner` is specified (non-null value) and this mutex
+     *        is already locked with the same token (same identity), this clause throws [IllegalStateException].
+     */
+    public fun Mutex.onLock(owner: Any? = null, block: suspend () -> R)
 }
 
 /**
@@ -107,6 +117,8 @@ public interface SelectInstance<in R> {
     public fun invokeOnCompletion(handler: CompletionHandler): Job.Registration
 
     public fun unregisterOnCompletion(registration: Job.Registration)
+
+    public fun removeOnCompletion(node: LockFreeLinkedListNode)
 }
 
 /**
@@ -133,6 +145,7 @@ public interface SelectInstance<in R> {
  * | [SendChannel]    | [send][SendChannel.send]                      | [onSend][SelectBuilder.onSend]                   | [offer][SendChannel.offer]
  * | [ReceiveChannel] | [receive][ReceiveChannel.receive]             | [onReceive][SelectBuilder.onReceive]             | [poll][ReceiveChannel.poll]
  * | [ReceiveChannel] | [receiveOrNull][ReceiveChannel.receiveOrNull] | [onReceiveOrNull][SelectBuilder.onReceiveOrNull] | [poll][ReceiveChannel.poll]
+ * | [Mutex]          | [lock][Mutex.lock]                            | [onLock][SelectBuilder.onLock]                   | [tryLock][Mutex.tryLock]
  *
  * 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].
@@ -224,7 +237,23 @@ internal class SelectBuilderImpl<in R>(
         registerSelectReceiveOrNull(this@SelectBuilderImpl, block)
     }
 
+    override fun Mutex.onLock(owner: Any?, block: suspend () -> R) {
+        registerSelectLock(this@SelectBuilderImpl, owner, block)
+    }
+
     override fun unregisterOnCompletion(registration: Job.Registration) {
         invokeOnCompletion(UnregisterOnCompletion(this, registration))
     }
+
+    override fun removeOnCompletion(node: LockFreeLinkedListNode) {
+        invokeOnCompletion(RemoveOnCompletion(this, node))
+    }
+}
+
+private class RemoveOnCompletion(
+    select: SelectBuilderImpl<*>,
+    @JvmField val node: LockFreeLinkedListNode
+) : JobNode<SelectBuilderImpl<*>>(select) {
+    override fun invoke(reason: Throwable?) { node.remove() }
+    override fun toString(): String = "RemoveOnCompletion[$node]"
 }

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/SelectUnbiased.kt

@@ -20,6 +20,7 @@ import kotlinx.coroutines.experimental.Deferred
 import kotlinx.coroutines.experimental.Job
 import kotlinx.coroutines.experimental.channels.ReceiveChannel
 import kotlinx.coroutines.experimental.channels.SendChannel
+import kotlinx.coroutines.experimental.sync.Mutex
 import java.util.*
 import kotlin.coroutines.experimental.Continuation
 import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
@@ -86,4 +87,8 @@ internal class UnbiasedSelectBuilderImpl<in R>(cont: Continuation<R>) : SelectBu
     override fun <E> ReceiveChannel<E>.onReceiveOrNull(block: suspend (E?) -> R) {
         clauses += { registerSelectReceiveOrNull(instance, block) }
     }
+
+    override fun Mutex.onLock(owner: Any?, block: suspend () -> R) {
+        clauses += { registerSelectLock(instance, owner, block) }
+    }
 }