[async] async data buffer: unified filters and better observability

[yueming-yuan]

When rollout production outruns training consumption in fully-async mode, finished groups pile up in what is effectively an unbounded queue (asyncio.Queue(maxsize=1000) = 125 training steps of backlog), and FIFO consumption means training always eats the oldest, stalest data — with no bound and no visibility.

Stacked on #1717 (yueming/fully-async-class-api), where fully_async_rollout.py lives.

Buffer capacity with staleness-ranked eviction

--async-data-buffer-max-batches caps the DataBuffer in multiples of rollout_batch_size (default 2). On overflow the producer does not block; the buffer evicts stalest-first — recycling the evicted prompts into the data source for regeneration (same path as the existing abort/staleness recycling) — until nothing in the buffer is beyond --max-weight-staleness (when set and the engine version is known) and the buffer is back within capacity. Ranking:

1. the group whose stalest sample is oldest (min weight version),
2. ties by the larger summed staleness,
3. remaining ties oldest-arrival-first.

The ranking needs no engine-version query: ranking by (min weight version, summed weight versions) is identical to ranking by (largest sample staleness, largest summed staleness) — with equal group sizes the current version is a constant offset that cancels. The version is only consulted for the --max-weight-staleness threshold.

0 disables the cap and restores the legacy behavior: a large bound that blocks the producer when full.

Evicting at insertion also relieves the drain-time --max-weight-staleness filter, which otherwise discovers stale groups only at consumption, one drain at a time, while training waits on the recycle-regenerate churn.

LIFO option

--async-data-buffer-order lifo consumes the freshest group first, keeping updates closest to on-policy without throttling production. Old groups then sink and only leave via eviction or a late (stale) drain, so lifo is meant to pair with the cap and/or --max-weight-staleness; the help text says so. Default stays fifo.

Metrics

• rollout/fully_async/evicted_stale_groups, evicted_overflow_groups, evict_rate (evictions / groups entering the buffer since the previous drain)
• rollout/fully_async/buffer_avg_staleness, buffer_max_staleness over the groups sitting in the buffer at drain time
• the consumed avg_staleness / max_staleness now report unconditionally, not only when --max-weight-staleness is set (previously the feature had to be on before you could see whether you needed it)

Sizing note

Measured on the current GLM-5.2 16-node run: steady-state queue is 0–5 groups and the post-eval burst is bounded by the in-flight cap (16 groups at --async-max-concurrent-samples 128), so ~2 training batches of capacity absorbs every observed burst while bounding queue-induced staleness to ~2 steps — hence the default of 2.

Tests

DataBuffer unit tests (overflow eviction, threshold-first eviction, summed-staleness tie-break, lifo order, staleness stats) plus a drain-level metrics-wiring test; existing tests updated for the two new args. 19 passed, 3 consecutive runs.

🤖 Generated with Claude Code

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[guapisolo]

Why the blocking logic removed here?

In the current impl. When the buffer is full, we just evicted stale groups

This design also force data buffer hold weight_version info, which introduce some semantic confusion for weight_version in DataBufferInput class. (it's actually observed engine weight version rather than the group's min weight version.).

What we achieve is just save some cpu memory when kick out staled groups.

[guapisolo]

I suggest the behavior change:

When the data buffer is full, block the data_source -- prompt --> generation_pool path to stop too many rollouts.

[guapisolo]

also a tiny fix on this #2241

[guapisolo]

LGTM except for the weight version handling.

[Shi-Dong]

LGTM