Skip to content

Commit 92beb9e

Browse files
sash-aclaude
andauthored
feat: better accumulator (#4)
* chore: remove double buffering simplify the accumulator by removing double buffering, it did nothing for us because .tobytes in the client is a copy and isn't async. If we do both of these in future then it may be worth adding back * feat: allow zero dimensional leaves with trajectory accumulator * feat: distinguish single-item from buffered timescales in TrajectoryAccumulator A timescale is now classified as either: - Buffered: every leaf shares leading dim N > 1; capacity = N, writes go to stored[s:s+1]. - Single-item: every leaf is 0-d or has shape[0] == 1. Capacity = 1, add() replaces the whole leaf via stored[...] = incoming. This makes the accumulator accept natural per-item shapes for one-shot trailing context (bootstrap step, episode return, param generation) without forcing callers to pad a leading 1 onto every leaf and squeeze it back out downstream. Other changes: - IndexError on add() past capacity now reports the timescale name, capacity, and offending index. - Buffered-timescale leading-dim mismatch raises with a hint that the caller can opt into single-item mode by making any leaf 0-d or all leaves shape (1, ...). - _write_slot simplified to a single stored[key] = incoming line, with key = Ellipsis for single-item and slice(s, s+1) for buffered. - Tests cover both detection paths, the mismatch error, the over-index error, and the single-item write semantics. - Docs (guide + API) describe both modes and the detection rule. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent d8de92a commit 92beb9e

4 files changed

Lines changed: 288 additions & 81 deletions

File tree

docs/src/api/trajectory-accumulator.md

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,4 +5,14 @@ client side before calling `client.send`. Useful when a single rollout step
55
produces several arrays at different rates (e.g. one transition per env
66
step, plus a single episode-level statistic per episode).
77

8+
Each timescale is either:
9+
10+
- **Buffered** — every leaf shares the same leading dim `N` (`N > 1`);
11+
that's the capacity, filled by `N` `add()` calls.
12+
- **Single-item** — capacity 1; detected when every leaf is 1-d or has
13+
at least one leaf that is 0-d. `add()` replaces the whole leaf.
14+
15+
See the [guide](../guides/trajectory-accumulator.md) for the rationale and
16+
worked examples.
17+
818
::: echo.trajectory_accumulator.TrajectoryAccumulator

docs/src/guides/trajectory-accumulator.md

Lines changed: 46 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,18 @@ samples.
1717
## How it works
1818

1919
Construct it with a dict whose top-level keys are timescale names. Each
20-
leaf's leading dimension is the number of `add()` calls expected before the
21-
buffer is ready to send.
20+
timescale is one of two kinds, inferred from the example pytree:
21+
22+
- **Buffered** — every leaf shares the same leading dim `N` (with `N > 1`).
23+
That leading dim is the timescale's capacity: the buffer fills after `N`
24+
`add()` calls and each call writes into `stored[s:s+1]`.
25+
26+
- **Single-item** — the timescale holds one trailing piece of context
27+
(e.g. an episode return, a bootstrap step) rather than a buffer of
28+
steps. Detected when at least one leaf is 0-d, *or* all leaves have
29+
`shape[0] == 1`. Capacity is `1` and `add()` replaces the whole leaf,
30+
so non-0-d leaves can carry any per-item shape (apart from the
31+
optional leading 1).
2232

2333
```python
2434
import numpy as np
@@ -27,12 +37,21 @@ from echo import TrajectoryAccumulator, TcpClient
2737
T = 64 # rollout length
2838

2939
example = {
40+
# Buffered timescale: leading dim T across every leaf.
3041
"step": {
3142
"obs": np.zeros((T, 4), dtype=np.float32),
32-
"reward": np.zeros((T, 1), dtype=np.float32),
43+
"reward": np.zeros((T,), dtype=np.float32),
3344
},
45+
# Single-item timescale: all leaves have shape (1, ...) — capacity 1,
46+
# `add()` replaces the whole leaf.
3447
"episode": {
3548
"return": np.zeros((1,), dtype=np.float32),
49+
"length": np.zeros((1,), dtype=np.float32)
50+
},
51+
# Single-item timescale: 0-d reward means add() replaces the whole leaf
52+
"final_step": {
53+
"obs": np.zeros((4,), dtype=np.float32),
54+
"reward": np.zeros((), dtype=np.int32),
3655
},
3756
}
3857

@@ -42,44 +61,47 @@ buf = TrajectoryAccumulator(example)
4261

4362
for _ in range(num_rollouts):
4463
episode_return = 0.0
64+
reward = 0.0
65+
obs = env.reset()
4566
for _ in range(T):
46-
obs, reward = env.step(...)
4767
buf.add("step", {"obs": obs, "reward": reward})
68+
obs, reward, ... = env.step(...)
4869
episode_return += float(reward)
4970

50-
buf.add("episode", {"return": np.array([episode_return], dtype=np.float32)})
71+
buf.add("episode", {"return": np.array([episode_return]), "length": np.array([length])})
72+
buf.add("final_step", {"obs": obs, "reward": reward})
5173
client.send(buf.build())
5274
```
5375

5476
## Mental model
5577

56-
`TrajectoryAccumulator` is just pre-allocated numpy arrays plus per-timescale slot
57-
counters. `add()` writes into the next free slot for that timescale via
58-
slice assignment; `build()` returns the filled pytree, flips the active
59-
buffer (so the next round writes into a fresh one without allocating), and
60-
resets the counters.
78+
`TrajectoryAccumulator` is just pre-allocated numpy arrays plus per-timescale
79+
slot counters. For buffered timescales, `add()` writes into the next free
80+
slot via slice assignment (`stored[s:s+1] = incoming`); for single-item
81+
timescales it replaces the whole leaf (`stored[...] = incoming`).
82+
`build()` returns the filled pytree and resets the counters. The buffer is
83+
reused — no allocation per rollout.
6184

6285
There's no network or Rust involvement; it's purely a way to amortise the
6386
flatten-and-send cost across many environment steps. The pytree it returns
6487
goes through the normal `client.send` path.
6588

66-
## Two buffers, no allocation per rollout
67-
68-
`TrajectoryAccumulator` double-buffers internally: two copies of the pytree, with
69-
`build()` swapping the active one. So while one buffer is being serialised
70-
and sent, the next rollout can start filling the other without any
71-
allocation. This matters when rollouts are short relative to flatten +
72-
network latency.
89+
The tree returned by `build()` aliases the accumulator's internal buffers,
90+
so the next `add()` will overwrite it. The usual `client.send(buf.build())`
91+
pattern is safe because `send` is synchronous and copies the bytes before
92+
returning — don't hold onto the returned tree across further `add()` calls.
7393

7494
## Common pitfalls
7595

76-
- **Leading-dimension mismatch within a timescale.** All leaves under one
77-
timescale key must share the same first axis size. That's what defines
78-
"how many `add` calls before the buffer is full". The constructor checks
79-
this and raises.
80-
- **Adding past capacity.** If you call `add` more than the leading
81-
dimension allows, you get `IndexError`. Call `reset()` if you want to
82-
abort a partial rollout.
96+
- **Leading-dimension mismatch in a *buffered* timescale.** Inside a
97+
buffered timescale, all leaves must share the same first axis size —
98+
that's what defines "how many `add` calls before the buffer is full".
99+
The constructor checks this and raises. If you actually meant a
100+
single-item timescale, make one leaf 0-d or add a leading dim to all leaves.
101+
- **Adding past capacity.** If you call `add` more times than the
102+
timescale's capacity, you get `IndexError` with the timescale name,
103+
capacity, and offending index. Call `reset()` to abort a partial
104+
rollout.
83105
- **Dict-only at the top level.** The top-level pytree must be a `dict`
84106
with timescale names. Below that, leaves can be any pytree shape that
85107
`optree` understands.

python/echo/trajectory_accumulator.py

Lines changed: 50 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -5,36 +5,56 @@
55

66

77
class TrajectoryAccumulator:
8-
"""Multi-timescale accumulator: fixed-size pytree buffer with double-buffering.
8+
"""Multi-timescale accumulator: fixed-size pytree buffer.
9+
10+
Per timescale, the example pytree determines how many ``add()`` calls
11+
fit before the buffer is full:
12+
13+
* **Buffered timescale** — every leaf shares the same leading dim ``N``
14+
(``N > 1``); the accumulator stores ``N`` per-add items into
15+
``stored[s:s+1] = incoming`` slot-by-slot. ``N`` becomes the
16+
timescale's capacity.
17+
18+
* **Single-item timescale** — the timescale holds one trailing piece of
19+
context (e.g. a bootstrap step, an episode return) rather than a
20+
buffer. Detected when at least one leaf is 0-d, or all leaves
21+
have ``shape[0] == 1``. Capacity is ``1``; ``add()`` replaces the
22+
whole leaf, so non-0-d leaves may have any per-item shape (apart
23+
from the optional leading 1).
924
1025
Args:
11-
example: Dict with timescale names as top-level keys. The leading
12-
dimension of each leaf array is the number of ``add()`` calls
13-
expected before the buffer is ready to send.
26+
example: Dict with timescale names as top-level keys. Each value is
27+
a pytree whose leaves declare the per-timescale layout per the
28+
rule above.
1429
"""
1530

1631
def __init__(self, example: dict[str, Any]):
1732
if not isinstance(example, dict):
1833
raise TypeError("example must be a dict with timescale names as top-level keys")
1934

2035
self._counts: dict[str, int] = {}
36+
self._single_item: dict[str, bool] = {}
2137
for name, subtree in example.items():
2238
leaves = optree.tree_leaves(subtree)
2339
if not leaves:
2440
raise ValueError(f"Timescale '{name}' has no array leaves")
25-
leading = leaves[0].shape[0] if leaves[0].ndim > 0 else 1
26-
if not all((leaf.shape[0] if leaf.ndim > 0 else 1) == leading for leaf in leaves):
27-
raise ValueError(
28-
f"All leaves in timescale '{name}' must share the same leading dimension"
29-
)
30-
self._counts[name] = leading
31-
32-
# Two copies of the pytree for double-buffering.
33-
self._trees: list[dict[str, Any]] = [
34-
{n: optree.tree_map(np.zeros_like, sub) for n, sub in example.items()},
35-
{n: optree.tree_map(np.zeros_like, sub) for n, sub in example.items()},
36-
]
37-
self._active = 0
41+
42+
# Single-item: any 0-d leaf OR every leaf with leading dim 1.
43+
if any(leaf.ndim == 0 for leaf in leaves) or all(leaf.shape[0] == 1 for leaf in leaves):
44+
self._counts[name] = 1
45+
self._single_item[name] = True
46+
else:
47+
leading = [leaf.shape[0] for leaf in leaves]
48+
if not all(s == leading[0] for s in leading):
49+
raise ValueError(
50+
f"All leaves in buffered timescale '{name}' must share the same "
51+
f"leading dimension (got {leading}); make any leaf 0-d or "
52+
f"all shape (1, ...) to mark the timescale single-item instead"
53+
)
54+
self._counts[name] = leading[0]
55+
self._single_item[name] = False
56+
57+
self._tree: dict[str, Any] = {n: optree.tree_map(np.zeros_like, sub) for n, sub in example.items()}
3858
self._slot: dict[str, int] = {name: 0 for name in example}
3959

4060
def add(self, name: str, data: Any) -> None:
@@ -43,24 +63,27 @@ def add(self, name: str, data: Any) -> None:
4363
raise KeyError(f"Unknown timescale '{name}'. Known: {list(self._counts)}")
4464
s = self._slot[name]
4565
if s >= self._counts[name]:
46-
raise IndexError(
47-
f"Timescale '{name}' is already full ({self._counts[name]} slots). "
48-
"Call reset() or build() before adding more."
49-
)
66+
raise IndexError(f"Timescale '{name}' has {self._counts[name]} slots, but you tried to add at index {s}")
67+
68+
# Single-item: replace the whole leaf
69+
# Buffered: write into the next slot of the leading dim.
70+
key = Ellipsis if self._single_item[name] else slice(s, s + 1)
5071

5172
def _write_slot(stored, incoming):
52-
stored[s:s + 1] = incoming
73+
stored[key] = incoming
5374
return stored
5475

55-
optree.tree_map_(_write_slot, self._trees[self._active][name], data)
76+
optree.tree_map_(_write_slot, self._tree[name], data)
5677
self._slot[name] += 1
5778

5879
def build(self) -> dict[str, Any]:
59-
"""Return the filled pytree and flip the active buffer."""
60-
tree = self._trees[self._active]
61-
self._active = 1 - self._active
80+
"""Return the filled pytree and reset slot counters.
81+
82+
The returned tree aliases internal buffers; callers must finish using
83+
it (e.g. complete the synchronous send) before the next ``add()``.
84+
"""
6285
self._slot = {name: 0 for name in self._slot}
63-
return tree
86+
return self._tree
6487

6588
def reset(self) -> None:
6689
"""Reset slot counters without sending (e.g. on episode abort)."""

0 commit comments

Comments
 (0)