Skip to content

[CI/Build][Bugfix] Fix AsyncLLM hanging issue due to unclosed ZMQ socket #22744

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

Closed
wants to merge 7 commits into from
Closed

[CI/Build][Bugfix] Fix AsyncLLM hanging issue due to unclosed ZMQ socket #22744

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6a4496d
check
Isotr0py Aug 12, 2025
a03755a
try
Isotr0py Aug 12, 2025
ea42547
revert
Isotr0py Aug 13, 2025
ee34d7d
fix
Isotr0py Aug 13, 2025
f012216
move assertion
Isotr0py Aug 13, 2025
a5b4240
oops typo
Isotr0py Aug 13, 2025
e90ac3a
Merge branch 'vllm-project:main' into dp-deadlock
Isotr0py Aug 13, 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
10 changes: 10 additions & 0 deletions vllm/v1/engine/core_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -326,6 +326,8 @@ class BackgroundResources:
output_socket: Optional[Union[zmq.Socket, zmq.asyncio.Socket]] = None
input_socket: Optional[Union[zmq.Socket, zmq.asyncio.Socket]] = None
first_req_send_socket: Optional[zmq.asyncio.Socket] = None
first_req_rcv_socket: Optional[zmq.asyncio.Socket] = None
stats_update_socket: Optional[zmq.asyncio.Socket] = None
output_queue_task: Optional[asyncio.Task] = None
stats_update_task: Optional[asyncio.Task] = None
shutdown_path: Optional[str] = None
Expand All @@ -344,6 +346,12 @@ def __call__(self):
self.coordinator.close()

cancel_task_threadsafe(self.output_queue_task)

# ZMQ context termination can hang if the sockets
# aren't explicitly closed first.
for socket in (self.first_req_rcv_socket, self.stats_update_socket):
if socket is not None:
socket.close(linger=0)
cancel_task_threadsafe(self.stats_update_task)

# ZMQ context termination can hang if the sockets
Expand Down Expand Up @@ -977,6 +985,8 @@ async def run_engine_stats_update_task():
bind=False) as first_req_rcv_socket:
assert isinstance(socket, zmq.asyncio.Socket)
assert isinstance(first_req_rcv_socket, zmq.asyncio.Socket)
self.resources.stats_update_socket = socket
self.resources.first_req_rcv_socket = first_req_rcv_socket
# Send subscription message.
await socket.send(b'\x01')

Expand Down