[P/D]Improve the performance of Layerwise Connector

[nwpu-zxr]

What this PR does / why we need it?

Improve the performance of Layerwise Connector, mainly includes the following points:

1. Use event synchronize to replace stream synchronize.
2. Access metaserver when scheduling.
3. Transfer kvcache each Chunk prefill segmentation.

This PR is related to [RFC]: CDCP Scheduling for Disaggregated Prefilling with KV Cache Layerwise Push Support #4842

Does this PR introduce any user-facing change?

No.

How was this patch tested?

By CI.

• vLLM version: release/v0.13.0
• vLLM main: vllm-project/vllm@5fbfa8d

[github-actions]

👋 Hi! Thank you for contributing to the vLLM Ascend project. The following points will speed up your PR merge:‌‌

• A PR should do only one thing, smaller PRs enable faster reviews.
• Every PR should include unit tests and end-to-end tests ‌to ensure it works and is not broken by other future PRs.
• Write the commit message by fulfilling the PR description to help reviewer and future developers understand.

If CI fails, you can run linting and testing checks locally according Contributing and Testing.

[gemini-code-assist]

Code Review

This pull request introduces several performance improvements to the Layerwise Connector. The main changes include replacing stream synchronization with event synchronization for better asynchrony, accessing a metaserver during scheduling, and adjusting KV cache transfer logic. The switch to event-based synchronization is a good optimization. However, I've found some critical issues related to resource management and error handling in the new metaserver access logic that need to be addressed.

[gemini-code-assist]

The _access_metaserver method does not handle HTTP errors correctly. The httpx.Client.post method does not raise an exception for HTTP error status codes (e.g., 4xx, 5xx). The current implementation will treat such responses as successful, which can lead to silent failures and incorrect behavior. The retry logic also lacks a backoff delay, which can put unnecessary load on the metaserver during failures. The function also fails silently if all retries fail without an exception being caught on the last attempt.

    def _access_metaserver(self, url, message):
        last_exception = None
        for attempt in range(1, 4):
            try:
                response = self.metaserver_client.post(url, json=message)
                response.raise_for_status()  # Raise an exception for 4xx/5xx status codes
                return
            except Exception as e:
                last_exception = e
                logger.error(
                    f"Failed to connect to metaserver: {url}, attempt {attempt}/3. Error: {e}"
                )
                if attempt < 3:
                    time.sleep(1)  # Add a 1-second delay before retrying
        if last_exception:
            raise last_exception

[gemini-code-assist]

The ThreadPoolExecutor and httpx.Client are initialized here but are never shut down or closed. This will lead to resource leaks (threads, file descriptors). It's crucial to add a shutdown method to the MooncakeLayerwiseConnectorScheduler class to clean up these resources.

Additionally, setting timeout=None for the httpx.Client is risky in a production system, as a request to the metaserver could hang indefinitely, blocking a thread from the pool. It's strongly recommended to set a reasonable timeout.

It's recommended to add a shutdown method to the MooncakeLayerwiseConnectorScheduler class to properly close the ThreadPoolExecutor and httpx.Client to prevent resource leaks.

def shutdown(self):
    self.executor.shutdown(wait=True)
    self.metaserver_client.close()

This method should be called when the scheduler is no longer needed.

Suggested change

	self.executor = ThreadPoolExecutor(32)
	self.metaserver_client = httpx.Client(
	limits=httpx.Limits(max_connections=100000),
	timeout=None)
	self.executor = ThreadPoolExecutor(32)
	self.metaserver_client = httpx.Client(
	limits=httpx.Limits(max_connections=100000),
	timeout=60.0)

[LCAIZJ]

Have you encountered any hang issues during testing? #4976

[liziyu179]

In our self-validation, we used the latest 8.5 version of the CANN package, and did not encounter any hanging issues with event synchronize. And we will add the relevant explanations.

[liziyu179]

It needs to be constrained such that the event is recorded only for P nodes and not merely for prefill.

[liziyu179]

Move the request for P nodes forward to the schedule stage, and delete the corresponding logic in the worker.

[wujinyuan1]

Does the long sequence CP mode need no this extra processing?

[github-actions]

This pull request has conflicts, please resolve those before we can evaluate the pull request.

[github-actions]

This pull request has conflicts, please resolve those before we can evaluate the pull request.

[github-actions]

This pull request has conflicts, please resolve those before we can evaluate the pull request.