goroutines.swift

goroutines.swift

import Foundation

protocol Channel: IteratorProtocol {
    func send(_ value: Element?)
}

/// A blocking channel for sending values.
///
/// `send` and `receive` must run in separate separate execution contexts, otherwise you get a deadlock.
final class BlockingChannel<A>: Channel {
    let producer = DispatchSemaphore(value: 0)
    let consumer = DispatchSemaphore(value: 0)
    private var value: A?
    var done: Bool = false
    
    func send(_ value: A?) {
        producer.wait()
        self.value = value
        if value == nil { done = true }
        consumer.signal()
    }
    
    func next() -> A? {
        guard !done else { return nil }
        producer.signal()
        consumer.wait()
        return value
    }
}

final class BufferedChannel<A>: Channel {
    let valueQueue = DispatchQueue(label: "sync value")
    let consumer = DispatchSemaphore(value: 0)
    private var values: [A] = []
    var done: Bool = false
    
    func send(_ value: A?) {
        if let v = value {
            self.valueQueue.sync {
                self.values.append(v)
            }
        } else {
            done = true
        }
        consumer.signal()
    }
    
    func next() -> A? {
        guard !done else {
            return values.isEmpty ? nil : values.removeFirst()
        }
        consumer.wait()
        var result: A? = nil
        valueQueue.sync {
            result = values.removeFirst()
        }
        return result
    }
}

func go(_ f: @escaping () -> ()) -> () {
    DispatchQueue.global().async {
        f()
    }
}

infix operator <-
func <-<C, A>(lhs: C, rhs: A?) where C: Channel, C.Element == A {
    lhs.send(rhs)
}

func test() {
    let channel = BufferedChannel<Int>()
    
    go {
        channel <- 1
        channel <- 2
        channel <- nil // done
    }
    
    sleep(1)
    while let value = channel.next() {
        print(value)
    }
}

test()