> ## Documentation Index
> Fetch the complete documentation index at: https://docs.reactor.inc/llms.txt
> Use this file to discover all available pages before exploring further.
# The Run Loop
> How run() works, emitting frames, and controlling the frame rate.
`run()` is where your model's logic lives. The runtime calls it once after `load()` and it runs for
the lifetime of the model. What happens inside is entirely yours.
## Basic pattern
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
async def run(self) -> None:
while True:
# ✅ Park here while nobody is watching
await self.connected.wait()
while self.connected.is_set():
frame = self.pipe.forward(prompt=self.prompt)
# ✅ Hand the frame to the transport
await self.emit(MyOutput(main_video=frame))
```
Each iteration of the inner loop runs your forward pass, emits the result, and re-checks whether
anyone is still connected.
The two-loop shape matters: the outer loop keeps the model alive across clients, while the inner one
runs only while someone is watching. Without it, an idle model would spin a GPU generating frames
nobody receives.
`run()` is required. A model whose whole job lives in `@event` handlers still needs one, and parks
instead of generating:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
async def run(self) -> None:
await asyncio.Event().wait()
```
## The connection signal
`self.connected` is an `asyncio.Event` the runtime owns. It is **set** while at least one client is
connected and **cleared** once the last one leaves.
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
await self.connected.wait() # block until someone is here
self.connected.is_set() # is anyone still here?
```
You never set or clear it yourself. It is set before the first `@connected` handler runs and cleared
before the `@disconnected` handler for the last client to leave. For a model serving one client at a
time, that is simply the moment they arrive and the moment they go.
## Emitting frames
`emit()` takes an instance of your `Output` class, with one payload per declared track:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
await self.emit(MyOutput(main_video=frame))
```
Every track on the class must be supplied. A video payload is an RGB `uint8` array shaped
`(H, W, 3)`; an audio payload is `int16` shaped `(1, samples)`.
`emit()` waits while the frames it already handed downstream are still playing, so a model that
generates faster than its playout rate is throttled to that rate instead of piling up latency. The
wait happens off the model's event loop, so commands and lifecycle hooks keep dispatching while it
holds. Rate limiting is the runtime's job.
A producer that would rather skip a frame than wait — anything driven by a live source, where the
newest frame is the only one worth sending — passes `drop=True`, and the overflow is discarded
downstream:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
await self.emit(MyOutput(main_video=frame), drop=True)
```
### Batches
Models that produce several frames per forward pass can emit them in one call. Pass a `(N, H, W, 3)`
array and the runtime splits it into individual frames downstream:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
frames = self.pipe.forward(prompt=self.prompt)
print(frames.shape) # (10, 720, 1280, 3)
# ✅ Emit the whole batch in one call — the runtime paces it out
await self.emit(MyOutput(main_video=frames))
```
Emit the full batch in a single call rather than looping over it. The runtime uses the batch as the
unit it paces against, so splitting it yourself produces choppier playback.
## Frame rate
Each emitted chunk carries the rate its frames should play out at. There are two ways to set it.
**Declare a fixed rate** with the `fps` class attribute when your model produces at a predictable
speed:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
class MyModel(ReactorModel):
fps = 24
```
It defaults to 30.
**Measure your own compute time** when throughput varies, and the runtime derives the rate from it:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
t0 = time.perf_counter()
frames = self.pipe.forward()
# ✅ The chunk plays back at the rate you actually produced it
await self.emit(MyOutput(main_video=frames), compute_time=time.perf_counter() - t0)
```
The playback rate becomes `n_frames / compute_time`, so the stream tracks your real throughput. A
model that speeds up after a warm-up, or slows down under a heavier setting, stays in sync without
you touching `fps`. When you pass `compute_time`, `fps` is ignored.
Pass the honest measured time or none at all. Playout follows the tag, so a doctored
`compute_time` makes the model permanently outrun its own playback.
## Controlling playout
`self.output` is the model's handle onto its outbound stream. It is bound for you, and every
operation on it fans out to each connected client — including clients that connect later.
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
self.output.fps = 24 # re-pace what is queued, and tag emits from here on
self.output.flush() # drop what is queued and cut playout to black
await self.output.emit(MyOutput(main_video=frame)) # self.emit() is an alias
```
Assigning `fps` re-paces frames that are already queued rather than waiting for the next emit, which
is what makes it usable from a command handler:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
@event(name="set_speed", description="Playback rate in frames per second")
async def set_speed(self, fps: float = InputField(default=24, ge=1, le=60)) -> None:
self.output.fps = fps
```
The assignment holds until a chunk emitted with `compute_time` supersedes it, so it is a durable
setting on a model that declares `fps` and a one-chunk nudge on a model that measures every pass.
Call `flush()` when generation resets or restarts — a new scene, a cleared prompt, a seek. It drops
everything queued, releases a producer waiting in `emit()`, and cuts the client to black, so none of
the old content plays after the reset:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
@event(name="reset", description="Clear the scene and start over")
async def reset(self) -> None:
self._step = 0
self.output.flush()
```
The recording keeps running across the cut, so a clip spanning a reset contains both sides of it.
## Buffered latency
`buffer_size` declares how many frames may sit between your model and each client. It is the
latency bound: a smaller value means a command takes effect on screen sooner, and a larger one
absorbs more variance in your generation speed.
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
class MyModel(ReactorModel):
fps = 24
buffer_size = 8
```
Leave it undeclared to accept the runtime's default. The bound is never applied below one emitted
chunk, so a model that emits batches always fits a whole batch no matter how small the number. A
declared value must be positive; zero or less fails at startup.
## Sending messages
Send structured data to clients from inside `run()`:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
await self.send(Progress(step=self._step, total=100))
```
See [Events & Messages](/deploy/development/reactor-model/events-and-messages#outbound-messages) for
defining message types.
## Concurrency
`@event` handlers run on the same event loop as `run()`, so a handler can fire at any `await` in
your loop. If you read an attribute, await something, and read it again, it may have changed in
between.
Snapshot what a forward pass depends on before you start it:
```python theme={"theme":{"light":"github-light","dark":"github-dark-high-contrast"}}
async def run(self) -> None:
while True:
await self.connected.wait()
while self.connected.is_set():
# ✅ Snapshot before the forward pass — handlers can fire at any await
prompt = self.prompt
step = self._step
frame = self.pipe.forward(prompt=prompt, step=step)
await self.emit(MyOutput(main_video=frame))
self._step += 1
```
This keeps one frame internally consistent. A command that lands mid-pass takes effect on the next
frame rather than half-applying to the current one.
## Next
Session hooks, multiple clients, and what resets when.
One event loop, shared by your loop and your handlers.