Function ray_worker_node_wait

autotest/utils/ray_distributed_utils.py:273–307

Worker node waits for Ray master node (Head Node) to be alive (by detecting GCS service port)

(manager: RayLMDeployManager, timeout_minutes: int = 60)

def ray_worker_node_wait(manager: RayLMDeployManager, timeout_minutes: int = 60):
    """Worker node waits for Ray master node (Head Node) to be alive (by
    detecting GCS service port)"""
    if manager.is_master:
        return

    print(f'⏸️ Worker node {manager.node_rank} entering wait mode...')
    max_checks = (timeout_minutes * 60) // WORKER_WAIT_INTERVAL
    consecutive_failures = 0
    max_consecutive_failures = 3

    for i in range(max_checks):
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.settimeout(CONNECTION_CHECK_TIMEOUT)
                if sock.connect_ex((manager.master_addr, RAY_PORT)) == 0:
                    consecutive_failures = 0
                else:
                    consecutive_failures += 1
        except Exception:
            consecutive_failures += 1

        if consecutive_failures >= max_consecutive_failures:
            print('📡 Ray master node GCS service unreachable, worker node exiting')
            break

        if i % 4 == 0:
            elapsed = (i * WORKER_WAIT_INTERVAL) // 60
            print(f'⏳ Worker node {manager.node_rank} waiting... Running for {elapsed} minutes')

        time.sleep(WORKER_WAIT_INTERVAL)
    else:
        print(f'⏰ Worker node {manager.node_rank} wait timeout ({timeout_minutes} minutes)')

    manager.cleanup()