Producer/Consumer, Pipelines und Timeouts

Was Sie erstellen werden

Grundlegender Producer/Consumer

// Producer pushes values; consumer iterates them asynchronously.
func makeNumbers(count: Int) -> AsyncStream<Int> {
    AsyncStream { continuation in
        Task {
            for i in 1...count {
                continuation.yield(i)
                try? await Task.sleep(nanoseconds: 40_000_000) // ~40ms
            }
            continuation.finish()
        }
    }
}

Task {
    let stream = makeNumbers(count: 5)
    for await n in stream {
        print("consume:", n)
    }
    print("done") // after finish()
}

Backpressure-Richtlinie

// Use a limited buffering policy to avoid memory spikes.
func makeBufferedNumbers(limit: Int) -> AsyncStream<Int> {
    AsyncStream(bufferingPolicy: .bufferingOldest(limit)) { cont in
        Task {
            for i in 1...20 {
                cont.yield(i)
                // Producer is fast; buffer prevents unbounded growth
            }
            cont.finish()
        }
    }
}

Task {
    let s = makeBufferedNumbers(limit: 5)
    for await n in s {
        // Simulate a slow consumer
        try? await Task.sleep(nanoseconds: 60_000_000)
        print("got:", n)
    }
}

Pipeline mit AsyncSequence

// Compose a pipeline: map -> filter -> reduce over an AsyncSequence.
func numbers(_ n: Int) -> AsyncStream<Int> {
    AsyncStream { cont in
        Task {
            for i in 1...n { cont.yield(i) }
            cont.finish()
        }
    }
}

Task {
    let evensSquaredSum = await numbers(10)
        .map { $0 * $0 }
        .filter { $0.isMultiple(of: 2) }
        .reduce(0, +)
    print("sum:", evensSquaredSum) // 220 (4+16+36+64+100)
}

Parallele Stufe

// Fan-out/fan-in: process items concurrently, then merge.
// Here we use a task group to transform elements in parallel.
func parallelUppercased(_ input: [String]) async -> [String] {
    await withTaskGroup(of: String.self) { group in
        for s in input {
            group.addTask {
                try? await Task.sleep(nanoseconds: 30_000_000)
                return s.uppercased()
            }
        }
        return await group.reduce(into: [String]()) { $0.append($1) }
    }
}

Task {
    let out = await parallelUppercased(["a","bb","ccc"])
    print(out) // ["A","BB","CCC"]
}

Timeout-Hilfsfunktion

enum TimeoutError: Error { case timedOut }

func withTimeout<T>(
    seconds: Double,
    operation: @escaping () async throws -> T
) async throws -> T {
    try await withThrowingTaskGroup(of: T.self) { group in
        // Child 1: actual work
        group.addTask { try await operation() }
        // Child 2: the timer
        group.addTask {
            try await Task.sleep(nanoseconds: UInt64(seconds * 1_000_000_000))
            throw TimeoutError.timedOut
        }
        // First child to finish wins; cancel the rest
        let result = try await group.next()!
        group.cancelAll()
        return result
    }
}

// Demo
func slowFetch() async throws -> String {
    try await Task.sleep(nanoseconds: 300_000_000) // 300ms
    return "OK"
}

Task {
    do {
        let v = try await withTimeout(seconds: 0.1) { try await slowFetch() }
        print("value:", v)
    } catch {
        print("timeout:", error) // expected
    }
}

Timeout-Muster (Wettlauf)

Zusammenfassung