Markdown

The RL orchestrator control loop¶

Scope: the coordinating process in an asynchronous RL post-training system, the one that hands prompts to inference, collects finished rollouts, assembles training batches, and keeps the two sides from drifting apart. This page covers the component decomposition, how a rollout is stamped with the policy version that produced it, the admission formula that bounds off-policy age, and what breaks that bound in practice. The systems framing and the algorithmic correction it enables are in async and disaggregated RL; the weight-transfer half of the loop is delta weight sync.

Code claims are checked against AReaL at commit d31701c (3 August 2026), which is the implementation read for this page. The Python block was executed with numpy. YAML snippets are unexecuted reference templates; pin the framework commit before relying on a field name.

What it is¶

An RL post-training run is three processes with different shapes. Inference turns a prompt into a rollout, which for a tool-using or multi-turn task means the whole trajectory: prompt, model turns, tool calls, environment replies. The trainer takes a batch of rollouts and applies a policy-gradient update. The orchestrator is what makes those two run at the same time instead of taking turns.

Nothing about that is optional at scale. Reasoning rollouts emit tens of thousands of tokens and finish at wildly different times, so a synchronous loop leaves one pool of GPUs idle while the other works. Turning asynchrony on typically halves wall-clock time; AReaL's own documentation puts the synchronous setting at "typically 2x slower".

The cost is that data goes off-policy. A rollout generated by policy version 3 may not reach the trainer until the policy is on version 5. The orchestrator's real job is not moving data. It is deciding how far apart the two sides are allowed to get, and enforcing that bound while keeping both busy.

Decomposed, the orchestrator is a small number of independent parts wired together by one process that owns them:

A dispatcher. Decides which prompts run next, keeps as much work in flight as capacity allows, and drops finished rollouts onto an output queue. In AReaL this is a BatchTaskDispatcher with separate input and output queues, a submission loop, a fetch loop, and explicit pause() and resume().
A sink that turns rollouts into training batches. Tokenize, compute advantages, filter, accumulate, and emit a batch when one is full. Eval rollouts go to a parallel sink with no advantage or filtering logic.
A weight watcher. Polls for a new trainer checkpoint, tells inference to reload it, and bumps a shared policy version that the rest of the system reads.
A capacity gate. Decides whether a new rollout may be admitted at all, given how many are already in the system and how far ahead of the trainer generation has run.
A metrics loop. Queue depths, stage timings, and the current lag, on a fixed interval, independent of the training path.

Why use it¶

Both pools stay busy. Generation and training overlap instead of alternating, which is the entire reason for the disaggregated layout.
Staleness becomes a number you set. Without an explicit bound, off-policy age is whatever falls out of your concurrency settings, so a throughput change silently changes your data distribution.
The trainer gets what it needs to correct. Stamping each rollout with the version that generated it is what lets the loss reweight stale samples, whether by truncated importance sampling or a decoupled objective.
Failures become visible. Queue depth and lag distinguish a slow environment from a slow trainer from a stalled weight sync, which look identical from a throughput graph alone.

When to use it (and when not)¶

Build or configure one when rollouts are long, variable, or tool-using, which is where the straggler problem dominates and synchronous batching wastes the most.
Set the staleness bound to zero when debugging. That degenerates to a synchronous loop and is the correct baseline for isolating an algorithmic bug from an asynchrony bug.
Keep it synchronous for short, uniform-length rollouts on a single node. The coordination machinery costs engineering time it will not repay.
Do not raise the staleness bound to fix a throughput problem caused by something else. If the trainer is the bottleneck, more in-flight rollouts just age on a queue.
Do not treat the bound as a free parameter. Typical published values are 2 to 8; the correction term in the loss has to be able to absorb whatever you pick.

Architecture¶

flowchart LR
  subgraph ORCH["Orchestrator process"]
    GATE{"Capacity gate"}
    DISP["Dispatcher: in-queue, out-queue, pause/resume"]
    SINK["Train sink: tokenize, advantages, filter, batch"]
    ESINK["Eval sink"]
    WATCH["Weight watcher: poll, reload, bump version"]
    LOG["Metrics loop: queue depth, lag, stage timings"]
  end
  GATE -->|"capacity > 0"| DISP
  DISP -->|"prompts"| INF["Inference engines (vLLM / SGLang)"]
  INF -->|"finished rollouts"| DISP
  DISP -->|"out-queue"| STAMP["Stamp env, group, episode, policy version"]
  STAMP --> SINK
  STAMP --> ESINK
  SINK -->|"full batch"| HOLD{"Trainer caught up?"}
  HOLD -->|"yes"| TR["Trainer"]
  HOLD -->|"no"| WAIT["Park until version advances"]
  WAIT --> HOLD
  TR -->|"new checkpoint"| WATCH
  WATCH -->|"reload"| INF
  WATCH -->|"version++"| GATE
  GATE -.-> LOG
  SINK -.-> LOG

