Skip to content

Task Flow: the Next Task by Condition or Probability

An entity rarely leaves after completing just one task. An order is checked, perhaps wrapped as a gift, packed, sometimes packed again, and shipped. This page shows how a finished task chooses the next one, and what the entity carries from task to task.

What It Is

A flow is the sequence of tasks one entity passes through. Each task names what follows it: always the same task, the first of several rules whose condition holds, or a task drawn at random by weight. A rule can send the entity back to a task it has passed, for rework. A task that names nothing ends the flow. The entity keeps its key and its attributes through the whole flow, and every task can read them.

A Flow of Tasks builds a pick, pack and ship flow step by step.

How to Use It

Choose the Next Task

next takes one of these forms:

  • A task name, next: pack_order: always that task.
  • Rules, when with an optional default: each rule is a condition and a task. Rules are tried in order. The first matching rule wins, and default applies if none matches. With no default and no matching rule, the flow ends.
  • Weights, below.
  • In the builder, a function that takes the Upstream and returns a task name or None.
docs/snippets/yaml/task_flow.yaml:condition
1
2
3
4
5
6
7
8
check_order:
  payload: {total: 80}
  # Rules are tried in order; the first match wins. Otherwise, use the default.
  next:
    when:
      - {condition: "level.stock == 0", task: backorder}
      - {condition: "payload.total >= 100 or attr.gift == 1", task: wrap_gift}
    default: pack_order
docs/snippets/pages/task_flow.py:builder-condition
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
# Rules are tried in order; the first match wins. Otherwise, use the default.
@app.task(
    next={
        "when": [
            {"condition": "level.stock == 0", "task": "backorder"},
            {
                "condition": "payload.total >= 100 or attr.gift == 1",
                "task": "wrap_gift",
            },
        ],
        "default": "pack_order",
    }
)
def check_order(task_id: int):
    return {"total": 80}
docs/snippets/pages/task_flow.py:low-level-condition
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# The same rules as next, as a Router; choose() returns the next task's name
route = Router(
    {
        "when": [{"condition": "attr.gift == 1", "task": "wrap_gift"}],
        "default": "pack_order",
    }
)
order = Entity("orders-1", "orders", {"gift": 0})  # key, arrival, attributes
finished = Upstream(0, order.key, order.attributes, {"total": 80})
assert route.choose(finished, env=None) == "pack_order"

A Router takes the same rules as next. When a task finishes, its process builds the Upstream, asks route.choose(upstream, env, levels=...) for the next task's name, and starts that task with the same entity. env lets a condition read registry[...] and t. levels, a mapping from store and container names to the objects, lets it read level.<name>.

The condition is evaluated when the task finishes, after its set. payload. reads the payload of the task that just finished, attr. the entity's attributes, level. a store's or container's level, and registry["..."] a live parameter. How It Works lists everything a condition can use.

Weights

weights maps task names to numbers, and the next task is drawn in proportion to them. The draw comes from a random stream of the routing task's own, route.<task>, so a seeded run repeats it and it changes no other draw:

docs/snippets/yaml/task_flow.yaml:weights
1
2
3
4
5
6
7
8
pack_order:
  service: packing
  resource: packers
  get: {container: stock, amount: 1}
  payload: {event_type: order_packed}
  # A task drawn in proportion to its weight: one in ten is packed again
  next:
    weights: {ship_order: 9, pack_order: 1}
docs/snippets/pages/task_flow.py:builder-weights
1
2
3
4
5
6
7
8
9
# A task drawn in proportion to its weight: one in ten is packed again
@app.task(
    service_id="packing",
    resource_id="packers",
    get={"container": "stock", "amount": 1},
    next={"weights": {"ship_order": 9, "pack_order": 1}},
)
def pack_order(task_id: int):
    return {"event_type": "order_packed"}
docs/snippets/pages/task_flow.py:low-level-weights
1
2
3
4
# Weights draw from a generator: the routing task's own stream, as the builder's do
rework = Router({"weights": {"ship_order": 9, "pack_order": 1}})
rework.rng = Sampler(seed=42).generator("route.pack_order")
choices = [rework.choose(finished) for _ in range(1000)]

Attributes

An arrival gives each new entity its starting attributes, and a task's set changes them when the task finishes. Each entity gets its own copy. Attributes are not published: a condition reads them as attr.<name>, and a payload includes them only when it asks.

