infrastructure-distributed-initiali-ray-worker-raises-an-exception-before-initializing-46eac589

THEN During distributed initialization, use a separate thread to monitor Ray worker exceptions via `ray.wait` while the main thread initializes the process group. If any worker fails, kill all workers and report the error. Alternatively, use a timeout or a custom error-handling wrapper around distributed setup.

# Pseudocode: run dist.init_process_group in one thread and ray.wait in another
import threading
import ray

def init_dist(rank, world_size, init_method):
    # worker code
    ...

handles = [worker.remote() for worker in workers]

def monitor_workers(handles):
    while True:
        ready, _ = ray.wait(handles, num_returns=1, timeout=5)
        if ready:
            try:
                ray.get(ready[0])
            except Exception:
                for h in handles:
                    ray.kill(h)
                raise

t = threading.Thread(target=monitor_workers, args=(handles,))
t.start()
# main process then proceeds with dist.init_process_group