The bound can be enforced on either side of the loop, and the choice is a real design decision. Admission-side gating refuses to start a rollout that would be too stale by the time it is consumed. Ship-side gating assembles the batch regardless and parks before handing it to the trainer until the trainer has caught up. Admission-side wastes no generation; ship-side is simpler and keeps the inference fleet saturated at the cost of work that may sit on a queue. AReaL takes the admission-side route.

How to use it¶

The admission gate is where the bound actually lives, and it is small enough to model exactly. AReaL's StalenessManager.get_capacity() computes:

capacity = min(max_concurrent - running, (S + version + 1) * B - (accepted + running))

where S is rollout.max_head_offpolicyness, B is the consumer batch size, and version is the current policy version. The block below drives that formula through a full run and tries to break it three ways.

# staleness_gate.py: executed model of the admission gate that bounds rollout age.
import numpy as np


class Gate:
    """capacity = min(concurrency slack, (S + version + 1) * B - in-system samples)."""

    def __init__(self, max_staleness, batch_size, max_concurrent, version=0, accepted=0):
        self.S, self.B, self.C = max_staleness, batch_size, max_concurrent
        self.version, self.accepted, self.running = version, accepted, 0

    def capacity(self):
        return min(self.C - self.running,
                   (self.S + self.version + 1) * self.B - (self.accepted + self.running))

    def pending_limit(self):
        return (self.S + 1) * self.B


def run(max_staleness, steps=60, batch=8, concurrent=64, latency=(3, 3),
        recover_at=0, fix_recovery=True, seed=0):
    """Admit, complete and train to `steps`; return observed ages and admissions/step."""
    rng = np.random.default_rng(seed)
    g = Gate(max_staleness, batch, concurrent, version=recover_at)
    if recover_at and fix_recovery:
        g.accepted = recover_at * batch                     # on_version_recovered
    inflight, done, ages, admitted = [], [], [], []
    lo, hi = latency
    for _ in range(steps):
        n = 0
        while g.capacity() > 0:
            wait = lo if hi <= lo else int(rng.integers(lo, hi))
            inflight.append((g.version, wait))              # (birth version, remaining)
            g.running += 1
            n += 1
        admitted.append(n)
        inflight = [(v, t - 1) for v, t in inflight]
        ripe = [v for v, t in inflight if t <= 0]
        inflight = [(v, t) for v, t in inflight if t > 0]
        g.running -= len(ripe)
        g.accepted += len(ripe)
        done.extend(ripe)                                   # consumed in completion order
        while len(done) >= batch:
            versions, done = done[:batch], done[batch:]
            ages.extend(g.version - v for v in versions)
            g.version += 1
    return np.array(ages), np.array(admitted)


# With uniform rollout latency the configured bound is exact: nothing is trained on
# more than max_staleness versions late, and the bound is actually reached.
for S in (0, 1, 2, 4):
    ages, _ = run(S)
    assert ages.size and ages.max() == S, (S, ages.max())
assert run(0)[0].max() == 0                      # S=0 degenerates to synchronous

# Heterogeneous rollout latency breaks that. Completions land out of admission order,
# so a slow rollout is consumed in a later batch than the gate accounted for and the
# observed age overshoots the configured bound. The mean still tracks S.
for S, observed in ((1, 2), (2, 4), (4, 7)):
    ages, _ = run(S, latency=(1, 6))
    assert ages.max() == observed > S
    assert abs(ages.mean() - S) < 1.0
# The overshoot scales with the spread of completion times, not with S alone.
assert run(2, latency=(1, 3))[0].max() < run(2, latency=(1, 12))[0].max()

# Remove the staleness term and the concurrency cap sets the age instead, at exactly
# one batch of slack per open batch. Staleness is then a side effect of a throughput
# knob: doubling generation capacity doubles how off-policy the data is.
ungated = {C: run(10**9, steps=80, concurrent=C)[0] for C in (16, 32, 64, 128, 256)}
for C, ages in ungated.items():
    assert ages.max() == C // 8 - 1, (C, ages.max())
