Skip to content

A Flow of Tasks

Until now an order was one task: it was packed and left. A real order is picked from the shelves, packed and shipped. Sometimes an item is out of stock, and sometimes a parcel fails its check and is packed again. This page asks: how do orders move through pick, pack and ship?

  • Decision: how much work to plan for beyond one pick and one pack per order.
  • Measures: two shares: how many orders go to backorder, and how many parcels are packed again.
  • Runs: one run of an hour with the seed 42, from an order's arrival to its shipment, with the supplier modelled only as a delay.
  • Assumptions: orders arrive at random, each picking time is equally likely to be anywhere from 1 to 3 seconds, and packing always takes 1.5 seconds.

What You Will Learn

This page shows how to:

  • chain tasks so an order is picked, packed and shipped;
  • route a share of the orders to backorder or back to packing;
  • carry an attribute, such as express, through the flow.

What You Add

This page adds three tasks in a row: pick_order, pack_order and ship_order. It also adds a backorder delay for an order picked while the shelf of boxes from Stock is empty, and an express attribute that each order carries. The shop is set up as follows:

  • It has two pickers and two packers.
  • A regular order arrives every 2.5 seconds on average (24 a minute, rate: 0.4), and an express order every 10 seconds (6 a minute, rate: 0.1).
  • A delivery of 53 boxes comes every 100 seconds, a little more than the orders take.
  • Picking time is uniform from 1 to 3 seconds only to keep the example short. Common Mistakes explains why a real study fits a distribution to measured times instead, as Choosing Inputs does.

Look at the flow first:

Arrivals spawn pick_order. When level.stock is 0 it leads to backorder, a timed step of 1 minute that sets backordered and leads back to pick_order. Otherwise it leads to pack_order, which takes a box from stock. pack_order leads to ship_order with weight 95, and back to itself with weight 5. ship_order ends the flow

Chain the Tasks

Choosing where an entity goes next is called routing. In desflow a task names what follows it with next. Tasks and Processes describes tasks. next takes one of three forms:

  • a task name, which always leads there;
  • a list of when rules, each a condition and a task: the first rule that holds wins, and default catches the rest;
  • weights, which Rework shows.

A task with no next ends the order's flow. Save the files from Full Files in one folder and run the commands there.

docs/snippets/yaml/flow_of_tasks.yaml:chain
1
2
3
4
5
6
7
8
9
pick_order:
  service: picking
  resource: pickers
  payload: {event_type: order_picked}
  # The first rule that holds wins; default catches the rest
  next:
    when:
      - {condition: "level.stock == 0", task: backorder}
    default: pack_order

Here the arrival loop passes the context to each task after its id. A task started by next receives the same arguments as the task before it. So every task function in the flow declares context.

docs/snippets/tutorials/flow_of_tasks.py:chain
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# The first rule that holds wins; default catches the rest
@app.task(
    service_id="picking",
    resource_id="pickers",
    next={
        "when": [{"condition": "level.stock == 0", "task": "backorder"}],
        "default": "pack_order",
    },
)
def pick_order(task_id: int, context: SimulationContext):
    return {"event_type": "order_picked"}
docs/snippets/tutorials/flow_of_tasks_low_level.py:chain
1
2
3
4
5
6
7
# The first rule that holds wins; default catches the rest
"pick_order": Router(
    {
        "when": [{"condition": "level.stock == 0", "task": "backorder"}],
        "default": "pack_order",
    }
),

A Router takes the same rules as next and returns the name of the task to start. When a task finishes, its process asks the router and starts the next task as a new process, with the same entity:

docs/snippets/tutorials/flow_of_tasks_low_level.py:next
1
2
3
4
5
6
7
8
entity.attributes.update(SETS.get(name, {}))
upstream = Upstream(task_id, entity.key, entity.attributes, finished)
following = ROUTES[name].choose(upstream, env, levels={"stock": shop["stock"]})
if following is None:
    env.stats.flow_finished(entity.arrival)  # the order leaves the shop
    return
instance = next(counters[following])
env.process(task(env, sampler, shop, following, entity, instance, counters))

In SimPy terms, a process that starts another process with env.process(...) is how one step leads to the next. SimPy and desflow describes processes.

