Merge pull request #14 from ReactiveCocoa/anders/fix-deadlock

Fix a deadlock when `Loop.producer` is started recursively.

Author: andersio

Loop/Floodgate.swift

@@ -27,7 +27,8 @@ final class Floodgate<State, Event>: FeedbackEventConsumer<Event> {
         }
     }
 
-    private let reducerLock = NSLock()
+    private let reducerLock = RecursiveLock()
+
     private var state: State
     private var hasStarted = false
 
@@ -53,48 +54,60 @@ final class Floodgate<State, Event>: FeedbackEventConsumer<Event> {
             )
         }
 
-        reducerLock.perform {
+        reducerLock.perform { reentrant in
+            assert(reentrant == false)
             drainEvents()
         }
     }
 
     override func process(_ event: Event, for token: Token) {
         enqueue(event, for: token)
 
-        if reducerLock.try() {
-            repeat {
-                drainEvents()
-                reducerLock.unlock()
-            } while queue.withValue({ $0.hasEvents }) && reducerLock.try()
-            // ^^^
-            // Restart the event draining after we unlock the reducer lock, iff:
-            //
-            // 1. the queue still has unprocessed events; and
-            // 2. no concurrent actor has taken the reducer lock, which implies no event draining would be started
-            //    unless we take active action.
-            //
-            // This eliminates a race condition in the following sequence of operations:
-            //
-            // |              Thread A              |              Thread B              |
-            // |------------------------------------|------------------------------------|
-            // |     concurrent dequeue: no item    |                                    |
-            // |                                    |         concurrent enqueue         |
-            // |                                    |         trylock lock: BUSY         |
-            // |            unlock lock             |                                    |
-            // |                                    |                                    |
-            // |             <<<  The enqueued event is left unprocessed. >>>            |
+        var continueToDrain = false
+
+        repeat {
+            // We use a recursive lock to guard the reducer, so as to allow state access via `withValue` to be
+            // reentrant. But in order to not deadlock in ReactiveSwift, reentrant calls MUST NOT drain the queue.
+            // Otherwise, `consume(_:)` will eventually invoke the reducer and send out a state via `changeObserver`,
+            // leading to a deadlock.
             //
-            // The trylock-unlock duo has a synchronize-with relationship, which ensures that Thread A must see any
-            // concurrent enqueue that *happens before* the trylock.
-        }
+            // If we know that the call is reentrant, we can confidently skip draining the queue anyway, because the
+            // outmost call — the one who first acquires the lock on the current lock owner thread — is already looping
+            // to exhaustively drain the queue.
+            continueToDrain = reducerLock.tryPerform { isReentrant in
+                guard isReentrant == false else { return false }
+                drainEvents()
+                return true
+            }
+        } while queue.withValue({ $0.hasEvents }) && continueToDrain
+        // ^^^
+        // Restart the event draining after we unlock the reducer lock, iff:
+        //
+        // 1. the queue still has unprocessed events; and
+        // 2. no concurrent actor has taken the reducer lock, which implies no event draining would be started
+        //    unless we take active action.
+        //
+        // This eliminates a race condition in the following sequence of operations:
+        //
+        // |              Thread A              |              Thread B              |
+        // |------------------------------------|------------------------------------|
+        // |     concurrent dequeue: no item    |                                    |
+        // |                                    |         concurrent enqueue         |
+        // |                                    |         trylock lock: BUSY         |
+        // |            unlock lock             |                                    |
+        // |                                    |                                    |
+        // |             <<<  The enqueued event is left unprocessed. >>>            |
+        //
+        // The trylock-unlock duo has a synchronize-with relationship, which ensures that Thread A must see any
+        // concurrent enqueue that *happens before* the trylock.
     }
 
     override func dequeueAllEvents(for token: Token) {
         queue.modify { $0.events.removeAll(where: { _, t in t == token }) }
     }
 
     func withValue<Result>(_ action: (State, Bool) -> Result) -> Result {
-        reducerLock.perform { action(state, hasStarted) }
+        reducerLock.perform { _ in action(state, hasStarted) }
     }
 
     func dispose() {

Loop/NSLock+Extensions.swift

@@ -1,10 +1,42 @@
 import Foundation
 
-extension NSLock {
-    internal func perform<Result>(_ action: () -> Result) -> Result {
-        lock()
-        defer { unlock() }
+final class RecursiveLock {
+    let lock = NSRecursiveLock()
+    private var isLockAcquired = false
 
-        return action()
+    init() {}
+
+    /// Try to acquire the lock, perform the given action, and return whether or not the action is successful.
+    ///
+    /// - returns: The boolean flag indicating successfulness as returned by `action`. If the lock cannot be acquired,
+    ///            `false` is returned.
+    @inlinable
+    func tryPerform(_ action: (_ isReentrant: Bool) -> Bool) -> Bool {
+        if lock.try() {
+            defer { lock.unlock() }
+            return run(action)
+        }
+
+        return false
+    }
+
+    /// Acquire the lock, and perform the given action. If the lock is contested, wait until it is free.
+    ///
+    /// - returns: Pass through the value produced by `action`.
+    @inlinable
+    func perform<Result>(_ action: (_ isReentrant: Bool) -> Result) -> Result {
+        lock.lock()
+        defer { lock.unlock() }
+        return run(action)
+    }
+
+    private func run<Result>(_ action: (_ isReentrant: Bool) -> Result) -> Result {
+        if isLockAcquired {
+            return action(true)
+        } else {
+            isLockAcquired = true
+            defer { isLockAcquired = false }
+            return action(false)
+        }
     }
 }

LoopTests/FeedbackLoopSystemTests.swift

@@ -336,7 +336,43 @@ class FeedbackLoopSystemTests: XCTestCase {
 
         expect(results) == [0, 2, 1002, 2004, 4006]
     }
-    
+
+    func test_should_not_deadlock_when_feedback_effect_starts_loop_producer_synchronously() {
+        var _loop: Loop<Int, Int>!
+
+        let loop = Loop<Int, Int>(
+            initial: 0,
+            reducer: { $0 += $1 },
+            feedbacks: [
+                .init(
+                    skippingRepeated: { $0 == 1 },
+                    effects: { isOne in
+                        isOne
+                            ? _loop.producer.map(value: 1000).take(first: 1)
+                            : .empty
+                    }
+                )
+            ]
+        )
+        _loop = loop
+
+        var results: [Int] = []
+        loop.producer.startWithValues { results.append($0) }
+
+        expect(results) == [0]
+
+        func evaluate() {
+            loop.send(1)
+            expect(results) == [0, 1, 1001]
+        }
+
+        #if arch(x86_64)
+        expect(expression: evaluate).notTo(throwAssertion())
+        #else
+        evaluate()
+        #endif
+    }
+
     func test_events_are_produced_in_correct_order() {
         let (feedback, input) = Loop<Int, Int>.Feedback.input
         var events: [Int] = []