Vanilla Reactive System

reactive.swift

// Credit Ryan Carniato: https://frontendmasters.com/courses/reactivity-solidjs/
// & Marc Grabanski: https://gist.github.com/1Marc/09e739caa6a82cc176ab4c2abd691814

extension Array<Observer> {
    fileprivate var current: Observer? {
        last
    }
}

private final class Observer {
    // FIXME: This is not thread safe
    // This is usually implemented as a thread local value in multi-threaded languages:
    //     - Example in Swift: https://github.com/unixzii/swift-signal/blob/main/Sources/SwiftSignal/Scope.swift#L8
    //     - Example in Rust: https://github.com/leptos-rs/leptos/blob/main/reactive_graph/src/graph/subscriber.rs#L5-L7
    static var context: [Observer] = []
        
    private var sideEffect: (() -> Void) = {}
    private var dependencies = Set<ReferenceSet<Observer>>()
    
    func registerEffect(_ executor: @escaping () -> Void) {
        sideEffect = executor
    }
    
    func execute() {
        // start a fresh run to capture new dependencies and remove stale dependencies
        cleanup()
        // Add ourselves to the context so any signals about to run can associate themselves with us
        Self.context.append(self)
        // Run the side effect and capture dependencies
        sideEffect()
        // Remove ourselves from the context
        Self.context.removeLast()
    }
    
    func observe<T>(_ signal: Signal<T>) {
        signal.subscriptions.insert(self)
        dependencies.insert(signal.subscriptions)
    }
    
    private func cleanup() {
        for dependency in dependencies {
            dependency.remove(self)
        }
        dependencies.removeAll()
    }
}

extension Observer: Hashable {
    static func == (lhs: Observer, rhs: Observer) -> Bool {
        lhs === rhs
    }

    func hash(into hasher: inout Hasher) {
        hasher.combine(ObjectIdentifier(self))
    }
}

// Reactive atomic value
// Reading from `value` inside of an `effect` will cause the effect to re-run when `value` is updated
public final class Signal<T> {
    fileprivate var subscriptions = ReferenceSet<Observer>()
    private var storage: T
    
    public var value: T {
        get {
            // Get the current Observer from the context and start observing ourself
            Observer.context.current?.observe(self)
            return storage
        }
        set {
            storage = newValue
            for observer in subscriptions {
                // Run the side effect for all of the Observers observing us
                observer.execute()
            }
        }
    }
    
    public init(_ initialValue: T) {
        storage = initialValue
    }
}

// Reactive side-effect
// Any signals read inside of `sideEffectFunction` will be observed and `sideEffectFunction`
// will be re-run when any observed signals update
public func effect(_ sideEffectFunction: @escaping () -> Void) {
    let observer = Observer()
    observer.registerEffect(sideEffectFunction)
    // Effects must run immediately to capture their dependencies
    observer.execute()
}

// Returns a signal that only updates when signals used inside `cachedExpression` update
public func memoize<T>(_ cachedExpression: @escaping @autoclosure () -> T) -> (() -> T) {
    let memoizedSignal = Signal<T?>(nil)
    // Only update `memoizedSignal` when signals used inside `cachedExpression` change
    effect { memoizedSignal.value = cachedExpression() }
    // return a read-only signal (a closure)
    return { memoizedSignal.value! }
}

// Reads the value returned by `nonReactiveReadsFn` without tracking any signals
// used inside `nonReactiveReadsFn`
public func untrack<T>(_ nonReactiveReadsFn: () -> T) -> T {
    let prevContext = Observer.context
    Observer.context = []
    let res = nonReactiveReadsFn()
    Observer.context = prevContext
    return res
}

// MARK: Usage Examples

let count = Signal(0)
let doubleCount = memoize(count.value * 2)

effect {
    print("2 × \(count.value) is \(doubleCount())")
}

for _ in 0...4 {
    try? await Task.sleep(.seconds(2))
    count.value = count.value + 1
}

// Prints:
// 2 × 0 is 0
// 2 × 1 is 2
// 2 × 2 is 4
// 2 × 3 is 6
// 2 × 4 is 8

ReferenceSet.swift

// A reference type Set
private final class ReferenceSet<Element: Hashable>: Hashable, Collection {
    typealias Element = Element
    typealias Iterator = Set<Element>.Iterator
    typealias Index = Set<Element>.Index
    typealias Indices = Set<Element>.Indices
    typealias SubSequence = Set<Element>.SubSequence
    
    private var inner = Set<Element>()
    
    func makeIterator() -> Iterator {
        inner.makeIterator()
    }
    
    var startIndex: Index {
        inner.startIndex
    }
    
    var endIndex: Index {
        inner.endIndex
    }
    
    var indices: Indices {
        inner.indices
    }
    
    func index(after i: Index) -> Index {
        inner.index(after: i)
    }
    
    subscript(position: Index) -> Element {
        inner[position]
    }
    
    subscript(bounds: Range<Index>) -> SubSequence {
        inner[bounds]
    }
    
    func insert(_ newMember: Element) {
        inner.insert(newMember)
    }
    
