[PD] Handle P/D failure and reconnect without affecting other instances

python/sglang/srt/disaggregation/decode.py

@@ -361,7 +361,7 @@ def pop_transferred(self) -> List[DecodeRequest]:
         indices_to_remove = set()
         for i, (decode_req, poll) in enumerate(zip(self.queue, polls)):
             if poll == KVPoll.Failed:
-                error_message = f"Decode transfer failed for request {decode_req.req.rid=} {decode_req.req.bootstrap_room=}"
+                error_message = f"Decode transfer failed for request rank={self.scheduler.tp_rank} {decode_req.req.rid=} {decode_req.req.bootstrap_room=}"
                 try:
                     decode_req.kv_receiver.failure_exception()
                 except Exception as e:
@@ -409,7 +409,8 @@ def pop_transferred(self) -> List[DecodeRequest]:
                             : decode_req.req.top_logprobs_num
                         ].tolist()
                     )
-
+                if hasattr(decode_req.kv_receiver, "clear"):
+                    decode_req.kv_receiver.clear()
                 transferred_reqs.append(decode_req.req)
                 indices_to_remove.add(i)
             elif poll in [