Skip to content

Add collective_rpc to llm engine #16999

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 6 commits into from
Apr 24, 2025
Merged

Add collective_rpc to llm engine #16999

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
75a4321
add collective_rpc to llm engine
yinghai Apr 22, 2025
9f5457a
format
yinghai Apr 22, 2025
9818053
fix
yinghai Apr 22, 2025
4b8a6ac
lint
yinghai Apr 22, 2025
23af73a
fix
yinghai Apr 22, 2025
c84051f
optional
yinghai Apr 22, 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
18 changes: 18 additions & 0 deletions vllm/engine/async_llm_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -530,6 +530,13 @@ async def check_health_async(self) -> None:
self.tokenizer.check_health()
self.model_executor.check_health()

async def collective_rpc_async(self,
method: str,
timeout: Optional[float] = None,
args: tuple = (),
kwargs: Optional[dict] = None):
raise NotImplementedError


async def build_guided_decoding_logits_processor_async(
sampling_params: SamplingParams, tokenizer: AnyTokenizer,
Expand Down Expand Up @@ -1238,6 +1245,17 @@ async def is_sleeping(self) -> bool:
async def add_lora(self, lora_request: LoRARequest) -> None:
self.engine.add_lora(lora_request)

async def collective_rpc(self,
method: str,
timeout: Optional[float] = None,
args: tuple = (),
kwargs: Optional[dict] = None):
"""
Perform a collective RPC call to the given path.
"""
return await self.engine.collective_rpc_async(method, timeout, args,
kwargs)
Comment on lines +1256 to +1257
Copy link
Member

Choose a reason for hiding this comment

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

Why add not just raise NotImplementedError here?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I added that in collective_rpc_async.



# TODO(v1): Remove this class proxy when V1 goes default.
if envs.is_set("VLLM_USE_V1") and envs.VLLM_USE_V1:
Expand Down
11 changes: 11 additions & 0 deletions vllm/v1/engine/async_llm.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -490,6 +490,17 @@ async def pin_lora(self, lora_id: int) -> bool:
"""Prevent an adapter from being evicted."""
return await self.engine_core.pin_lora_async(lora_id)

async def collective_rpc(self,
method: str,
timeout: Optional[float] = None,
args: tuple = (),
kwargs: Optional[dict] = None):
"""
Perform a collective RPC call to the given path.
"""
return await self.engine_core.collective_rpc_async(
method, timeout, args, kwargs)

@property
def is_running(self) -> bool:
# Is None before the loop is started.
Expand Down