Skip to content

ray_distributed_executor

Classes

fastvideo.worker.ray_distributed_executor.RayDistributedExecutor 

RayDistributedExecutor(fastvideo_args: FastVideoArgs, *, log_queue=None)

Bases: Executor

Ray-based distributed executor

Source code in fastvideo/worker/executor.py 
18
19
20
21
22
23
24
25
26
27
def __init__(
    self,
    fastvideo_args: FastVideoArgs,
    *,
    log_queue=None,
):
    self.fastvideo_args = fastvideo_args
    self._log_queue = log_queue

    self._init_executor()

Methods:

fastvideo.worker.ray_distributed_executor.RayDistributedExecutor.set_log_queue 
set_log_queue(log_queue: Queue | None) -> None

Keep the driver-side queue locally.

multiprocessing.Queue is not picklable across Ray nodes, so worker logs stay in the Ray session log dir instead of being forwarded.

Source code in fastvideo/worker/ray_distributed_executor.py 
369
370
371
372
373
374
375
def set_log_queue(self, log_queue: Queue | None) -> None:
    """Keep the driver-side queue locally.

    ``multiprocessing.Queue`` is not picklable across Ray nodes, so worker
    logs stay in the Ray session log dir instead of being forwarded.
    """
    self._log_queue = log_queue

fastvideo.worker.ray_distributed_executor.RayWorkerMetaData dataclass 

RayWorkerMetaData(worker: ActorHandle, created_rank: int, adjusted_rank: int = -1, ip: str = '')

Metadata for a Ray worker. The order of ray worker creation can be random, and we need to reset the rank after creating all workers.

Functions:

fastvideo.worker.ray_distributed_executor.should_use_gloo_loopback 

should_use_gloo_loopback(worker_ips: list[str]) -> bool

Loopback is only safe when every worker shares one host IP.

Single-node Ray (one or many GPUs on the same box) can dial the Gloo store on loopback. Two Sparks already have distinct worker IPs, so this returns False and Gloo stays on the fabric address. Per-node NIC names are a separate issue: do not copy NCCL_SOCKET_IFNAME / GLOO_SOCKET_IFNAME from the driver onto those workers.

Source code in fastvideo/worker/ray_distributed_executor.py 
39
40
41
42
43
44
45
46
47
48
def should_use_gloo_loopback(worker_ips: list[str]) -> bool:
    """Loopback is only safe when every worker shares one host IP.

    Single-node Ray (one or many GPUs on the same box) can dial the Gloo store
    on loopback. Two Sparks already have distinct worker IPs, so this returns
    False and Gloo stays on the fabric address. Per-node NIC names are a
    separate issue: do not copy ``NCCL_SOCKET_IFNAME`` / ``GLOO_SOCKET_IFNAME``
    from the driver onto those workers.
    """
    return len(set(worker_ips)) <= 1