docs/snippets/yaml/task_flow.yaml:attributes
1
2
3
4
5
6
arrivals:
  # Each order starts with these attributes, read as attr.gift
  # An order every 2 s on average (rate 0.5)
  orders: {dist: exponential, rate: 0.5, spawn: check_order, attributes: {gift: 0}}
  # A gift order every 10 s on average (rate 0.1)
  gifts: {dist: exponential, rate: 0.1, spawn: check_order, attributes: {gift: 1}}
docs/snippets/yaml/task_flow.yaml:set
1
2
# Applied to the order's attributes when the task finishes
set: {backordered: 1}

A task's attributes list adds the named attributes to a mapping payload, as ship_order does in the full blueprint, task_flow.yaml.

docs/snippets/pages/task_flow.py:builder-attributes
1
2
3
4
5
# Each order starts with these attributes, read as attr.gift
# An order every 2 s on average (rate 0.5)
app.add_arrival("orders", dist="exponential", rate=0.5, attributes={"gift": 0})
# A gift order every 10 s on average (rate 0.1)
app.add_arrival("gifts", dist="exponential", rate=0.1, attributes={"gift": 1})
docs/snippets/pages/task_flow.py:builder-set
1
2
3
4
# `set` is applied to the order's attributes when the task finishes
@app.task(service_id="supplying", set={"backordered": 1}, next="check_order")
def backorder(task_id: int):
    return {"event_type": "order_backordered"}

An Entity(key, arrival, attributes) carries them. The process updates entity.attributes itself after a task, as the low-level tab of A Flow of Tasks shows.

Entity Key

Each arrival starts a new entity, keyed by the arrival's name and a counter from 1: orders-1, orders-2. Every event of every task in its flow carries that key as its record key, so a sink can follow one order through the whole flow. Kafka sends records with the same key to one partition, so it keeps one order's events in order. The builder can replace the key with a function of the counter:

A blueprint uses the default key.

docs/snippets/pages/task_flow.py:builder-key
1
2
3
# Keys ret-00001, ret-00002 instead of returns-1, from the counter, which starts at 1
# A return every 20 s on average (rate 0.05)
app.add_arrival("returns", dist="exponential", rate=0.05, key=lambda n: f"ret-{n:05d}")

The process that creates each Entity gives it its key.

Below are the finished events of gifts-5 in task_flow.yaml, the blueprint whose parts this page shows. The gift order is checked while the shelf is empty, backordered, checked again, wrapped, packed and shipped, all under one key:

1
2
3
4
5
6
[EVT] {'sim_ts': 19.216, 'timestamp': '...', 'key': 'gifts-5', 'task': 'check_order', 'instance': 13, 'value': {'total': 80}}
[EVT] {'sim_ts': 49.216, 'timestamp': '...', 'key': 'gifts-5', 'task': 'backorder', 'instance': 1, 'value': {'event_type': 'order_backordered'}}
[EVT] {'sim_ts': 49.216, 'timestamp': '...', 'key': 'gifts-5', 'task': 'check_order', 'instance': 34, 'value': {'total': 80}}
[EVT] {'sim_ts': 52.216, 'timestamp': '...', 'key': 'gifts-5', 'task': 'wrap_gift', 'instance': 7, 'value': {'event_type': 'gift_wrapped'}}
[EVT] {'sim_ts': 54.362, 'timestamp': '...', 'key': 'gifts-5', 'task': 'pack_order', 'instance': 29, 'value': {'event_type': 'order_packed'}}
[EVT] {'sim_ts': 54.362, 'timestamp': '...', 'key': 'gifts-5', 'task': 'ship_order', 'instance': 27, 'value': {'event_type': 'order_shipped', 'gift': 1, 'backordered': 1}}

Upstream

A task started by next receives what the task before it handed on, the Upstream: the finished task's instance id, the entity's key and attributes, and the finished task's payload. A builder function gets it by declaring a parameter called upstream, and gets the entity by declaring entity. A function that declares neither is called without them. The first task of a flow gets upstream=None.

A blueprint task has no function to declare upstream. Its payload names the attributes it carries, and its next conditions read the finished task's payload as payload.<field> and the entity's attributes as attr.<name>.

