Skip to content

Caches should_stop so we're not hammering Ray. #973

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Aug 30, 2025
Merged

Caches should_stop so we're not hammering Ray. #973

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
87556b4
Now, caches should_stop value.
finbarrtimbers Aug 30, 2025
79bda33
Merge branch 'main' into cache-stop
finbarrtimbers Aug 30, 2025
ae8a4d8
Updated code to maintain last value on timeout.
finbarrtimbers Aug 30, 2025
1b87348
Cleaned up conditional.
finbarrtimbers Aug 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions open_instruct/vllm_utils3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,17 +368,30 @@ def __init__(
self.eval_results_queue = eval_results_queue
self.actor_manager = actor_manager

# For caching should_stop status.
self._last_should_stop_update = float("-inf")
self._should_stop_value = False
self._should_stop_timeout_s = 5

def _should_stop(self) -> bool:
if (time.perf_counter() - self._last_should_stop_update) > self._should_stop_timeout_s:
should_stop_ref = self.actor_manager.should_stop.remote()
ready_refs, _ = ray.wait([should_stop_ref], timeout=0.1)
if ready_refs:
self._should_stop_value = ray.get(ready_refs[0])
self._last_should_stop_update = time.perf_counter()
else:
ray.cancel(should_stop_ref)
Copy link
Collaborator

@hamishivi hamishivi Aug 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you think its worth adding some logging here? It seems like if the actor manager doesn't return at all / times out for this call its a sign something is a bit off.

return self._should_stop_value

def process_from_queue(self, timeout: float = 60.0):
"""Run generation loop using LLMEngine directly, with optional tool support.

Returns:
int: Number of requests processed (0 or 1)
"""
while True:
# Non-blocking check for should_stop using ray.wait
should_stop_ref = self.actor_manager.should_stop.remote()
ready_refs, _ = ray.wait([should_stop_ref], timeout=0.1)
if ready_refs and ray.get(ready_refs[0]):
if self._should_stop():
return 0

try:
Expand Down