Repository navigation
Extract pause_resume_in_place kit; rename test_abort to test_scheduler_control - #22647
Conversation
|
/rerun-test test_disaggregation_basic.py |
|
❌
Please provide the full path, e.g. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a reusable PauseResumeInPlaceMixin within a new test kit to standardize pause and resume functionality testing. The TestDisaggregationAccuracy class has been refactored to inherit from this mixin, effectively removing duplicate test logic. Review feedback suggests enhancing the mixin's configurability by converting the request timeout constant into a class attribute. Additionally, it is recommended to replace the use of as_completed with concurrent.futures.wait to more robustly handle timeouts and ensure that final assertions are executed even when requests fail to resolve.
| _REQUEST_TIMEOUT = 180 | ||
|
|
||
|
|
||
| class PauseResumeInPlaceMixin: | ||
| """Test pause/resume with in_place mode. | ||
|
|
||
| Sends concurrent requests, pauses mid-generation, verifies no progress | ||
| during the pause window, then resumes and verifies all requests complete. | ||
|
|
||
| Subclass must set: | ||
| - pause_generate_url: URL to send /generate requests (or falls back to self.base_url) | ||
| - pause_target_urls: list of URLs to send /pause_generation and /continue_generation | ||
| """ | ||
|
|
||
| pause_num_requests: int = 32 | ||
| pause_max_new_tokens: int = 512 | ||
| pause_duration: float = 5 | ||
| pause_generate_url: str = "" | ||
| pause_target_urls: list = [] |
There was a problem hiding this comment.
To improve the configurability of this test kit, consider moving the _REQUEST_TIMEOUT constant into a class attribute (e.g., pause_request_timeout: int = 180). This would allow subclasses to override the timeout for specific test environments, consistent with how pause_num_requests and pause_duration are exposed.
| for future in as_completed(futures, timeout=_REQUEST_TIMEOUT): | ||
| prompt_id = futures[future] | ||
| try: | ||
| resp = future.result() | ||
| if resp.status_code == 200: | ||
| body = resp.json() | ||
| self.assertIn("text", body) | ||
| self.assertGreater(len(body["text"]), 0) | ||
| completed += 1 | ||
| else: | ||
| errors.append(f"Request {prompt_id}: status={resp.status_code}") | ||
| except Exception as e: | ||
| errors.append(f"Request {prompt_id}: exception={e}") | ||
|
|
||
| self.assertEqual( | ||
| completed + len(errors), | ||
| num_requests, | ||
| "Some requests did not resolve within the timeout -- likely hung during pause.", | ||
| ) | ||
| self.assertEqual( | ||
| completed, | ||
| num_requests, | ||
| f"Some requests failed: {completed}/{num_requests} succeeded. Errors: {errors}", | ||
| ) |
There was a problem hiding this comment.
The current implementation using as_completed with a timeout will raise a concurrent.futures.TimeoutError if any requests fail to complete within the specified time. This exception propagates out of the method, skipping the summary assertions at the end (lines 102-111) which are intended to provide a descriptive failure report. Additionally, if the loop finishes normally, the assertion at line 102 is redundant as the sum of completed and errored requests will always equal num_requests.
Using concurrent.futures.wait is a more robust pattern here. It allows you to collect results from all completed futures and explicitly identify which ones timed out, ensuring that the final assertions are always reached and provide a comprehensive error message.
from concurrent.futures import wait
# Use a slightly longer timeout for wait to allow individual request timeouts to trigger first
done, not_done = wait(futures, timeout=_REQUEST_TIMEOUT + 5)
completed = 0
errors = []
for future in done:
prompt_id = futures[future]
try:
resp = future.result()
if resp.status_code == 200:
body = resp.json()
self.assertIn("text", body)
self.assertGreater(len(body["text"]), 0)
completed += 1
else:
errors.append(f"Request {prompt_id}: status={resp.status_code}")
except Exception as e:
errors.append(f"Request {prompt_id}: exception={e}")
for future in not_done:
prompt_id = futures[future]
errors.append(f"Request {prompt_id}: timed out")
self.assertEqual(
completed + len(errors),
num_requests,
"Some requests did not resolve within the timeout -- likely hung during pause.",
)
self.assertEqual(
completed,
num_requests,
f"Some requests failed: {completed}/{num_requests} succeeded. Errors: {errors}",
)
|
/rerun-test test/registered/scheduler/test_scheduler_control.py test/registered/disaggregation/test_disaggregation_basic.py |
|
✅ ✅ |
|
/tag-and-rerun-ci |
hnyls2002
changed the title
1 participant
Summary
test_pause_resume_in_placefromtest_disaggregation_basic.pyinto reusablePauseResumeInPlaceMixinkit (python/sglang/test/kits/pause_generation_kit.py)TestSchedulerControl(formerlyTestAbortAll) for single-server pause/resume coveragetest_abort.py->test_scheduler_control.pyandTestAbortAll->TestSchedulerControlto reflect broader scheduler control scopeTest plan
test_scheduler_control.pypassed (359s)test_disaggregation_basic.pypassed (395s)