MPTransport

class torchrl.weight_update.MPTransport(pipe_connection, timeout: float = 10.0)

Multiprocessing transport using pipes.

Parameters:

• pipe_connection (mp.Pipe) – The pipe connection to use for communication.

• timeout (float) – The timeout for waiting for acknowledgment. Default is 10 seconds.

check_ack(message: str = 'updated') → None

Check for acknowledgment.

receive_weights(timeout: float = 1.0) → tuple[str, Any] | None

Receive weights from the pipe (used in worker process).

This method only handles weight update messages. Other messages (like “close”, “continue”, etc.) are ignored and should be handled by the main worker loop.

Returns:

Tuple of (model_id, weights) if weights were received, None if no data available or if a non-weight message was received.

send_ack(message: str = 'updated') → None

Send acknowledgment back to sender.

send_weights(model_id: str, weights: Any) → None

Send weights through the pipe.

Sends weights and waits for acknowledgment to ensure delivery.

send_weights_async(model_id: str, weights: Any) → None

Send weights through the pipe without waiting for acknowledgment.

Use wait_ack() to wait for acknowledgment after sending to all workers.

wait_ack() → None

Wait for acknowledgment from worker.