The rule reads level.stock, the level of the container stock, at the moment picking finishes. Stores and Containers describes containers. Each name in a condition says where its value comes from: level. for a store or container, attr. for the entity's attributes, payload. for the finished task's payload, and registry["..."] for a live parameter. Task Flow lists everything a condition can use.

Every event of every task carries the order's key, orders-1, so one order can be followed through its whole flow. task names the task, and instance numbers its runs. Each task has one counter, shared by its arrivals and by next, so a re-pick after a backorder gets a new id. In the full builder file the arrival loops take the id from context.next_instance("pick_order"), and the low-level file keeps one counters entry per task.

Out of Stock

An order picked while the shelf is empty waits for the supplier for a minute, then is picked again. The wait uses no picker or packer, so backorder has a service and no resource: a timed step. It publishes started at once, with no queued event, waits the sampled time, and publishes its finished event:

docs/snippets/yaml/flow_of_tasks.yaml:backorder
1
2
3
4
5
6
7
backorder:
  # A timed step: a service and no resource, so it starts at once
  service: supplying
  payload: {event_type: order_backordered}
  # Applied to the order's attributes when the step finishes
  set: {backordered: 1}
  next: pick_order
docs/snippets/tutorials/flow_of_tasks.py:backorder
1
2
3
4
5
# A timed step: a service and no resource, so it starts at once. `set` is applied
# to the order's attributes when the step finishes.
@app.task(service_id="supplying", set={"backordered": 1}, next="pick_order")
def backorder(task_id: int, context: SimulationContext):
    return {"event_type": "order_backordered"}
docs/snippets/tutorials/flow_of_tasks_low_level.py:backorder
1
"backorder": Router("pick_order"),
docs/snippets/tutorials/flow_of_tasks_low_level.py:set
1
2
# Applied to the order's attributes when the step finishes
SETS = {"backorder": {"backordered": 1}}

next: pick_order sends the order back to an earlier task. A route back is allowed anywhere in a flow.

Rework

One parcel in 20 fails its check and is packed again. Sending work back to a task it has passed is called rework. weights chooses the next task at random, in proportion to the weights. It draws from the routing task's own random number stream, route.pack_order. So a seeded run repeats, and the choice does not change any other draw:

docs/snippets/yaml/flow_of_tasks.yaml:rework
1
2
3
4
5
6
7
8
pack_order:
  service: packing
  resource: packers
  get: {container: stock, amount: 1}
  payload: {event_type: order_packed}
  # One parcel in 20 fails its check and is packed again
  next:
    weights: {ship_order: 95, pack_order: 5}
docs/snippets/tutorials/flow_of_tasks.py:rework
1
2
3
4
5
6
7
8
9
# One parcel in 20 fails its check and is packed again
@app.task(
    service_id="packing",
    resource_id="packers",
    get={"container": "stock", "amount": 1},
    next={"weights": {"ship_order": 95, "pack_order": 5}},
)
def pack_order(task_id: int, context: SimulationContext):
    return {"event_type": "order_packed"}
docs/snippets/tutorials/flow_of_tasks_low_level.py:rework
1
2
3
# One parcel in 20 fails its check and is packed again; the generator is set
# in run below
"pack_order": Router({"weights": {"ship_order": 95, "pack_order": 5}}),

The low-level router needs a generator to draw from: ROUTES["pack_order"].rng = sampler.generator("route.pack_order") in run, the stream the builder uses.

Carry Attributes

An order is an entity with attributes: values it carries from task to task, such as whether it is express. The arrival gives each new entity its starting attributes, and set on a task changes them when the task finishes, as backorder does with backordered: 1:

docs/snippets/yaml/flow_of_tasks.yaml:attributes
1
2
3
4
arrivals:
  # Each order starts with the attributes given here, read as attr.<name>
  orders: {dist: exponential, rate: 0.4, spawn: pick_order, attributes: {express: 0}}
  express: {dist: exponential, rate: 0.1, spawn: pick_order, attributes: {express: 1}}
docs/snippets/tutorials/flow_of_tasks.py:attributes
1
2
3
# Each order starts with the attributes given here, read as attr.<name>
app.add_arrival("orders", dist="exponential", rate=0.4, attributes={"express": 0})
app.add_arrival("express", dist="exponential", rate=0.1, attributes={"express": 1})
docs/snippets/tutorials/flow_of_tasks_low_level.py:attributes
1
2
# Each order starts with the attributes given here, read as attr.<name>
ATTRIBUTES = {"orders": {"express": 0}, "express": {"express": 1}}

