【0.13.0】【bugfix】Resolved memory deallocation failure in the pooling layer under re-computation workloads.

[fems14]

What this PR does / why we need it?

[releases/v0.13.0] Resolved a double-free memory vulnerability in the pooling layer under re-computation scenarios.
PR for main branch: #6045

Does this PR introduce any user-facing change?

How was this patch tested?

[gemini-code-assist]

Code Review

This pull request addresses a memory deallocation failure by introducing preempted_req_ids into the AscendConnectorMetadata and refining how stored requests are tracked and decremented. The changes correctly propagate the preempted_req_ids from the scheduler to the worker. However, there are critical type safety and logical issues related to how AscendConnectorMetadata is handled and how stored_requests are checked, which could lead to runtime errors or incorrect behavior.

[gemini-code-assist]

The self._get_connector_metadata() method is inherited from KVConnectorBase_V1. The get_finished method in KVPoolWorker now explicitly expects an AscendConnectorMetadata object, which includes preempted_req_ids. If the inherited _get_connector_metadata() returns a generic KVConnectorMetadata (which it likely does, as AscendConnectorMetadata is a specific implementation), accessing meta.preempted_req_ids in KVPoolWorker will result in an AttributeError. To ensure type safety and correct functionality, AscendStoreConnector should override _get_connector_metadata to explicitly return an AscendConnectorMetadata instance, ensuring it contains the necessary preempted_req_ids from the scheduler.

[gemini-code-assist]

The check if req_id not in self.stored_requests: might not behave as expected. self.stored_requests is a defaultdict(int). If a request's count has been decremented to 0 by dec_stored_request but the key has not been explicitly removed by delete_finished_stored_request, the req_id will still be present in self.stored_requests (with a value of 0). In such a case, this condition would be False, and the method would proceed to process a request that has no pending blocks to store, potentially leading to incorrect behavior or resource issues. Consider checking if self.stored_requests.get(req_id, 0) <= 0: to correctly identify requests that are logically finished from the sending perspective.

Suggested change

	if req_id not in self.stored_requests:
	self.request_queue.task_done()
	return
	if self.stored_requests.get(req_id, 0) <= 0:
	self.request_queue.task_done()
	return