infrastructure-distributed-deadlock-when-a-ray-worker-raises-an-exception-before-calli-b3d60ffe

THEN Use multithreading to concurrently wait for worker exceptions via ray.wait and initialize the process group. Alternatively, implement a timeout or kill all workers upon the first exception to prevent deadlock.

import threading
import ray

def worker_fn():
    raise RuntimeError("early error")

ray.init()
handle = ray.remote(worker_fn).remote()

def wait_for_worker():
    try:
        ray.get(handle)
    except Exception as e:
        print(f"Worker failed: {e}")
        # abort main process group init

thread = threading.Thread(target=wait_for_worker, daemon=True)
thread.start()
# proceed with dist.init_process_group with timeout or check thread status