Cutting the SignalProducer overhead further.

CHANGELOG.md

@@ -1,6 +1,8 @@
 # master
 *Please add new entries at the top.*
 
+1. The `SignalProducer` internals have undergone a significant refactoring, which bootstraps the effort to reduce the overhead of constant producers and producer compositions. (#487, kudos to @andersio)
+
 # 2.0.1
 1. Addressed the exceptionally high build time. (#495)
 

Sources/Atomic.swift

@@ -42,9 +42,7 @@ internal struct UnsafeAtomicState<State: RawRepresentable> where State.RawValue
 	/// - returns: `true` if the current state matches the expected state.
 	///            `false` otherwise.
 	internal func `is`(_ expected: State) -> Bool {
-		return OSAtomicCompareAndSwap32Barrier(expected.rawValue,
-		                                       expected.rawValue,
-		                                       value)
+		return expected.rawValue == value.pointee
 	}
 
 	/// Try to transition from the expected current state to the specified next
@@ -82,7 +80,7 @@ internal struct UnsafeAtomicState<State: RawRepresentable> where State.RawValue
 	/// - returns: `true` if the current state matches the expected state.
 	///            `false` otherwise.
 	internal func `is`(_ expected: State) -> Bool {
-		return value.modify { $0 == expected.rawValue }
+		return value.value == expected.rawValue
 	}
 
 	/// Try to transition from the expected current state to the specified next

Sources/Disposable.swift

@@ -38,6 +38,31 @@ extension UnsafeAtomicState where State == DisposableState {
 	}
 }
 
+/// A disposable that does not have side effect upon disposal.
+internal final class _SimpleDisposable: Disposable {
+	private let state = UnsafeAtomicState<DisposableState>(.active)
+
+	var isDisposed: Bool {
+		return state.is(.disposed)
+	}
+
+	func dispose() {
+		_ = state.tryDispose()
+	}
+
+	deinit {
+		state.deinitialize()
+	}
+}
+
+/// A disposable that has already been disposed.
+internal final class NopDisposable: Disposable {
+	static let shared = NopDisposable()
+	var isDisposed = true
+	func dispose() {}
+	private init() {}
+}
+
 /// A type-erased disposable that forwards operations to an underlying disposable.
 public final class AnyDisposable: Disposable {
 	private final class ActionDisposable: Disposable {

Sources/Event.swift

@@ -178,3 +178,89 @@ extension Signal.Event: EventProtocol {
 		return self
 	}
 }
+
+extension Signal.Event {
+	internal static func filter(_ isIncluded: @escaping (Value) -> Bool) -> (@escaping Signal<Value, Error>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void {
+		return { action in
+			return { event in
+				switch event {
+				case let .value(value):
+					if isIncluded(value) {
+						action(.value(value))
+					}
+
+				case .completed:
+					action(.completed)
+
+				case let .failed(error):
+					action(.failed(error))
+
+				case .interrupted:
+					action(.interrupted)
+				}
+			}
+		}
+	}
+
+	internal static func filterMap<U>(_ transform: @escaping (Value) -> U?) -> (@escaping Signal<U, Error>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void {
+		return { action in
+			return { event in
+				switch event {
+				case let .value(value):
+					if let newValue = transform(value) {
+						action(.value(newValue))
+					}
+
+				case .completed:
+					action(.completed)
+
+				case let .failed(error):
+					action(.failed(error))
+
+				case .interrupted:
+					action(.interrupted)
+				}
+			}
+		}
+	}
+
+	internal static func map<U>(_ transform: @escaping (Value) -> U) -> (@escaping Signal<U, Error>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void {
+		return { action in
+			return { event in
+				switch event {
+				case let .value(value):
+					action(.value(transform(value)))
+
+				case .completed:
+					action(.completed)
+
+				case let .failed(error):
+					action(.failed(error))
+
+				case .interrupted:
+					action(.interrupted)
+				}
+			}
+		}
+	}
+
+	internal static func mapError<E>(_ transform: @escaping (Error) -> E) -> (@escaping Signal<Value, E>.Observer.Action) -> (Signal<Value, Error>.Event) -> Void {
+		return { action in
+			return { event in
+				switch event {
+				case let .value(value):
+					action(.value(value))
+
+				case .completed:
+					action(.completed)
+
+				case let .failed(error):
+					action(.failed(transform(error)))
+
+				case .interrupted:
+					action(.interrupted)
+				}
+			}
+		}
+	}
+}

Sources/Observer.swift

@@ -18,6 +18,24 @@ extension Signal {
 		/// Whether the observer should send an `interrupted` event as it deinitializes.
 		private let interruptsOnDeinit: Bool
 
+		/// The target observer of `self`.
+		private let wrapped: AnyObject?
+
+		/// An initializer that transforms the action of the given observer with the
+		/// given transform.
+		///
+		/// If the given observer would perform side effect on deinitialization, the
+		/// created observer would retain it.
+		///
+		/// - parameters:
+		///   - observer: The observer to transform.
+		///   - transform: The transform.
+		internal init<U, E: Swift.Error>(_ observer: Signal<U, E>.Observer, _ transform: @escaping (@escaping Signal<U, E>.Observer.Action) -> Action) {
+			self.action = transform(observer.action)
+			self.wrapped = observer.interruptsOnDeinit ? observer : nil
+			self.interruptsOnDeinit = false
+		}
+
 		/// An initializer that accepts a closure accepting an event for the
 		/// observer.
 		///
@@ -27,6 +45,7 @@ extension Signal {
 		///                         event as it deinitializes. `false` otherwise.
 		internal init(action: @escaping Action, interruptsOnDeinit: Bool) {
 			self.action = action
+			self.wrapped = nil
 			self.interruptsOnDeinit = interruptsOnDeinit
 		}
 
@@ -37,6 +56,7 @@ extension Signal {
 		///   - action: A closure to lift over received event.
 		public init(_ action: @escaping Action) {
 			self.action = action
+			self.wrapped = nil
 			self.interruptsOnDeinit = false
 		}
 

Sources/Signal.swift

@@ -527,6 +527,22 @@ extension Signal where Error == NoError {
 }
 
 extension Signal {
+	/// Perform an action upon every event from `self`. The action may generate zero or
+	/// more events.
+	///
+	/// - precondition: The action must be synchronous.
+	///
+	/// - parameters:
+	///   - transform: A closure that creates the said action from the given event
+	///                closure.
+	///
+	/// - returns: A signal that forwards events yielded by the action.
+	internal func flatMapEvent<U, E>(_ transform: @escaping (@escaping Signal<U, E>.Observer.Action) -> (Event) -> Void) -> Signal<U, E> {
+		return Signal<U, E> { observer in
+			return self.observe(.init(observer, transform))
+		}
+	}
+
 	/// Map each value in the signal to a new value.
 	///
 	/// - parameters:
@@ -535,11 +551,7 @@ extension Signal {
 	///
 	/// - returns: A signal that will send new values.
 	public func map<U>(_ transform: @escaping (Value) -> U) -> Signal<U, Error> {
-		return Signal<U, Error> { observer in
-			return self.observe { event in
-				observer.action(event.map(transform))
-			}
-		}
+		return flatMapEvent(Signal.Event.map(transform))
 	}
 
 #if swift(>=3.2)
@@ -562,11 +574,7 @@ extension Signal {
 	///
 	/// - returns: A signal that will send new type of errors.
 	public func mapError<F>(_ transform: @escaping (Error) -> F) -> Signal<Value, F> {
-		return Signal<Value, F> { observer in
-			return self.observe { event in
-				observer.action(event.mapError(transform))
-			}
-		}
+		return flatMapEvent(Signal.Event.mapError(transform))
 	}
 
 	/// Maps each value in the signal to a new value, lazily evaluating the
@@ -599,18 +607,7 @@ extension Signal {
 	///
 	/// - returns: A signal that forwards the values passing the given closure.
 	public func filter(_ isIncluded: @escaping (Value) -> Bool) -> Signal<Value, Error> {
-		return Signal { observer in
-			return self.observe { (event: Event) -> Void in
-				guard let value = event.value else {
-					observer.action(event)
-					return
-				}
-
-				if isIncluded(value) {
-					observer.send(value: value)
-				}
-			}
-		}
+		return flatMapEvent(Signal.Event.filter(isIncluded))
 	}
 
 	/// Applies `transform` to values from `signal` and forwards values with non `nil` results unwrapped.
@@ -620,22 +617,7 @@ extension Signal {
 	///
 	/// - returns: A signal that will send new values, that are non `nil` after the transformation.
 	public func filterMap<U>(_ transform: @escaping (Value) -> U?) -> Signal<U, Error> {
-		return Signal<U, Error> { observer in
-			return self.observe { (event: Event) -> Void in
-				switch event {
-				case let .value(value):
-					if let mapped = transform(value) {
-						observer.send(value: mapped)
-					}
-				case let .failed(error):
-					observer.send(error: error)
-				case .completed:
-					observer.sendCompleted()
-				case .interrupted:
-					observer.sendInterrupted()
-				}
-			}
-		}
+		return flatMapEvent(Signal.Event.filterMap(transform))
 	}
 }