docs/snippets/pages/task_flow.py:builder-upstream
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# A function declaring `upstream` receives what the previous task handed on,
# and `entity` the order itself; a payload includes attributes only when it asks
@app.task()
def ship_order(task_id: int, upstream, entity):
    return {
        "event_type": "order_shipped",
        "packed_by": upstream.task_id,  # the instance of pack_order before it
        "gift": entity.attributes.get("gift"),
        "backordered": entity.attributes.get("backordered"),
    }


# A rule can also be a function of the Upstream, returning a task name or None
@app.task(next=lambda upstream: "check_order" if upstream.payload["ok"] else None)
def inspect_return(task_id: int):
    return {"ok": False}

The low-level API has no Upstream: a process holds the order and its attributes itself, and decides the next step in Python.

How It Works

Conditions. A condition is a Python expression, parsed with Python's ast module and checked against a list of what is allowed before it is ever evaluated. Nothing goes through eval. It may use:

  • numbers, True and False, and the arithmetic +, -, *, /, // and %;
  • the comparisons <, <=, >, >=, == and !=, strings compared with == or !=, and in and not in with items(<store>) on the right;
  • and, or, not, and the conditional form a if c else b;
  • names with their source as a prefix:
    • payload.<field>, attr.<name> and level.<name>;
    • users.<name>, how many units of a resource are held, and queue.<name>, how many requests wait for it (Resources);
    • putters.<name> and getters.<name>, how many puts and gets wait at a store or container;
    • count(<store>) and items(<store>), how many items a store holds and the list of them (Stores and Containers);
    • item.<field>, the item the task's store get took (Stores and Containers);
    • wait.<branch>.done, .ok and .value, the task's wait, and signal.<name>.count, .ok and .value (Signals and Waiting);
    • process.<name>, .instance and .target (Tasks and Processes);
    • registry["<path>"];
  • t, the simulation time in seconds.

A name carries its source, so adding a payload field cannot change what an attribute of the same name means. The registry[...] form lets a registry path, which contains dots, be written whole. Anything else is refused with a message that names it:

A blueprint condition that uses anything else is refused when the blueprint is built, with the same message.

docs/snippets/pages/task_flow.py:refused
1
2
3
4
5
for text in ("payload.total ** 2", "x > 1", "open('orders.csv')"):
    try:
        Expression(text)
    except ValueError as error:
        print(error)

The low-level API has no conditions: a process decides in Python.

1
2
3
'payload.total ** 2' uses the operator Pow, which an expression cannot
'x > 1' uses the name 'x' (write payload.<field>, attr.<name>, level.<name>, users.<name>, queue.<name>, putters.<name>, getters.<name>, item.<field>, count(<store>), items(<store>), wait.<branch>.<part>, signal.<name>.<part>, process.<name>, registry["<path>"] or t), which an expression cannot
'open('orders.csv')' uses a Call other than count(<store>) or items(<store>), which an expression cannot

A blueprint is checked when it loads, and each error names its YAML line. Every condition must parse, and every task a rule names must exist. Every level.<name> must name a store or container, every users.<name> and queue.<name> a resource, every putters.<name> and getters.<name> a store or container, and every count(...) and items(...) a store. Every registry["..."] must be a path of the model. The builder makes the same checks when run() starts. A name that is missing when a condition is evaluated, such as a payload field, raises an error naming the expression.

Weights. The weights are turned into cumulative shares once. Each choice draws one uniform number from the routing task's stream and picks the task whose share it falls in.

Starting the Next Task. When a task finishes, it applies set, builds the Upstream and asks its rules for the next task. It starts that task as a new SimPy process, with the same entity and the next instance id from the task's counter. An arrival that spawns the task uses the same counter, so no two runs of a task share an id. In SimPy terms, one process starting the next is the whole mechanism. SimPy and desflow describes processes.

Stability and Limits

  • Flow balance. The rate into each task follows from the rates into the tasks before it and the routing shares. With weights of 9 to ship and 1 back to packing, one packing in ten sends the order back to be packed again. Every order is packed once, one in ten a second time, one in a hundred a third time, and so on. An order is therefore packed 1 + 0.1 + 0.01 + ... = 1 รท 0.9 = 1.11 times on average, so the packers see 11% more work than there are orders.
  • Rework raises utilisation. Work the load out with the rework included, as Queues and Capacity does, or a rework loop can push a resource past full use. Queueing Basics explains utilisation.
  • Refused expressions. **, function calls, indexing other than registry["..."], list and dictionary literals, and bare names other than t are refused.
  • One entity, one flow. A task leads to at most one next task. Splitting one order into several, or joining several into one, comes after 1.0.
  • decide is kept free. The task key decide is kept for decisions sent from outside for one entity, planned after 1.0. Today it is refused like any unknown key.