Entity(key, arrival, attributes) is the order, created by the arrival process; the full file below shows it.

Attributes are not published. A condition reads them as attr.express, and a payload includes them only when it names them. ship_order names express and backordered, so its finished event carries both. This is orders-172, which was picked while the shelf was empty, backordered, picked again and packed twice:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
[EVT] {'sim_ts': 397.158, 'timestamp': '...', 'key': 'orders-172', 'task': 'pick_order', 'instance': 222, 'value': {'path_id': 'Shop.service.picking', 'status': 'queued'}}
[EVT] {'sim_ts': 398.017, 'timestamp': '...', 'key': 'orders-172', 'task': 'pick_order', 'instance': 222, 'value': {'path_id': 'Shop.service.picking', 'status': 'started'}}
[EVT] {'sim_ts': 399.138, 'timestamp': '...', 'key': 'orders-172', 'task': 'pick_order', 'instance': 222, 'value': {'event_type': 'order_picked'}}
[EVT] {'sim_ts': 399.138, 'timestamp': '...', 'key': 'orders-172', 'task': 'backorder', 'instance': 24, 'value': {'path_id': 'Shop.service.supplying', 'status': 'started'}}
[EVT] {'sim_ts': 459.138, 'timestamp': '...', 'key': 'orders-172', 'task': 'backorder', 'instance': 24, 'value': {'event_type': 'order_backordered'}}
[EVT] {'sim_ts': 459.138, 'timestamp': '...', 'key': 'orders-172', 'task': 'pick_order', 'instance': 258, 'value': {'path_id': 'Shop.service.picking', 'status': 'queued'}}
[EVT] {'sim_ts': 461.427, 'timestamp': '...', 'key': 'orders-172', 'task': 'pick_order', 'instance': 258, 'value': {'path_id': 'Shop.service.picking', 'status': 'started'}}
[EVT] {'sim_ts': 463.376, 'timestamp': '...', 'key': 'orders-172', 'task': 'pick_order', 'instance': 258, 'value': {'event_type': 'order_picked'}}
[EVT] {'sim_ts': 463.376, 'timestamp': '...', 'key': 'orders-172', 'task': 'pack_order', 'instance': 243, 'value': {'path_id': 'Shop.service.packing', 'status': 'queued'}}
[EVT] {'sim_ts': 463.376, 'timestamp': '...', 'key': 'orders-172', 'task': 'pack_order', 'instance': 243, 'value': {'path_id': 'Shop.service.packing', 'status': 'started'}}
[EVT] {'sim_ts': 464.876, 'timestamp': '...', 'key': 'orders-172', 'task': 'pack_order', 'instance': 243, 'value': {'event_type': 'order_packed'}}
[EVT] {'sim_ts': 464.876, 'timestamp': '...', 'key': 'orders-172', 'task': 'pack_order', 'instance': 245, 'value': {'path_id': 'Shop.service.packing', 'status': 'queued'}}
[EVT] {'sim_ts': 464.876, 'timestamp': '...', 'key': 'orders-172', 'task': 'pack_order', 'instance': 245, 'value': {'path_id': 'Shop.service.packing', 'status': 'started'}}
[EVT] {'sim_ts': 466.376, 'timestamp': '...', 'key': 'orders-172', 'task': 'pack_order', 'instance': 245, 'value': {'event_type': 'order_packed'}}
[EVT] {'sim_ts': 466.376, 'timestamp': '...', 'key': 'orders-172', 'task': 'ship_order', 'instance': 233, 'value': {'event_type': 'order_shipped', 'express': 0, 'backordered': 1}}

The events of orders-172 in order: pick_order, backorder, pick_order again, pack_order twice, and ship_order with express 0 and backordered 1, all under the key orders-172

What You Get and Why

How do orders move through pick, pack and ship? Most go straight through, and a few go to backorder or are packed again. The run publishes the finished count of every task at the end, with stats: basic:

Task Finished in the hour
pick_order 1,857
backorder 80
pack_order 1,849
ship_order 1,775

4.3% of picks led to backorder (80 of 1,857), and 4.0% of parcels were packed again (74 of 1,849 packs did not lead to a shipment). The weights set 5%, and this one run of the hour, with the seed 42, drew a few fewer.

