NVIDIA-NeMo / NVIDIA-NeMo/Curator

Add way to wait for ray client to start() before yielding

Open
#1,494 2 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

enhancement
Dominant language
Python
Stars
1.8k
Forks
328
Avg merge
4d 5h
Merged PRs (30d)
30

Description

While trying to run Ray nightly I ran into this issue https://github.com/ray-project/ray/issues/60991

If that doesn't get fixed, we can do something like (but I'm not sure how hacky this would be for multi node)

#core/client.py

    def _wait_for_cluster_ready(self, gcs_address: str, timeout: int = 120) -> None:
        """Wait for the Ray cluster's GCS and raylet to be fully ready.

        Ray 3.0 introduced find_node_ids() which discovers raylets via psutil.
        If ray.init() is called before our raylet registers with GCS, it may
        pick up foreign raylets from other clusters on shared machines, causing
        a ConnectionError. This method blocks until at least one node has
        registered with our cluster's GCS.
        """
        from ray._raylet import GcsClient

        start = time.time()

        # Phase 1: Wait for GCS to be reachable
        client = None
        while time.time() - start < timeout:
            try:
                client = GcsClient(address=gcs_address)
                client.internal_kv_get(b"cluster_ready_check", None)
                break
            except Exception:
                time.sleep(1)
        else:
            msg = f"Ray GCS at {gcs_address} did not become reachable within {timeout}s"
            raise TimeoutError(msg)

        logger.info(f"Ray GCS at {gcs_address} is reachable, waiting for raylet to register...")

        # Phase 2: Wait for at least one node (raylet) to register with GCS.
        # GcsNodeInfo is a protobuf: state=0 means ALIVE, state=1 means DEAD.
        while time.time() - start < timeout:
            try:
                nodes = client.get_all_node_info()
                alive_nodes = [n for n in nodes.values() if n.state == 0]
                if alive_nodes:
                    elapsed = time.time() - start
                    logger.info(f"Ray cluster ready: {len(alive_nodes)} node(s) registered with GCS ({elapsed:.1f}s)")
                    return
            except Exception:
                pass
            time.sleep(1)

        msg = f"No raylet registered with Ray GCS at {gcs_address} within {timeout}s"
        raise TimeoutError(msg)

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start in core/client.py and trace the Ray client start() path to see where it yields before the cluster is ready. Review Ray issue 60991 and the proposed GCS and raylet readiness checks, then validate that startup waits correctly for multi-node use before yielding.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
distributed-systems
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
50/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.