Streaming

Use streaming when Python produces values over time: token generation, progress updates, sensor data, long transforms, search results, or incremental parsing.

All pool streaming APIs return CancellableStream, a custom AsyncSequence that sends a cooperative cancel signal when the consumer stops iterating.

Value Streams

let stream: CancellableStream<Int> = try await pool.evalStream("range(10)")

for try await value in stream {
    print(value)
}

For a module function:

let tokens: CancellableStream<String> = try await pool.invokeStream(
    module: "my_inference",
    function: "generate_tokens",
    args: [.python("Write a haiku")]
)

For a method on a remote object:

let model = try await pool.invokeOwned(
    module: "my_inference",
    function: "load_model",
    args: [.python("/models/model.bin")]
)

let tokens: CancellableStream<String> = try await pool.methodStream(
    handle: model,
    name: "stream",
    args: [.python("Hello")]
)

Decode Tiers

Each stream verb has two public forms:

The custom decoder receives raw pickled data. Do not pass it directly to JSONDecoder. Use a decoder appropriate to the actual wire value, or first return a Python JSON string through a typed CancellableStream<String> and then decode that string in your application.

StreamOptions

StreamOptions carries per-stream timeout, worker affinity, and channel capacity.

let stream: CancellableStream<Int> = try await pool.evalStream(
    "slow_generator()",
    options: .longRunning(timeout: 1800)
)

let pinned: CancellableStream<Double> = try await pool.methodStream(
    handle: model,
    name: "predict_stream",
    options: .pinned(worker: 0)
)

let custom = StreamOptions(
    timeout: 120,
    workerAffinity: 0,
    channelCapacity: 8
)

WorkerContext stream helpers are already pinned to the context worker.

Stopping a Stream

Breaking out of the loop cancels the stream promptly.

let stream: CancellableStream<Double> = try await pool.evalStream("""
import itertools
itertools.count(0.0, 0.1)
""")

for try await value in stream {
    if value >= 1.0 {
        break
    }
}

Python generators can cooperate with cancellation:

from swift_bridge import check_cancel

def generate():
    for item in expensive_source():
        check_cancel()
        yield item

check_cancel() raises KeyboardInterrupt when the Swift consumer has stopped iterating. Outside an active stream it is a no-op.

For code without checkpoints, use a timeout and then replace the worker if the workload stays wedged:

try await pool.respawnWorker(0, reason: .userInitiated, force: true)

C extensions blocked in syscalls may not stop until the syscall returns. Design long-running Python iterators with checkpoints when you need graceful cancel.

Timeouts and Keepalive

let ipc = IPCConfiguration(
    receiveTimeout: 30,
    streamKeepaliveInterval: 5,
    respawnOnTimeout: true
)

receiveTimeout is the maximum time the pool waits for stream activity. streamKeepaliveInterval lets the worker prove the stream is alive even when no user value is ready. With respawnOnTimeout, a timed-out stream can replace the wedged worker so later work starts fresh.

For long-running but healthy streams, prefer StreamOptions.longRunning(timeout:) or a custom timeout rather than disabling timeouts globally.

Timeout cleanup is stream-scoped. When a stream times out, the runtime aborts that stream and drains stale frames through the stream's demux channel before returning, so a per-call timeout override does not fall through to the pool-wide receive timeout.

Progress Events

Use swift_bridge.progress in Python to report semantic progress.

from swift_bridge import progress

def train():
    progress("loading data")
    data = load_data()
    for epoch in range(10):
        train_one_epoch(data, epoch)
        progress(f"epoch {epoch + 1}/10")
        yield epoch

Value-only *Stream APIs drop progress frames. Use *Events APIs to receive both values and progress.

let events: CancellableStream<StreamEvent<Int>> = try await pool.evalEvents("train()")

for try await event in events {
    switch event {
    case .value(let epoch):
        print("completed epoch \(epoch)")
    case .progress(_, let hint):
        print(hint ?? "working")
    }
}

The event-stream entry points are:

Each also has a custom decode: form.

Ordering Contract

For *Events, progress and values are delivered in the order Python produced them. If Python calls progress("a"), yields 1, calls progress("b"), and yields 2, Swift observes those four events in that order.

Streaming From a Specific Worker

let worker = pool.worker(1)
let stream: CancellableStream<String> = try await worker.invokeStream(
    module: "my_jobs",
    function: "tail_logs",
    args: [.python("/tmp/job.log")],
    timeout: 300
)

Worker contexts are useful when the stream depends on a handle or Python state that exists on a specific worker.

Out-of-Band Streaming

evalStream / invokeStream / methodStream send each chunk over the worker's main IPC socket. For the duration of the stream, that socket is held by the streaming command — any other eval / invoke on the same worker has to wait its turn.

When you need a long-lived stream that does not tie up the worker, ask the runtime for an opaque ManagedOutputBuffer. The runtime selects and owns the transport; the public API does not expose mapping names, ring headers, socket framing, or isolation-specific endpoints.

let output = try await pool.startOutputStream(
    generatorCode: "(json.dumps(frame) for frame in frames())",
    worker: 0,
    capacity: 65_536
)

while !output.isFinished {
    let chunk = output.readAvailable()
    if !chunk.isEmpty { handle(chunk) }
    try await Task.sleep(nanoseconds: 10_000_000)
}

generatorCode must be a Python expression that evaluates to an iterable of str or bytes. It can reference names installed in that worker's persistent namespace. The worker's main command channel remains available for regular calls while output is produced.

Polling Pattern

let output = try await pool.startOutputStream(
    generatorCode: "telemetry(n=200)",
    worker: 0,
    capacity: 64 * 1024
)

var lineBuffer = Data()
while !output.isFinished || !lineBuffer.isEmpty {
    let chunk = output.readAvailable()
    if !chunk.isEmpty {
        lineBuffer.append(chunk)
        while let nl = lineBuffer.firstIndex(of: 0x0A) {
            let line = Data(lineBuffer[..<nl])
            lineBuffer.removeSubrange(...nl)
            handle(line) // parse one framed message
        }
    }
    try await Task.sleep(nanoseconds: 10_000_000) // ~10 ms
}

To stop early, call output.cancel(). Inspect output.failed after completion when the application needs to distinguish a producer failure.

When to Reach for OOB Streaming vs evalStream

Common Pitfalls