Flow balance is the rule that, in a flow, each task's arrival rate follows from the rates into it and the routing weights. Every pack leads to a shipment with probability 0.95, so each order is packed 1 ÷ 0.95 = 1.05 times on average, and the packers see 5% more work than there are orders. A rework loop therefore raises utilisation, so it belongs in the load you work out before running. Queueing Basics shows why waits grow quickly near full use. Task Flow describes routing in full.

Key Takeaways

  • A task names the task that follows it, by a condition or by weights.
  • In a flow, each task's rate follows from the rates into it and the routing weights.
  • A rework loop adds work to the resources it returns to, so include it in the load you work out before running.

Full Files

Full files
docs/snippets/yaml/flow_of_tasks.yaml
 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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# The A Flow of Tasks tutorial: orders are picked, packed and shipped.
# Run it with: desflow run flow_of_tasks.yaml
simulation:
  sim_id: Shop
  factor: 0
  random_seed: 42

resources:
  pickers: {current_cap: 2, max_cap: 4}
  packers: {current_cap: 2, max_cap: 4}

containers:
  # Boxes on the shelf, restocked with 53 every 100 seconds
  stock: {current_cap: 200, max_cap: 200, level: 50}

services:
  picking: {dist: uniform, low: 1 s, high: 3 s}
  packing: {dist: constant, value: 1.5 s}
  # The supplier takes a minute to send an item that is out of stock
  supplying: {dist: constant, value: 1 min}

arrivals:
  # Each order starts with the attributes given here, read as attr.<name>
  orders: {dist: exponential, rate: 0.4, spawn: pick_order, attributes: {express: 0}}
  express: {dist: exponential, rate: 0.1, spawn: pick_order, attributes: {express: 1}}
  deliveries: {dist: constant, value: 100 s, spawn: restock}

tasks:
  pick_order:
    service: picking
    resource: pickers
    payload: {event_type: order_picked}
    # The first rule that holds wins; default catches the rest
    next:
      when:
        - {condition: "level.stock == 0", task: backorder}
      default: pack_order
  backorder:
    # A timed step: a service and no resource, so it starts at once
    service: supplying
    payload: {event_type: order_backordered}
    # Applied to the order's attributes when the step finishes
    set: {backordered: 1}
    next: pick_order
  pack_order:
    service: packing
    resource: packers
    get: {container: stock, amount: 1}
    payload: {event_type: order_packed}
    # One parcel in 20 fails its check and is packed again
    next:
      weights: {ship_order: 95, pack_order: 5}
  ship_order:
    payload: {event_type: order_shipped}
    # A payload includes attributes only when it names them
    attributes: [express, backordered]
  restock:
    put: {container: stock, amount: 53}
    payload: {event_type: stock_delivered}

# Finished counts of every task and flow, at the end
stats: {level: basic, interval: 1 h}

egress:
  - type: Console

run:
  until: 1 h
docs/snippets/tutorials/flow_of_tasks.py
 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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
"""The A Flow of Tasks tutorial: orders are picked, packed and shipped, builder.

Two pickers, two packers and a shelf of boxes restocked every 100 seconds. An order
picked while the shelf is empty goes to backorder for a minute and is picked again,
and one parcel in 20 is packed again. Every record is printed to the terminal.
"""

import logging

from desflow import ConsoleEgress, SimulationContext

logging.basicConfig(
    level=logging.INFO, format="%(levelname)s [%(asctime)s] %(message)s"
)

app = (
    SimulationContext(sim_id="Shop", factor=0, random_seed=42)
    .add_resource("pickers", current_cap=2, max_cap=4)
    .add_resource("packers", current_cap=2, max_cap=4)
    # Boxes on the shelf, restocked with 53 every 100 seconds
    .add_container("stock", current_cap=200, max_cap=200, level=50)
    .add_service("picking", dist="uniform", low=1, high=3)
    .add_service("packing", dist="constant", value=1.5)
    # The supplier takes a minute to send an item that is out of stock
    .add_service("supplying", dist="constant", value=60)
    .add_arrival("deliveries", dist="constant", value=100)
    # Finished counts of every task and flow, at the end
    .with_stats("basic", interval=3600)
    .add_egress(ConsoleEgress())
)
# Each order starts with the attributes given here, read as attr.<name>
app.add_arrival("orders", dist="exponential", rate=0.4, attributes={"express": 0})
app.add_arrival("express", dist="exponential", rate=0.1, attributes={"express": 1})


