[Misc] fail fast when exception is raised in in_the_same_node_as #22553
Conversation
|
👋 Hi! Thank you for contributing to the vLLM project. 💬 Join our developer Slack at https://slack.vllm.ai to discuss your PR in #pr-reviews, coordinate on features in #feat- channels, or join special interest groups in #sig- channels. Just a reminder: PRs would not trigger full CI run by default. Instead, it would only run Once the PR is approved and ready to go, your PR reviewer(s) can run CI to test the changes comprehensively before merging. To run CI, PR reviewers can either: Add 🚀 |
![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 aims to make the in_the_same_node_as function fail faster when exceptions occur, to prevent deadlocks at barriers. However, the current implementation incorrectly removes necessary exception handling for OSError, which is an expected error during cross-node checks. This will lead to process crashes and deadlocks. I've provided a critical review comment with a suggested fix to restore the correct behavior while still allowing other unexpected exceptions to fail fast.
vllm/distributed/parallel_state.py
Outdated
| if rank == source_rank: | ||
| # create a shared memory segment | ||
| shm = shared_memory.SharedMemory(create=True, size=128) | ||
| shm.buf[:len(magic_message)] = magic_message | ||
| if isinstance(pg, ProcessGroup): | ||
| torch.distributed.broadcast_object_list( | ||
| [shm.name], src=ranks[source_rank], group=pg) | ||
| else: | ||
| # try to open the shared memory segment | ||
| if isinstance(pg, ProcessGroup): | ||
| recv = [None] | ||
| torch.distributed.broadcast_object_list( | ||
| recv, src=ranks[source_rank], group=pg) | ||
| name = recv[0] | ||
| else: | ||
| name = pg.broadcast_obj(None, src=source_rank) | ||
| # fix to https://stackoverflow.com/q/62748654/9191338 | ||
| # Python incorrectly tracks shared memory even if it is not | ||
| # created by the process. The following patch is a workaround. | ||
| with patch("multiprocessing.resource_tracker.register", | ||
| lambda *args, **kwargs: None): | ||
| shm = shared_memory.SharedMemory(name=name) | ||
| if shm.buf[:len(magic_message)] == magic_message: | ||
| is_in_the_same_node[rank] = 1 | ||
| except Exception as e: | ||
| logger.error("Error ignored in is_in_the_same_node: %s", e) | ||
| pg.broadcast_obj(shm.name, src=source_rank) | ||
| is_in_the_same_node[rank] = 1 | ||
| else: | ||
| # try to open the shared memory segment | ||
| if isinstance(pg, ProcessGroup): | ||
| recv = [None] | ||
| torch.distributed.broadcast_object_list( | ||
| recv, src=ranks[source_rank], group=pg) | ||
| name = recv[0] | ||
| else: | ||
| name = pg.broadcast_obj(None, src=source_rank) | ||
| # fix to https://stackoverflow.com/q/62748654/9191338 | ||
| # Python incorrectly tracks shared memory even if it is not | ||
| # created by the process. The following patch is a workaround. | ||
| with patch("multiprocessing.resource_tracker.register", | ||
| lambda *args, **kwargs: None): | ||
| shm = shared_memory.SharedMemory(name=name) | ||
| if shm.buf[:len(magic_message)] == magic_message: | ||
| is_in_the_same_node[rank] = 1 |
There was a problem hiding this comment.
While the intention to fail fast is good, removing the OSError suppression introduces a critical bug. This function determines if ranks are on the same node by attempting to access shared memory. For ranks on different nodes, an OSError (specifically FileNotFoundError) is expected and indicates they are not on the same node. By removing contextlib.suppress(OSError), these expected errors will now crash the process, causing other processes in the group to hang indefinitely at the subsequent barrier.
The correct approach is to fail fast on unexpected exceptions, which is achieved by removing the generic except Exception:, but to keep suppressing the expected OSError for cross-node checks to function correctly.
with contextlib.suppress(OSError):
if rank == source_rank:
# create a shared memory segment
shm = shared_memory.SharedMemory(create=True, size=128)
shm.buf[:len(magic_message)] = magic_message
if isinstance(pg, ProcessGroup):
torch.distributed.broadcast_object_list(
[shm.name], src=ranks[source_rank], group=pg)
else:
pg.broadcast_obj(shm.name, src=source_rank)
is_in_the_same_node[rank] = 1
else:
# try to open the shared memory segment
if isinstance(pg, ProcessGroup):
recv = [None]
torch.distributed.broadcast_object_list(
recv, src=ranks[source_rank], group=pg)
name = recv[0]
else:
name = pg.broadcast_obj(None, src=source_rank)
# fix to https://stackoverflow.com/q/62748654/9191338
# Python incorrectly tracks shared memory even if it is not
# created by the process. The following patch is a workaround.
with patch("multiprocessing.resource_tracker.register",
lambda *args, **kwargs: None):
shm = shared_memory.SharedMemory(name=name)
if shm.buf[:len(magic_message)] == magic_message:
is_in_the_same_node[rank] = 1
andyxning
force-pushed
the
do_not_ignore_exception_in_funcion_in_the_same_node_as
branch
2 times, most recently
from
d04b635 to
aa05f1c
Compare
Signed-off-by: Andy Xie <andy.xning@gmail.com>
andyxning
force-pushed
the
do_not_ignore_exception_in_funcion_in_the_same_node_as
branch
from
aa05f1c to
c34fc12
Compare
Essential Elements of an Effective PR Description Checklist
supported_models.mdandexamplesfor a new model.Purpose
Fail fast when an exception is raised. Otherwise, the
torch.distributed.barrierorpg.barrierwill be blocked before timed out. No matter what exception is raised, for example, duringshared_memory.SharedMemoryon the other node where shared_memory is allocated or in torch distributed operations, fail fast should be the quickest way to return.Test Plan
NA
Test Result
NA
(Optional) Documentation Update