[fix] Fail fast when SimpleStorage lacks Ray CPU capacity - #165
[fix] Fail fast when SimpleStorage lacks Ray CPU capacity#1650xAstraZhang wants to merge 2 commits into
Conversation
Signed-off-by: Astra Zhang <zhangyv7412@gmail.com>
CLA Signature Guide@0xAstraZhang , thanks for your pull request. The following commit(s) are not associated with a signed Contributor License Agreement (CLA).
To sign CLA, click here. To check if your email is configured correctly, refer to the FAQs. Once you've signed the CLA or updating your email, please comment |
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
There was a problem hiding this comment.
🟡 Changes recommended
A few issues could undermine the “fail fast” goal and error reporting reliability (per-handler timeout scaling, timeout cleanup masking errors, and potentially confusing CPU-capacity messaging).
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
This PR adds a bounded startup path for the SimpleStorage backend so TransferQueue doesn’t hang indefinitely when requested SimpleStorageUnit actors cannot be scheduled due to insufficient Ray CPU capacity. It introduces a timeout while retrieving ZMQ server info, cleans up partially created actors on failure, and rolls back controller/process state so users can retry safely.
Changes:
- Add an optional
timeouttoprocess_zmq_server_info()and propagate it toray.get(...). - In SimpleStorage bootstrap, record available CPU capacity, enforce a 60s startup timeout, kill created units on timeout, and raise an actionable error.
- In
transfer_queue.interface.init(), roll back controller/storage state viaclose()if storage initialization fails; add tests for timeout cleanup and rollback.
File summaries
| File | Description |
|---|---|
| transfer_queue/utils/zmq_utils.py | Adds a timeout parameter to ZMQ server-info retrieval. |
| transfer_queue/storage/bootstrap/simple_storage_bootstrap.py | Enforces a startup timeout for SimpleStorage and reports actionable CPU-capacity errors with cleanup. |
| transfer_queue/interface.py | Ensures init() rolls back controller state when storage initialization fails. |
| tests/test_simple_storage_scheduling.py | Adds coverage for timeout cleanup/error messaging and init() rollback behavior. |
Review details
- Files reviewed: 4/4 changed files
- Comments generated: 4
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| for storage_node in simple_storage_handles.values(): | ||
| ray.kill(storage_node) |
| else: | ||
| server_info = {} | ||
| for name, handler in handlers.items(): | ||
| server_info[name] = ray.get(handler.get_zmq_server_info.remote()) # type: ignore[union-attr, attr-defined] | ||
| server_info[name] = ray.get( # type: ignore[union-attr, attr-defined] | ||
| handler.get_zmq_server_info.remote(), timeout=timeout |
| f"Ray CPU capacity of {num_data_storage_units}, but Ray reported " | ||
| f"available CPU capacity of {available_cpus:g} " | ||
| "before startup. " |
| Args: | ||
| handlers: Dictionary of handler objects (controllers, storage managers or storage units), | ||
| or a single handler object | ||
| timeout: Maximum seconds to wait for each handler to return its server information. |
Suggested by Copilot Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
CLA Signature Guide@0xAstraZhang , thanks for your pull request. The following commit(s) are not associated with a signed Contributor License Agreement (CLA).
To sign CLA, click here. To check if your email is configured correctly, refer to the FAQs. Once you've signed the CLA or updating your email, please comment |
Closes #164
What does this PR do?
This PR prevents SimpleStorage startup deadlocks by timing out on insufficient Ray CPU resources, reporting an actionable error, and cleaning up partial initialization so users can retry safely.
Motivation
Related to verl-project/verl#7656.
According to the report, due to the absence of an exception for exceeding the timeout limit, when the number of CPU cores required by the TransferQueue at a certain node is greater than the actual number of available CPU cores, the process will be indefinitely suspended.
The older TransferQueue version in the report hangs on
placement_group.ready(). Current TransferQueuemainuses hard node affinity instead, but retains the same user-visible failure: an unscheduled SimpleStorageUnit waits indefinitely, whileprocess_zmq_server_info()blocks inray.get()without a timeout.TransferQueue must still handle an explicitly configured but unschedulable storage-unit count without hanging.
Modification
backend.SimpleStorage.num_data_storage_unitsCompatibility
Test
python -m compileall -q transfer_queue tutorial testspytest -q tests/test_simple_storage_scheduling.pyopenyuanrong-datasystemis unavailable in the local macOS environment; upstream CI installs theyuanrongextra on Ubuntu.