Unserialized and Reentrant Signals (#797)

* Offer Signal variants for advanced users.

* Switch `flatMapEvent` to use `nonSerializing`.

* Test coverage.

* Changelog

* Expand the changelog.

* Fix indentation.

* Extend `nonSerializing()` to properties and producers too.

* Make `on()` and `withLatest` non-serializing

* Signals made by Producer can be non-serializing too

* NonSerializing -> Unserialized

* Remove use of @_specialize

* reentrant -> reentrantUnserialized

* Update changelog

* Revive @_specialize

Author: andersio

CHANGELOG.md

@@ -1,6 +1,20 @@
 # master
 *Please add new entries at the top.*
 
+1. `Signal` offers two special variants for advanced users: unserialized and reentrant-unserialized. (#797)
+
+   The input observer of these variants assume that mutual exclusion has been enforced by its callers.
+
+   You can create these variants through four `Signal` static methods: `unserialized(_:)`, `unserializedPipe(_:)`, `reentrantUnserialized(_:)` and `reentrantUnserializedPipe(_:)`. These would be adopted by ReactiveCocoa UIKit bindings to improve interoperability with Loop, to tackle some legitimate recursive delivery scenarios (e.g. around first responder management), and also to reduce fine-grained locking in ReactiveCocoa.
+
+   Note that the default behavior of `Signal` has not been changed — event serialization remains the default behavior.
+
+1. `SignalProducer` offers an unserialized variant via `SignalProducer.unserialized(_:)`. (#797)
+
+1. `Signal` and Properties now use fewer locks, which should translate into minor performance improvements. (#797)
+
+1. Added the `interval` operator (#810, kudos to @mluisbrown)
+
 1. `TestScheduler` can `advance` by `TimeInterval`. (#828)
 
 1. Fixed spelling error in `Token` class documentation.
@@ -19,6 +33,7 @@
 1. Bumped deployment target to iOS 9.0, per Xcode 12 warnings. (#818, kudos to @harleyjcooper)
 
 1. Fixed a few deprecation warning when the project is being built. (#819, kudos to @apps4everyone)
+>>>>>>> origin/master
 
 # 6.5.0
 

Sources/Atomic.swift

@@ -104,7 +104,7 @@ internal struct UnsafeAtomicState<State: RawRepresentable> where State.RawValue
 
 /// `Lock` exposes `os_unfair_lock` on supported platforms, with pthread mutex as the
 /// fallback.
-internal class Lock {
+internal class Lock: LockProtocol {
 	#if os(macOS) || os(iOS) || os(tvOS) || os(watchOS)
 	@available(iOS 10.0, *)
 	@available(macOS 10.12, *)
@@ -195,14 +195,14 @@ internal class Lock {
 		}
 	}
 
-	static func make() -> Lock {
+	static func make() -> Self {
 		#if os(macOS) || os(iOS) || os(tvOS) || os(watchOS)
 		if #available(*, iOS 10.0, macOS 10.12, tvOS 10.0, watchOS 3.0) {
-			return UnfairLock()
+			return UnfairLock() as! Self
 		}
 		#endif
 
-		return PthreadLock()
+		return PthreadLock() as! Self
 	}
 
 	private init() {}
@@ -212,6 +212,22 @@ internal class Lock {
 	func `try`() -> Bool { fatalError() }
 }
 
+internal protocol LockProtocol {
+	static func make() -> Self
+	
+	func lock()
+	func unlock()
+	func `try`() -> Bool
+}
+
+internal struct NoLock: LockProtocol {
+	static func make() -> NoLock { NoLock() }
+
+	func lock() {}
+	func unlock() {}
+	func `try`() -> Bool { true }
+}
+
 /// An atomic variable.
 public final class Atomic<Value> {
 	private let lock: Lock

Sources/Property.swift

@@ -85,6 +85,11 @@ extension PropertyProtocol {
 		return Property(unsafeProducer: transform(producer))
 	}
 
+	/// Lifts a unary SignalProducer operator to operate upon PropertyProtocol instead.
+	fileprivate func liftUnserialized<U>(_ transform: @escaping (SignalProducer<Value, Never>) -> SignalProducer<U, Never>) -> Property<U> {
+		return Property(unsafeProducer: transform(producer), unserialized: true)
+	}
+
 	/// Lifts a binary SignalProducer operator to operate upon PropertyProtocol instead.
 	fileprivate func lift<P: PropertyProtocol, U>(_ transform: @escaping (SignalProducer<Value, Never>) -> (SignalProducer<P.Value, Never>) -> SignalProducer<U, Never>) -> (P) -> Property<U> {
 		return { other in
@@ -102,7 +107,7 @@ extension PropertyProtocol {
 	///
 	/// - returns: A property that holds a mapped value from `self`.
 	public func map<U>(_ transform: @escaping (Value) -> U) -> Property<U> {
-		return lift { $0.map(transform) }
+		return liftUnserialized { $0.map(transform) }
 	}
 
 	/// Map the current value and all susequent values to a new constant property.
@@ -112,7 +117,7 @@ extension PropertyProtocol {
 	///
 	/// - returns: A property that holds a mapped value from `self`.
 	public func map<U>(value: U) -> Property<U> {
-		return lift { $0.map(value: value) }
+		return liftUnserialized { $0.map(value: value) }
 	}
 
 	/// Maps the current value and all subsequent values to a new property
@@ -123,7 +128,7 @@ extension PropertyProtocol {
 	///
 	/// - returns: A property that holds a mapped value from `self`.
 	public func map<U>(_ keyPath: KeyPath<Value, U>) -> Property<U> {
-		return lift { $0.map(keyPath) }
+		return liftUnserialized { $0.map(keyPath) }
 	}
 
 	/// Passes only the values of the property that pass the given predicate
@@ -176,7 +181,7 @@ extension PropertyProtocol {
 	/// - returns: A property that holds tuples that contain previous and
 	///            current values of `self`.
 	public func combinePrevious(_ initial: Value) -> Property<(Value, Value)> {
-		return lift { $0.combinePrevious(initial) }
+		return liftUnserialized { $0.combinePrevious(initial) }
 	}
 
 	/// Forward only values from `self` that are not considered equivalent to its
@@ -189,7 +194,7 @@ extension PropertyProtocol {
 	///
 	/// - returns: A property which conditionally forwards values from `self`.
 	public func skipRepeats(_ isEquivalent: @escaping (Value, Value) -> Bool) -> Property<Value> {
-		return lift { $0.skipRepeats(isEquivalent) }
+		return liftUnserialized { $0.skipRepeats(isEquivalent) }
 	}
 }
 
@@ -200,7 +205,7 @@ extension PropertyProtocol where Value: Equatable {
 	///
 	/// - returns: A property which conditionally forwards values from `self`.
 	public func skipRepeats() -> Property<Value> {
-		return lift { $0.skipRepeats() }
+		return liftUnserialized { $0.skipRepeats() }
 	}
 }
 
@@ -243,7 +248,7 @@ extension PropertyProtocol {
 	///
 	/// - returns: A property that sends unique values during its lifetime.
 	public func uniqueValues<Identity: Hashable>(_ transform: @escaping (Value) -> Identity) -> Property<Value> {
-		return lift { $0.uniqueValues(transform) }
+		return liftUnserialized { $0.uniqueValues(transform) }
 	}
 }
 
@@ -257,7 +262,7 @@ extension PropertyProtocol where Value: Hashable {
 	///
 	/// - returns: A property that sends unique values during its lifetime.
 	public func uniqueValues() -> Property<Value> {
-		return lift { $0.uniqueValues() }
+		return liftUnserialized { $0.uniqueValues() }
 	}
 }
 
@@ -420,7 +425,7 @@ extension PropertyProtocol where Value == Bool {
 	///
 	/// - returns: A property that contains the logical NOT results.
 	public func negate() -> Property<Value> {
-		return self.lift { $0.negate() }
+		return liftUnserialized { $0.negate() }
 	}
 
 	/// Create a property that computes a logical AND between the latest values of `self`
@@ -618,7 +623,7 @@ public final class Property<Value>: PropertyProtocol {
 	///
 	/// - parameters:
 	///   - unsafeProducer: The composed producer for creating the property.
-	fileprivate init(unsafeProducer: SignalProducer<Value, Never>) {
+	fileprivate init(unsafeProducer: SignalProducer<Value, Never>, unserialized: Bool = false) {
 		// The ownership graph:
 		//
 		// ------------     weak  -----------    strong ------------------
@@ -634,7 +639,9 @@ public final class Property<Value>: PropertyProtocol {
 		// A composed property tracks its active consumers through its relay signal, and
 		// interrupts `unsafeProducer` if the relay signal terminates.
 		let disposable = SerialDisposable()
-		let (relay, observer) = Signal<Value, Never>.pipe(disposable: disposable)
+		let (relay, observer) = unserialized
+			? Signal<Value, Never>.unserializedPipe(disposable: disposable)
+			: Signal<Value, Never>.pipe(disposable: disposable)
 
 		disposable.inner = unsafeProducer.start { [weak box] event in
 			// `observer` receives `interrupted` only as a result of the termination of
@@ -666,7 +673,7 @@ public final class Property<Value>: PropertyProtocol {
 		_value = { box.value! }
 		signal = relay
 
-		producer = SignalProducer { [box, relay] observer, lifetime in
+		producer = SignalProducer.unserialized { [box, relay] observer, lifetime in
 			box.withValue { value in
 				observer.send(value: value!)
 				lifetime += relay.observe(Signal.Observer(mappingInterruptedToCompleted: observer))
@@ -739,7 +746,7 @@ public final class MutableProperty<Value>: ComposableMutablePropertyProtocol {
 	/// followed by all changes over time, then complete when the property has
 	/// deinitialized.
 	public var producer: SignalProducer<Value, Never> {
-		return SignalProducer { [box, signal] observer, lifetime in
+		return SignalProducer.unserialized { [box, signal] observer, lifetime in
 			box.withValue { value in
 				observer.send(value: value)
 				lifetime += signal.observe(Signal.Observer(mappingInterruptedToCompleted: observer))
@@ -752,7 +759,7 @@ public final class MutableProperty<Value>: ComposableMutablePropertyProtocol {
 	/// - parameters:
 	///   - initialValue: Starting value for the mutable property.
 	public init(_ initialValue: Value) {
-		(signal, observer) = Signal.pipe()
+		(signal, observer) = Signal.unserializedPipe()
 		(lifetime, token) = Lifetime.make()
 
 		/// Need a recursive lock around `value` to allow recursive access to