refactor e2e tests with concurrent polling and add Buildkite CI config - #250
Conversation
Signed-off-by: fangyuchu <fangyuchu@qq.com>
![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 adds a Buildkite pipeline step for fault tolerance E2E tests on H100 GPUs and refactors the fault tolerance tests to support concurrent polling and execution across multiple servers. The review feedback highlights that the new _wait_for_engines helper still polls servers sequentially, which could block the loop for up to 10 seconds if a server is unresponsive. The reviewer suggests using a ThreadPoolExecutor to poll the servers concurrently to avoid test flakiness.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| results: dict[int, dict[str, Any]] = {} | ||
| pending = dict(enumerate(servers)) | ||
| start = time.time() | ||
| while time.time() - start < deadline_s: | ||
| with contextlib.suppress(Exception): | ||
| for engine in _get_ft_status(server)["engines"]: | ||
| if predicate(engine): | ||
| return engine | ||
| time.sleep(1.0) | ||
| return None | ||
|
|
||
|
|
||
| def _wait_for_status(server, statuses, deadline_s: int = FAULT_DETECTION_DEADLINE_S): | ||
| """Poll until an engine reports one of ``statuses`` (a set of status strings).""" | ||
| return _poll_status(server, deadline_s, lambda e: e["status"] in statuses) | ||
| while pending and time.time() - start < deadline_s: | ||
| for i, server in list(pending.items()): | ||
| with contextlib.suppress(Exception): | ||
| for engine_status in _get_ft_status(server)["engines"]: | ||
| if engine_status.get(match_key) in match_values: | ||
| results[i] = engine_status | ||
| del pending[i] | ||
| break | ||
| if pending: | ||
| time.sleep(1.0) | ||
| return [results.get(i) for i in range(len(servers))] |
There was a problem hiding this comment.
The current implementation of _wait_for_engines polls the servers sequentially. Since _get_ft_status has a 10-second timeout, any unresponsive or slow server will block the entire polling loop for up to 10 seconds, preventing other servers from being polled. This can lead to test flakiness or timeouts, and defeats the PR's goal of "concurrent polling".
We should poll the pending servers concurrently using a ThreadPoolExecutor (similar to _in_parallel).
results: dict[int, dict[str, Any]] = {}
pending = dict(enumerate(servers))
start = time.time()
with ThreadPoolExecutor(max_workers=max(1, len(servers))) as ex:
while pending and time.time() - start < deadline_s:
items = list(pending.items())
def poll(item):
idx, server = item
try:
for engine_status in _get_ft_status(server)["engines"]:
if engine_status.get(match_key) in match_values:
return idx, engine_status
except Exception:
pass
return idx, None
for idx, engine_status in ex.map(poll, items):
if engine_status is not None:
results[idx] = engine_status
del pending[idx]
if pending:
time.sleep(1.0)
return [results.get(i) for i in range(len(servers))]
#250) Signed-off-by: fangyuchu <fangyuchu@qq.com>
…ft-simplify This squashes the following commits into one: - Fault Tolerance Framework (fangyuchu#229) - unify param name for nixl and deepep and destroy old cpu_group in retry (fangyuchu#230) - Surface exception to status - change status report mode from pull to push - simplify fault tolerance config args - use existing get_all2all_manager implementation - support tp>1 (fangyuchu#241) - Enhance the pass of fault tolerance results (fangyuchu#242) - clean states for model runner v2 - add support for fault detection through mask for model runner v2 - [FT] validate single API server, fix clean_buffers ordering and worker state cleanup - [FT] make apply endpoint async and gate recovery by engine status - add e2e test for fault tolerance - add e2e test for retry recovery - refactor e2e tests with concurrent polling and add Buildkite CI config (fangyuchu#250) - set cpu timeout to default value of nixl-ep in test (fangyuchu#251) Co-Authored-By: Claude <noreply@anthropic.com>
#250) Signed-off-by: fangyuchu <fangyuchu@qq.com>
1 participant
PLEASE FILL IN THE PR DESCRIPTION HERE ENSURING ALL CHECKLIST ITEMS (AT THE BOTTOM) HAVE BEEN CONSIDERED.
Purpose
Test Plan
python -m pytest tests/v1/fault_tolerance/test_fault_tolerance_e2e.py -v -s
Test Result
通过
Essential Elements of an Effective PR Description Checklist
supported_models.mdandexamplesfor a new model.BEFORE SUBMITTING, PLEASE READ https://docs.vllm.ai/en/latest/contributing (anything written below this line will be removed by GitHub Actions)