[BugFix][EPD] fix embedding req_id transfer error (#23481)
This commit is contained in:
@@ -343,7 +343,12 @@ class MultiModalEmbeddingData(EmbeddingData):
|
|||||||
return kwargs
|
return kwargs
|
||||||
|
|
||||||
def add(self, embedding_data: EmbeddingData):
|
def add(self, embedding_data: EmbeddingData):
|
||||||
assert self.req_id == embedding_data.req_id
|
if self.req_id != embedding_data.req_id:
|
||||||
|
logger.warning(
|
||||||
|
f"Dropping embedding data with mismatched req_id: "
|
||||||
|
f"expected {self.req_id}, got {embedding_data.req_id}"
|
||||||
|
)
|
||||||
|
return False
|
||||||
assert not self.ready_list[embedding_data.part_idx]
|
assert not self.ready_list[embedding_data.part_idx]
|
||||||
pid = embedding_data.part_idx
|
pid = embedding_data.part_idx
|
||||||
self.ready_list[pid] = True
|
self.ready_list[pid] = True
|
||||||
@@ -521,6 +526,17 @@ class WaitingImageRequest:
|
|||||||
self.recv_socket.close()
|
self.recv_socket.close()
|
||||||
return
|
return
|
||||||
|
|
||||||
|
# Extract original req_id from part_req_id and drop stale payloads
|
||||||
|
# that may arrive on a reused ZMQ port after a prior request aborted.
|
||||||
|
original_req_id = extract_original_req_id(recv_obj.req_id)
|
||||||
|
if original_req_id != self.recv_req.rid:
|
||||||
|
logger.warning(
|
||||||
|
f"Dropping stale embedding data: expected rid={self.recv_req.rid}, "
|
||||||
|
f"got rid={recv_obj.req_id} (likely from ZMQ port reuse)"
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
recv_obj.req_id = original_req_id
|
||||||
|
|
||||||
buffer = parts[1].buffer if hasattr(parts[1], "buffer") else parts[1]
|
buffer = parts[1].buffer if hasattr(parts[1], "buffer") else parts[1]
|
||||||
recv_obj.embedding = (
|
recv_obj.embedding = (
|
||||||
torch.frombuffer(buffer, dtype=recv_obj.dtype)
|
torch.frombuffer(buffer, dtype=recv_obj.dtype)
|
||||||
@@ -528,12 +544,6 @@ class WaitingImageRequest:
|
|||||||
.clone()
|
.clone()
|
||||||
)
|
)
|
||||||
|
|
||||||
# Extract original req_id from part_req_id
|
|
||||||
part_req_id = recv_obj.req_id
|
|
||||||
original_req_id = extract_original_req_id(part_req_id)
|
|
||||||
# Update recv_obj.req_id to original for aggregation
|
|
||||||
recv_obj.req_id = original_req_id
|
|
||||||
|
|
||||||
if self.recv_embedding_data is None:
|
if self.recv_embedding_data is None:
|
||||||
self.recv_embedding_data = MultiModalEmbeddingData.from_embedding_data(
|
self.recv_embedding_data = MultiModalEmbeddingData.from_embedding_data(
|
||||||
recv_obj
|
recv_obj
|
||||||
|
|||||||
Reference in New Issue
Block a user