Implementation:Danijar Dreamerv3 Parallel Replay

Knowledge Sources	DreamerV3
Domains	Reinforcement_Learning, Distributed_Systems, Experience_Replay
Last Updated	2026-02-15 09:00 GMT

Overview

Concrete tool for running the distributed replay process that serves experience data via RPC with rate-limited insertion and sampling provided by the DreamerV3 parallel module.

Description

The parallel_replay() function in embodied/run/parallel.py deserializes factory functions, creates train and eval replay buffers plus three data streams, sets up a SamplesPerInsert rate limiter, and starts a portal.Server with five RPC endpoints: add_batch, sample_batch_train, sample_batch_report, sample_batch_eval, and update.

The function runs an infinite loop that periodically saves checkpoints and logs replay statistics.

Usage

Spawned as a separate process by combined() or launched independently via config.script == 'parallel_replay'.

Code Reference

Source Location

• Repository: dreamerv3
• File: embodied/run/parallel.py
• Lines: L221-314

Signature

def parallel_replay(make_replay_train, make_replay_eval, make_stream, args):
    """
    Run the distributed replay process.

    Args:
        make_replay_train: bytes or Callable -> Replay (training buffer)
        make_replay_eval: bytes or Callable -> Replay (eval buffer)
        make_stream: bytes or Callable(replay, mode) -> Stream
        args: elements.Config with replay_addr, train_ratio, batch_size,
              batch_length, log_every, save_every, logger_addr
    """

Import

import embodied
embodied.run.parallel.parallel_replay(make_replay_train, make_replay_eval, make_stream, args)

I/O Contract

Inputs

Name	Type	Required	Description
make_replay_train	bytes or Callable	Yes	Factory for training replay buffer (cloudpickle-serialized)
make_replay_eval	bytes or Callable	Yes	Factory for evaluation replay buffer
make_stream	bytes or Callable	Yes	Factory for data stream iterators
args	elements.Config	Yes	replay_addr, train_ratio, batch_size, batch_length, save_every, log_every, logger_addr

Outputs

Name	Type	Description
RPC Server	portal.Server	Running at args.replay_addr with add_batch, sample_batch_train/report/eval, update endpoints
Checkpoints	Files	Periodic saves of replay_train, replay_eval, and limiter state

Usage Examples

# Typically spawned by combined(), but can be run standalone:
import embodied
from functools import partial as bind

embodied.run.parallel.parallel_replay(
    bind(make_replay, config, 'replay'),
    bind(make_replay, config, 'replay_eval', 'eval'),
    bind(make_stream, config),
    args)

Related Pages

Implements Principle

• Principle:Danijar_Dreamerv3_Distributed_Replay_Management