Fixed StackOverflow on Mutex.unlock convoy for Unconfined coroutines

Fixes #80

Author: elizarov

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

@@ -96,6 +96,15 @@ public interface Job : CoroutineContext.Element {
          */
         @Deprecated("Replaced with top-level function", level = DeprecationLevel.HIDDEN)
         public operator fun invoke(parent: Job? = null): Job = Job(parent)
+
+        init {
+            /*
+             * Here we make sure that CoroutineExceptionHandler is always initialized in advance, so
+             * that if a coroutine fails due to StackOverflowError we don't fail to report this error
+             * trying to initialize CoroutineExceptionHandler
+             */
+            CoroutineExceptionHandler
+        }
     }
 
     // ------------ state query ------------

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/sync/Mutex.kt

@@ -136,11 +136,19 @@ internal class MutexImpl(locked: Boolean) : Mutex {
     @Volatile
     private var _state: Any? = if (locked) EmptyLocked else EmptyUnlocked // shared objects while we have no waiters
 
+    // resumeNext is: RESUME_QUIESCENT | RESUME_ACTIVE | ResumeReq
+    @Volatile
+    private var resumeNext: Any = RESUME_QUIESCENT
+
     private companion object {
         @JvmField
         val STATE: AtomicReferenceFieldUpdater<MutexImpl, Any?> =
             AtomicReferenceFieldUpdater.newUpdater(MutexImpl::class.java, Any::class.java, "_state")
 
+        @JvmField
+        val RESUME_NEXT: AtomicReferenceFieldUpdater<MutexImpl, Any> =
+            AtomicReferenceFieldUpdater.newUpdater(MutexImpl::class.java, Any::class.java, "resumeNext")
+
         @JvmField
         val LOCK_FAIL = Symbol("LOCK_FAIL")
 
@@ -165,6 +173,11 @@ internal class MutexImpl(locked: Boolean) : Mutex {
         @JvmField
         val EmptyUnlocked = Empty(UNLOCKED)
 
+        @JvmField
+        val RESUME_QUIESCENT = Symbol("RESUME_QUIESCENT")
+
+        @JvmField
+        val RESUME_ACTIVE = Symbol("RESUME_ACTIVE")
     }
 
     public override val isLocked: Boolean get() {
@@ -347,9 +360,17 @@ internal class MutexImpl(locked: Boolean) : Mutex {
                         if (STATE.compareAndSet(this, state, op) && op.perform(this) == null) return
                     } else {
                         val token = (waiter as LockWaiter).tryResumeLockWaiter()
-                        if (token != null) { // successfully resumed waiter that now is holding the lock
+                        if (token != null) {
+                            // successfully resumed waiter that now is holding the lock
+                            // we must immediately transfer ownership to the next waiter, because this coroutine
+                            // might try to lock it again after unlock returns do to StackOverflow avoidance code
+                            // and its attempts to take a lock must be queued.
                             state.owner = waiter.owner ?: LOCKED
-                            waiter.completeResumeLockWaiter(token)
+                            // StackOverflow avoidance code
+                            if (startResumeNext(waiter, token)) {
+                                waiter.completeResumeLockWaiter(token)
+                                finishResumeNext()
+                            }
                             return
                         }
                     }
@@ -359,6 +380,44 @@ internal class MutexImpl(locked: Boolean) : Mutex {
         }
     }
 
+    private class ResumeReq(
+        @JvmField val waiter: LockWaiter,
+        @JvmField val token: Any
+    )
+
+    private fun startResumeNext(waiter: LockWaiter, token: Any): Boolean {
+        while (true) { // lock-free loop on resumeNext
+            val resumeNext = this.resumeNext
+            when {
+                resumeNext === RESUME_QUIESCENT -> {
+                    // this is never concurrent, because only one thread is holding mutex and trying to resume
+                    // next waiter, so no need to CAS here
+                    this.resumeNext = RESUME_ACTIVE
+                    return true
+                }
+                resumeNext === RESUME_ACTIVE ->
+                    if (RESUME_NEXT.compareAndSet(this, resumeNext, ResumeReq(waiter, token))) return false
+                else -> error("Cannot happen")
+            }
+        }
+    }
+
+    private fun finishResumeNext() {
+        while (true) { // lock-free loop on resumeNext, also a resumption loop to fulfill requests of inner resume invokes
+            val resumeNext = this.resumeNext
+            when {
+                resumeNext === RESUME_ACTIVE ->
+                    if (RESUME_NEXT.compareAndSet(this, resumeNext, RESUME_QUIESCENT)) return
+                resumeNext is ResumeReq -> {
+                    // this is never concurrently, only one thread is finishing, so no need to CAS here
+                    this.resumeNext = RESUME_ACTIVE
+                    resumeNext.waiter.completeResumeLockWaiter(resumeNext.token)
+                }
+                else -> error("Cannot happen")
+            }
+        }
+    }
+
     override fun toString(): String {
         while (true) {
             val state = this._state

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/sync/MutexTest.kt

@@ -16,8 +16,8 @@
 
 package kotlinx.coroutines.experimental.sync
 
-import guide.sync.example06.mutex
 import kotlinx.coroutines.experimental.*
+import org.hamcrest.core.IsEqual
 import org.junit.Assert.*
 import org.junit.Test
 
@@ -91,4 +91,20 @@ class MutexTest : TestBase() {
         println("Shared value = $shared")
         assertEquals(n * k, shared)
     }
+
+    @Test
+    fun testUnconfinedStackOverflow() {
+        val waiters = 10000
+        val mutex = Mutex(true)
+        var done = 0
+        repeat(waiters) {
+            launch(Unconfined) {  // a lot of unconfined waiters
+                mutex.withLock {
+                    done++
+                }
+            }
+        }
+        mutex.unlock() // should not produce StackOverflowError
+        assertThat(done, IsEqual(waiters))
+    }
 }