Streams

This release introduces streams - ordered sequences of values produced by a task, and consumed by one or more other tasks. In their simplest form, this allows the partial result of a task to be processed before its execution has completed. Or a task can be used to monitor an external source, with values being distributed amongst multiple consumers. Consumers can use suspense to suspend execution when the stream goes quiet, such that they're automatically restarted when values arrive.

The release also introduces checkpoints - a way to share state between executions of a step. This can be used, for example, to keep track of a cursor for consuming from an upstream provider.

Streams

To define a producer, simply yield values from the task:

import coflux as cf

@cf.task()
def fetch_pages(url: str):
    for page in paginate(url):
        yield page

Calling fetch_pages(url) will return a cf.Stream handle, which can be consumed directly, or passed to another task to iterate over:

@cf.task()
def index(pages: cf.Stream[dict]):
    for page in pages:
        add_to_index(page)

@cf.workflow()
def crawl(url: str):
    pages = fetch_pages(url)
    return index(pages)

Each item is delivered to the consumer as it's produced, so index starts on the first page while fetch_pages is still working on the rest.

Items are stored by the server as they're produced, so a stream can be read more than once, and by more than one consumer, including a consumer that starts long after the producer finished. Views let you split a stream across parallel consumers:

handles = [index.submit(pages.partition(4, i)) for i in range(4)]

partition(n, i) delivers every n-th item starting at i, and slice(start, stop) restricts to a range. Views compose, and can be passed around like any other handle.

By default a producer runs in lockstep with its slowest consumer, so a fast producer can't run away from a slow consumer. Configure how far ahead it may run with buffer:

@cf.task(streams=cf.Streams(buffer=100))
def fetch_pages(url: str): ...

To return more than one stream, or a stream alongside other values, register a generator explicitly with cf.stream() and put the handle wherever you like in the return value.

See the documentation for details, including idle timeouts, and what happens to a stream when its producer fails or is cancelled.

Checkpoints

When a step suspends, retries, recurs, or is re-run, its code starts again from the top. For state that's the result of some work, memoising a child task is the answer. But some state isn't a task result - a cursor, a page token, a running total - and previously there was nowhere to keep it.

A checkpoint is a named value scoped to a step, carried across all of those restarts:

cursor = cf.Checkpoint("cursor", default=0)

@cf.workflow(recurrent=True, delay=60)
def poll_orders():
    since = cursor.get()
    orders, next_since = fetch_orders(since)
    for order in orders:
        process_order.submit(order)
    cursor.set(next_since)

Each iteration reads what the previous one wrote, so the poller only fetches what it hasn't seen. Reads are served locally (the state arrives with the execution), and writes are delivered in the background - cf.flush() waits for them to be acknowledged, for when a side effect depends on it.

Checkpoints are scoped to a workspace as well as to a step. Reads fall back through the workspace's bases, so re-running a step in a development workspace sees production's real state, while writes only ever land in the workspace doing the writing.

See the documentation for details.

Suspending streams

Streams and checkpoints combine with suspense to make long-lived pipelines cheap.

A stream belongs to the step that produces it, not to any one execution. So a producer can suspend from inside its generator, and the execution that resumes the step continues the same stream - consumers just wait through the pause:

cursor = cf.Checkpoint("cursor", default=0)

@cf.task()
def tail_events():
    since = cursor.get()
    for event in fetch_events(since):
        yield event
        cursor.set(event.id)
    cf.suspend(60)

This is how to write a producer that keeps going indefinitely without holding a worker slot while it waits for more data.

The same works on the consuming side. Iterating inside a cf.suspense scope gives up the worker slot when the stream goes quiet, and the consumer is resumed when the next item arrives - picking up where it left off rather than re-reading from the start:

@cf.task()
def handle_events(events: cf.Stream[dict]):
    with cf.suspense(30):
        for event in events:
            store.submit(event)

Put the two together (and submit the consumer rather than waiting on it) and a whole pipeline can sit idle holding no worker slots at all: producer, consumer and workflow all suspended between bursts, with the consumer only scheduled again once there's something for it to read.

Feedback welcome! (joe@coflux.com)