[Bug][Update][Upstream] Fix AscendPDBacked init for v0.4.5

[marcobarlo]

Summary

This PR fixes issue #264 .

• Fix Ascend PD prefiller crash on peer connect:
  AttributeError: 'AscendPDBackend' object has no attribute '_nixl_agent_lock'.
• Replace stub super()._ensure_peer_connection(...) with an HCCL-native
  peer init (lazy channel connect + alloc REQ socket). Do not call
  upstream NIXL _ensure_peer_connection.
• Scope: AscendPDSenderMixin._ensure_peer_connection only
  (lmcache_ascend/v1/storage_backend/pd/sender_mixin.py).

Why

• Upstream LMCache 7f60057c (#2972, first in v0.4.5) added
  _nixl_agent_lock for the async NIXL worker thread.
• Ascend AscendPDBackend.__init__ never calls PDBackend.__init__, so
  that lock is never created. HCCL already serializes via
  transfer_channel._state_lock; Ascend does not need _nixl_agent_lock.
• The old stub from Enable Direct rH2D / D2D KVtransfer in P2P and PD backend  #173 worked on LMCache ≤ v0.4.4 and broke after ≥ v0.4.5.

Test plan

• Apply only this change (see docs/fix_missing_nixl_agent_lock.diff).
• Launch Ascend PD disagg (HCCL, prefiller + decoder).
• Send a request through the disagg proxy.
• Prefiller no longer raises _nixl_agent_lock; peer connect succeeds.

[gemini-code-assist]

Code Review

This pull request overrides the _ensure_peer_connection method in sender_mixin.py to establish a decoder peer connection over HCCL directly, bypassing the upstream implementation that relies on an unavailable lock. Feedback highlights a potential race condition where concurrent calls to this method could result in duplicate ZMQ socket creation and file descriptor leaks. It is recommended to wrap the initialization logic in a lock to ensure thread safety.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The _ensure_peer_connection method can be called concurrently by multiple threads (e.g., during concurrent request processing). Since checking and updating self.initialized_peers and self.mem_alloc_sockets is not atomic, a race condition exists where multiple threads could concurrently attempt to initialize the same peer connection. This can lead to duplicate ZMQ socket creation, overwriting existing sockets in self.mem_alloc_sockets, and leaking file descriptors.

To prevent this, wrap the connection initialization logic in a lock. We can reuse self._peer_alloc_backoff_lock which is already initialized and available on the instance.

        with self._peer_alloc_backoff_lock:
            if receiver_id in self.initialized_peers:
                return

            receiver_init_url = f"{receiver_host}:{receiver_init_port}"
            receiver_mem_alloc_url = f"{receiver_host}:{receiver_alloc_port}"

            self.transfer_channel.lazy_init_peer_connection(
                local_id=self.local_id,
                peer_id=receiver_id,
                peer_init_url=receiver_init_url,
            )

            mem_alloc_socket = get_zmq_socket(
                self.zmq_context,
                receiver_mem_alloc_url,
                "tcp",
                zmq.REQ,
                "connect",
            )
            self.mem_alloc_sockets[receiver_id] = mem_alloc_socket
            self.initialized_peers.add(receiver_id)

[marcobarlo]

@matthewygf I think this is not a valid comment as the peer_id is unique on sender side for any given rank of a D instance. Am I right?