Multi-GPU inference

Run your parallel model across GPUs without building your own worker infrastructure. Reactor handles worker startup, GPU assignment, coordinated calls, shared-memory transfer, and failure reporting. Workers load once and keep weights and model state between calls.

All workers cooperate on the same request. Each GPU runs a separate instance of the same worker class. Every step calls the same generate(input) implementation with the same input on every worker. Your model uses rank to divide the computation and coordinate GPU-to-GPU communication. The runner waits for all workers and returns rank 0’s complete result. If ranks produce separate parts of the output, your model must gather them onto rank 0 inside generate() before returning. The runner does not assemble outputs.

This fits models that already implement tensor or sequence/context parallelism within a generation step. Ranks can hold different weights or tensors. The runner does not partition the model, route independent requests, or schedule pipeline stages. All workers run on one machine.

An application calls DistributedRunner, which sends the same input to two GPU workers on one machine. The model coordinates their computation. Rank 0 returns the complete result. — One call reaches every GPU worker. The runner waits for all workers and returns rank 0's result.

Development preview: use a compatible runtime build, such as 3.5.0.dev99. Stable 3.5.0 and the linked LingBot recipe’s current runtime pin do not include these APIs.

Connect your model

Extend DistributedWorker and implement load(), generate(), and reset(). The base class declares rank, world_size, and device and provides is_leader for rank-0 checks. The runner assigns the attributes before load().

Define your worker

This adapter wraps the parallel model from your video_model.py module. Here, ParallelVideoModel accepts a weights path, device, rank, and worker count. It provides generate() and reset(). Adapt those calls to your model’s API.

video_worker.py

from pathlib import Path

from reactor_runtime.distributed import DistributedWorker
from video_model import ParallelVideoModel, RolloutExhausted, VideoInput, VideoResult


class VideoWorker(DistributedWorker):
    def load(self, weights_root: Path) -> None:
        self.model = ParallelVideoModel(
            weights_root=weights_root,
            device=self.device,
            rank=self.rank,
            world_size=self.world_size,
        )

    def generate(self, input: VideoInput) -> VideoResult:
        return self.model.generate(input)

    def reset(self) -> None:
        self.model.reset()

VideoInput(prompt=...), VideoResult, and RolloutExhausted also come from your model module. The result holds CPU RGB frames in result.frames. In this example, every rank raises RolloutExhausted when a rollout ends, and resetting lets the next prompt start another rollout. This is a model-specific recovery contract. Only use that recovery branch if your model supports it. Keep the worker at module level with a no-argument constructor. Use picklable inputs and results, with contiguous CPU NumPy arrays for large payloads. Run one call at a time, on one machine.

Use the runner in a ReactorApp

Start the runner in load(), then call it from generate(). Client commands and tracks stay in the application. The model keeps its inference code and session state.

video_app.py

from pathlib import Path

from reactor_runtime import (
    ApplicationError,
    InputField,
    InputState,
    Output,
    ReactorApp,
    StepOutcome,
    TrackPayload,
    Video,
    event,
    get_weights_path,
    session_ended,
)
from reactor_runtime.distributed import DistributedRunner
from video_worker import RolloutExhausted, VideoInput, VideoResult, VideoWorker


class VideoState(InputState):
    prompt: str = InputField(default="A walk through a forest.")
    paused: bool = InputField(default=False)


class VideoOutput(Output):
    main_video: Video


class VideoApp(ReactorApp):
    state: VideoState

    def load(self, config_path: Path | None) -> None:
        self.runner = DistributedRunner(
            VideoWorker,
            world_size=2,
            load_kwargs={"weights_root": get_weights_path()},
        )
        self.runner.start()

    async def process_input(self) -> VideoInput:
        if self.state.paused:
            raise ApplicationError("paused")
        return VideoInput(prompt=self.state.prompt)

    def generate(self, input: VideoInput) -> VideoResult:
        return self.runner.generate(input)

    async def process_output(self, outcome: StepOutcome) -> VideoOutput | None:
        if isinstance(outcome.error, RolloutExhausted):
            self.reset()
            return None
        if outcome.error is not None:
            self.runner.shutdown()
            raise outcome.error
        return VideoOutput(main_video=TrackPayload(outcome.result.frames))

    @event(name="reset", description="Start a new rollout.")
    def reset(self) -> None:
        self.runner.reset()
        self.output.flush()

    @session_ended
    def on_session_ended(self) -> None:
        if self.runner.healthy:
            self.reset()
        else:
            self.runner.shutdown()

The reset command and session-end hook clear model state while keeping weights loaded. Commands and hooks run between steps. In process_output(), the expected model error resets the rollout and skips that output. Other errors shut down the runner and end the session. The change from an in-process model is in load(): construct and start DistributedRunner instead of constructing and loading the model directly. Calls to generate() and reset() keep the same interface. This isolates the model. Using multiple GPUs also requires the model’s own parallel implementation.

Allocate GPUs

Match the runner’s world_size to the GPU count in reactor.yaml:

model:
  resources:
    gpu:
      type: NVIDIA_B200
      count: 2

Follow Deploying models, or test locally.

Handle a failed step

The application example handles failures in process_output(). If every rank raises the same exception type, the runner stays healthy. Reset and continue only for a model error that you know is recoverable, such as the example’s RolloutExhausted. A crash, timeout, or rank disagreement leaves the runner unusable. The example shuts it down and re-raises the error. This ends the session and stops the model loop. A new runner is required before inference resumes. The runner does not restart workers or replay failed steps. See the error contract.

Complete example

LingBot uses the runner in place of a custom subprocess adapter, JSON request/reply loop, and temporary frame files. Its existing sequence-parallel inference stays in the model. See the worker methods and runner options for the full API.

Get Started

CLI & Platform

Build a Model

Media & Transport

Releases

Multi-GPU inference

Connect your model

Define your worker

Use the runner in a ReactorApp

Allocate GPUs

Handle a failed step

Complete example

​Connect your model

​Define your worker

​Use the runner in a ReactorApp

​Allocate GPUs

​Handle a failed step

​Complete example

Connect your model

Define your worker

Use the runner in a ReactorApp

Allocate GPUs

Handle a failed step

Complete example