# The first rule that holds wins; default catches the rest
@app.task(
    service_id="picking",
    resource_id="pickers",
    next={
        "when": [{"condition": "level.stock == 0", "task": "backorder"}],
        "default": "pack_order",
    },
)
def pick_order(task_id: int, context: SimulationContext):
    return {"event_type": "order_picked"}


# A timed step: a service and no resource, so it starts at once. `set` is applied
# to the order's attributes when the step finishes.
@app.task(service_id="supplying", set={"backordered": 1}, next="pick_order")
def backorder(task_id: int, context: SimulationContext):
    return {"event_type": "order_backordered"}


# One parcel in 20 fails its check and is packed again
@app.task(
    service_id="packing",
    resource_id="packers",
    get={"container": "stock", "amount": 1},
    next={"weights": {"ship_order": 95, "pack_order": 5}},
)
def pack_order(task_id: int, context: SimulationContext):
    return {"event_type": "order_packed"}


@app.task()
def ship_order(task_id: int, context: SimulationContext, entity):
    # A payload includes attributes only when it asks for them
    return {
        "event_type": "order_shipped",
        "express": entity.attributes.get("express"),
        "backordered": entity.attributes.get("backordered"),
    }


@app.task(put={"container": "stock", "amount": 53})
def restock(task_id: int, context: SimulationContext):
    return {"event_type": "stock_delivered"}


def spawn(arrival, task, name):
    def loop(context: SimulationContext):
        while True:
            entity = yield context.wait_for_arrival(arrival)
            # One counter per task, shared by both order arrivals and by next
            context.spawn(task(context.next_instance(name), context, entity=entity))

    return loop


app.arrival_loop("orders")(spawn("orders", pick_order, "pick_order"))
app.arrival_loop("express")(spawn("express", pick_order, "pick_order"))
app.arrival_loop("deliveries")(spawn("deliveries", restock, "restock"))

if __name__ == "__main__":
    app.run(until="1 h")
docs/snippets/tutorials/flow_of_tasks_low_level.py
  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
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
"""The A Flow of Tasks tutorial: orders are picked, packed and shipped, low-level.

Two pickers, two packers and a shelf of boxes restocked every 100 seconds. An order
picked while the shelf is empty goes to backorder for a minute and is picked again,
and one parcel in 20 is packed again. Every record is printed to the terminal.
"""

import contextlib
import itertools
import logging

from desflow import (
    CapacityConfig,
    ConsoleEgress,
    ContainerConfig,
    DistributionConfig,
    DynamicContainer,
    DynamicRealtimeEnvironment,
    DynamicResource,
    Entity,
    Router,
    Sampler,
    SimParameter,
    Upstream,
)

logging.basicConfig(
    level=logging.INFO, format="%(levelname)s [%(asctime)s] %(message)s"
)

PARAMS = SimParameter(
    sim_id="Shop",
    arrival={
        "orders": DistributionConfig(dist="exponential", rate=0.4),
        "express": DistributionConfig(dist="exponential", rate=0.1),
        "deliveries": DistributionConfig(dist="constant", value=100),
    },
    service={
        "picking": DistributionConfig(dist="uniform", low=1, high=3),
        "packing": DistributionConfig(dist="constant", value=1.5),
        # The supplier takes a minute to send an item that is out of stock
        "supplying": DistributionConfig(dist="constant", value=60),
    },
    resources={
        "pickers": CapacityConfig(current_cap=2, max_cap=4),
        "packers": CapacityConfig(current_cap=2, max_cap=4),
    },
    # Boxes on the shelf, restocked with 53 every 100 seconds
    containers={"stock": ContainerConfig(current_cap=200, max_cap=200, level=50)},
)
# Each order starts with the attributes given here, read as attr.<name>
ATTRIBUTES = {"orders": {"express": 0}, "express": {"express": 1}}

