goroutines.swift

goroutines.swift

import Foundation

enum Message<T> {
    case value(T)
    case finished
}

protocol Channel: IteratorProtocol {
    func send(_ value: Message<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: Message<A> = .finished

    func send(_ value: Message<A>) {
        producer.wait()
        self.value = value
        consumer.signal()
    }

    func next() -> A? {
        guard case let .value(v) = self.value else { return nil }
        producer.signal()
        consumer.wait()
        return v
    }
}

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: Message<A>) {
        switch value {
        case let .value(v):
            self.valueQueue.sync {
                self.values.append(v)
            }
        case .finished:
            self.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 <- .value(rhs)
}

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

func test() {
    let channel = BufferedChannel<Int>()

    go {
        channel <- 1
        channel <- 2
        channel <- .finished // done
    }

    sleep(1)
    while let value = channel.next() {
        print(value)
    }
}

test()