Example: A Flow of Tasks

In the flow example, orders are picked, gift orders are wrapped, one parcel in ten is packed again, and every order is shipped. Wrapping and shipping are timed steps. It needs no Docker service and ends on its own.

The tutorial page A Flow of Tasks teaches these features step by step.

Quick Start

desflow examples get downloads the example at the release you install, into a folder of its own. Pick uv or pip:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# 1. Install
uv tool install desflow

# 2. Download the example
desflow examples get flow --api yaml
cd flow/yaml

# 3. Find the bottleneck, with no run
desflow explain flow.yaml

# 4. Run it
desflow run flow.yaml

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# 1. Install
pip install desflow

# 2. Download the example
desflow examples get flow --api yaml
cd flow/yaml

# 3. Find the bottleneck, with no run
desflow explain flow.yaml

# 4. Run it
desflow run flow.yaml
examples/yaml/flow.yaml:routes
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
tasks:
  pick:
    service: picking
    resource: pickers
    payload: {event_type: order_picked}
    # The first rule that holds wins; default catches the rest.
    next:
      when:
        - {condition: "attr.gift == 1", task: wrap}
      default: pack
  wrap:
    # A timed step: a service and no resource, so no queued event and no wait.
    service: wrapping
    payload: {event_type: gift_wrapped}
    # Applied to the entity's attributes when the task finishes.
    set: {wrapped: 1}
    next: pack
  pack:
    service: packing
    resource: packers
    payload: {event_type: order_packed}
    # Drawn from this task's own random sequence: one parcel in ten is packed again.
    next:
      weights: {ship: 9, pack: 1}

The whole file is examples/yaml/flow.yaml.

Quick Start

desflow examples get downloads the example at the release you install, into a folder of its own. Pick uv or pip:

1
2
3
4
5
6
7
8
9
# 1. Install
uv tool install desflow

# 2. Download the example
desflow examples get flow --api builder
cd flow/builder

# 3. Run it
uv run --no-project --with desflow flow_example.py

1
2
3
4
5
6
7
8
9
# 1. Install
pip install desflow

# 2. Download the example
desflow examples get flow --api builder
cd flow/builder

# 3. Run it
python flow_example.py
examples/builder/flow_example.py:routes
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@app.task(
    service_id="picking",
    resource_id="pickers",
    # The first rule that holds wins; default catches the rest
    next={"when": [{"condition": "attr.gift == 1", "task": "wrap"}], "default": "pack"},
)
def pick(task_id, context):
    """Picks the order's items."""
    return {"event_type": "order_picked"}


# A timed step: a service and no resource, so no queued event and no wait.
# `set` is applied to the entity's attributes when the task finishes.
@app.task(service_id="wrapping", set={"wrapped": 1}, next="pack")
def wrap(task_id, context):
    """Wraps a gift."""
    return {"event_type": "gift_wrapped"}


# Drawn from this task's own random sequence: one parcel in ten is packed again
@app.task(
    service_id="packing",
    resource_id="packers",
    next={"weights": {"ship": 9, "pack": 1}},
)
def pack(task_id, context):
    """Packs the order."""
    return {"event_type": "order_packed"}

The whole file is examples/builder/flow_example.py.

Quick Start

desflow examples get downloads the example at the release you install, into a folder of its own. Pick uv or pip:

1
2
3
4
5
6
7
8
9
# 1. Install
uv tool install desflow

# 2. Download the example
desflow examples get flow --api low-level
cd flow/low-level

# 3. Run it
uv run --no-project --with desflow flow_example.py

1
2
3
4
5
6
7
8
9
# 1. Install
pip install desflow

# 2. Download the example
desflow examples get flow --api low-level
cd flow/low-level

# 3. Run it
python flow_example.py
examples/low-level/flow_example.py:routes
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
ROUTES = {
    "pick": Router(
        {"when": [{"condition": "attr.gift == 1", "task": "wrap"}], "default": "pack"}
    ),
    "wrap": Router("pack"),
    # Weights draw from a generator, set in processes()
    "pack": Router({"weights": {"ship": 9, "pack": 1}}),
    "ship": Router(None),
}
SETS = {"wrap": {"wrapped": 1}}

The whole file is examples/low-level/flow_example.py.