Skip to content
Technical preview. This site is published for review. Everything on it, including the API, tokens and module protocol, is subject to change.
Guides
Futures and gather

Futures and gather

Nothing in a workflow definition returns a result directly if producing it means leaving the interpreter. An action call, timer(), poll() and w.execute_workflow() all return a future, the handle for a result that is not there yet. gather is how the workflow waits for it, and blocking on a future or a channel is the only way a workflow waits at all.

What a future is

A future stands for one result that arrives later. It resolves once, to a permanent value, and gathering a resolved future again returns the same value. A future can be stored in a list or handed to a helper inside the workflow, but it cannot leave it: a future has no encoding in a result, an action argument or a child workflow’s arguments.

load("module:gitlab", "call_api")

def main(w, project_id, token):
    headers = {"Authorization": "Bearer " + token}
    f = call_api("GET", "/api/v4/projects/%d" % project_id, headers = headers)
    status, _, _, err = gather(f)
    return status

call_api(...) returns at once with a future; the HTTP request runs outside the interpreter. gather(f) blocks the workflow until the result is recorded, then returns it.

gather 

gather(input, timeout = None, timeout_value = None)

input is one future or channel, or a list of them, and the result mirrors that shape: one value for one input, a list of values in the same order for a list. Gathering a list waits for every element.

load("module:gitlab", "call_api")

def main(w, project_id, token):
    headers = {"Authorization": "Bearer " + token}
    base = "/api/v4/projects/%d" % project_id
    project, pipelines = gather([
        call_api("GET", base, headers = headers),
        call_api("GET", base + "/pipelines", headers = headers),
    ])
    return project[0], pipelines[0]

With a timeout, gather returns timeout_value when the timeout fires before the input is ready, and consumes nothing. The default timeout_value is None, so a legitimate False or 0 result is never mistaken for a timeout; pass your own when None is a possible result.

answer = gather(f, timeout = 10 * time.minute, timeout_value = "slow")

select

select(cases, timeout = None, timeout_value = None) takes a dict of name to future or channel and returns the name of a case that is ready, or timeout_value once the timeout fires. It reports readiness and consumes nothing: a channel value stays in the channel until a gather reads it, and a ready future is still gathered afterwards. When several cases are ready in the same round, the choice is deterministic but unspecified, so do not encode priority in the order of the dict. A timeout never wins over a case that is ready.

def main(w, reply):
    ready = select(
        {"reply": reply, "reminder": timer(30 * time.minute)},
        timeout = 2 * time.hour,
        timeout_value = "gave up",
    )
    if ready == "reply":
        return gather(reply)
    fail("no answer: %s" % ready)

Channels are streams, futures are values

A channel is a stream of values sent into the workflow from outside, created with channel() and handed to a module, or bound to a main parameter by the caller. Where a future resolves once, a channel is never closed and each gather on it consumes exactly one value. A gather without a timeout on a channel nobody sends to blocks until the workflow times out.

def main(w, reply):
    answer = gather(
        reply,
        timeout = 7 * 24 * time.hour,
        timeout_value = "no answer",
    )
    if answer == "no answer":
        fail("nobody answered within a week")
    return answer

Here the caller bound a channel to reply by passing a channel_value in the kwargs of StartWorkflow, and received a channel token for it in the response. Policy enforcement shows the other direction: a channel the workflow creates and hands out.

A future nobody gathers

An action error surfaces where its future is gathered, not where the action was called, so a failing action nobody gathers never fails the workflow. And an operation scheduled in the round that ends the workflow is dropped rather than run. Gather a future even when its value is of no interest:

gather(call_api("POST", path, headers = headers, body = body))

Without the gather, main can return before the request is made, and the call never happens.

Futures on replay

A workflow is replayed from the top against its history whenever it resumes. A future whose result is recorded resolves from history at once: the action is not run again and the timer does not wait again. Only a future with no recorded result blocks, and that is where live execution continues. Crash and replay covers the mechanics.

Waiting is blocking

A workflow waits only by blocking: gather, select and sleep are the blocking calls, and each of them blocks on a future or a channel; sleep(d) blocks on a durable timer. Nothing runs while the workflow is blocked, and nothing about it stays in memory. Workflow time follows the same rule: time.now() reads the time recorded in history and advances only as results are consumed, so a loop that checks the clock without blocking never sees time move. Waiting for time itself is covered in Durable waits.

Related

Last updated on
Durable waits