API reference

The API card is the one-page version. This page is generated from the docstrings.

Tolquane: parallel programming with composable building blocks.

Nodes speak on channels. Pipelines, farms and all-to-all blocks compose them, and the same graph runs on threads, processes or across a network. The design lives in DESIGN.md at the repository root.

import tolquane as tq

@tq.source
def numbers():
    yield from range(1, 101)

@tq.node
def double(x):
    return x * 2

out = tq.to_list()
tq.run(numbers >> tq.farm(double, workers=4) >> out)
print(sorted(out.items))

Context

What a node can do while it runs: send items, see where they came from, stop.

cancelled property

True once the run is being cancelled; long-running raw nodes should return.

is_feedback property

True when the current item arrived on a feedback edge.

input_waiting property

True when another item is already queued for this node (raw nodes batch on it).

feedback_inputs property

Input indexes that are feedback edges, for recv(source=...) in raw heads.

send(item, *, to=None)

Send item downstream: round robin by default, or to output to.

broadcast(item)

Send the same object to every output. Receivers must treat it as read-only.

feedback(item, *, to=None)

Send item back to the start of the enclosing tq.feedback block.

flush()

Send any batched output now. Raw nodes that block outside Tolquane (a socket, a queue) should call this before blocking, so downstream is not kept waiting.

stop()

Finish this node after the current item. Its outputs are closed normally.

recv(source=None)

Raw nodes: next (source, item), or None once every input has ended.

With source=i only that input is read (others are held back with their backpressure) and None means that input has ended.

inputs()

Raw nodes: iterate (source, item) pairs until every input has ended.

Report dataclass

What happened during a run. print(report) shows a table.

busiest(n=3)

Node names with the most busy seconds, most busy first: the bottleneck first.

to_dict()

The report as JSON-ready data. Edges are keyed "src->dst".

node(fn=None, *, name=None, distribute='round_robin')

A processing node: def f(item) returns what to send, def f(item, ctx) sends.

source(fn=None, *, name=None)

A node with no inputs that yields (or returns an iterable of) items.

sink(fn=None, *, name=None)

A node with no outputs. Its return value is ignored.

raw(fn=None, *, name=None)

Full control: def f(ctx) reads with ctx.recv()/ctx.inputs() and sends itself.

farm(worker, workers=4, **options)

Emitter, workers copies of worker and a collector. See Farm for options.

comb(first, second)

Fuse two nodes into one so they run on one thread with no channel between them.

pipeline(*blocks)

Connect blocks left to right; the same thing as a >> b >> c.

all2all(left, right, *, R=None, G=None, merge=False)

Join two farms worker to worker, removing the collector and emitter between them.

R is fused after every left worker, G before every right worker. With merge=True the two farms stay in a pipeline through one node (R, G or comb(R, G)), or worker to worker when neither is given.

feedback(block, *, name=None)

Wire a block's outputs back to its inputs. Send back with ctx.feedback(item).

The loop closes by itself once every outside input has ended and nothing is in flight; ctx.stop() in the first stage ends it earlier.

check(block)

Expand and validate a block without running it. Raises GraphError with a fix.

explain(block)

One line per node and per edge saying what was inferred and which rule wired it.

to_list(name='to_list')

A sink for tests and quick scripts: run the graph, then read sink.items.

from_iterable(items, name='from_iterable')

A source that yields the given items.

Bases: Block

Emitter, N workers and a collector.

A worker is a function, a class, a comb() or any block: a pipeline, a farm, a feedback loop or an all-to-all. A block worker is copied workers times, each copy named <farm>.<i>.<node>, and must have one input and at most one output.

clone(**changes)

A copy of this farm with some options replaced.

A running graph you feed with put and read with get or iteration.

Use it as a context manager; leaving the block closes the input, waits for the graph to drain and re-raises any failure. on_progress, progress_interval, tap and stop work as they do in run; a session that is stopped raises RunCancelled when the block is left.

put(item)

Send one item into the graph.

close()

End the input stream; the graph drains and finishes.

get(timeout=None)

Next result. Blocks up to timeout; raises SessionClosed when done.

Read a deploy file (TOML) or an equivalent dict.

[groups.G1]
endpoint = "10.0.0.1:7000"
nodes = ["numbers", "double"]          # node names, farm names, or glob patterns
[groups.G2]
endpoint = "10.0.0.2:7000"
nodes = ["show"]
ssh = "me@10.0.0.2"                    # optional; where `tolquane launch` starts it
[options]
secret = "change-me"                   # optional; HMAC handshake on every connection
connect_timeout = 60                   # seconds to wait for a peer to come up
reconnect_timeout = 30                 # seconds to tolerate a dropped connection
python = "python3"                     # optional; interpreter `tolquane launch` uses
workdir = "/srv/flow"                  # optional; directory it starts each group in

ssh, python and workdir may also be given per group.

One-shot: build, check and run a flow for description; returns the result.