From 32470b69f6704396c7ea117300c96be84d02ff9c Mon Sep 17 00:00:00 2001 From: sufubao Date: Mon, 24 Aug 2026 09:28:53 +0800 Subject: [PATCH 1/2] perf(pd): reduce disaggregated serving latency --- lightllm/server/core/objs/shm_req_manager.py | 9 ++++++ lightllm/server/detokenization/manager.py | 4 ++- lightllm/server/router/manager.py | 32 ++++++++++++------- .../mode_backend/pd/base_kv_move_manager.py | 3 +- .../decode_node_impl/decode_trans_process.py | 6 ++-- .../prefill_trans_process.py | 10 +++--- 6 files changed, 45 insertions(+), 19 deletions(-) diff --git a/lightllm/server/core/objs/shm_req_manager.py b/lightllm/server/core/objs/shm_req_manager.py index fd9106d59c..5d0fe0b1ff 100644 --- a/lightllm/server/core/objs/shm_req_manager.py +++ b/lightllm/server/core/objs/shm_req_manager.py @@ -140,6 +140,15 @@ def put_back_req_obj(self, req: Req): req.ref_count = req.ref_count - 1 self.proc_private_get_state[req_index_in_mem] = 0 + # 请求槽位会被复用，但 prompt/logprob 共享内存句柄仅属于当前请求。 + # 如果将已连接的句柄保留在进程本地的 ctypes 包装对象上， + # 每次复用槽位时都会泄漏文件描述符。 + for attr_name in ("shm_prompt_ids", "shm_logprobs"): + shm_array = getattr(req, attr_name, None) + if shm_array is not None: + shm_array.detach_shm() + delattr(req, attr_name) + async def async_put_back_req_obj(self, req: Req): return self.put_back_req_obj(req) diff --git a/lightllm/server/detokenization/manager.py b/lightllm/server/detokenization/manager.py index 58f9324859..f7d58aac98 100644 --- a/lightllm/server/detokenization/manager.py +++ b/lightllm/server/detokenization/manager.py @@ -1,3 +1,4 @@ +import os import uvloop import asyncio import setproctitle @@ -20,6 +21,7 @@ from lightllm.utils.shm_port_args import get_shm_port_args logger = init_logger(__name__) +DETOKENIZATION_POLL_INTERVAL_S = float(os.getenv("LIGHTLLM_DETOKENIZATION_POLL_INTERVAL_S", "0.002")) class DeTokenizationManager: @@ -93,7 +95,7 @@ def handle_loop(self): logger.info(f"detokenize batch cost time {cost_time} ms") if not exist_need_detoken: - time.sleep(0.002) + time.sleep(DETOKENIZATION_POLL_INTERVAL_S) except Exception as e: logger.exception(f"detoken process has exception {str(e)}") diff --git a/lightllm/server/router/manager.py b/lightllm/server/router/manager.py index 01634d962e..a93507e220 100644 --- a/lightllm/server/router/manager.py +++ b/lightllm/server/router/manager.py @@ -1,3 +1,4 @@ +import os import time import uvloop import asyncio @@ -52,6 +53,7 @@ def __init__(self, args: StartArgs): self.node_rank = args.node_rank self.dp_size = args.dp self.schedule_time_interval = args.schedule_time_interval # 默认30ms 的调度周期 + self.metric_gauge_step_interval = max(1, int(os.getenv("LIGHTLLM_ROUTER_GAUGE_STEP_INTERVAL", "1"))) # 兼容多机纯tp的运行模式，这时候 1 // 2 == 0, 需要兼容 self.dp_size_in_node = max(1, args.dp // self.nnodes) self.dp_world_size = self.world_size // self.dp_size @@ -248,16 +250,17 @@ async def loop_for_fwd( self.metric_client.gauge_set("lightllm_batch_pause_size", self._get_paused_req_num()) # pd decode mode need to update token_load more frequently self.req_queue.update_token_load(self.running_batch, force_update=self.is_pd_decode_mode) - self.metric_client.gauge_set("lightllm_batch_current_size", len(self.running_batch.reqs)) - self.metric_client.gauge_set("lightllm_num_running_reqs", len(self.running_batch.reqs)) - self.metric_client.gauge_set("lightllm_queue_size", self.req_queue.get_wait_req_num()) - self.metric_client.gauge_set( - "lightllm_batch_current_max_tokens", - int( - sum(self.shared_token_load.get_dynamic_max_load(d_i) for d_i in range(self.dp_size_in_node)) - * self.max_total_token_num - ), - ) + if counter_count % self.metric_gauge_step_interval == 0: + self.metric_client.gauge_set("lightllm_batch_current_size", len(self.running_batch.reqs)) + self.metric_client.gauge_set("lightllm_num_running_reqs", len(self.running_batch.reqs)) + self.metric_client.gauge_set("lightllm_queue_size", self.req_queue.get_wait_req_num()) + self.metric_client.gauge_set( + "lightllm_batch_current_max_tokens", + int( + sum(self.shared_token_load.get_dynamic_max_load(d_i) for d_i in range(self.dp_size_in_node)) + * self.max_total_token_num + ), + ) else: self.req_queue.update_token_load(self.running_batch, force_update=True) if counter_count % 300 == 0: @@ -272,7 +275,14 @@ async def loop_for_fwd( estimated_peak_token_count = self.shared_token_load.get_estimated_peak_token_count(dp_i) logger.debug(f"dp_i {dp_i} estimated_peak_token_count: {estimated_peak_token_count} \n") - await asyncio.sleep(self._get_schedule_time_interval()) + schedule_time_interval = self._get_schedule_time_interval() + if schedule_time_interval < 0.001: + # uvloop 会将亚毫秒级 sleep 向下取整为忙等待。 + # 此处使用短暂的阻塞式 sleep 更准确，也能避免 + # router/metrics 循环退化为占满 CPU 的轮询循环。 + time.sleep(schedule_time_interval) + else: + await asyncio.sleep(schedule_time_interval) async def _step(self): """ diff --git a/lightllm/server/router/model_infer/mode_backend/pd/base_kv_move_manager.py b/lightllm/server/router/model_infer/mode_backend/pd/base_kv_move_manager.py index 125edede25..b3b117d201 100644 --- a/lightllm/server/router/model_infer/mode_backend/pd/base_kv_move_manager.py +++ b/lightllm/server/router/model_infer/mode_backend/pd/base_kv_move_manager.py @@ -13,6 +13,7 @@ from lightllm.utils.error_utils import log_exception logger = init_logger(__name__) +PD_KV_RETURN_POLL_INTERVAL_S = float(os.getenv("LIGHTLLM_PD_KV_RETURN_POLL_INTERVAL_S", "0.01")) class BaseKVMoveManager: @@ -77,7 +78,7 @@ def task_ret_upload_loop(self): self.shm_pd_trans_io_buffer.set_ready() break else: - time.sleep(0.01) + time.sleep(PD_KV_RETURN_POLL_INTERVAL_S) ret_objs.extend(self._collect_return_objects()) def _collect_return_objects(self): diff --git a/lightllm/server/router/model_infer/mode_backend/pd/decode_node_impl/decode_trans_process.py b/lightllm/server/router/model_infer/mode_backend/pd/decode_node_impl/decode_trans_process.py index cd9b015b11..4a57a4c84c 100644 --- a/lightllm/server/router/model_infer/mode_backend/pd/decode_node_impl/decode_trans_process.py +++ b/lightllm/server/router/model_infer/mode_backend/pd/decode_node_impl/decode_trans_process.py @@ -1,8 +1,9 @@ -import torch +import os import time import inspect import threading import setproctitle +import torch import torch.multiprocessing as mp import queue import pickle @@ -24,6 +25,7 @@ from lightllm.utils.process_check import start_parent_check_thread logger = init_logger(__name__) +PD_KV_POLL_INTERVAL_S = float(os.getenv("LIGHTLLM_PD_KV_POLL_INTERVAL_S", "0.001")) def start_decode_trans_process( @@ -337,7 +339,7 @@ def accept_peer_task_loop( self._check_tasks_time_out() if not notifies_dict: - time.sleep(0.001) + time.sleep(PD_KV_POLL_INTERVAL_S) def _check_tasks_time_out(self): with self.waiting_dict_lock: diff --git a/lightllm/server/router/model_infer/mode_backend/pd/prefill_node_impl/prefill_trans_process.py b/lightllm/server/router/model_infer/mode_backend/pd/prefill_node_impl/prefill_trans_process.py index 40bd2e42af..29c29e8ecc 100644 --- a/lightllm/server/router/model_infer/mode_backend/pd/prefill_node_impl/prefill_trans_process.py +++ b/lightllm/server/router/model_infer/mode_backend/pd/prefill_node_impl/prefill_trans_process.py @@ -1,8 +1,9 @@ -import torch +import os import time import inspect import threading import setproctitle +import torch import torch.multiprocessing as mp import queue import pickle @@ -19,6 +20,7 @@ logger = init_logger(__name__) +PD_KV_POLL_INTERVAL_S = float(os.getenv("LIGHTLLM_PD_KV_POLL_INTERVAL_S", "0.001")) def start_prefill_trans_process( @@ -283,7 +285,7 @@ def accept_decode_write_task_loop(self): self._check_tasks_time_out() if not notifies_dict: - time.sleep(0.001) + time.sleep(PD_KV_POLL_INTERVAL_S) return def _check_tasks_time_out(self): @@ -328,7 +330,7 @@ def update_task_status_loop( ): while True: if len(self.waiting_dict) == 0: - time.sleep(0.001) + time.sleep(PD_KV_POLL_INTERVAL_S) continue with self.waiting_dict_lock: @@ -372,7 +374,7 @@ def update_task_status_loop( trans_task.error_info = "time out in update_task_status_loop" self.failed_queue.put(trans_task) - time.sleep(0.001) + time.sleep(PD_KV_POLL_INTERVAL_S) @log_exception def success_loop(self): From 054eb9385bcb3d1c390f2a90491957bf2a194f4b Mon Sep 17 00:00:00 2001 From: sufubao Date: Mon, 24 Aug 2026 11:29:26 +0800 Subject: [PATCH 2/2] chore: split shared-memory cleanup into #1482 --- lightllm/server/core/objs/shm_req_manager.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/lightllm/server/core/objs/shm_req_manager.py b/lightllm/server/core/objs/shm_req_manager.py index 5d0fe0b1ff..fd9106d59c 100644 --- a/lightllm/server/core/objs/shm_req_manager.py +++ b/lightllm/server/core/objs/shm_req_manager.py @@ -140,15 +140,6 @@ def put_back_req_obj(self, req: Req): req.ref_count = req.ref_count - 1 self.proc_private_get_state[req_index_in_mem] = 0 - # 请求槽位会被复用，但 prompt/logprob 共享内存句柄仅属于当前请求。 - # 如果将已连接的句柄保留在进程本地的 ctypes 包装对象上， - # 每次复用槽位时都会泄漏文件描述符。 - for attr_name in ("shm_prompt_ids", "shm_logprobs"): - shm_array = getattr(req, attr_name, None) - if shm_array is not None: - shm_array.detach_shm() - delattr(req, attr_name) - async def async_put_back_req_obj(self, req: Req): return self.put_back_req_obj(req)