Merge pull request #123 from ReactiveCocoa/interrupt-fix

Mitigate race conditions in `Signal` interrupt handling.

Sources/Signal.swift

@@ -40,58 +40,113 @@ public final class Signal<Value, Error: Swift.Error> {
 	public init(_ generator: (Observer) -> Disposable?) {
 		state = Atomic(SignalState())
 
+		/// Holds the final signal state captured by an `interrupted` event. If it
+		/// is set, the Signal should interrupt as soon as possible. Implicitly
+		/// protected by `state` and `sendLock`.
+		var interruptedState: SignalState<Value, Error>? = nil
+
+		/// Used to track if the signal has terminated. Protected by `sendLock`.
+		var terminated = false
+
 		/// Used to ensure that events are serialized during delivery to observers.
 		let sendLock = NSLock()
 		sendLock.name = "org.reactivecocoa.ReactiveSwift.Signal"
 
-		/// When set to `true`, the Signal should interrupt as soon as possible.
-		let interrupted = Atomic(false)
-
 		let observer = Observer { [weak self] event in
 			guard let signal = self else {
 				return
 			}
 
-			func interrupt() {
-				if let state = signal.state.swap(nil) {
-					for observer in state.observers {
-						observer.sendInterrupted()
-					}
+			@inline(__always)
+			func interrupt(_ observers: Bag<Observer>) {
+				for observer in observers {
+					observer.sendInterrupted()
 				}
+				terminated = true
+				interruptedState = nil
 			}
 
 			if case .interrupted = event {
-				// Normally we disallow recursive events, but `interrupted` is
-				// kind of a special snowflake, since it can inadvertently be
-				// sent by downstream consumers.
+				// Recursive events are generally disallowed. But `interrupted` is kind
+				// of a special snowflake, since it can inadvertently be sent by
+				// downstream consumers.
 				//
-				// So we'll flag Interrupted events specially, and if it
-				// happened to occur while we're sending something else,  we'll
-				// wait to deliver it.
-				interrupted.value = true
-
-				if sendLock.try() {
-					interrupt()
-					sendLock.unlock()
-
-					signal.generatorDisposable?.dispose()
+				// So we would treat `interrupted` events specially. If it happens
+				// to occur while the `sendLock` is acquired, the observer call-out and
+				// the disposal would be delegated to the current sender, or
+				// occasionally one of the senders waiting on `sendLock`.
+				if let state = signal.state.swap(nil) {
+					// Writes to `interruptedState` are implicitly synchronized. So we do
+					// not need to guard it with locks.
+					//
+					// Specifically, senders serialized by `sendLock` can react to and
+					// clear `interruptedState` only if they see the write made below.
+					// The write can happen only once, since `state` being swapped with
+					// `nil` is a point of no return.
+					//
+					// Even in the case that both a previous sender and its successor see
+					// the write (the `interruptedState` check before & after the unlock
+					// of `sendLock`), the senders are still bound to the `sendLock`.
+					// So whichever sender loses the battle of acquring `sendLock` is
+					// guaranteed to be blocked.
+					interruptedState = state
+
+					if sendLock.try() {
+						if !terminated, let state = interruptedState {
+							interrupt(state.observers)
+						}
+						sendLock.unlock()
+						signal.generatorDisposable?.dispose()
+					}
 				}
 			} else {
-				if let state = (event.isTerminating ? signal.state.swap(nil) : signal.state.value) {
+				let isTerminating = event.isTerminating
+
+				if let observers = (isTerminating ? signal.state.swap(nil)?.observers : signal.state.value?.observers) {
+					var shouldDispose = false
+
 					sendLock.lock()
 
-					for observer in state.observers {
-						observer.action(event)
-					}
+					if !terminated {
+						for observer in observers {
+							observer.action(event)
+						}
+
+						// Check if a downstream consumer or a concurrent sender has
+						// interrupted the signal.
+						if !isTerminating, let state = interruptedState {
+							interrupt(state.observers)
+							shouldDispose = true
+						}
 
-					let shouldInterrupt = !event.isTerminating && interrupted.value
-					if shouldInterrupt {
-						interrupt()
+						if isTerminating {
+							terminated = true
+							shouldDispose = true
+						}
 					}
 
 					sendLock.unlock()
 
-					if event.isTerminating || shouldInterrupt {
+					// Based on the implicit memory order, any updates to the
+					// `interruptedState` should always be visible after `sendLock` is
+					// released. So we check it again and handle the interruption if
+					// it has not been taken over.
+					if !shouldDispose && !terminated && !isTerminating, let state = interruptedState {
+						sendLock.lock()
+
+						// `terminated` before acquring the lock could be a false negative,
+						// since it might race against other concurrent senders until the
+						// lock acquisition above succeeds. So we have to check again if the
+						// signal is really still alive.
+						if !terminated {
+							interrupt(state.observers)
+							shouldDispose = true
+						}
+
+						sendLock.unlock()
+					}
+
+					if shouldDispose {
 						// Dispose only after notifying observers, so disposal
 						// logic is consistently the last thing to run.
 						signal.generatorDisposable?.dispose()

Tests/ReactiveSwiftTests/SignalSpec.swift

@@ -7,6 +7,7 @@
 //
 
 import Foundation
+import Dispatch
 
 import Result
 import Nimble
@@ -213,6 +214,104 @@ class SignalSpec: QuickSpec {
 			}
 		}
 
+		describe("interruption") {
+			it("should not send events after sending an interrupted event") {
+				let queue: DispatchQueue
+				let counter = Atomic<Int>(0)
+
+				if #available(macOS 10.10, *) {
+					queue = DispatchQueue.global(qos: .userInitiated)
+				} else {
+					queue = DispatchQueue.global(priority: .high)
+				}
+
+				let (signal, observer) = Signal<Int, NoError>.pipe()
+
+				var hasSlept = false
+				var events: [Event<Int, NoError>] = []
+
+				// Used to synchronize the `interrupt` sender to only act after the
+				// chosen observer has started sending its event, but before it is done.
+				let semaphore = DispatchSemaphore(value: 0)
+
+				signal.observe { event in
+					if !hasSlept {
+						semaphore.signal()
+						// 100000 us = 0.1 s
+						usleep(100000)
+						hasSlept = true
+					}
+					events.append(event)
+				}
+
+				let group = DispatchGroup()
+
+				DispatchQueue.concurrentPerform(iterations: 10) { index in
+					queue.async(group: group) {
+						observer.send(value: index)
+					}
+
+					if index == 0 {
+						semaphore.wait()
+						observer.sendInterrupted()
+					}
+				}
+
+				group.wait()
+
+				expect(events.count) == 2
+
+				if events.count >= 2 {
+					expect(events[1].isTerminating) == true
+				}
+			}
+
+			it("should interrupt concurrently") {
+				let queue: DispatchQueue
+				let counter = Atomic<Int>(0)
+				let executionCounter = Atomic<Int>(0)
+
+				if #available(macOS 10.10, *) {
+					queue = DispatchQueue.global(qos: .userInitiated)
+				} else {
+					queue = DispatchQueue.global(priority: .high)
+				}
+
+				let iterations = 1000
+				let group = DispatchGroup()
+
+				queue.async(group: group) {
+					DispatchQueue.concurrentPerform(iterations: iterations) { _ in
+						let (signal, observer) = Signal<(), NoError>.pipe()
+
+						var isInterrupted = false
+						signal.observeInterrupted { counter.modify { $0 += 1 } }
+
+						// Used to synchronize the `value` sender and the `interrupt`
+						// sender, giving a slight priority to the former.
+						let semaphore = DispatchSemaphore(value: 0)
+
+						queue.async(group: group) {
+							semaphore.signal()
+							observer.send(value: ())
+							executionCounter.modify { $0 += 1 }
+						}
+
+						queue.async(group: group) {
+							semaphore.wait()
+							observer.sendInterrupted()
+							executionCounter.modify { $0 += 1 }
+						}
+					}
+				}
+
+				group.wait()
+
+				expect(executionCounter.value) == iterations * 2
+				expect(counter.value).toEventually(equal(iterations), timeout: 5)
+			}
+		}
+
 		describe("observe") {
 			var testScheduler: TestScheduler!