    func remove(_ member: Element) {
        inner.remove(member)
    }
    
    static func == (lhs: ReferenceSet, rhs: ReferenceSet) -> Bool {
        lhs.inner == rhs.inner
    }

    func hash(into hasher: inout Hasher) {
        inner.hash(into: &hasher)
    }
}

It's getting harder for me to offer good help here. I don't know how the system is supposed to behave! But, I still am extremely skeptical of using a thread-local unless I really understand why it is necessary.

Another problem is you have structured your use of Task here like how GCD queues might be used. But that is not how isolation works. Top-level swift code is isolated to the MainActor. So, your first Task is not actually in the background. And your other Tasks would all be MainActor even without your annotations.

Finally, doesn't this program terminate before finishing? There's nothing waiting on these tasks.

Ah I didn't realize that the top level was already MainActor isolated. In that case, would something like this do what I'm expecting?

// MARK: Usage Example

func someExpensiveCalculation(_ number: Int) -> Int {
    // Pretend...
    number * number
}

let count = Signal(0)
let doubleCount = memoize(count.value * 2)

Task(priority: .background) {
    // Do expensive work on a background thread/thread pool/task
    let doubleCountSquared = memoize(someExpensiveCalculation(doubleCount()))
    
    effect {
        // Some sort of non-UI logging code that doesn't need to happen on the main thread
        // and can run in parallel to the UI effect
        print("doubleCountSquared changed: \(doubleCountSquared())")
    }
    
    Task { @MainActor in
        // Receive updates from the background thread on the main thread
        // to update the UI
        effect {
            print("Render UI: \(doubleCountSquared())")
        }
    }
}

// Pretend this is us responding to user input
for _ in 0..<4 {
    try? await Task.sleep(for: .seconds(2))
    count.value += 1
}

Or do I really just want a nonisolated async function instead of a task at all?

// MARK: Usage Example

func someExpensiveCalculation(_ number: Int) -> Int {
    // Pretend...
    number * number
}

let count = Signal(0)
let doubleCount = memoize(count.value * 2)

nonisolated func background() async {
    // Do expensive work on a background thread/thread pool/task
    let doubleCountSquared = memoize(someExpensiveCalculation(doubleCount()))
    
    effect {
        // Some sort of non-UI logging code that doesn't need to happen on the main thread
        // and can run in parallel to the UI effect
        print("doubleCountSquared changed: \(doubleCountSquared())")
    }
    
    Task { @MainActor in
        // Receive updates from the background thread on the main thread
        // to update the UI
        effect {
            print("Render UI: \(doubleCountSquared())")
        }
    }
}

Task { await background() }

// Pretend this is us responding to user input
for _ in 0..<4 {
    try? await Task.sleep(for: .seconds(2))
    count.value += 1
}

Or a TaskGroup?

// MARK: Usage Example

func someExpensiveCalculation(_ number: Int) -> Int {
    // Pretend...
    number * number
}

let count = Signal(0)
let doubleCount = memoize(count.value * 2)

await withTaskGroup(of: Void.self) { group in
    group.addTask {
        // Do expensive work on a background thread/thread pool/task
        let doubleCountSquared = memoize(someExpensiveCalculation(doubleCount()))
        
        effect {
            // Some sort of non-UI logging code that doesn't need to happen on the main thread
            // and can run in parallel to the UI effect
            print("doubleCountSquared changed: \(doubleCountSquared())")
        }
        
        group.addTask { @MainActor in
            // Receive updates from the background thread on the main thread
            // to update the UI
            effect {
                print("Render UI: \(doubleCountSquared())")
            }
        }
    }
    
    // Pretend this is us responding to user input
    for _ in 0..<4 {
        try? await Task.sleep(for: .seconds(2))
        count.value += 1
    }
}

@mattmassicotte after reading your latest post, I think this is correct?

// MARK: Usage Example

func someExpensiveCalculation(_ number: Int) -> Int {
    // Pretend...
    number * number
}

let count = Signal(0)
let doubleCount = memoize(count.value * 2)

nonisolated func background() async {
    // Do expensive work in the background
    let doubleCountSquared = memoize(someExpensiveCalculation(doubleCount()))
    
    @MainActor
    func updateUI() {
        // Receive updates from the background thread on the main thread
        // to update the UI
        effect {
            print("Render UI: \(doubleCountSquared())")
        }
    }
    
    // Start the background effect
    effect {
        // Some sort of non-UI logging code that doesn't need to happen on the main thread
        // and can run in parallel to the UI effect
        print("doubleCountSquared changed: \(doubleCountSquared())")
    }
    
    // Start the MainActor effect
    await updateUI()
}

// Start the background and MainActor effect without suspending
async let _ = background()

// Pretend this is us responding to user input
for _ in 0..<4 {
    try? await Task.sleep(for: .seconds(2))
    count.value += 1
}

This does look good, except for one thing. I actually don't know what happens when you ignore the result of an async let like that! I guess it is equivalent to but I'm not actually sure.

Task {
  await background()
}