assert ungated[256].max() == 31 > ungated[16].max() == 1
# With the term in place the two knobs separate: concurrency sizes the fleet, S sets
# the age, and the smaller of the two binds.
for C in (16, 64, 256):
    assert run(2, steps=80, concurrent=C)[0].max() == min(2, C // 8 - 1)

# Throughput is what the bound costs. S=0 keeps one batch in flight; each extra
# version of slack buys another batch of overlap between generation and training.
for S in (0, 1, 2, 4):
    g = Gate(S, 8, 10**6)
    assert g.pending_limit() == (S + 1) * 8 == g.capacity()
assert Gate(4, 8, 12).capacity() == 12           # concurrency can bind first

# Resume is where the formula bites. The version jumps to V while the accepted counter
# is still zero, so the staleness term opens (S + V + 1) * B slots at once and the
# bound is lost for the rest of the run.
assert Gate(2, 8, 10**6, version=40, accepted=0).capacity() == 344
assert Gate(2, 8, 10**6, version=40, accepted=40 * 8).capacity() == 24
bad_ages, bad_admits = run(2, recover_at=40, fix_recovery=False, concurrent=10**6)
good_ages, good_admits = run(2, recover_at=40, fix_recovery=True, concurrent=10**6)
assert (bad_admits[0], good_admits[0]) == (344, 24)
assert bad_ages.max() == 42 and good_ages.max() == 2
# A concurrency cap masks the size of the burst without restoring the bound: the run
# still trains on data seven versions old where two was configured.
clipped_ages, clipped_admits = run(2, recover_at=40, fix_recovery=False, concurrent=64)
assert clipped_admits[0] == 64 and clipped_ages.max() == 7 > 2

print(
    f"uniform latency: max age == S for S in 0,1,2,4 | "
    f"jittered latency S=1/2/4 -> "
    f"{[int(run(S, latency=(1, 6))[0].max()) for S in (1, 2, 4)]} "
    f"| ungated max age by concurrency "
    f"{ {C: int(a.max()) for C, a in ungated.items()} } | "
    f"resume burst {bad_admits[0]} vs {good_admits[0]} admissions, "
    f"max age {bad_ages.max()} vs {good_ages.max()}"
)

Executed output:

uniform latency: max age == S for S in 0,1,2,4 | jittered latency S=1/2/4 -> [2, 4, 7] | ungated max age by concurrency {16: 1, 32: 3, 64: 7, 128: 15, 256: 31} | resume burst 344 vs 24 admissions, max age 42 vs 2

Four things fall out, and the second and fourth are the ones worth carrying:

With uniform completion times the bound is exact and tight. Nothing exceeds S, and S is reached, so the parameter means what it says.
With variable completion times, observed age overshoots the configured bound. At S=4 the model trains on data 7 versions old. Rollouts finish out of admission order, so a slow trajectory lands in a later batch than the gate accounted for. The mean still tracks S, which is why this hides in aggregate metrics. Long-tail agentic rollouts are exactly the regime where the spread is widest.
Without the staleness term, concurrency sets the age. Max age comes out at concurrency / batch_size - 1 exactly, so a fleet-sizing decision silently becomes a data-distribution decision. The staleness term is what separates the two knobs.
Resuming a checkpoint can silently unbound the run. The formula reads the live version, so restarting at version 40 with a zeroed accepted counter opens 344 slots instead of 24 and produces data 42 versions stale where 2 was configured. AReaL handles this with an explicit on_version_recovered adjustment whose docstring calls out "a burst of submissions and unbounded staleness growth". A concurrency cap hides the size of the burst without restoring the bound.

How to develop with it¶

Stamp versions per token, not per rollout. When inference reloads weights mid-generation, one trajectory spans policy versions. AReaL computes a head (oldest), a tail (newest), and a run-length encoding of versions across the loss-masked tokens, and its config help for max_head_offpolicyness reads "Maximum off-policyness for the head". The bound is on the oldest contributing version, not on an average or on the trajectory as a whole.

# Reference template: AReaL rollout and actor config at commit d31701c.
rollout:
  max_head_offpolicyness: 4   # 0 reverts to synchronous RL; published range is 2-8
  consumer_batch_size: 8      # batch size for consuming rollouts from the queue
  max_concurrent_rollouts: 64 # defaults to consumer_batch_size when unset
  queue_size: 1024            # input/output queue size for async rollout
actor:
  use_decoupled_loss: true    # off-policy objective
  recompute_logprobs: true    # required when use_decoupled_loss is true

Keep the components ignorant of each other. The dispatcher should not know what a sink is; the sink should not know how weights are transferred. The orchestrator holds the shared objects (the policy handle, the engine pool, the environment registry) and every component reads the version off the same handle, so a version bump propagates without a message.

Two invariants belong in the loop itself rather than in a monitoring dashboard, because both fail silently:

A drain phase. When the run reaches its step target, stop admitting new training rollouts but keep the loop turning so in-flight work finishes and gets checkpointed. Killing the loop at the step count throws away GPU-hours of completed generation and produces a checkpoint that does not match the data on disk.
An empty-batch guard. If a batch arrives with nothing trainable after filtering, count it, and stop the run after a small number of consecutive empties. A broken environment or a reward function returning a constant produces exactly this, and without a guard the run spins for hours consuming budget and emitting a flat loss curve rather than an error.

How to maintain it¶

Report the age distribution, not the configured bound. Log the maximum and the p99 of the observed per-token head offset. The executed model shows these diverge from the configured value whenever completion times vary, which is always.
Alarm on the resume path specifically. The first step after a restart is where the bound is most likely to be lost. Assert that admissions in step one are at most (S + 1) * B and fail the run if they are not.
Change one knob at a time. Concurrency, consumer batch size, and staleness all move the same quantity through the same formula. Changing two at once makes a regression unattributable.
Pin the trainer and inference-engine commits together. The weight-transfer contract and the version-bump semantics are a single compatibility unit, the same way delta weight sync requires it.
Exclude rollouts that did not come from the live policy. Replayed, cached, or externally supplied trajectories carry no meaningful policy version. Averaging them into the age statistic drags it toward zero and hides real staleness in the rollouts that do come from generation.

How to run it in production¶

Instrument the loop as a pipeline, because that is what it is, and the interesting failures are always a specific stage rather than the whole thing:

Signal	Reads as healthy	What a regression means
Dispatcher out-queue depth	non-empty, stable	empty means generation is the bottleneck; growing means the trainer is
Admissions per step	near `(S + 1) * B` at steady state	zero means the gate is closed; a spike means the counters are wrong
Observed head off-policy age (max, p99)	at or slightly above `S`	a large gap means completion-time spread, not a config error
Weight-watcher poll to reload latency	bounded, well under a step	a growing value stalls the version and therefore the gate
Trainable samples per batch after filtering	stable fraction	trending to zero is a broken environment or reward
Time parked waiting for the trainer	small and steady	growing means generation is running ahead of what the bound permits

Operationally, three things pay for themselves. Persist the rollouts that were actually trained on, alongside the batch, before shipping it; when a run diverges, the question is always which data caused it. Make the shutdown path bounded, so cleanup gets a fixed window and then the process is killed, on the premise that everything important is already on disk. And keep a synchronous configuration (max_head_offpolicyness: 0) runnable at all times, because it is the only clean way to test whether a suspected algorithmic bug is actually an asynchrony bug.

Failure modes¶

Configured bound read as observed bound. The parameter bounds admission. Variable completion times push observed age above it, by 75% at S=4 in the executed model. Measure it.
Resume without adjusting the accepted counter. The single highest-impact defect in this design: a large burst of admissions and a bound that never re-establishes for the rest of the run.
Concurrency raised for throughput. Without a staleness term this raises off-policy age one-for-one. With one, it does nothing until it exceeds the staleness limit, at which point extra capacity buys nothing and the fleet sits idle.
Staleness raised to hide a slow trainer. More in-flight rollouts do not make the trainer faster; they make the data older while the same queue grows.
Per-rollout version stamping under partial rollouts. A trajectory spanning versions 3 through 5 stamped as "5" underreports age exactly on the tokens most in need of correction.
No drain at the step target. In-flight generation discarded, and a final checkpoint that does not correspond to the last data trained on.
No empty-batch guard. A broken environment produces a run that consumes GPU-hours and emits no error.
Weight-watcher stall. If the version stops advancing, an admission-side gate closes and generation stops; a ship-side gate parks the trainer instead. Either way the symptom is a quiet halt, not a crash.
Off-policy correction not matched to the bound. Raising the staleness limit without a decoupled objective or importance-ratio cap moves instability from the systems layer into the loss.

References¶

AReaL (asynchronous RL system; the implementation read for this page): https://github.com/inclusionAI/AReaL
AReaL, "AReaL: A Large-Scale Asynchronous Reinforcement Learning System for Language Reasoning": https://arxiv.org/abs/2505.24298
AReaL asynchronous-RL guide (off-policyness control, decoupled PPO objective, partial rollouts): https://github.com/inclusionAI/AReaL/blob/main/docs/en/algorithms/async.md
StalenessManager, the capacity formula and the checkpoint-recovery adjustment modelled above: https://github.com/inclusionAI/AReaL/blob/main/areal/infra/staleness_manager.py
Noukhovitch et al., "Asynchronous RLHF: Faster and More Efficient Off-Policy RL for Language Models": https://arxiv.org/abs/2410.18252
DeepSeek-AI, "DeepSeek-R1: Incentivizing Reasoning Capability in LLMs via Reinforcement Learning": https://arxiv.org/abs/2501.12948
Kwon et al., "Efficient Memory Management for Large Language Model Serving with PagedAttention" (vLLM, the rollout backend): https://arxiv.org/abs/2309.06180
Zheng et al., "SGLang: Efficient Execution of Structured Language Model Programs": https://arxiv.org/abs/2312.07104