From 7de035386fbd44398211339cf51776ed152010d1 Mon Sep 17 00:00:00 2001 From: Abatom Date: Wed, 23 Jul 2025 18:20:06 +0800 Subject: [PATCH 1/3] pop->get Signed-off-by: Abatom --- .../disagg_proxy_p2p_nccl_xpyd.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py b/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py index ec58a183061e..1c964be45950 100644 --- a/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py +++ b/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py @@ -46,7 +46,7 @@ def _listen_for_register(poller, router_socket): global prefill_instances global prefill_cv with prefill_cv: - node = prefill_instances.pop(data["http_address"], None) + node = prefill_instances.get(data["http_address"], None) prefill_instances[data["http_address"]] = ( data["zmq_address"], time.time() + DEFAULT_PING_SECONDS, @@ -57,7 +57,7 @@ def _listen_for_register(poller, router_socket): global decode_instances global decode_cv with decode_cv: - node = decode_instances.pop(data["http_address"], None) + node = decode_instances.get(data["http_address"], None) decode_instances[data["http_address"]] = ( data["zmq_address"], time.time() + DEFAULT_PING_SECONDS, From 1b3aa8892308755db282723b974310761e364217 Mon Sep 17 00:00:00 2001 From: Abatom Date: Tue, 29 Jul 2025 17:57:23 +0800 Subject: [PATCH 2/3] bugfix Signed-off-by: Abatom --- .../disagg_proxy_p2p_nccl_xpyd.py | 1 + 1 file changed, 1 insertion(+) diff --git a/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py b/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py index 168caad26624..4ccbc290f814 100644 --- a/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py +++ b/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py @@ -64,6 +64,7 @@ def _listen_for_register(poller, router_socket): ) _remove_oldest_instances(decode_instances) else: + node = None print( "Unexpected, Received message from %s, data: %s", remote_address, From 654088485de4cfcc98255fdb4f115227b5733aa3 Mon Sep 17 00:00:00 2001 From: Abatom Date: Tue, 29 Jul 2025 17:58:34 +0800 Subject: [PATCH 3/3] bugfix Signed-off-by: Abatom --- .../disagg_proxy_p2p_nccl_xpyd.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py b/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py index 4ccbc290f814..9062b9784589 100644 --- a/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py +++ b/examples/online_serving/disaggregated_serving_p2p_nccl_xpyd/disagg_proxy_p2p_nccl_xpyd.py @@ -64,12 +64,12 @@ def _listen_for_register(poller, router_socket): ) _remove_oldest_instances(decode_instances) else: - node = None print( "Unexpected, Received message from %s, data: %s", remote_address, data, ) + return if node is None: print(f"🔵Add [HTTP:{data['http_address']}, ZMQ:{data['zmq_address']}]")