# Each task's service and resource; None for a timed step or a task that takes no time
TASKS = {
    "pick_order": ("picking", "pickers"),
    "backorder": ("supplying", None),
    "pack_order": ("packing", "packers"),
    "ship_order": (None, None),
}
EVENTS = {
    "pick_order": "order_picked",
    "backorder": "order_backordered",
    "pack_order": "order_packed",
}
ROUTES = {
    # The first rule that holds wins; default catches the rest
    "pick_order": Router(
        {
            "when": [{"condition": "level.stock == 0", "task": "backorder"}],
            "default": "pack_order",
        }
    ),
    "backorder": Router("pick_order"),
    # One parcel in 20 fails its check and is packed again; the generator is set
    # in run below
    "pack_order": Router({"weights": {"ship_order": 95, "pack_order": 5}}),
    "ship_order": Router(None),
}
# Applied to the order's attributes when the step finishes
SETS = {"backorder": {"backordered": 1}}


def payload(name, entity):
    if name == "ship_order":
        # A payload includes attributes only when it names them
        return {
            "event_type": "order_shipped",
            "express": entity.attributes.get("express"),
            "backordered": entity.attributes.get("backordered"),
        }
    return {"event_type": EVENTS[name]}


def task(env, sampler, shop, name, entity, task_id, counters):
    """Runs one task for an order, then starts the task its router chooses."""
    service, resource = TASKS[name]
    path_id = f"Shop.service.{service}"

    def publish(value):
        env.publish_event(entity.key, value, task=name, instance=task_id)

    if resource is not None:
        publish({"path_id": path_id, "status": "queued"})
        env.stats.task_queued(name, task_id)
    unit = shop[resource].request() if resource else contextlib.nullcontext()
    with unit as request:
        if request is not None:
            yield request
        if name == "pack_order":
            yield shop["stock"].get(1)
        if service is not None:
            publish({"path_id": path_id, "status": "started"})
            env.stats.task_started(name, task_id)
        finished = payload(name, entity)
        if service is not None:
            yield env.timeout(sampler.sample(env.registry.get_config(path_id)))
        publish(finished)
        env.stats.task_finished(name, task_id)

    entity.attributes.update(SETS.get(name, {}))
    upstream = Upstream(task_id, entity.key, entity.attributes, finished)
    following = ROUTES[name].choose(upstream, env, levels={"stock": shop["stock"]})
    if following is None:
        env.stats.flow_finished(entity.arrival)  # the order leaves the shop
        return
    instance = next(counters[following])
    env.process(task(env, sampler, shop, following, entity, instance, counters))


def order_generator(env, sampler, shop, arrival, counters):
    """Starts an order for every arrival, keyed orders-1, express-1 and so on."""
    gap = env.registry.get_config(f"Shop.arrival.{arrival}")
    for number in itertools.count(1):
        yield env.timeout(sampler.sample(gap))
        entity = Entity(f"{arrival}-{number}", arrival, dict(ATTRIBUTES[arrival]))
        instance = next(counters["pick_order"])
        env.process(task(env, sampler, shop, "pick_order", entity, instance, counters))


def delivery_generator(env, sampler, shop):
    gap = env.registry.get_config("Shop.arrival.deliveries")
    for number in itertools.count(1):
        yield env.timeout(sampler.sample(gap))
        yield shop["stock"].put(53)
        value = {"event_type": "stock_delivered"}
        env.publish_event(
            f"deliveries-{number}", value, task="restock", instance=number - 1
        )
        env.stats.task_finished("restock", number - 1)
        env.stats.flow_finished("deliveries")


def run(until=3600, factor=0.0):
    env = DynamicRealtimeEnvironment(factor=factor)
    env.registry.register_sim_parameter(PARAMS)
    env.setup_egress([ConsoleEgress()])
    # Finished counts of every task and flow, at the end
    env.setup_stats("Shop", "basic", interval=3600)
    shop = {
        "pickers": DynamicResource(env, "Shop", "pickers"),
        "packers": DynamicResource(env, "Shop", "packers"),
        "stock": DynamicContainer(env, "Shop", "stock"),
    }
    sampler = Sampler(seed=42)
    # Weights draw from the routing task's own sequence, as the builder's do
    ROUTES["pack_order"].rng = sampler.generator("route.pack_order")
    # One counter per task, shared by the arrivals and the rules
    counters = {name: itertools.count() for name in TASKS}
    for arrival in ("orders", "express"):
        env.process(order_generator(env, sampler, shop, arrival, counters))
    env.process(delivery_generator(env, sampler, shop))
    try:
        env.run(until=until)
    finally:
        env.teardown()
    return env


if __name__ == "__main__":
    run()