fix: improve RPC/cache-store diagnostics and xgrammar loading - #1344
xinfei-shi wants to merge 12 commits into
Conversation
LLLLKKKK
left a comment
There was a problem hiding this comment.
AI Code Review - PR #1344
Status: BLOCKING
Summary: P0/1 · P1/6 · P2/13 · P3/3
Reviewed: commit 1f34a8aa9505 · 2026-08-28 15:52 UTC+8
Blocking Issues
P0
load_xgrammar未绑定到模块全局,GrammarValidator 必然构建失败并静默关闭语法准入校验 @rtp_llm/dash_sc/inference/grammar_validator.py:454- 建议:
:454改回return xgr.GrammarCompiler(——:445的xgr已是load_xgrammar()的记忆化结果,语义等价且不重复加载;若确需显式调用,请在文件顶部(:36 附近)补模块级from rtp_llm.config.grammar_tokenizer_info import load_xgrammar。同时补一条_build_backend()成功路径单测:现有grammar_validator_test.py:34用GrammarValidator.__new__(GrammarValidator)绕过__init__,正是该缺陷逃过 CI 的直接原因,新测试不得以 mock 替换 xgrammar 这一生产边界(P.P.G.3)。另建议不要用宽except Exception把NameError这类编程错误与真实的 xgrammar 初始化失败混成同一条 RuntimeError 文案;上层降级为None时建议把 warning 升级为可告警指标,避免准入校验静默关闭。
- 建议:
P1
- decode 阶段计时被双重推进,四个耗时指标全部错位且 stage 越出枚举上界 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.cc:1307- 建议:删除
:1307与:1310两行,让 allocate 阶段计时完全由allocateResourceFunc内层的EXECUTE_STAGE_FUNC负责(finishStage()的begin_time == 0幂等短路已保证重试不重复累加),与 prefill 写法一致;若目的是把多次 retry 合并成一段耗时,应在EXECUTE_WITH_RETRY内累加而非额外推进 stage。同时清理:1331这条在新语义下已多余、且使 stage 越界的裸nextStage(),并在nextStage()内加断言禁止 stage 超过finish(当前虽无后续finishStage()调用未触发defaultabort,但属潜在隐患)。请在已有的rtp_llm/cpp/model_rpc/test/DecodeRpcServerTest.cc补一个DecodeStatInfo单测:模拟 prepare → allocate(含 1 次重试)→ loadCache → localGenerate,断言四个*_rt_us各自非零、互不串位且 stage 终值不越界。
- 建议:删除
- decode KV cache 分配失败的对外错误码从 DECODE_MALLOC_FAILED(8211) 退化为 MALLOC_FAILED(602) @
rtp_llm/cpp/model_rpc/PrefillRpcServer.cc:149- 建议:在 decode 侧构造
error_info时就固化角色语义,例如allocateResource中把MALLOC_FAILED显式映射为DECODE_MALLOC_FAILED后再写入(这样 8211 也能经 details 原样跨节点存活);或在 prefill 的 details 命中分支保留补偿映射status.error_code() == RESOURCE_EXHAUSTED && remote_code == MALLOC_FAILED → DECODE_MALLOC_FAILED。请在 PR description 中列出本次对外错误码取值域的变化,并确认上游按 8211 做重试/降级判定的逻辑仍成立;建议在已有的PrefillRpcServerTest.cc补一条用例,断言 decode 返回 602 + RESOURCE_EXHAUSTED 时 prefill 对外仍报 8211。
- 建议:在 decode 侧构造
- REPORT_GAUGE 移除零值守卫,PD 两侧互相向对方独有的阶段耗时 gauge 写 0 @
rtp_llm/cpp/metrics/RtpLLMMetrics.cc:43- 建议:不要用一个宏统一改变 69 个调用点的语义:保留条件版
REPORT_GAUGE,另增REPORT_GAUGE_ALWAYS,仅对onflight_request、total_rt_us等每周期确实有效的状态量使用;阶段耗时类(全部 PD 阶段耗时、retry_times)维持「未发生即不上报」。顺带收敛本文件并存的三套约定——REPORT_QPS(:38-41) 仍带守卫、REPORT_GAUGE(:43) 无守卫、REPORT_NON_ZERO_MUTABLE_METRIC(:751) 专门表达大于 0,另有多处裸REPORT_MUTABLE_METRIC——统一为语义明确的两个宏并在文件头注释写清适用场景。请在 PR description 中列出语义变更的 metric 清单,便于看板与告警阈值同步调整。
- 建议:不要用一个宏统一改变 69 个调用点的语义:保留条件版
- 基类 stopStream() 在正常成功路径上每请求输出一条 WARNING @
rtp_llm/cpp/model_rpc/GenerateContext.cc:98- 建议:基类复用同一判定:
stream_->hasEvent(StreamEvents::GenerateDone) && !stream_->hasError()时视为正常收尾,降级为RTP_LLM_LOG_DEBUG且不再reportError(CANCELLED)(这同时能避免误报错误干扰 decode 侧releaseResource() → insertIntoCache()的 KV 复用写入,并顺带修正成功请求被计入cancel_qps的既有偏差);context_error/client_cancel两条 WARNING 保留,仅在确实无 GenerateDone 时才输出 context_cleanup WARNING。这样可消除 local(非 PD)与 decode 两条最高 QPS 路径上的日志洪泛。
- 建议:基类复用同一判定:
- flush() 由 NOOP 改为无上界 Queue.join() 并被 close() 调用,反转模块自身 never-block 契约 @
rtp_llm/access_logger/async_log_handler.py:205- 建议:
flush()开头增加if self._stop_event.is_set(): self._file_handler.flush(); return短路;把Queue.join()换成有上界的等待(threading.Condition或按flush_interval轮询unfinished_tasks),总超时后记录剩余条数并放弃,等待预算与部署 SIGTERM 宽限期对齐;close()中先_stop_event.set()再 flush,消除入队竞态窗口;不要在flush()内重启 worker,把重启职责留给emit()。__del__中不要做阻塞等待——join()不抛异常,:248的except Exception: pass无法兜住挂死。
- 建议:
- 22 个变更文件零测试,且现有测试写法结构性无法发现本次回归 @
rtp_llm/cpp/model_rpc/GenerateContext.h:103- 建议:优先补三条能直接拦住本 PR 阻塞项的用例:(1) 以合法
tokenizer_info_json真实构造GrammarValidator并断言_backend is not None,不得用 mock 替换 xgrammar 这一生产边界;(2)DecodeStatInfo/PrefillStatInfo在「正常顺序 / retry 后 restoreStage / 提前 return」三种路径下各*_rt_us的归属与 stage 上界断言;(3)serializeErrorMsg→deserializeErrorDetails往返能还原原始ErrorCode,并断言 decode 返回 602 时 prefill 对外仍为 8211。其次补低成本高收益的分支用例:response 不设error_code时回调收到LoadErrorUnknown、callback/execNoBlockCopy抛异常时回调恰好一次且 closure 被释放(配合 ASAN)、local_partition_count == 0与 buffer 未找到均返回 nullptr 且不崩溃、blocks.size() > unfinished_count_不变量分支。
- 建议:优先补三条能直接拦住本 PR 阻塞项的用例:(1) 以合法
Non-blocking Suggestions
P2
- deserializeErrorDetails 无值域与 NONE_ERROR 校验,且与坏连接淘汰分支互斥 @
rtp_llm/cpp/model_rpc/RpcErrorCode.h:49- 建议:在 helper 内补
details.error_code() != static_cast<int64_t>(ErrorCode::NONE_ERROR)判断并校验数值落在已知枚举集合中,未知值回退UNKNOWN_ERROR(可复用白名单映射思路);把两类判别改为互补而非互斥:先按传输层 error_message 决定是否closeGrpcConnection(),再用远端 details 覆盖error_code/message。同时把PrefillRpcServer.cc:425与PrefillBatchRpcServer.cc:142两处重复实现改为调用该 helper,避免三份同语义、强度不一致的代码继续分叉,并补「details 为空 / 非法 blob / code=0 / 正常 code」四类输入单测。
- 建议:在 helper 内补
- closure end() 自毁与 Run() 的 catch 重入构成 use-after-free 与 double free 窗口 @
rtp_llm/cpp/disaggregate/cache_store/TcpCacheStoreLoadServiceClosure.cpp:107- 建议:给两个 closure 各加幂等哨兵:成员
bool ended_ = false;,end()入口if (ended_) { return; } ended_ = true;;并把end()声明为noexcept,把collector_->markEnd与全部日志/字符串构造移入内部try/catch(...),使「self-destroy」与「异常兜底」两个职责不再互相重入。同时把Run()catch 内取 request_id 的动作改为在 try 之前先拷贝到局部std::string,避免在 catch 中触碰可能已释放的成员。建议把「自持生命周期 + 单次执行守卫 + callback 异常兜底」抽成共用基类或小工具,让同目录仍用裸delete this且 callback 无兜底的TcpBlockReadClosure.cpp:84一并复用(该文件不在本次 diff 内,如不便扩大范围请在 PR description 中记录为已知遗留项)。
- 建议:给两个 closure 各加幂等哨兵:成员
- notifyDone 忽略 success 形参,失败可被判为成功并回 EC_SUCCESS @
rtp_llm/cpp/disaggregate/cache_store/CacheTransferServiceImplContext.cpp:62- 建议:将
:62判定改为if (!success || error_code != CacheStoreErrorCode::None),让success形参重新参与决策;并在runFailed()(:98) 入口对ec == None兜底为LoadErrorUnknown,保证「走失败路径」与「回非 EC_SUCCESS」这一不变量不可被破坏。RemoteStoreTaskImpl.cpp:214同样加兜底:internal_error_code == NONE_ERROR时替换为CACHE_STORE_LOAD_UNKNOWN_ERROR,并对(false, None)打 ERROR 日志暴露上游违约。
- 建议:将
- cache_store 响应错误码取值域扩大且 has_error_code 语义翻转,混部期错误分类与重试行为会漂移 @
rtp_llm/cpp/disaggregate/cache_store/CacheTransferServiceImplContext.cpp:99- 建议:在 PR description 中明确这批 wire 错误码取值域与 presence 语义的变化,并确认按
CACHE_STORE_LOAD_*分类的重试判定对新增取值仍成立;必要时为CallPrefillTimeout、PushWorkerItemFailed等在toKvCacheStoreErrorCode中补专属枚举而非落 default,避免不同原因被压平到同一EC_FAILED_INTERNAL。若判定逻辑确有分歧,建议加环境变量开关保留旧的压平行为作为运维回滚手段。
- 建议:在 PR description 中明确这批 wire 错误码取值域与 presence 语义的变化,并确认按
- 运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉 @
rtp_llm/config/grammar_tokenizer_info.py:154- 建议:对齐既有范式:import 前
ctypes.CDLL(str(binding_path), mode=ctypes.RTLD_GLOBAL)(包在try/except OSError中)显式预加载绑定库,成功即继续 import,失败再回退裸import xgrammar;这样对当前进程真实有效,也不再污染子进程环境。建议把该预加载抽成共用的preload_shared_library(path)工具函数,避免与tilelang_kernels.py两处各写一套(R.I.1)。若确认第三方 loader 会读取该变量而必须保留,请在 docstring 写明这一外部实现依赖及「当前进程 + spawn 子进程」的实际生效范围,并用 try/finally 在 import 后恢复原值。
- 建议:对齐既有范式:import 前
- TP load cache 报错日志用 rank 直接索引 peer_addrs,非对称 TP 下指向错误对端 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.cc:676- 建议:抽出一个与
constructRemoteLoadRequest/constructRemoteLoadRequestForMla共用的 rank→peer 映射函数,或直接复用已构造的load_request.peer_addrs()内容作为日志字段,避免同一映射规则在两处各写一遍而漂移。
- 建议:抽出一个与
- loading_cache_request 改为析构期采样后几乎恒为 0,叠加零值上报会拉平该指标 @
rtp_llm/cpp/model_rpc/DecodeGenerateContext.cc:74- 建议:在
loadCacheFromPrefill(AtomicGuard 生效期间)把loading_cache_requests_->load()快照进stat_info,reportTime()直接上报该快照;prefill 侧同样把采样点移入 load 窗口。若确实想要「结束时刻并发度」,请改指标名或在 PR description 中说明语义变更,避免沿用旧看板。同时记录rtp_llm_rpc_onflight_request的采样时点变化(入口快照 → 结束时读数),便于告警阈值同步调整。
- 建议:在
- close() 在 worker join 超时后仍关闭底层流,日志丢失既无计数也无告警 @
rtp_llm/access_logger/async_log_handler.py:212- 建议:用
if self._worker_thread.is_alive():检查 join 后线程状态,仅在确实退出时才_file_handler.close();超时分支输出一条带剩余qsize()的 warning 并跳过 close,把底层流交给进程退出回收,避免与仍在写入的 worker 竞争。为_stats增加dropped_after_close之类计数,让关闭期丢弃对外可见。
- 建议:用
- _start_worker() 无锁保护,flush() 与 emit() 并发会产生重复 worker 线程 @
rtp_llm/access_logger/async_log_handler.py:95- 建议:新增一个专用
threading.Lock包住_start_worker()的「检查 + 创建 + 启动」整段使其原子;或按前述建议移除flush()中的 worker 重启逻辑,从源头消除该竞态。同时把worker_restarts自增(:187-188)移到_start_worker()确认线程已启动之后,避免_stop_event置位时计数虚高。
- 建议:新增一个专用
- None 记录被消费但未 task_done,可永久破坏新的阻塞式 flush() @
rtp_llm/access_logger/async_log_handler.py:127- 建议:把
task_done()移到「每次成功get()之后」的finally中,与取出动作一一对应,而不是与「是否为有效 record」耦合;或直接删掉这几处不可达的None判断,让入队契约保持为「只放 LogRecord」。
- 建议:把
- 新增统计计数器与 get_stats() 无任何消费方,却在 emit() 热路径增加 qsize 与两次加锁 @
rtp_llm/access_logger/async_log_handler.py:224- 建议:要么把
get_stats()真正接出去(在已有 metrics 上报或调试接口中周期性读取),要么先精简为 worker 侧确实需要跨线程保护的written/write_errors,并把max_queue_depth采样移到 worker 线程(_process_batch每批一次),避免逐条调用qsize()。
- 建议:要么把
- del 触发的 close() 实际不可达,孤儿 handler 仍泄漏且 finalizer 内会阻塞与建线程 @
rtp_llm/access_logger/async_log_handler.py:244- 建议:让
init_logger/init_dash_sc_grpc_access_logger在handlers.clear()之前显式对旧 handler 调用close(),把生命周期交给调用方而非 GC;__del__内不要再调用会阻塞或建线程的close(),如需兜底改用weakref.finalize且只做置位_stop_event这类非阻塞动作;异常至少降级为 debug 日志而非完全静默。
- 建议:让
- 单个 PR 混合五件互不相关的变更,难以定位与回滚 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.h:70- 建议:建议至少拆成三个 PR:(1) xgrammar 加载修复(含上面的 P0);(2) cache_store 闭包/错误码加固;(3) model_rpc 计时与错误码透传 + metrics 宏。指标与错误码语义变更请单独成 commit,并在 PR description 中逐项列出受影响的 metric、error code 清单及回滚手段(目前均无环境变量开关),便于看板、告警和运维决策。
P3
- allocateResource 失败时同一错误被重复打印两条日志,且每次重试都成对翻倍 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.cc:207- 建议:删除
:207的RTP_LLM_LOG_ERROR,统一依赖serializeErrorMsg的日志;确实需要 ERROR 级别时,只在:1311之后(重试耗尽的最终失败点)打印一次。
- 建议:删除
- 删除 loadCacheSyncForTp 后 thread_pool_ 与 initThreadPool 成为完全不可达的死代码 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.h:103- 建议:在本次清理中一并删除
initThreadPool()的声明/定义、thread_pool_成员及析构中的相关代码;若后续确有异步 load 线程池需求,届时按实际语义重新引入并在 PR description 中说明。
- 建议:在本次清理中一并删除
- 新增与修改的行未通过 black / clang-format @
rtp_llm/config/grammar_tokenizer_info.py:164- 建议:对本次改动的文件跑一次仓库 pre-commit(black + isort + clang-format),使格式化结果与逻辑改动在同一提交内保持一致,避免占用 CI 轮次或后续产生纯格式化噪声 diff。
Checklist Findings (23 fail / 121 total)
General Principles Checklist
- [6.1] Architecture — 兼容性:外部 HTTP/RPC API、持久数据、配置、环境迁移安全 → issue
loading_cache_request 改为析构期采样后几乎恒为 0,叠加零值上报会拉平该指标
改为const std::atomic<size_t>*后取值时机从「RemoteGenerate 入口值快照」变成析构期reportTime()时load()(DecodeGenerateContext.cc:74-75、PrefillGenerateContext.cc:319-320)。但AtomicGuard request_guard(loading_cache_requests_)的作用域仅限loadCacheFromPrefill函数体(DecodeRpcServer.cc:226)与 prefill 对应窗口(PrefillRpcServer.cc:399),析构时该计数早已回落,低并发下采样值几乎必然为 0。叠加同 PR 的REPORT_GAUGE无条件上报,rtp_llm_rpc_loading_cache_request会被大量 0 样本拉平,「load cache 并发度」看板语义失效。相较之下onflight_requests_的 guard 覆盖整个RemoteGenerate(:1294), - [6.1] Architecture — 分层边界:新概念在正确层级,不泄漏内部 → issue
运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉
:154-159在:161 import xgrammar之前把包目录前插到os.environ["LD_LIBRARY_PATH"]。glibc 的 ld.so 在进程启动时即固化该变量的搜索路径,进程内改写通常不影响后续 C 层dlopen(),能否生效完全取决于 xgrammar/tvm-ffi loader 是否在 Python 层自行读取该变量(第三方实现,本仓不可见),而 docstring:148直接声称 "including from Bazel rules_python runfiles"。仓库内models_py/modules/dsv4/tilelang_kernels.py:54-62对完全同类的 bazel runfiles 裸名 dlopen 问题,注释明确写着 "only works if the lib is on LD_LIBRARY_PATH or loaded eagerly",并在:68采用ctypes.CDLL(path, mode=ctypes.RTLD_GLOBAL)预加载。 - [6.1] Architecture — 可观测性:日志/指标/超时可操作、非噪声 → issue
allocateResource 失败时同一错误被重复打印两条日志,且每次重试都成对翻倍
:207先RTP_LLM_LOG_ERROR("request [%s] allocate resource failed, error code [%s], error message [%s]", ...),:211的serializeErrorMsg内部又RTP_LLM_LOG_WARNING("%s, error code [%s], error message [%s]", ...)(LocalRpcServer.cc:91-94)输出同一 request_key / error_code / error_message。该函数由EXECUTE_WITH_RETRY驱动,每次重试都会成对打印;旧代码在此只打一条 ERROR。 - [6.1] Architecture — 回滚路径:风险行为存在运维回滚手段 → issue
单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个改动文件横跨五个彼此独立的关注点:xgrammar 加载入口(2 个 Python 文件)、cache_store 闭包与错误码(6 个 C++ 文件)、model_rpc 阶段计时与错误码透传(12 个 C++ 文件)、访问日志 handler 生命周期(1 个 Python 文件)、REPORT_GAUGE宏语义(1 个 C++ 文件)。分支名fix/cache-store-rpc-xgrammar本身即三个主题的拼接。其中 grammar 的 P0 与 model_rpc 的阶段计时 P1 无任何因果关联,而至少四项各自带有独立的对外可见语义变更(错误码 8211→602、cache_store wire error_code 取值域与 presence、69 个 gauge 的零值上报、两个 gauge 的采样时点),任一项出问题都只能整体回滚,CI 二分定位也无法进行。 - [6.1] Architecture — 状态不变量:创建/更新/失败/重试/回滚路径有效 → issue
__del__ 触发的 close() 实际不可达,孤儿 handler 仍泄漏且 finalizer 内会阻塞与建线程
:100-104 threading.Thread(target=self._worker_loop)持有 bound method 对 self 的强引用,而 worker 循环在_stop_event未置位时永不退出,因此 handler 永远不会被 GC,新增的__del__ -> close()清理路径在正常运行下不可达;access_logger.py:31与dash_sc/access_log.py:63/88的handlers.clear()仍会泄漏线程与文件句柄。反之若 worker 因致命错误退出而触发__del__,close() -> flush() -> _start_worker()会在 finalizer 里新建线程从而复活对象。:248 except Exception: pass静默吞掉全部异常,且对join()挂死无效。 - [6.1] Architecture — 错误语义:fail-fast/retry/fallback/silent 行为显式 → issue
close() 在 worker join 超时后仍关闭底层流,日志丢失既无计数也无告警
:219 join(timeout=max(1.0, self._flush_interval * 2))之后无任何状态判断,:220直接self._file_handler.close()。join 带超时即意味着 worker 可能仍停在_write_record(:155) 或:149的_file_handler.flush(),磁盘卡顿或 rotation 慢时尤为可能。此时底层流被关闭,后续写入抛ValueError: I/O operation on closed file,被吞成write_errors并刷屏logging.error,剩余记录静默丢失,与 docstring:9的 "graceful shutdown ensures important logs are not lost" 矛盾。emit()在_stop_event置位后于:182-183静默 return,这部分丢弃也未计入dropped。 - [6.1] Quality — Commit 原子、message 与行为匹配 → issue
单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个改动文件横跨五个彼此独立的关注点:xgrammar 加载入口(2 个 Python 文件)、cache_store 闭包与错误码(6 个 C++ 文件)、model_rpc 阶段计时与错误码透传(12 个 C++ 文件)、访问日志 handler 生命周期(1 个 Python 文件)、REPORT_GAUGE宏语义(1 个 C++ 文件)。分支名fix/cache-store-rpc-xgrammar本身即三个主题的拼接。其中 grammar 的 P0 与 model_rpc 的阶段计时 P1 无任何因果关联,而至少四项各自带有独立的对外可见语义变更(错误码 8211→602、cache_store wire error_code 取值域与 presence、69 个 gauge 的零值上报、两个 gauge 的采样时点),任一项出问题都只能整体回滚,CI 二分定位也无法进行。 - [6.1] Quality — Mega-PR 已拆分为独立变更 → issue
单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个改动文件横跨五个彼此独立的关注点:xgrammar 加载入口(2 个 Python 文件)、cache_store 闭包与错误码(6 个 C++ 文件)、model_rpc 阶段计时与错误码透传(12 个 C++ 文件)、访问日志 handler 生命周期(1 个 Python 文件)、REPORT_GAUGE宏语义(1 个 C++ 文件)。分支名fix/cache-store-rpc-xgrammar本身即三个主题的拼接。其中 grammar 的 P0 与 model_rpc 的阶段计时 P1 无任何因果关联,而至少四项各自带有独立的对外可见语义变更(错误码 8211→602、cache_store wire error_code 取值域与 presence、69 个 gauge 的零值上报、两个 gauge 的采样时点),任一项出问题都只能整体回滚,CI 二分定位也无法进行。 - [6.1] Quality — PR description 说明动机与设计 → issue
单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个改动文件横跨五个彼此独立的关注点:xgrammar 加载入口(2 个 Python 文件)、cache_store 闭包与错误码(6 个 C++ 文件)、model_rpc 阶段计时与错误码透传(12 个 C++ 文件)、访问日志 handler 生命周期(1 个 Python 文件)、REPORT_GAUGE宏语义(1 个 C++ 文件)。分支名fix/cache-store-rpc-xgrammar本身即三个主题的拼接。其中 grammar 的 P0 与 model_rpc 的阶段计时 P1 无任何因果关联,而至少四项各自带有独立的对外可见语义变更(错误码 8211→602、cache_store wire error_code 取值域与 presence、69 个 gauge 的零值上报、两个 gauge 的采样时点),任一项出问题都只能整体回滚,CI 二分定位也无法进行。 - [6.1] Quality — 无 per-forward 调试日志 / 噪声热路径输出 → issue
基类 stopStream() 在正常成功路径上每请求输出一条 WARNING
GenerateStreamTest.cc:277-299显式断言:nextOutput()返回ErrorCode::FINISHED后getStatus()仍为StreamState::RUNNING(StreamState::FINISHED只由 scheduler 后续moveToNext()设置)。pollStreamOutput正是在该点 break(LocalRpcServer.cc:124-128),随后 context 被销毁。此时getStatus() != FINISHED、hasError()==false、error_info为空、未取消,必然落进:97-101的 else 分支,对每个成功请求打一条stopping unfinished stream with terminal source=context_cleanupWARNING。PrefillGenerateContext::stopStream(:89) 正是为这一竞态才加了 `hasEvent(Generat - [6.1] Quality — 逻辑变更未混入无关格式化 → issue
新增与修改的行未通过 black / clang-format
Python 侧::163 return xgr之后仅 1 个空行(:164)即接:165顶层定义def build_grammar_tokenizer_info_json(,同文件其余顶层函数之间(如:144→:147)均为 2 个空行;仓库 pre-commit 启用 black 且该路径不在排除列表(仅排除rtp_llm/ops、3rdparty),会被black --check判为需重新格式化。C++ 侧:.clang-format启用AlignConsecutiveDeclarations/AlignConsecutiveAssignments,而DecodeGenerateContext.h:72的const std::atomic<size_t>* loading_cache_requests打断了:68-73的成员声明对齐块,DecodeRpcServer.cc:1297-1299三行连续赋值的=分处三个不同列位,Messager.cpp:80的 `local - [6.1] Software Engineering — DRY:重复非平凡逻辑被抽取或显式复用 → issue
运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉
:154-159在:161 import xgrammar之前把包目录前插到os.environ["LD_LIBRARY_PATH"]。glibc 的 ld.so 在进程启动时即固化该变量的搜索路径,进程内改写通常不影响后续 C 层dlopen(),能否生效完全取决于 xgrammar/tvm-ffi loader 是否在 Python 层自行读取该变量(第三方实现,本仓不可见),而 docstring:148直接声称 "including from Bazel rules_python runfiles"。仓库内models_py/modules/dsv4/tilelang_kernels.py:54-62对完全同类的 bazel runfiles 裸名 dlopen 问题,注释明确写着 "only works if the lib is on LD_LIBRARY_PATH or loaded eagerly",并在:68采用ctypes.CDLL(path, mode=ctypes.RTLD_GLOBAL)预加载。 - [6.1] Software Engineering — KISS/YAGNI:无投机性抽象 → issue
删除 loadCacheSyncForTp 后 thread_pool_ 与 initThreadPool 成为完全不可达的死代码
loadCacheSyncForTp中的thread_pool_->async(...)是thread_pool_的唯一使用者,已随本 PR 删除(全仓搜索loadCacheSyncForTp零命中)。删除后全仓搜索确认:DecodeRpcServer::initThreadPool()(声明 DecodeRpcServer.h:70,定义 DecodeRpcServer.cc:127-135)本身没有任何调用点(唯一同名命中是PrefillBatchRpcServer::initThreadPools,不同类不同函数且有调用点),成员thread_pool_(.h:103)及析构中的stop()/reset()(.cc:137-142)全部成为死代码,已无任何pushTask/async调用点。initThreadPool()内if (resource_.workers.size() > 0) return;的判断方向还与函数名语义相反。 - [6.1] Software Engineering — LSP:子类/重写保持基类契约 → issue
基类 stopStream() 在正常成功路径上每请求输出一条 WARNING
GenerateStreamTest.cc:277-299显式断言:nextOutput()返回ErrorCode::FINISHED后getStatus()仍为StreamState::RUNNING(StreamState::FINISHED只由 scheduler 后续moveToNext()设置)。pollStreamOutput正是在该点 break(LocalRpcServer.cc:124-128),随后 context 被销毁。此时getStatus() != FINISHED、hasError()==false、error_info为空、未取消,必然落进:97-101的 else 分支,对每个成功请求打一条stopping unfinished stream with terminal source=context_cleanupWARNING。PrefillGenerateContext::stopStream(:89) 正是为这一竞态才加了 `hasEvent(Generat - [6.1] Software Engineering — OCP:本地扩展点优先于修改中心逻辑 → issue
REPORT_GAUGE 移除零值守卫,PD 两侧互相向对方独有的阶段耗时 gauge 写 0
宏由if (collector->name) { REPORT_MUTABLE_METRIC(...) }改为无条件上报,本文件 69 个REPORT_GAUGE(调用点全部受影响,RpcMetrics::report(:152-193)独占 28 个。而RpcMetricsCollector(RtpLLMMetrics.h:21-66)中 prefill 字段(:37-47,11 个)与 decode 阶段字段(:50-53)+ tp 字段(:63-65)互不相交且默认 0:DecodeGenerateContext::reportTime(:69-88) 只填 decode 字段,PrefillGenerateContext::reportTime(:314-335) 只填 prefill 字段,基类只填 basic。改后 decode 进程每请求向multimodal_process_rt_us、remote_generate_rt_us等 11 个 prefill-only gauge 写 0,prefill 侧向 `prepare - [6.1] Tests — 分布式/跨平台变更有对应覆盖 → issue
22 个变更文件零测试,且现有测试写法结构性无法发现本次回归
diff_paths22 个文件中零测试文件,而 harness 齐备:rtp_llm/cpp/model_rpc/test/已有DecodeRpcServerTest.cc/PrefillRpcServerTest.cc/LocalRpcServerTest.cc,cache_store/test/TcpCacheStoreLoadServiceClosureTest.cpp亦存在。在model_rpc/test/检索stat_info|deserializeErrorDetails|onflight_requests|nextStage|finishStage|stopStream零命中。更关键的是现有写法结构性屏蔽本次回归:grammar_validator_test.py:34用GrammarValidator.__new__(GrammarValidator)完全跳过__init__→_build_backend,故上述 P0 的NameError不可能被发现;`TcpCacheStoreLoad - [6.1] Tests — 新逻辑有聚焦单测 + 相关集成/smoke 测试 → issue
22 个变更文件零测试,且现有测试写法结构性无法发现本次回归
diff_paths22 个文件中零测试文件,而 harness 齐备:rtp_llm/cpp/model_rpc/test/已有DecodeRpcServerTest.cc/PrefillRpcServerTest.cc/LocalRpcServerTest.cc,cache_store/test/TcpCacheStoreLoadServiceClosureTest.cpp亦存在。在model_rpc/test/检索stat_info|deserializeErrorDetails|onflight_requests|nextStage|finishStage|stopStream零命中。更关键的是现有写法结构性屏蔽本次回归:grammar_validator_test.py:34用GrammarValidator.__new__(GrammarValidator)完全跳过__init__→_build_backend,故上述 P0 的NameError不可能被发现;`TcpCacheStoreLoad - [6.1] Tests — 边界 case 覆盖(空、单元素、最大值) → issue
None 记录被消费但未 task_done,可永久破坏新的阻塞式 flush()
_process_batch:127与:136取出的元素若为None不会进入records_batch,因此:142-146的循环不会为它调用task_done();_drain_queue:170-174的if record is not None:同样把task_done()挡在分支内。这些None判断的存在说明作者认为哨兵值可能入队(关闭队列的常见写法)。一旦后续有人向队列放入None,unfinished_tasks将永久大于 0,本次新增的无超时flush()会永久阻塞。当前emit()只放LogRecord,属潜在缺陷。
RTP-LLM Checklist
- [I] 代码质量 — 删除或重命名内部 file、registry entry、model name、metric enum、op binding、plugin symbol 时,必须全仓搜索消费者,并提供替代实现、迁移说明或 smoke 覆盖;只有暴露到 HTTP/RPC/config/persisted format 时才按外部兼容性处理 → issue
删除 loadCacheSyncForTp 后 thread_pool_ 与 initThreadPool 成为完全不可达的死代码
loadCacheSyncForTp中的thread_pool_->async(...)是thread_pool_的唯一使用者,已随本 PR 删除(全仓搜索loadCacheSyncForTp零命中)。删除后全仓搜索确认:DecodeRpcServer::initThreadPool()(声明 DecodeRpcServer.h:70,定义 DecodeRpcServer.cc:127-135)本身没有任何调用点(唯一同名命中是PrefillBatchRpcServer::initThreadPools,不同类不同函数且有调用点),成员thread_pool_(.h:103)及析构中的stop()/reset()(.cc:137-142)全部成为死代码,已无任何pushTask/async调用点。initThreadPool()内if (resource_.workers.size() > 0) return;的判断方向还与函数名语义相反。 - [I] 代码质量 — 同一功能用统一工具函数 → issue
TP load cache 报错日志用 rank 直接索引 peer_addrs,非对称 TP 下指向错误对端
新增日志把peer_addrs[rank]当作该 rank 的对端(:676-678、:687-689),但真实映射在constructRemoteLoadRequest中是peer_addrs[index / part_cnt](D≥P,:468)、peer_addrs[index % peer_cnt](prefill CP,:461)或多地址组(P≥D,:473-476)。decode TP=8、prefill TP=2 时 rank=5 实际连的是peer_addrs[1],而日志因rank < peer_addrs.size()不成立输出<missing>;rank 落在 size 内时更会打印一个从未连接过的 peer,把排障引向无关节点——恰好是本 PR 想改善的场景。
Python Static-First Checklist
- [P.B] 错误处理 — 禁止 bare except 或静默吞异常 → issue
__del__ 触发的 close() 实际不可达,孤儿 handler 仍泄漏且 finalizer 内会阻塞与建线程
:100-104 threading.Thread(target=self._worker_loop)持有 bound method 对 self 的强引用,而 worker 循环在_stop_event未置位时永不退出,因此 handler 永远不会被 GC,新增的__del__ -> close()清理路径在正常运行下不可达;access_logger.py:31与dash_sc/access_log.py:63/88的handlers.clear()仍会泄漏线程与文件句柄。反之若 worker 因致命错误退出而触发__del__,close() -> flush() -> _start_worker()会在 finalizer 里新建线程从而复活对象。:248 except Exception: pass静默吞掉全部异常,且对join()挂死无效。 - [P.F] 语言陷阱 — 禁止模块级 import 副作用 → issue
运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉
:154-159在:161 import xgrammar之前把包目录前插到os.environ["LD_LIBRARY_PATH"]。glibc 的 ld.so 在进程启动时即固化该变量的搜索路径,进程内改写通常不影响后续 C 层dlopen(),能否生效完全取决于 xgrammar/tvm-ffi loader 是否在 Python 层自行读取该变量(第三方实现,本仓不可见),而 docstring:148直接声称 "including from Bazel rules_python runfiles"。仓库内models_py/modules/dsv4/tilelang_kernels.py:54-62对完全同类的 bazel runfiles 裸名 dlopen 问题,注释明确写着 "only works if the lib is on LD_LIBRARY_PATH or loaded eagerly",并在:68采用ctypes.CDLL(path, mode=ctypes.RTLD_GLOBAL)预加载。 - [P.G] 测试规范 — mock/fake/stub 不得替代本次声称覆盖的生产边界 → issue
22 个变更文件零测试,且现有测试写法结构性无法发现本次回归
diff_paths22 个文件中零测试文件,而 harness 齐备:rtp_llm/cpp/model_rpc/test/已有DecodeRpcServerTest.cc/PrefillRpcServerTest.cc/LocalRpcServerTest.cc,cache_store/test/TcpCacheStoreLoadServiceClosureTest.cpp亦存在。在model_rpc/test/检索stat_info|deserializeErrorDetails|onflight_requests|nextStage|finishStage|stopStream零命中。更关键的是现有写法结构性屏蔽本次回归:grammar_validator_test.py:34用GrammarValidator.__new__(GrammarValidator)完全跳过__init__→_build_backend,故上述 P0 的NameError不可能被发现;`TcpCacheStoreLoad
Strengths
Messager.cpp:73-85拆分了原先的短路判断,修掉 buffer 未找到时在日志里解引用空指针的必然崩溃,并新增local_partition_count == 0前置校验消除len % 0除零 UB(x86 上 SIGFPE)。这是本 PR 中最扎实的真实缺陷修复。DecodeStatInfo/PrefillStatInfo引入begin_time == 0哨兵与配套finishStage(),使EXECUTE_WITH_RETRY的restoreStage()不再把重试 sleep 与上一段耗时重复累加;prefill 侧逐段核对(PrefillRpcServer.cc:616-654共 9 个EXECUTE_STAGE_FUNC+ 收尾nextStage(),恰好到finish=10)归属完全正确,可作为正确用法对照。finishStage()把start/finish显式纳入 switch 分支(DecodeGenerateContext.cc:38-40),消除了旧实现落到default触发RTP_LLM_CHECK_WITH_INFO(false)的 abort 面。- decode allocate 失败改走
serializeErrorMsg(LocalRpcServer.cc:96-101),复用既有ErrorDetailsPB(proto 未改),真实ErrorCode得以跨 P→D 存活;对NONE_ERROR(转UNKNOWN_ERROR,DecodeRpcServer.cc:198-201)与空 message(转ErrorCodeToString)各有兜底,消除了「stream 报错但没有错误码」的语义空洞,替代了脆弱的 error_message 子串匹配。 CacheTransferServiceImplContext.cpp:76-84新增blocks.size() > unfinished_count_不变量校验,阻止unfinished_count_减为负数后永不命中== 0、进而done_永不触发的挂死路径。TcpCacheStoreLoadServiceClosure.cpp:92的execNoBlockCopy是真实可抛异常点(torch 算子),此前异常逃出Run()会导致 closure 泄漏且 callback 永不触发(调用方挂到超时),新增 try/catch 让它收敛到一次end(false, LoadErrorUnknown)。- 错误日志从裸字符串
RTP_LLM_LOG_ERROR(error_msg)改为带格式串的形式,消除了远端可控字符串直接作为 format string 的隐患。 stopStream()新增!stream_->hasError()守卫(GenerateContext.cc:86),不再用 CANCELLED 覆盖 stream 上已有的真实错误,并区分 context_error / client_cancel / context_cleanup 三类终态来源。loadCacheAsyncForTp失败/超时/取消三条路径均补齐 rank/worker/peer/cq/grpc_code/finished 计数,且用resource_.grpc_workers.at(rank)(DecodeRpcServer.cc:675/686)而非[],越界显式抛出而非 UB。loadCacheSyncForTp删除干净:全仓搜索零残留引用,同时移除了原实现中min/max_response_done_time_us被多线程 lambda 无锁并发写入的数据竞争。- 并发计数改为
const std::atomic<size_t>*后,四个 server 均已赋值(DecodeRpcServer.cc:1298-1299、PrefillRpcServer.cc:709-710、PrefillBatchRpcServer.cc:899-900、LocalRpcServer.cc:199),? :兜底完备,propagation 无遗漏。 _process_batch/_drain_queue用try/finally包裹_write_record(async_log_handler.py:142-146、170-174),写入抛异常时仍递减unfinished_tasks;_stats读写统一收敛到_stats_lock。
| self._tokenizer_info_json | ||
| ) | ||
| return xgr.GrammarCompiler( | ||
| return load_xgrammar().GrammarCompiler( |
There was a problem hiding this comment.
[P0] load_xgrammar 未绑定到模块全局,GrammarValidator 必然构建失败并静默关闭语法准入校验
全仓 grep load_xgrammar 仅 4 处:定义在 grammar_tokenizer_info.py:147,本文件只有 :786 在 _xgrammar() 函数体内的局部 import 与 :788 调用,模块级 import 段(:17-36)无此名字且无 import *。故 :454 的 load_xgrammar().GrammarCompiler(...) 全局查找必然 NameError,被 :460 except Exception 包成 RuntimeError。_build_backend() 由 _initialize_compiler 在 :405 无条件调用(注释写明 "Fail service startup..."),:445-449 已保证 xgr 非 None、无分支可绕过。上层 app.py:621-636 用 except Exception 兜住、仅打 warning 并置 grammar_validator = None;`_spawned_san...
建议: :454 改回 return xgr.GrammarCompiler(——:445 的 xgr 已是 load_xgrammar() 的记忆化结果,语义等价且不重复加载;若确需显式调用,请在文件顶部(:36 附近)补模块级 from rtp_llm.config.grammar_tokenizer_info import load_xgrammar。同时补一条 _build_backend() 成功路径单测:现有 grammar_validator_test.py:34 用 GrammarValidator.__new__(GrammarValidator) 绕过 __init__,正是该缺陷逃过 CI 的直接原因,新测试不得以 mock 替换 xgrammar 这一生产边界(P.P.G.3)。另建议不要用宽 except Exception 把 NameError 这类编程错误与真实的 xgrammar 初始化失败混成同一条 RuntimeError 文案;上层降级为 None 时建议把 warning 升级为可告警指标,避免准入校验静默关闭。
|
|
||
| try { | ||
| EXECUTE_STAGE_FUNC(prepareGenerateContext, decode_context); | ||
| decode_context.stat_info.nextStage(); |
There was a problem hiding this comment.
[P1] decode 阶段计时被双重推进,四个耗时指标全部错位且 stage 越出枚举上界
新语义为「自结算」:finishStage() 把耗时记入 stage 自身桶(DecodeGenerateContext.cc:19-44),EXECUTE_STAGE_FUNC 因此自带 nextStage()+finishStage() 配对(GenerateContext.h:101-103),而 allocateResourceFunc 本身就是 EXECUTE_STAGE_FUNC(allocateResource,...)(:1264)。逐帧推演::1307 额外一次 nextStage() 使 stage 多推进一格,allocate_resource_rt_us 只记 ≈0 的空隙;allocateResource 真实耗时落入 load_cache_from_prefill_rt_us;loadCacheFromPrefill 落入 local_generate_rt_us;localGenerate 命中 case finish 被整体丢弃。:1310 因内层已把 begin_time 置 0 恒为空操作;...
建议: 删除 :1307 与 :1310 两行,让 allocate 阶段计时完全由 allocateResourceFunc 内层的 EXECUTE_STAGE_FUNC 负责(finishStage() 的 begin_time == 0 幂等短路已保证重试不重复累加),与 prefill 写法一致;若目的是把多次 retry 合并成一段耗时,应在 EXECUTE_WITH_RETRY 内累加而非额外推进 stage。同时清理 :1331 这条在新语义下已多余、且使 stage 越界的裸 nextStage(),并在 nextStage() 内加断言禁止 stage 超过 finish(当前虽无后续 finishStage() 调用未触发 default abort,但属潜在隐患)。请在已有的 rtp_llm/cpp/model_rpc/test/DecodeRpcServerTest.cc 补一个 DecodeStatInfo 单测:模拟 prepare → allocate(含 1 次重试)→ loadCache → localGenerate,断言四个 *_rt_us 各自非零、互不串位且 stage 终值不越界。
| new_error_msg += error_msg; \ | ||
| if (status.error_code() == grpc::StatusCode::RESOURCE_EXHAUSTED) { \ | ||
| new_error_code = ErrorCode::DECODE_MALLOC_FAILED; \ | ||
| if (!remote_error_info.hasError()) { \ |
There was a problem hiding this comment.
[P1] decode KV cache 分配失败的对外错误码从 DECODE_MALLOC_FAILED(8211) 退化为 MALLOC_FAILED(602)
decode 侧 KV 分配失败时 stream 错误码为 MALLOC_FAILED(602)(GenerateStateMachine.cc:73/116 均为 ErrorInfo(MALLOC_FAILED, "LACK MEM"))。新代码在 DecodeRpcServer.cc:206/211 把它写入 error_info 并经 serializeErrorMsg 序列化进 ErrorDetailsPB(LocalRpcServer.cc:96-101)。prefill 侧 :127 的 deserializeErrorDetails 命中后 new_error_code = 602,而新增的 :149 if (!remote_error_info.hasError()) 使原本无条件执行的 RESOURCE_EXHAUSTED → DECODE_MALLOC_FAILED(:151-153)整体被跳过。改动前 decode 只回不带 details 的 `grpc::Status(RESOURCE_EXHAUSTED, msg...
建议: 在 decode 侧构造 error_info 时就固化角色语义,例如 allocateResource 中把 MALLOC_FAILED 显式映射为 DECODE_MALLOC_FAILED 后再写入(这样 8211 也能经 details 原样跨节点存活);或在 prefill 的 details 命中分支保留补偿映射 status.error_code() == RESOURCE_EXHAUSTED && remote_code == MALLOC_FAILED → DECODE_MALLOC_FAILED。请在 PR description 中列出本次对外错误码取值域的变化,并确认上游按 8211 做重试/降级判定的逻辑仍成立;建议在已有的 PrefillRpcServerTest.cc 补一条用例,断言 decode 返回 602 + RESOURCE_EXHAUSTED 时 prefill 对外仍报 8211。
| if (collector->name) { \ | ||
| REPORT_MUTABLE_METRIC(name##_metric, collector->name); \ | ||
| } | ||
| #define REPORT_GAUGE(name) REPORT_MUTABLE_METRIC(name##_metric, collector->name) |
There was a problem hiding this comment.
[P1] REPORT_GAUGE 移除零值守卫,PD 两侧互相向对方独有的阶段耗时 gauge 写 0
宏由 if (collector->name) { REPORT_MUTABLE_METRIC(...) } 改为无条件上报,本文件 69 个 REPORT_GAUGE( 调用点全部受影响,RpcMetrics::report(:152-193)独占 28 个。而 RpcMetricsCollector(RtpLLMMetrics.h:21-66)中 prefill 字段(:37-47,11 个)与 decode 阶段字段(:50-53)+ tp 字段(:63-65)互不相交且默认 0:DecodeGenerateContext::reportTime(:69-88) 只填 decode 字段,PrefillGenerateContext::reportTime(:314-335) 只填 prefill 字段,基类只填 basic。改后 decode 进程每请求向 multimodal_process_rt_us、remote_generate_rt_us 等 11 个 prefill-only gauge 写 0,prefill 侧向 `prep...
建议: 不要用一个宏统一改变 69 个调用点的语义:保留条件版 REPORT_GAUGE,另增 REPORT_GAUGE_ALWAYS,仅对 onflight_request、total_rt_us 等每周期确实有效的状态量使用;阶段耗时类(全部 PD 阶段耗时、retry_times)维持「未发生即不上报」。顺带收敛本文件并存的三套约定——REPORT_QPS(:38-41) 仍带守卫、REPORT_GAUGE(:43) 无守卫、REPORT_NON_ZERO_MUTABLE_METRIC(:751) 专门表达大于 0,另有多处裸 REPORT_MUTABLE_METRIC——统一为语义明确的两个宏并在文件头注释写清适用场景。请在 PR description 中列出语义变更的 metric 清单,便于看板与告警阈值同步调整。
Checklist: [6.1] OCP:本地扩展点优先于修改中心逻辑
| request_key.c_str()); | ||
| stream_->reportError(ErrorCode::CANCELLED, "request cancelled by client"); | ||
| } else { | ||
| RTP_LLM_LOG_WARNING("request [%s] stopping unfinished stream with terminal source=context_cleanup", |
There was a problem hiding this comment.
[P1] 基类 stopStream() 在正常成功路径上每请求输出一条 WARNING
GenerateStreamTest.cc:277-299 显式断言:nextOutput() 返回 ErrorCode::FINISHED 后 getStatus() 仍为 StreamState::RUNNING(StreamState::FINISHED 只由 scheduler 后续 moveToNext() 设置)。pollStreamOutput 正是在该点 break(LocalRpcServer.cc:124-128),随后 context 被销毁。此时 getStatus() != FINISHED、hasError()==false、error_info 为空、未取消,必然落进 :97-101 的 else 分支,对每个成功请求打一条 stopping unfinished stream with terminal source=context_cleanup WARNING。PrefillGenerateContext::stopStream(:89) 正是为这一竞态才加了 `hasEvent(Gene...
建议: 基类复用同一判定:stream_->hasEvent(StreamEvents::GenerateDone) && !stream_->hasError() 时视为正常收尾,降级为 RTP_LLM_LOG_DEBUG 且不再 reportError(CANCELLED)(这同时能避免误报错误干扰 decode 侧 releaseResource() → insertIntoCache() 的 KV 复用写入,并顺带修正成功请求被计入 cancel_qps 的既有偏差);context_error / client_cancel 两条 WARNING 保留,仅在确实无 GenerateDone 时才输出 context_cleanup WARNING。这样可消除 local(非 PD)与 decode 两条最高 QPS 路径上的日志洪泛。
Checklist: [6.1] 无 per-forward 调试日志 / 噪声热路径输出;[6.1] LSP:子类/重写保持基类契约
| if dropped % 10 == 1: # Reduce logging frequency | ||
| logging.warning(f"AsyncLogHandler: dropped {dropped} log records (queue full)") | ||
|
|
||
| def flush(self) -> None: |
There was a problem hiding this comment.
[P1] flush() 由 NOOP 改为无上界 Queue.join() 并被 close() 调用,反转模块自身 never-block 契约
:209 self._queue.join() 无超时、无 _stop_event 短路,其唯一自愈手段 _start_worker() 在 _stop_event 置位时于 :97-98 直接 return。logging.Handler.__init__ 会把 handler 注册进 logging._handlerList,atexit 的 logging.shutdown() 逐个 flush() 再 close(),故该路径每次进程退出必经;log_utils.py:116 配置 max_queue_size=****** 且写入为同步文件 I/O,磁盘变慢或 rotation 阻塞时退出会被整个 backlog 的排空时长无界拖住。另注意 close() 有 :213 if self._stop_event.is_set(): return 守卫而 flush() 没有,故「先显式 close(:219 的 join 带 timeout,worker 可能仍在写而留下 unfinished_tasks)→...
建议: flush() 开头增加 if self._stop_event.is_set(): self._file_handler.flush(); return 短路;把 Queue.join() 换成有上界的等待(threading.Condition 或按 flush_interval 轮询 unfinished_tasks),总超时后记录剩余条数并放弃,等待预算与部署 SIGTERM 宽限期对齐;close() 中先 _stop_event.set() 再 flush,消除入队竞态窗口;不要在 flush() 内重启 worker,把重启职责留给 emit()。__del__ 中不要做阻塞等待——join() 不抛异常,:248 的 except Exception: pass 无法兜住挂死。
| CHECK_REQUEST_STOP(generate_context) \ | ||
| generate_context.stat_info.nextStage(); \ | ||
| func(generate_context); \ | ||
| generate_context.stat_info.finishStage(); \ |
There was a problem hiding this comment.
[P1] 22 个变更文件零测试,且现有测试写法结构性无法发现本次回归
diff_paths 22 个文件中零测试文件,而 harness 齐备:rtp_llm/cpp/model_rpc/test/ 已有 DecodeRpcServerTest.cc/PrefillRpcServerTest.cc/LocalRpcServerTest.cc,cache_store/test/TcpCacheStoreLoadServiceClosureTest.cpp 亦存在。在 model_rpc/test/ 检索 stat_info|deserializeErrorDetails|onflight_requests|nextStage|finishStage|stopStream 零命中。更关键的是现有写法结构性屏蔽本次回归:grammar_validator_test.py:34 用 GrammarValidator.__new__(GrammarValidator) 完全跳过 __init__→_build_backend,故上述 P0 的 NameError 不可能被发现;`TcpCacheStoreL...
建议: 优先补三条能直接拦住本 PR 阻塞项的用例:(1) 以合法 tokenizer_info_json 真实构造 GrammarValidator 并断言 _backend is not None,不得用 mock 替换 xgrammar 这一生产边界;(2) DecodeStatInfo/PrefillStatInfo 在「正常顺序 / retry 后 restoreStage / 提前 return」三种路径下各 *_rt_us 的归属与 stage 上界断言;(3) serializeErrorMsg → deserializeErrorDetails 往返能还原原始 ErrorCode,并断言 decode 返回 602 时 prefill 对外仍为 8211。其次补低成本高收益的分支用例:response 不设 error_code 时回调收到 LoadErrorUnknown、callback/execNoBlockCopy 抛异常时回调恰好一次且 closure 被释放(配合 ASAN)、local_partition_count == 0 与 buffer 未找到均返回 nullptr 且不崩溃、blocks.size() > unfinished_count_ 不变量分支。
Checklist: [6.1] 分布式/跨平台变更有对应覆盖;[6.1] 新逻辑有聚焦单测 + 相关集成/smoke 测试;[P.G] mock/fake/stub 不得替代本次声称覆盖的生产边界
| if (!details.ParseFromString(status.error_details())) { | ||
| return false; | ||
| } | ||
| *error_info = ErrorInfo(static_cast<ErrorCode>(details.error_code()), |
There was a problem hiding this comment.
[P2] deserializeErrorDetails 无值域与 NONE_ERROR 校验,且与坏连接淘汰分支互斥
:49 直接 static_cast<ErrorCode>(details.error_code()) 后无条件 return true,未像同文件 transRPCErrorCode(:54-97)/transErrorCodeToRPC(:99-142) 那样用白名单映射 + UNKNOWN_ERROR 兜底,也未校验 error_code != NONE_ERROR(proto3 int64 无 presence,缺失即 0)。滚动升级或对端新增 code 时会得到值域外枚举,并沿 reportEvent(Error, code)(PrefillRpcServer.cc:161) 与 RpcMetrics 的 error_code tag 外溢。更关键的是该 helper 置于 PrefillRpcServer.cc:127 判别链最前且用 else if 串联,命中后 Connect Failed/Connection reset by peer 等分支被短路,closeGrpcConnection() 不再执行,坏 c...
建议: 在 helper 内补 details.error_code() != static_cast<int64_t>(ErrorCode::NONE_ERROR) 判断并校验数值落在已知枚举集合中,未知值回退 UNKNOWN_ERROR(可复用白名单映射思路);把两类判别改为互补而非互斥:先按传输层 error_message 决定是否 closeGrpcConnection(),再用远端 details 覆盖 error_code/message。同时把 PrefillRpcServer.cc:425 与 PrefillBatchRpcServer.cc:142 两处重复实现改为调用该 helper,避免三份同语义、强度不一致的代码继续分叉,并补「details 为空 / 非法 blob / code=0 / 正常 code」四类输入单测。
| end(true, CacheStoreErrorCode::None); | ||
| } | ||
|
|
||
| void TcpCacheStoreLoadServiceClosure::end(bool success, CacheStoreErrorCode ec) { |
There was a problem hiding this comment.
[P2] closure end() 自毁与 Run() 的 catch 重入构成 use-after-free 与 double free 窗口
end()(:107-122) 首行 std::unique_ptr<...> self(this) 接管自身,函数返回即 delete this;而本 PR 新增的 Run() catch(:95-104) 在异常路径上再次调用 end(false, LoadErrorUnknown)。end() 并非全程不抛:collector_->markEnd(success)(:109) 在 try 之外,两个 catch 处理器内 CacheStoreErrorCodeToString(ec)(:115/:120) 与 getRequestId() 会构造 std::string。一旦 end() 自身抛出,栈展开时 self 已 delete this,异常传到 Run() 的 catch 后先读 request_block_buffer_(:97,已释放内存)再第二次调用 end(),构成 UAF + double free 并使 markEnd 重复上报。`CacheTransferServiceClosure...
建议: 给两个 closure 各加幂等哨兵:成员 bool ended_ = false;,end() 入口 if (ended_) { return; } ended_ = true;;并把 end() 声明为 noexcept,把 collector_->markEnd 与全部日志/字符串构造移入内部 try/catch(...),使「self-destroy」与「异常兜底」两个职责不再互相重入。同时把 Run() catch 内取 request_id 的动作改为在 try 之前先拷贝到局部 std::string,避免在 catch 中触碰可能已释放的成员。建议把「自持生命周期 + 单次执行守卫 + callback 异常兜底」抽成共用基类或小工具,让同目录仍用裸 delete this 且 callback 无兜底的 TcpBlockReadClosure.cpp:84 一并复用(该文件不在本次 diff 内,如不便扩大范围请在 PR description 中记录为已知遗留项)。
| @@ -60,12 +60,26 @@ void CacheTransferServiceImplContext::notifyDone(bool | |||
| } | |||
|
|
|||
| if (error_code != CacheStoreErrorCode::None) { | |||
There was a problem hiding this comment.
[P2] notifyDone 忽略 success 形参,失败可被判为成功并回 EC_SUCCESS
notifyDone(success, error_code, blocks)(:47-49) 完全忽略 success 形参,仅以 error_code != CacheStoreErrorCode::None(:62) 判定失败。若任一 TransferConnection::read 实现回调 (false, None),则走「成功」分支、unfinished_count_ 归零后 runSuccess() 向 decode 侧回 EC_SUCCESS,而 KV block 实际未写入;CacheStoreUtil.cpp:38-39 亦显示 toKvCacheStoreErrorCode(None) == EC_SUCCESS,故 runFailed(None)(:98-101) 同样会回成功码。RemoteStoreTaskImpl.cpp:210-216 同构:transCacheStoreErrorCode(None) == NONE_ERROR(ErrorCodeUtil.h:10),会让 `all_succes...
建议: 将 :62 判定改为 if (!success || error_code != CacheStoreErrorCode::None),让 success 形参重新参与决策;并在 runFailed()(:98) 入口对 ec == None 兜底为 LoadErrorUnknown,保证「走失败路径」与「回非 EC_SUCCESS」这一不变量不可被破坏。RemoteStoreTaskImpl.cpp:214 同样加兜底:internal_error_code == NONE_ERROR 时替换为 CACHE_STORE_LOAD_UNKNOWN_ERROR,并对 (false, None) 打 ERROR 日志暴露上游违约。
| @@ -86,4 +100,4 @@ void CacheTransferServiceImplContext::runFailed(CacheStoreErrorCode ec) { | |||
| done_->Run(); | |||
There was a problem hiding this comment.
📍 实际位置 rtp_llm/cpp/disaggregate/cache_store/CacheTransferServiceImplContext.cpp:99(不在 diff 展示范围内,就近挂载)
[P2] cache_store 响应错误码取值域扩大且 has_error_code 语义翻转,混部期错误分类与重试行为会漂移
改动前失败一律回 toKvCacheStoreErrorCode(LoadRdmaWriteFailed) 即 EC_FAILED_RDMA_WRITE;改动后按真实码映射,而 CacheStoreUtil.cpp:36-51 的 switch 只显式覆盖 5 个枚举,CallPrefillTimeout/PushWorkerItemFailed/LoadConnectFailed/LoadErrorUnknown 全部落到 :48 的 default 分支 EC_FAILED_INTERNAL。对端 fromKvCacheStoreErrorCode(:32-33) 把 EC_FAILED_INTERNAL 解成 LoadErrorUnknown,最终内部 ErrorCode 由 CACHE_STORE_LOAD_RDMA_WRITE_FAILED 变为 CACHE_STORE_LOAD_UNKNOWN_ERROR。同时 closure 侧新增 !response_->has_error_code() 判失败(Tc...
建议: 在 PR description 中明确这批 wire 错误码取值域与 presence 语义的变化,并确认按 CACHE_STORE_LOAD_* 分类的重试判定对新增取值仍成立;必要时为 CallPrefillTimeout、PushWorkerItemFailed 等在 toKvCacheStoreErrorCode 中补专属枚举而非落 default,避免不同原因被压平到同一 EC_FAILED_INTERNAL。若判定逻辑确有分歧,建议加环境变量开关保留旧的压平行为作为运维回滚手段。
| package_dir = Path(spec.origin).parent | ||
| binding_path = package_dir / "libxgrammar_bindings.so" | ||
| if binding_path.is_file(): | ||
| library_paths = os.environ.get("LD_LIBRARY_PATH", "").split(":") |
There was a problem hiding this comment.
[P2] 运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉
:154-159 在 :161 import xgrammar 之前把包目录前插到 os.environ["LD_LIBRARY_PATH"]。glibc 的 ld.so 在进程启动时即固化该变量的搜索路径,进程内改写通常不影响后续 C 层 dlopen(),能否生效完全取决于 xgrammar/tvm-ffi loader 是否在 Python 层自行读取该变量(第三方实现,本仓不可见),而 docstring :148 直接声称 "including from Bazel rules_python runfiles"。仓库内 models_py/modules/dsv4/tilelang_kernels.py:54-62 对完全同类的 bazel runfiles 裸名 dlopen 问题,注释明确写着 "only works if the lib is on LD_LIBRARY_PATH or loaded eagerly",并在 :68 采用 ctypes.CDLL(path, mode=ctypes.RTLD_GLOBAL) 预...
建议: 对齐既有范式:import 前 ctypes.CDLL(str(binding_path), mode=ctypes.RTLD_GLOBAL)(包在 try/except OSError 中)显式预加载绑定库,成功即继续 import,失败再回退裸 import xgrammar;这样对当前进程真实有效,也不再污染子进程环境。建议把该预加载抽成共用的 preload_shared_library(path) 工具函数,避免与 tilelang_kernels.py 两处各写一套(R.I.1)。若确认第三方 loader 会读取该变量而必须保留,请在 docstring 写明这一外部实现依赖及「当前进程 + spawn 子进程」的实际生效范围,并用 try/finally 在 import 后恢复原值。
Checklist: [6.1] 分层边界:新概念在正确层级,不泄漏内部;[6.1] DRY:重复非平凡逻辑被抽取或显式复用;[P.F] 禁止模块级 import 副作用
| @@ -194,4 +243,7 @@ def setLevel(self, level) -> None: | |||
|
|
|||
| def __del__(self): | |||
There was a problem hiding this comment.
[P2] del 触发的 close() 实际不可达,孤儿 handler 仍泄漏且 finalizer 内会阻塞与建线程
:100-104 threading.Thread(target=self._worker_loop) 持有 bound method 对 self 的强引用,而 worker 循环在 _stop_event 未置位时永不退出,因此 handler 永远不会被 GC,新增的 __del__ -> close() 清理路径在正常运行下不可达;access_logger.py:31 与 dash_sc/access_log.py:63/88 的 handlers.clear() 仍会泄漏线程与文件句柄。反之若 worker 因致命错误退出而触发 __del__,close() -> flush() -> _start_worker() 会在 finalizer 里新建线程从而复活对象。:248 except Exception: pass 静默吞掉全部异常,且对 join() 挂死无效。
建议: 让 init_logger / init_dash_sc_grpc_access_logger 在 handlers.clear() 之前显式对旧 handler 调用 close(),把生命周期交给调用方而非 GC;__del__ 内不要再调用会阻塞或建线程的 close(),如需兜底改用 weakref.finalize 且只做置位 _stop_event 这类非阻塞动作;异常至少降级为 debug 日志而非完全静默。
Checklist: [6.1] 状态不变量:创建/更新/失败/重试/回滚路径有效;[P.B] 禁止 bare except 或静默吞异常
| @@ -82,7 +82,6 @@ class DecodeRpcServer: public RemoteRpcServer { | |||
| ErrorInfo loadCache(const LoadKVCacheContext& load_context); | |||
There was a problem hiding this comment.
📍 实际位置 rtp_llm/cpp/model_rpc/DecodeRpcServer.h:70(不在 diff 展示范围内,就近挂载)
[P2] 单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个改动文件横跨五个彼此独立的关注点:xgrammar 加载入口(2 个 Python 文件)、cache_store 闭包与错误码(6 个 C++ 文件)、model_rpc 阶段计时与错误码透传(12 个 C++ 文件)、访问日志 handler 生命周期(1 个 Python 文件)、REPORT_GAUGE 宏语义(1 个 C++ 文件)。分支名 fix/cache-store-rpc-xgrammar 本身即三个主题的拼接。其中 grammar 的 P0 与 model_rpc 的阶段计时 P1 无任何因果关联,而至少四项各自带有独立的对外可见语义变更(错误码 8211→602、cache_store wire error_code 取值域与 presence、69 个 gauge 的零值上报、两个 gauge 的采样时点),任一项出问题都只能整体回滚,CI 二分定位也无法进行。
建议: 建议至少拆成三个 PR:(1) xgrammar 加载修复(含上面的 P0);(2) cache_store 闭包/错误码加固;(3) model_rpc 计时与错误码透传 + metrics 宏。指标与错误码语义变更请单独成 commit,并在 PR description 中逐项列出受影响的 metric、error code 清单及回滚手段(目前均无环境变量开关),便于看板、告警和运维决策。
Checklist: [6.1] 回滚路径:风险行为存在运维回滚手段;[6.1] Commit 原子、message 与行为匹配;[6.1] Mega-PR 已拆分为独立变更;[6.1] PR description 说明动机与设计
| RTP_LLM_LOG_ERROR(error_msg); | ||
| decode_context.error_status = grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED, error_msg); | ||
| decode_context.error_info = ErrorInfo(error_code, error_msg); | ||
| RTP_LLM_LOG_ERROR("request [%s] allocate resource failed, error code [%s], error message [%s]", |
There was a problem hiding this comment.
[P3] allocateResource 失败时同一错误被重复打印两条日志,且每次重试都成对翻倍
:207 先 RTP_LLM_LOG_ERROR("request [%s] allocate resource failed, error code [%s], error message [%s]", ...),:211 的 serializeErrorMsg 内部又 RTP_LLM_LOG_WARNING("%s, error code [%s], error message [%s]", ...)(LocalRpcServer.cc:91-94)输出同一 request_key / error_code / error_message。该函数由 EXECUTE_WITH_RETRY 驱动,每次重试都会成对打印;旧代码在此只打一条 ERROR。
建议: 删除 :207 的 RTP_LLM_LOG_ERROR,统一依赖 serializeErrorMsg 的日志;确实需要 ERROR 级别时,只在 :1311 之后(重试耗尽的最终失败点)打印一次。
Checklist: [6.1] 可观测性:日志/指标/超时可操作、非噪声
| ErrorInfo loadCacheSyncForTp(DecodeGenerateContext& decode_context, LoadKVCacheContext& load_context); | ||
| BroadcastLoadRequestPB constructRemoteLoadRequest(const LoadKVCacheContext& load_context, | ||
| int index, | ||
| const std::vector<std::string>& peer_ips) const; |
There was a problem hiding this comment.
📍 实际位置 rtp_llm/cpp/model_rpc/DecodeRpcServer.h:103(不在 diff 展示范围内,就近挂载)
[P3] 删除 loadCacheSyncForTp 后 thread_pool_ 与 initThreadPool 成为完全不可达的死代码
loadCacheSyncForTp 中的 thread_pool_->async(...) 是 thread_pool_ 的唯一使用者,已随本 PR 删除(全仓搜索 loadCacheSyncForTp 零命中)。删除后全仓搜索确认:DecodeRpcServer::initThreadPool()(声明 DecodeRpcServer.h:70,定义 DecodeRpcServer.cc:127-135)本身没有任何调用点(唯一同名命中是 PrefillBatchRpcServer::initThreadPools,不同类不同函数且有调用点),成员 thread_pool_(.h:103)及析构中的 stop()/reset()(.cc:137-142)全部成为死代码,已无任何 pushTask/async 调用点。initThreadPool() 内 if (resource_.workers.size() > 0) return; 的判断方向还与函数名语义相反。
建议: 在本次清理中一并删除 initThreadPool() 的声明/定义、thread_pool_ 成员及析构中的相关代码;若后续确有异步 load 线程池需求,届时按实际语义重新引入并在 PR description 中说明。
Checklist: [6.1] KISS/YAGNI:无投机性抽象;[I] 删除或重命名内部 file、registry entry、model name、metric enum、op binding、plugin symbol 时,必须全仓搜索消费者,并提供替代实现、迁移说明或 smoke 覆盖;只有暴露到 HTTP/RPC/config/persisted format 时才按外部兼容性处理
| import xgrammar as xgr | ||
|
|
||
| return xgr | ||
|
|
There was a problem hiding this comment.
[P3] 新增与修改的行未通过 black / clang-format
Python 侧::163 return xgr 之后仅 1 个空行(:164)即接 :165 顶层定义 def build_grammar_tokenizer_info_json(,同文件其余顶层函数之间(如 :144→:147)均为 2 个空行;仓库 pre-commit 启用 black 且该路径不在排除列表(仅排除 rtp_llm/ops、3rdparty),会被 black --check 判为需重新格式化。C++ 侧:.clang-format 启用 AlignConsecutiveDeclarations/AlignConsecutiveAssignments,而 DecodeGenerateContext.h:72 的 const std::atomic<size_t>* loading_cache_requests 打断了 :68-73 的成员声明对齐块,DecodeRpcServer.cc:1297-1299 三行连续赋值的 = 分处三个不同列位,Messager.cpp:80 的 `lo...
建议: 对本次改动的文件跑一次仓库 pre-commit(black + isort + clang-format),使格式化结果与逻辑改动在同一提交内保持一致,避免占用 CI 轮次或后续产生纯格式化噪声 diff。
Checklist: [6.1] 逻辑变更未混入无关格式化
1f34a8a to
f365444
Compare
LLLLKKKK
left a comment
There was a problem hiding this comment.
AI Code Review - PR #1344
Status: BLOCKING
Summary: P0/1 · P1/6 · P2/14 · P3/5
Reviewed: commit f365444634a4 · 2026-08-28 17:57 UTC+8
Blocking Issues
P0
- load_xgrammar 未绑定到模块全局,GrammarValidator 必然构建失败并静默关闭语法准入校验 @
rtp_llm/dash_sc/inference/grammar_validator.py:454- 建议:
:454改为return xgr.GrammarCompiler(...),复用:445已加载并经lru_cache记忆的模块对象——既修掉NameError,也避免同一函数内并存两种取模块方式、绕过_xgrammar()的None语义契约;若确需显式调用则必须在模块顶部补from rtp_llm.config.grammar_tokenizer_info import load_xgrammar。同时补一条覆盖_build_backend()的聚焦单测:注入 fake xgrammar 模块后走真实_initialize_compiler路径,断言可用时返回非 None compiler、不可用时抛RuntimeError。现有grammar_validator_test.py:34用GrammarValidator.__new__绕过__init__,对该生产边界零覆盖,正是本缺陷逃过 CI 的直接原因(P.P.G.3)。另建议评估「配置要求开启但初始化失败」是否应改为启动期 fail-fast(:403-404注释本就如此声明),而非降级为 warning 后静默置空。
- 建议:
P1
- decode 阶段计时被双重推进,四个耗时指标全部错位且 stage 越出枚举上界 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.cc:1307- 建议:删除
:1307与:1310两行,让EXECUTE_STAGE_FUNC独占阶段推进(重试累加由restoreStage+begin_time哨兵天然处理),与PrefillRpcServer::syncPrefix(:627的EXECUTE_WITH_RETRY前无任何手工 bracket)保持同一范式;删除后末尾nextStage()正好停在finish不越界。若确实希望单独统计「含重试的整体 allocate 耗时」,请新增独立字段而不要复用 stage 状态机。请在已存在的rtp_llm/cpp/model_rpc/test/DecodeRpcServerTest.cc补一个纯DecodeStatInfo单测(无需 GPU):按prepare → allocate(含 N 次重试) → loadCache → localGenerate → finish驱动,断言四个*_rt_us各自落桶正确、终态 stage 恰为finish,并覆盖中途提前 return 的路径(stage 越界后任何finishStage()都会命中default的RTP_LLM_CHECK_WITH_INFO(false))。
- 建议:删除
- decode KV cache 分配失败的对外错误码从 DECODE_MALLOC_FAILED(8211) 退化为 MALLOC_FAILED(602) @
rtp_llm/cpp/model_rpc/PrefillRpcServer.cc:151- 建议:保留「远端码优先」的大方向,但对 decode 资源不足显式归一:解析成功后判断
remote_error_info.code() == ErrorCode::MALLOC_FAILED时映射为DECODE_MALLOC_FAILED;或更彻底地由 decode 在allocateResource内就以DECODE_MALLOC_FAILED构造ErrorInfo再serializeErrorMsg,使跨节点语义稳定为 8211。若这是有意的口径变更,请在 PR description 中明确说明,并同步核对rtp_llm/config/exceptions.py及依赖 8211 的监控看板、告警阈值与 FlexLB 调度归因,给出迁移说明。另请一并说明混部灰度期行为:新 decode + 旧 prefill 时旧端无deserializeErrorDetails,非 malloc 类错误的 gRPC status code 也已从固定RESOURCE_EXHAUSTED变为按transErrorCodeToGrpc映射,分类结果会随之变化。建议补一条断言「decode 报 602 时 prefill 对外仍为 8211」的往返单测。
- 建议:保留「远端码优先」的大方向,但对 decode 资源不足显式归一:解析成功后判断
- REPORT_GAUGE 移除零值守卫,68 处 gauge 语义静默失真且无回滚开关 @
rtp_llm/cpp/metrics/RtpLLMMetrics.cc:43- 建议:保留「值是否有效」的判定,而不是整体删除。本文件已有更合适的既有范式可复用:
PrefillRecentCacheKeyMetrics::report(:135)的has_value与RtpLLMExecutorMetrics::report(:364)的context_batch_size != 0。推荐为 collector 字段增加显式 set/has 标记(或用std::optional),既能上报真实测得的 0,又不会把「本部署形态/本阶段不适用」当成 0。若确实希望恒定上报,请拆为独立提交、提供环境变量开关与回滚说明,并在 PR description 中列出受影响指标清单及看板/告警迁移方案;同时说明RtpLLMExecutorMetrics::report(:361-379)位于 per-step 热路径、无条件上报 14 个 gauge 带来的上报量增长是否可接受。
- 建议:保留「值是否有效」的判定,而不是整体删除。本文件已有更合适的既有范式可复用:
- 基类 stopStream() 缺少 prefill 已有的 GenerateDone 守卫,正常成功路径每请求输出一条 WARNING @
rtp_llm/cpp/model_rpc/GenerateContext.cc:98- 建议:把「按 error_info / cancel / cleanup 判定终止原因 + GenerateDone 守卫」下沉为一个 protected 辅助函数(如
terminalErrorForCleanup()返回ErrorInfo),基类与PrefillGenerateContext::stopStream()共用;命中GenerateDone窗口时降为 DEBUG 且不上报 CANCELLED,仅对context_error/client_cancel保留 WARNING(G.6.1.22)。另注意该函数会在基类析构期执行,其中:93调用的虚函数isRequestCancelled()只会解析到基类实现,建议改为进入析构前由派生类写入的非虚成员快照。请在rtp_llm/cpp/model_rpc/test/LocalRpcServerTest.cc补一条断言:正常完成的请求在 context 析构后不产生 CANCELLED 终态、也不触发 WARNING 分支。
- 建议:把「按 error_info / cancel / cleanup 判定终止原因 + GenerateDone 守卫」下沉为一个 protected 辅助函数(如
- flush() 由 NOOP 改为无上界 Queue.join() 并被 close()/del 调用,进程退出可能永久挂死 @
rtp_llm/access_logger/async_log_handler.py:205- 建议:三点收敛:(1)
close()中先_stop_event.set()截断新日志、消除emit()与close()的 TOCTOU 窗口,再join(timeout=...)等待 worker,未按期退出时用_drain_queue()同步兜底,最后关文件;(2)flush()改为有 deadline 的等待(轮询unfinished_tasks,或用threading.Condition.wait_for(..., timeout=...)配合自维护的enqueued/written计数),超时记录告警后返回,任何情况下不在close()/__del__中调用无超时queue.join();(3)__del__只做 best-effort 的_stop_event.set(),不触发可能阻塞或新建线程的close()。同步更新模块 docstring(:2-27仍写 "avoiding main thread blocking")与类注释(:37仍写 "runs forever"),明确emit()非阻塞、flush()/close()会阻塞及其上限、close 后emit()静默丢弃。并补单测:emit 后 flush 返回时记录已落盘、close 幂等、worker 不可用时 flush 能在有限时间内返回、队列满时dropped递增且主流程不阻塞。
- 建议:三点收敛:(1)
- 22 个变更文件零测试,而三处现成 fixture 本可直接拦住本次 P0 与多处回归 @
rtp_llm/cpp/model_rpc/GenerateContext.h:103- 建议:按既有 fixture 补最小集合(均为纯 CPU 逻辑,无需 GPU):(1)
DecodeStatInfo/PrefillStatInfo阶段归因表驱动单测(含重试与提前 return);(2)deserializeErrorDetails覆盖合法码往返、空 details、解析失败、error_code=0、error_message为空回落status.error_message();(3) closure 侧覆盖 response 不设error_code、blocks_size()为 0/超预期、Run()与callback_抛异常,断言end()只执行一次;(4)makeTransferRequest在 buffer 缺失与local_partition_count == 0下返回nullptr且不崩溃;(5)notifyDone的超发与重复回调不变量;(6) Python 侧为load_xgrammar()(找到/未找到.so、重复调用幂等)与_build_backend()各补一条——上述 P0 只需一条_build_backend()用例即可拦住。
- 建议:按既有 fixture 补最小集合(均为纯 CPU 逻辑,无需 GPU):(1)
Non-blocking Suggestions
P2
- deserializeErrorDetails 缺少取值域与 NONE_ERROR 校验,可把失败降级为无错误并绕过坏连接淘汰 @
rtp_llm/cpp/model_rpc/RpcErrorCode.h:49- 建议:补两道校验:
details.error_code() == static_cast<int64_t>(ErrorCode::NONE_ERROR)时返回 false(details 只承载真实错误),使其与同文件既有契约一致——PrefillRpcServer.cc:426-427显式要求error_code() != NONE_ERROR、PrefillBatchRpcServer.cc:142-147要求== PRIORITY_PREEMPTED;并把裸static_cast<ErrorCode>换成集中式白名单校验(复用/扩展transRPCErrorCode的映射表,未知取值回落UNKNOWN_ERROR),避免非本仓 peer 的错误码原样穿透到 PythonExceptionType。随后用该函数统一替换PrefillRpcServer.cc:425-436、PrefillGenerateContext.cc:239、PrefillBatchRpcServer.cc:142的内联解析,避免同一 wire 契约出现第四份漂移实现(R.I.1)。若要彻底消除与google.rpc.Status的歧义,可在ErrorDetailsPB增加魔数/版本字段。
- 建议:补两道校验:
- closure end() 自毁与 Run() 新增 catch 重入构成 use-after-free 与 double free 窗口 @
rtp_llm/cpp/disaggregate/cache_store/TcpCacheStoreLoadServiceClosure.cpp:107- 建议:两处 closure 统一加幂等与不抛保证:给
end()标noexcept(让markEnd/日志异常直接终止而非退化为内存破坏),并增加bool ended_ = false;哨兵,end()入口if (ended_) return; ended_ = true;,Run()兜底 catch 仅在!ended_时调用;同时把collector_->markEnd(success)移入 try 内,并统一transfer_request_/request_block_buffer_的空指针假设(Run()的 catch 做了保护、end()内直接解引用)。请在已有的test/TcpCacheStoreLoadServiceClosureTest.cppfixture 上补两条 case:Run()体内抛异常、callback_抛异常,断言end()只执行一次且无重复释放。
- 建议:两处 closure 统一加幂等与不抛保证:给
- notifyDone 忽略 success 形参,完成不变量仅按数量校验,重复上报仍会误判成功并回 EC_SUCCESS @
rtp_llm/cpp/disaggregate/cache_store/CacheTransferServiceImplContext.cpp:76- 建议:把计数式判定改为按 key 去重:以
std::set<std::string>(或复用local_blocks_的 key 集合)记录待完成 block,回调时逐个erase,返回 0 即为「重复或未知 key」的不变量违反,集合为空才runSuccess()——这样同时覆盖超发与重复上报。同时让success参与判定(if (!success || error_code != None)走失败路径),或删除该未使用形参以消除误导(推荐用RTP_LLM_CHECK断言其与error_code一致);并给:34的runFailed()前补finished_ = true,使finished_成为唯一的完成态真值来源。
- 建议:把计数式判定改为按 key 去重:以
- !success 且 error_code 为 None 时 error_info_ 退化成 NONE_ERROR,失败被读成无错误 @
rtp_llm/cpp/disaggregate/cache_store/RemoteStoreTaskImpl.cpp:214- 建议:在赋值前补一道收敛:
if (internal_error_code == ErrorCode::NONE_ERROR) { internal_error_code = ErrorCode::UNKNOWN_ERROR; },把「失败必须带错误码」固定在唯一写入点,避免后续新增 callback 站点误传None时静默劣化。错误码透传本身是好改进,此处只需补齐兜底;这与上一条notifyDone的建议属同一类不变量加固,建议一并处理。
- 建议:在赋值前补一道收敛:
- 运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉 @
rtp_llm/config/grammar_tokenizer_info.py:154- 建议:复用仓内既有且已验证的范式:
find_spec已拿到binding_path绝对路径,直接ctypes.CDLL(str(binding_path), mode=ctypes.RTLD_GLOBAL)预加载后再import xgrammar(捕获OSError后回退裸 import),不改os.environ——预加载后同名依赖会命中已加载镜像,在当前进程即刻生效。建议把该逻辑抽成rtp_llm/utils下共用 helper 供 z3 与 xgrammar 复用,避免第三份实现(R.I.1)。若确实只为 spawn 子进程服务,请把函数名与 docstring 改为表达该语义、至少改为追加而非前置以降低遮蔽风险,并用contextlib在 import 前后恢复原值。无论哪种方案,都需要一个在 Bazel runfiles 布局下断言load_xgrammar()加载成功的测试,否则本 PR 的主目标无法验证。
- 建议:复用仓内既有且已验证的范式:
- GRPC_RET_IF_ERROR 不写 error_info,真实 INTERNAL 失败被归为 context_cleanup @
rtp_llm/cpp/model_rpc/GenerateContext.cc:87- 建议:把
GRPC_RET_IF_ERROR扩展为同时填充error_info(新增ErrorCode参数,或统一改走serializeErrorMsg+error_info双写),使新增的三态分类真正覆盖 decode 的主要失败路径;否则新增的source=日志会长期给出误导性归因,反而增加排障成本。同时把终态错误上报(:86-102的 if/else 块)移到meta->dequeue(...)之前,保持「先定终态、再落终态记录」的顺序,并在rtp_llm/cpp/model_rpc/test/RpcServerRuntimeMetaTest.cc补一条断言:GRPC_RET_IF_ERROR失败场景下finished_streams_携带非零 error_code。
- 建议:把
- stopStream 非虚被派生类隐藏,新增终止分类对 prefill 完全失效 @
rtp_llm/cpp/model_rpc/GenerateContext.cc:82- 建议:将
GenerateContext::stopStream()声明为virtual,让PrefillGenerateContext::stopStream()显式override,并把「按 error_info / cancel / cleanup 决定终止原因」下沉为一个 protected 辅助函数供两个实现复用,prefill 仅保留 scheduler-owned 等待、超时强制取消与markRequestEnd()的差异部分(与前述 P1 的修复建议合并实施)。同时在GenerateContext.cc:82处注明「本函数可能在基类析构中执行,虚函数只会解析到基类实现」,避免后续有人在其中新增依赖虚派发的逻辑。
- 建议:将
- 把二进制 error_details 拼进日志与对外错误消息 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.cc:682- 建议:改用本 PR 新增的
deserializeErrorDetails(status, &remote_info)解析后输出remote_code=%s remote_message=%s,与:683-692分支保持一致(R.I.1);解析失败时只输出grpc_details_size=%zu,或做 hex/base64 编码后再拼接,且只写日志、不放进对外返回的 message。
- 建议:改用本 PR 新增的
- loadCacheAsyncForTp 提前 return 未 Shutdown completion queue,in-flight 调用引用即将销毁的栈对象 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.cc:629- 建议:把清理动作收进 RAII guard 或
absl::Cleanup风格的作用域退出钩子:先对每个client_context->TryCancel(),再Shutdown()全部 completion queue 并 drain 干净残余事件,最后才允许all_context析构;或把all_context改为shared_ptr并让 drain 逻辑持有一份引用。建议本 PR 顺手修掉而非留待后续。
- 建议:把清理动作收进 RAII guard 或
- onflight/loading_cache gauge 采样点由请求到达时改为请求结束时,指标语义变更未说明 @
rtp_llm/cpp/model_rpc/DecodeGenerateContext.cc:74- 建议:如需保留完成时采样,请在 PR description 中明确指标语义变更并同步核对相关看板阈值;若目的只是修复
LocalRpcServer恒为 0 的问题,更小的改法是保留入口处快照赋值、仅补上LocalRpcServer.cc缺失的那一行,避免同时改动 decode/prefill 两条既有链路的采样语义。两个 gauge 的取值口径(是否含自身请求、采样时刻)建议在字段处补一行注释固定下来,并与REPORT_GAUGE的零值决策一起评估——两处叠加会使该 gauge 与历史数据不可比。
- 建议:如需保留完成时采样,请在 PR description 中明确指标语义变更并同步核对相关看板阈值;若目的只是修复
- close() 中 flush() 抛异常会导致 stop_event 未置位与 fd 泄漏,但 handler 已被摘除 @
rtp_llm/access_logger/async_log_handler.py:212- 建议:调整为先置位
_stop_event再尽力 flush,并把_file_handler.close()放入finally保证释放:self._stop_event.set()→try: self.flush() except Exception: logging.error(...)→finally: self._file_handler.close(); super().close()。这样即使 flush 失败也能保证停止信号已发出、worker 可正常退出、fd 被释放,错误语义为「尽力落盘 + 必定释放资源」而非「落盘失败即泄漏」。join 超时后请用_drain_queue()兜底并对实际丢弃条数计数告警,避免静默丢日志。该调整与前述 P1 的close()顺序修复可一并实施。
- 建议:调整为先置位
- None 记录被消费但未 task_done,可永久破坏新的阻塞式 flush() @
rtp_llm/access_logger/async_log_handler.py:127- 建议:让每次成功的
get()/get_nowait()都在 try/finally 中配对一次task_done()(包含被丢弃的None),把task_done()移出if record is not None分支;或直接删除这两处不可达的None防御分支,使 put 与 task_done 的计数严格守恒,避免后续有人真的往队列里塞哨兵值时触发死等。
- 建议:让每次成功的
- emit 热路径为每条日志新增 qsize 与两次加锁,但唯一出口 get_stats() 无消费方 @
rtp_llm/access_logger/async_log_handler.py:193- 建议:二选一:(1) 在本次改动内补齐消费侧,把
get_stats()接入现有 kmonitor 上报或 worker status/debug 接口,让这些计数真正可观测、可告警;(2) 若消费侧尚未规划,先收敛为最小集合(仅保留低频事件计数dropped、write_errors、worker_restarts),去掉enqueued/written的逐条精确计数,并把max_queue_depth/qsize()这类采集移到后台 worker 侧周期性或采样更新,避免在每请求路径上为无人读取的计数付出锁开销。
- 建议:二选一:(1) 在本次改动内补齐消费侧,把
- 单个 PR 混合五件互不相关的变更,难以定位与回滚 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.h:70- 建议:建议至少拆成三个 PR:(1) cache_store + model_rpc 的错误传递与崩溃修复(同一故障域);(2)
REPORT_GAUGE上报语义变更(需独立的受影响指标清单与回滚开关);(3) access log 生命周期 + xgrammar 加载(后者可独立先合以尽快修掉 P0)。同时在 PR description 中补充各项的动机与设计:特别是 xgrammar 导入失败的具体触发环境(哪个 Bazel target、哪条报错栈)、REPORT_GAUGE去守卫的目标与受影响指标清单、以及 decode 对外错误码口径变更的说明,并保证 commit 粒度原子、message 与行为一致。
- 建议:建议至少拆成三个 PR:(1) cache_store + model_rpc 的错误传递与崩溃修复(同一故障域);(2)
P3
- allocateResource 失败路径重复打印同一错误并丢失 request_info @
rtp_llm/cpp/model_rpc/DecodeRpcServer.cc:207- 建议:去掉
:207的显式 ERROR 日志(错误码与消息已由serializeErrorMsg统一打印),或反之让serializeErrorMsg不再打日志、由调用方决定;同时把:211改为三参重载serializeErrorMsg(decode_context.request_key, decode_context.request_info, decode_context.error_info),与同文件其它错误路径保持一致的日志标签。
- 建议:去掉
- 删除 loadCacheSyncForTp 后 thread_pool_ 与 initThreadPool 成为完全不可达的死代码 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.h:103- 建议:在同一 PR 内一并删除
DecodeRpcServer::thread_pool_成员、initThreadPool()声明与定义、以及析构中的 stop 分支,使死代码清理完整(已确认零消费者,无需替代实现或迁移说明);并同步清理 BUILD 中不再需要的autil线程池依赖(若无其他使用者)。若后续确有异步 load cache 计划,请以注释或 TODO 说明保留原因,避免读者误以为该线程池仍在服务某条路径。
- 建议:在同一 PR 内一并删除
- TP load cache 报错日志用 rank 直接索引 peer_addrs,非对称 TP 下指向错误对端 @
rtp_llm/cpp/model_rpc/DecodeRpcServer.cc:676- 建议:抽一个
peerAddrForRank(rank)辅助函数复用constructRemoteLoadRequest/ForMla的映射规则(含 1:1、workers 多、peers 多、prefill_cp_size > 1四种情形),在:590、:676、:687三处统一调用;无法确定映射时输出peer=<unmapped>而非可能错误的地址,避免新增的诊断字段反而误导定位。
- 建议:抽一个
- 同一文件内"是否上报"存在三套并存且不一致的约定 @
rtp_llm/cpp/metrics/RtpLLMMetrics.cc:38- 建议:统一为单一约定:保留一个带显式 has/optional 判断的上报入口(如
REPORT_GAUGE_IF_SET),把无条件上报单独命名(如REPORT_GAUGE_ALWAYS),使调用点自解释,并在宏定义处补一行注释说明两者取舍与适用场景。若短期不做统一,至少将REPORT_GAUGE重命名以反映「无条件上报」语义,避免与仍带真值判断的REPORT_QPS形成误导性对称。
- 建议:统一为单一约定:保留一个带显式 has/optional 判断的上报入口(如
- 新增与修改的行未通过 black / clang-format @
rtp_llm/config/grammar_tokenizer_info.py:164- 建议:提交前对改动文件跑一次仓库 pre-commit(
black、isort --profile=black、clang-format),使逻辑改动与格式化彻底分离;否则这些行会在后续任意提交中被自动重排,把无关格式化 churn 混入其他变更,干扰 blame 与 CI 二分定位。
- 建议:提交前对改动文件跑一次仓库 pre-commit(
Checklist Findings (20 fail / 54 total)
General Principles Checklist
- [6.1] Architecture — 兼容性:外部 HTTP/RPC API、持久数据、配置、环境迁移安全 → issue
onflight/loading_cache gauge 采样点由请求到达时改为请求结束时,指标语义变更未说明
改前decode_context.onflight_requests = onflight_requests_;与loading_cache_requests = loading_cache_requests_;是把std::atomic<size_t>隐式转成int64_t,即在RemoteGenerate入口取快照;改后字段变为const std::atomic<size_t>*(GenerateContext.h:49、DecodeGenerateContext.h:72),在reportTime()(析构期)才load()(DecodeGenerateContext.cc:74-75、GenerateContext.cc:63)。rtp_llm_rpc_onflight_request与rtp_llm_rpc_loading_cache_request两个既有 gauge 的采样点因此从「到达时并发度」变为「完成时并发度」,同一负载下分布明显不同,依赖历史基线的看板与阈值告警会漂移。叠加同 P - [6.1] Architecture — 分层边界:新概念在正确层级,不泄漏内部 → issue
运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉
load_xgrammar()在:154-159把包目录前置进os.environ["LD_LIBRARY_PATH"],随后:161在同一进程内import xgrammar as xgr。glibc 的ld.so在进程启动时一次性解析并缓存该变量,进程内后续修改不影响本进程dlopen;只有 xgrammar/tvm-ffi 自己在 Python 层读该变量时才生效,而 docstring 声称的 "including from Bazel rules_python runfiles" 无任何测试覆盖。仓内models_py/modules/dsv4/tilelang_kernels.py:54-71处理同一问题时明确注释 bare-name dlopen "only works if the lib is on LD_LIBRARY_PATH or loaded eagerly",并改用ctypes.CDLL(..., RTLD_GLOBAL)。该写入还是进程级持久副作用,会被 spawn 出的 bac - [6.1] Architecture — 可观测性:日志/指标/超时可操作、非噪声 → issue
TP load cache 报错日志用 rank 直接索引 peer_addrs,非对称 TP 下指向错误对端
:676-678与:687-689用rank < decode_context.peer_addrs.size() ? decode_context.peer_addrs[rank] : "<missing>"取对端地址。越界已有保护(不会 OOB),但索引语义不对:同文件constructRemoteLoadRequestForMla(:403-417)表明 worker→peer 的正确映射为peer_addrs[index / part_cnt](workers 多于 peers)或peer_addrs[index * group_num](peers 多于 workers),只有 1:1 时才等于peer_addrs[rank]。因此在非对称 TP 部署下这条新增的peer=字段会稳定指向错误对端,且因rank < size()成立而不会退化为<missing>,可能把排障引向无关机器。影响仅限日志文本,不改变控制流。 - [6.1] Architecture — 回滚路径:风险行为存在运维回滚手段 → issue
单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个文件覆盖五个彼此无依赖的主题:cache_store 传输错误码与崩溃修复(6 文件)、model_rpc 阶段计时重构与结构化错误传递(12 文件)、REPORT_GAUGE全局上报语义变更(1 文件,影响 68 处调用点)、access log handler 生命周期补全(1 文件,反转 never-block 契约)、xgrammar 动态库加载(2 文件)。其中后两项与 PR 分支主题(cache-store / RPC / xgrammar)无关,却都改变线上可观测性与进程退出行为;PR description 未说明其动机。任一项出问题只能整体回滚,CI 二分定位也失去粒度;本次 7 个阻塞项分散在四个不同主题中,正说明混合提交放大了评审与回滚成本。 - [6.1] Architecture — 状态不变量:创建/更新/失败/重试/回滚路径有效 → issue
None 记录被消费但未 task_done,可永久破坏新的阻塞式 flush()
_process_batch在:126/:135取出的record为None时既不放入records_batch也不调用task_done();task_done()只在:146的 finally 与_drain_queue的:174中、且均在if record is not None之后执行。一旦队列出现None,unfinished_tasks永久 +1,本次新增的flush()/close()中的queue.join()即永久阻塞。当前emit()只入LogRecord,该分支实际不可达,属潜在陷阱:防御式None判断与新增的join()语义互相矛盾。 - [6.1] Architecture — 错误语义:fail-fast/retry/fallback/silent 行为显式 → issue
None 记录被消费但未 task_done,可永久破坏新的阻塞式 flush()
_process_batch在:126/:135取出的record为None时既不放入records_batch也不调用task_done();task_done()只在:146的 finally 与_drain_queue的:174中、且均在if record is not None之后执行。一旦队列出现None,unfinished_tasks永久 +1,本次新增的flush()/close()中的queue.join()即永久阻塞。当前emit()只入LogRecord,该分支实际不可达,属潜在陷阱:防御式None判断与新增的join()语义互相矛盾。 - [6.1] Quality — Commit 原子、message 与行为匹配 → issue
单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个文件覆盖五个彼此无依赖的主题:cache_store 传输错误码与崩溃修复(6 文件)、model_rpc 阶段计时重构与结构化错误传递(12 文件)、REPORT_GAUGE全局上报语义变更(1 文件,影响 68 处调用点)、access log handler 生命周期补全(1 文件,反转 never-block 契约)、xgrammar 动态库加载(2 文件)。其中后两项与 PR 分支主题(cache-store / RPC / xgrammar)无关,却都改变线上可观测性与进程退出行为;PR description 未说明其动机。任一项出问题只能整体回滚,CI 二分定位也失去粒度;本次 7 个阻塞项分散在四个不同主题中,正说明混合提交放大了评审与回滚成本。 - [6.1] Quality — Mega-PR 已拆分为独立变更 → issue
单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个文件覆盖五个彼此无依赖的主题:cache_store 传输错误码与崩溃修复(6 文件)、model_rpc 阶段计时重构与结构化错误传递(12 文件)、REPORT_GAUGE全局上报语义变更(1 文件,影响 68 处调用点)、access log handler 生命周期补全(1 文件,反转 never-block 契约)、xgrammar 动态库加载(2 文件)。其中后两项与 PR 分支主题(cache-store / RPC / xgrammar)无关,却都改变线上可观测性与进程退出行为;PR description 未说明其动机。任一项出问题只能整体回滚,CI 二分定位也失去粒度;本次 7 个阻塞项分散在四个不同主题中,正说明混合提交放大了评审与回滚成本。 - [6.1] Quality — PR description 说明动机与设计 → issue
单个 PR 混合五件互不相关的变更,难以定位与回滚
22 个文件覆盖五个彼此无依赖的主题:cache_store 传输错误码与崩溃修复(6 文件)、model_rpc 阶段计时重构与结构化错误传递(12 文件)、REPORT_GAUGE全局上报语义变更(1 文件,影响 68 处调用点)、access log handler 生命周期补全(1 文件,反转 never-block 契约)、xgrammar 动态库加载(2 文件)。其中后两项与 PR 分支主题(cache-store / RPC / xgrammar)无关,却都改变线上可观测性与进程退出行为;PR description 未说明其动机。任一项出问题只能整体回滚,CI 二分定位也失去粒度;本次 7 个阻塞项分散在四个不同主题中,正说明混合提交放大了评审与回滚成本。 - [6.1] Quality — 无 per-forward 调试日志 / 噪声热路径输出 → issue
基类 stopStream() 缺少 prefill 已有的 GenerateDone 守卫,正常成功路径每请求输出一条 WARNING
新 else 分支条件为getStatus() != FINISHED && !hasError() && !cancelled() && !isRequestCancelled(),命中即打source=context_cleanupWARNING 并reportError(CANCELLED, ...)(:98-100)。该窗口有既有测试直接证明:engine_base/stream/test/GenerateStreamTest.cc:264-273在reportEvent(GenerateDone)后断言消费者拿到ErrorCode::FINISHED而stream->getStatus() == StreamState::RUNNING(GenerateStateMachine.h:63的checkFinished()即如此设计)。本 PR 在PrefillGenerateContext.cc:89专门为此加了!(hasEvent(GenerateDone) && !hasError())守卫并 - [6.1] Quality — 逻辑变更未混入无关格式化 → issue
新增与修改的行未通过 black / clang-format
grammar_tokenizer_info.py::163为load_xgrammar()的return xgr,:164仅一个空行,:165即顶层def build_grammar_tokenizer_info_json(;Black/PEP 8 E302 与本文件其余顶层定义(:134-135、:145-147)均为两个空行,black --check会报改动。Messager.cpp:80:同一个RTP_LLM_LOG_WARNING的实参列表中:77-79与:81为 16 空格缩进,而:80的local_key.c_str(),为 32 空格,仓库根.clang-format(4 空格、120 列、实参对齐)不会产生这种同列表内不一致的对齐。仓库 pre-commit 已启用 black + isort + clang-format。 - [6.1] Software Engineering — DRY:重复非平凡逻辑被抽取或显式复用 → issue
同一文件内"是否上报"存在三套并存且不一致的约定
改动后同一文件并存三套约定:REPORT_QPS(:38-41)保留if (collector->name)真值判断;PrefillRecentCacheKeyMetrics::report(:135)用has_value、RtpLLMExecutorMetrics::report(:364)用context_batch_size != 0做 ad-hoc 显式判断;REPORT_GAUGE(:43)改为完全无条件上报。宏名仍叫REPORT_GAUGE,无法体现「无条件」这一新语义,维护者容易按旧直觉误以为仍有值判断。另min_response_done_time_us默认值为1lu << 60(.h:59),在旧宏下即为真值恒被上报,也印证纯真值判断本身不可靠。 - [6.1] Software Engineering — KISS/YAGNI:无投机性抽象 → issue
删除 loadCacheSyncForTp 后 thread_pool_ 与 initThreadPool 成为完全不可达的死代码
loadCacheSyncForTp是thread_pool_的唯一使用者。全仓检索rtp_llm/cpp/model_rpc确认,删除该函数后DecodeRpcServer::thread_pool_只剩initThreadPool()内的创建(DecodeRpcServer.cc:131-133)与析构里的stop()(:138-141),再无任何pushTask/async;而initThreadPool()本身也已零调用点(仅.h:70声明 +.cc:127定义)。即便被调用其逻辑也已失效:if (resource_.workers.size() > 0) { return; }意味着只有 workers 为空时才建池,而池大小恰为resource_.workers.size() * 8即 0。 - [6.1] Software Engineering — LSP:子类/重写保持基类契约 → issue
stopStream 非虚被派生类隐藏,新增终止分类对 prefill 完全失效
GenerateContext::stopStream()是 protected 非虚函数(GenerateContext.h:65),而PrefillGenerateContext在 private 段声明了同名非虚void stopStream();(PrefillGenerateContext.h:129),属名字隐藏而非重写。~PrefillGenerateContext()(PrefillGenerateContext.cc:65-69)静态绑定到自己那份实现(:78-112),仍只上报CANCELLED, "cancel stream";随后基类析构时stream_已在:110reset,基类版本成为 no-op。因此本次三态细化只对 Local 与 Decode 路径生效,两侧语义已分叉。 - [6.1] Tests — 分布式/跨平台变更有对应覆盖 → issue
22 个变更文件零测试,而三处现成 fixture 本可直接拦住本次 P0 与多处回归
22 个改动文件中无任何测试文件,而补测基础设施完备:rtp_llm/cpp/model_rpc/test/有 14 个文件(含DecodeRpcServerTest.cc、PrefillRpcServerTest.cc、RpcServerRuntimeMetaTest.cc),cache_store/test/TcpCacheStoreLoadServiceClosureTest.cpp有 7 条 case 与可复用 fixture。但全仓检索DecodeStatInfo/allocate_resource_rt_us/deserializeErrorDetails在测试目录零命中。改变行为的语义全部无覆盖:EXECUTE_STAGE_FUNC新增的finishStage()与 decode 阶段归因、deserializeErrorDetails取值边界、stopStream()三态分类、!has_error_code() → LoadErrorUnknown(既有 case 因:42始终显式 - [6.1] Tests — 新逻辑有聚焦单测 + 相关集成/smoke 测试 → issue
运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉
load_xgrammar()在:154-159把包目录前置进os.environ["LD_LIBRARY_PATH"],随后:161在同一进程内import xgrammar as xgr。glibc 的ld.so在进程启动时一次性解析并缓存该变量,进程内后续修改不影响本进程dlopen;只有 xgrammar/tvm-ffi 自己在 Python 层读该变量时才生效,而 docstring 声称的 "including from Bazel rules_python runfiles" 无任何测试覆盖。仓内models_py/modules/dsv4/tilelang_kernels.py:54-71处理同一问题时明确注释 bare-name dlopen "only works if the lib is on LD_LIBRARY_PATH or loaded eagerly",并改用ctypes.CDLL(..., RTLD_GLOBAL)。该写入还是进程级持久副作用,会被 spawn 出的 bac - [6.1] Tests — 边界 case 覆盖(空、单元素、最大值) → issue
notifyDone 忽略 success 形参,完成不变量仅按数量校验,重复上报仍会误判成功并回 EC_SUCCESS
notifyDone(bool success, ...)(:47-91)全函数从不读取success,只按error_code != None分支(:62):调用方传(false, None)会走成功路径并可能runSuccess()回EC_SUCCESS。新增校验blocks.size() > static_cast<size_t>(unfinished_count_)(:76)只比数量、不校验 block key 身份:以总量 4 为例,首次上报 A/B(4→2)后若同一批被重复回调,2 > 2不成立,unfinished_count_归零并runSuccess(),而 C/D 从未传输,decode 侧会读到未初始化 KV,比计数变负更难发现。另run()的:30-36在lock()失败走runFailed()时未置finished_ = true。TransferConnection::read的分批回调实现不在本仓视图内,重复回调是否可达无法确认,故按残留
RTP-LLM Checklist
- [I] 代码质量 — 删除或重命名内部 file、registry entry、model name、metric enum、op binding、plugin symbol 时,必须全仓搜索消费者,并提供替代实现、迁移说明或 smoke 覆盖;只有暴露到 HTTP/RPC/config/persisted format 时才按外部兼容性处理 → issue
删除 loadCacheSyncForTp 后 thread_pool_ 与 initThreadPool 成为完全不可达的死代码
loadCacheSyncForTp是thread_pool_的唯一使用者。全仓检索rtp_llm/cpp/model_rpc确认,删除该函数后DecodeRpcServer::thread_pool_只剩initThreadPool()内的创建(DecodeRpcServer.cc:131-133)与析构里的stop()(:138-141),再无任何pushTask/async;而initThreadPool()本身也已零调用点(仅.h:70声明 +.cc:127定义)。即便被调用其逻辑也已失效:if (resource_.workers.size() > 0) { return; }意味着只有 workers 为空时才建池,而池大小恰为resource_.workers.size() * 8即 0。 - [I] 代码质量 — 同一功能用统一工具函数 → issue
同一文件内"是否上报"存在三套并存且不一致的约定
改动后同一文件并存三套约定:REPORT_QPS(:38-41)保留if (collector->name)真值判断;PrefillRecentCacheKeyMetrics::report(:135)用has_value、RtpLLMExecutorMetrics::report(:364)用context_batch_size != 0做 ad-hoc 显式判断;REPORT_GAUGE(:43)改为完全无条件上报。宏名仍叫REPORT_GAUGE,无法体现「无条件」这一新语义,维护者容易按旧直觉误以为仍有值判断。另min_response_done_time_us默认值为1lu << 60(.h:59),在旧宏下即为真值恒被上报,也印证纯真值判断本身不可靠。
Python Static-First Checklist
- [P.G] 测试规范 — mock/fake/stub 不得替代本次声称覆盖的生产边界 → issue
load_xgrammar 未绑定到模块全局,GrammarValidator 必然构建失败并静默关闭语法准入校验
:454的return load_xgrammar().GrammarCompiler(...)使用裸名,但全仓检索该名字只有两处绑定:grammar_tokenizer_info.py:147的定义与grammar_validator.py:786位于_xgrammar()函数体内的 import(仅函数局部)。已核对模块级 import 段:17-36无此名、无 star import,故:454必抛NameError,被:460的except Exception包成误导性RuntimeError。而:445的xgr = self._xgrammar()已拿到可用模块并在:446校验非 None。调用链__init__:122 → _initialize_compiler:405 → _build_backend无兜底,dash_sc/app.py:629只 warning 后置grammar_validator = None。
Strengths
has_error_code()守卫是真实的静默正确性修复:cache_store_service.proto:1为proto2、:7定义EC_SUCCESS = 0、:51/:77/:87三处error_code均为optional,字段缺失时读到的默认值恰是「成功」,等于把未写入的 KV buffer 当成功交付上层。Messager::makeTransferRequest拆成两段独立校验(Messager.cpp:73、:85),修掉了原代码在block_buffer == nullptr分支仍打印block_buffer->len的空指针解引用,并新增local_partition_count == 0判断消除len % 0整数除零 UB;两条错误分支都正确delete transfer_request,未引入泄漏。finishStage()以begin_time == 0作幂等哨兵(DecodeGenerateContext.cc:19-24),语义比原先「进入新阶段时隐式归算上一阶段」清晰得多,并顺带修掉EXECUTE_WITH_RETRY重试时把耗时错记进前一阶段的老问题;prefill 与 batch prefill 两条链路在新语义下逐段推演归因全部正确。- decode
allocateResource(DecodeRpcServer.cc:194-212)不再把所有分配失败硬编码为RESOURCE_EXHAUSTED+"malloc kv cache block failed",改为透传 stream 真实ErrorCode并显式兜底NONE_ERROR → UNKNOWN_ERROR、空 message →ErrorCodeToString;同时把RTP_LLM_LOG_ERROR(error_msg)(把std::string当格式串,含%时为格式串漏洞)改为带"%s"的安全形式。 CacheTransferServiceImplContext::notifyDone新增完成不变量校验(:76),避免unfinished_count_被减成负数后永远等不到== 0(旧行为一路挂到超时且失败原因无从定位)。fromKvCacheStoreErrorCode/fromArpcErrorCode/toKvCacheStoreErrorCode均有default兜底,错误码透传没有引入未受控的值域穿透。- TP load cache 失败日志补齐
rank / worker / peer / cq / grpc_code / remote_code / finished=x/y / cost_ms,peer_addrs索引做了rank < size()越界保护,resource_.grpc_workers.at(rank)用.at()而非[]。 onflight_requests改为指向 server 端std::atomic<size_t>的常量指针,LocalRpcServer.cc:199首次补上该路径的赋值,修复rtp_llm_rpc_onflight_request恒为 0;指针生命周期经核对安全(context 为 handler 栈对象,先于 server 成员析构)。stopStream()新增!stream_->hasError()前置守卫,不再用CANCELLED覆盖 stream 上已有的真实错误码,并删除了~GenerateContext()中与之语义重复的一次reportError。- 删除
loadCacheSyncForTp前经全仓检索确认零残留引用,属真死代码清理。 - access log 侧修复了
flush/close/__del__三者此前均为空实现的问题,并把统计计数统一收敛到_stats_lock之下,修正了原先self._stats['dropped'] += 1的非原子读改写。
| self._tokenizer_info_json | ||
| ) | ||
| return xgr.GrammarCompiler( | ||
| return load_xgrammar().GrammarCompiler( |
There was a problem hiding this comment.
[P0] load_xgrammar 未绑定到模块全局,GrammarValidator 必然构建失败并静默关闭语法准入校验
:454 的 return load_xgrammar().GrammarCompiler(...) 使用裸名,但全仓检索该名字只有两处绑定:grammar_tokenizer_info.py:147 的定义与 grammar_validator.py:786 位于 _xgrammar() 函数体内的 import(仅函数局部)。已核对模块级 import 段 :17-36 无此名、无 star import,故 :454 必抛 NameError,被 :460 的 except Exception 包成误导性 RuntimeError。而 :445 的 xgr = self._xgrammar() 已拿到可用模块并在 :446 校验非 None。调用链 __init__:122 → _initialize_compiler:405 → _build_backend 无兜底,dash_sc/app.py:629 只 warning 后置 grammar_validator = None。
建议: :454 改为 return xgr.GrammarCompiler(...),复用 :445 已加载并经 lru_cache 记忆的模块对象——既修掉 NameError,也避免同一函数内并存两种取模块方式、绕过 _xgrammar() 的 None 语义契约;若确需显式调用则必须在模块顶部补 from rtp_llm.config.grammar_tokenizer_info import load_xgrammar。同时补一条覆盖 _build_backend() 的聚焦单测:注入 fake xgrammar 模块后走真实 _initialize_compiler 路径,断言可用时返回非 None compiler、不可用时抛 RuntimeError。现有 grammar_validator_test.py:34 用 GrammarValidator.__new__ 绕过 __init__,对该生产边界零覆盖,正是本缺陷逃过 CI 的直接原因(P.P.G.3)。另建议评估「配置要求开启但初始化失败」是否应改为启动期 fail-fast(:403-404 注释本就如此声明),而非降级为 warning 后静默置空。
Checklist: [P.G] mock/fake/stub 不得替代本次声称覆盖的生产边界
|
|
||
| try { | ||
| EXECUTE_STAGE_FUNC(prepareGenerateContext, decode_context); | ||
| decode_context.stat_info.nextStage(); |
There was a problem hiding this comment.
[P1] decode 阶段计时被双重推进,四个耗时指标全部错位且 stage 越出枚举上界
本 PR 为 EXECUTE_STAGE_FUNC 新增了 finishStage()(GenerateContext.h:103),同时新增 :1307/:1310 的手工 nextStage()/finishStage(),而 allocateResourceFunc(:1263-1266)本身就是 EXECUTE_STAGE_FUNC(allocateResource)。逐段推演:allocate_resource_rt_us 只累加 :1307 到宏之间的几微秒;allocateResource 真实耗时(含每次重试)落入 load_cache_from_prefill_rt_us;loadCacheFromPrefill 耗时落入 local_generate_rt_us;localGenerate 耗时命中 case finish: break 被丢弃;:1331 后 stage=6,越出 ExecuteStage(finish=5,DecodeGenerateContext.h:13-20)。`...
建议: 删除 :1307 与 :1310 两行,让 EXECUTE_STAGE_FUNC 独占阶段推进(重试累加由 restoreStage + begin_time 哨兵天然处理),与 PrefillRpcServer::syncPrefix(:627 的 EXECUTE_WITH_RETRY 前无任何手工 bracket)保持同一范式;删除后末尾 nextStage() 正好停在 finish 不越界。若确实希望单独统计「含重试的整体 allocate 耗时」,请新增独立字段而不要复用 stage 状态机。请在已存在的 rtp_llm/cpp/model_rpc/test/DecodeRpcServerTest.cc 补一个纯 DecodeStatInfo 单测(无需 GPU):按 prepare → allocate(含 N 次重试) → loadCache → localGenerate → finish 驱动,断言四个 *_rt_us 各自落桶正确、终态 stage 恰为 finish,并覆盖中途提前 return 的路径(stage 越界后任何 finishStage() 都会命中 default 的 RTP_LLM_CHECK_WITH_INFO(false))。
| new_error_code = ErrorCode::DECODE_MALLOC_FAILED; \ | ||
| if (!remote_error_info.hasError()) { \ | ||
| new_error_msg += error_msg; \ | ||
| if (status.error_code() == grpc::StatusCode::RESOURCE_EXHAUSTED) { \ |
There was a problem hiding this comment.
[P1] decode KV cache 分配失败的对外错误码从 DECODE_MALLOC_FAILED(8211) 退化为 MALLOC_FAILED(602)
两处改动叠加产生的跨节点契约错配。改前 decode allocateResource 返回 grpc::Status(RESOURCE_EXHAUSTED, error_msg),不带 details,prefill 走 if (!remote_error_info.hasError()) 分支由 :151-152 映射为 DECODE_MALLOC_FAILED(8211)。改后 DecodeRpcServer.cc:211 改走 serializeErrorMsg,其内部 LocalRpcServer.cc:96-101 必定写入 ErrorDetailsPB.error_code;decode kv block 分配失败由 GenerateStateMachine.cc:73/116/183 报 MALLOC_FAILED(602)。于是 deserializeErrorDetails 命中(位于 :127 else-if 链首位),8211 兜底整块被跳过,对外码变为 602——与 prefill 本地 OOM...
建议: 保留「远端码优先」的大方向,但对 decode 资源不足显式归一:解析成功后判断 remote_error_info.code() == ErrorCode::MALLOC_FAILED 时映射为 DECODE_MALLOC_FAILED;或更彻底地由 decode 在 allocateResource 内就以 DECODE_MALLOC_FAILED 构造 ErrorInfo 再 serializeErrorMsg,使跨节点语义稳定为 8211。若这是有意的口径变更,请在 PR description 中明确说明,并同步核对 rtp_llm/config/exceptions.py 及依赖 8211 的监控看板、告警阈值与 FlexLB 调度归因,给出迁移说明。另请一并说明混部灰度期行为:新 decode + 旧 prefill 时旧端无 deserializeErrorDetails,非 malloc 类错误的 gRPC status code 也已从固定 RESOURCE_EXHAUSTED 变为按 transErrorCodeToGrpc 映射,分类结果会随之变化。建议补一条断言「decode 报 602 时 prefill 对外仍为 8211」的往返单测。
| if (collector->name) { \ | ||
| REPORT_MUTABLE_METRIC(name##_metric, collector->name); \ | ||
| } | ||
| #define REPORT_GAUGE(name) REPORT_MUTABLE_METRIC(name##_metric, collector->name) |
There was a problem hiding this comment.
[P1] REPORT_GAUGE 移除零值守卫,68 处 gauge 语义静默失真且无回滚开关
宏由 if (collector->name) { REPORT_MUTABLE_METRIC(...) } 改为无条件上报,影响本文件 68 处调用点(REPORT_GAUGE( 共 69 处匹配,含 #define 自身)。RpcMetrics::report(:152-193)单次 RPC 无条件上报 28 个 gauge:非 PD 部署下 remote_*、multimodal_process_rt_us 恒为默认 0(.h:38);PD 分离下 prefill 为 decode 独有的 prepare_generate_context_rt_us/allocate_resource_rt_us/local_generate_rt_us 补 0,decode 为 prefill 独有的 get_rpc_connection_rt_us/remote_* 补 0。RtpLLMStreamMetrics::report(:226-253)同理会上报 first_token_latency_us=0、`timeou...
建议: 保留「值是否有效」的判定,而不是整体删除。本文件已有更合适的既有范式可复用:PrefillRecentCacheKeyMetrics::report(:135)的 has_value 与 RtpLLMExecutorMetrics::report(:364)的 context_batch_size != 0。推荐为 collector 字段增加显式 set/has 标记(或用 std::optional),既能上报真实测得的 0,又不会把「本部署形态/本阶段不适用」当成 0。若确实希望恒定上报,请拆为独立提交、提供环境变量开关与回滚说明,并在 PR description 中列出受影响指标清单及看板/告警迁移方案;同时说明 RtpLLMExecutorMetrics::report(:361-379)位于 per-step 热路径、无条件上报 14 个 gauge 带来的上报量增长是否可接受。
| request_key.c_str()); | ||
| stream_->reportError(ErrorCode::CANCELLED, "request cancelled by client"); | ||
| } else { | ||
| RTP_LLM_LOG_WARNING("request [%s] stopping unfinished stream with terminal source=context_cleanup", |
There was a problem hiding this comment.
[P1] 基类 stopStream() 缺少 prefill 已有的 GenerateDone 守卫,正常成功路径每请求输出一条 WARNING
新 else 分支条件为 getStatus() != FINISHED && !hasError() && !cancelled() && !isRequestCancelled(),命中即打 source=context_cleanup WARNING 并 reportError(CANCELLED, ...)(:98-100)。该窗口有既有测试直接证明:engine_base/stream/test/GenerateStreamTest.cc:264-273 在 reportEvent(GenerateDone) 后断言消费者拿到 ErrorCode::FINISHED 而 stream->getStatus() == StreamState::RUNNING(GenerateStateMachine.h:63 的 checkFinished() 即如此设计)。本 PR 在 PrefillGenerateContext.cc:89 专门为此加了 !(hasEvent(GenerateDone) && !hasError()) ...
建议: 把「按 error_info / cancel / cleanup 判定终止原因 + GenerateDone 守卫」下沉为一个 protected 辅助函数(如 terminalErrorForCleanup() 返回 ErrorInfo),基类与 PrefillGenerateContext::stopStream() 共用;命中 GenerateDone 窗口时降为 DEBUG 且不上报 CANCELLED,仅对 context_error / client_cancel 保留 WARNING(G.6.1.22)。另注意该函数会在基类析构期执行,其中 :93 调用的虚函数 isRequestCancelled() 只会解析到基类实现,建议改为进入析构前由派生类写入的非虚成员快照。请在 rtp_llm/cpp/model_rpc/test/LocalRpcServerTest.cc 补一条断言:正常完成的请求在 context 析构后不产生 CANCELLED 终态、也不触发 WARNING 分支。
Checklist: [6.1] 无 per-forward 调试日志 / 噪声热路径输出
| if dropped % 10 == 1: # Reduce logging frequency | ||
| logging.warning(f"AsyncLogHandler: dropped {dropped} log records (queue full)") | ||
|
|
||
| def flush(self) -> None: |
There was a problem hiding this comment.
[P1] flush() 由 NOOP 改为无上界 Queue.join() 并被 close()/del 调用,进程退出可能永久挂死
三个方法改前均为 pass。现 flush() 用无超时 self._queue.join()(:209)等待 unfinished_tasks 归零,且被 close():216 与 __del__():247 调用。确定的挂死路径:线程 A 通过 emit() 的 :182 stop 检查后被抢占;close() 完成 drain、_stop_event.set()(:217)、worker join 退出;A 恢复后 put_nowait(:192)使 unfinished_tasks 变 1 且再无消费者。此后 logging.shutdown() 依次调 flush()/close():flush() 中 _start_worker() 因 :97 的 if self._stop_event.is_set(): return 不再建线程,:209 的 join() 永久阻塞。access log 为每请求路径,高 QPS 下也可能长期不归零。__del__ 的 `exce...
建议: 三点收敛:(1) close() 中先 _stop_event.set() 截断新日志、消除 emit() 与 close() 的 TOCTOU 窗口,再 join(timeout=...) 等待 worker,未按期退出时用 _drain_queue() 同步兜底,最后关文件;(2) flush() 改为有 deadline 的等待(轮询 unfinished_tasks,或用 threading.Condition.wait_for(..., timeout=...) 配合自维护的 enqueued/written 计数),超时记录告警后返回,任何情况下不在 close()/__del__ 中调用无超时 queue.join();(3) __del__ 只做 best-effort 的 _stop_event.set(),不触发可能阻塞或新建线程的 close()。同步更新模块 docstring(:2-27 仍写 "avoiding main thread blocking")与类注释(:37 仍写 "runs forever"),明确 emit() 非阻塞、flush()/close() 会阻塞及其上限、close 后 emit() 静默丢弃。并补单测:emit 后 flush 返回时记录已落盘、close 幂等、worker 不可用时 flush 能在有限时间内返回、队列满时 dropped 递增且主流程不阻塞。
| CHECK_REQUEST_STOP(generate_context) \ | ||
| generate_context.stat_info.nextStage(); \ | ||
| func(generate_context); \ | ||
| generate_context.stat_info.finishStage(); \ |
There was a problem hiding this comment.
[P1] 22 个变更文件零测试,而三处现成 fixture 本可直接拦住本次 P0 与多处回归
22 个改动文件中无任何测试文件,而补测基础设施完备:rtp_llm/cpp/model_rpc/test/ 有 14 个文件(含 DecodeRpcServerTest.cc、PrefillRpcServerTest.cc、RpcServerRuntimeMetaTest.cc),cache_store/test/TcpCacheStoreLoadServiceClosureTest.cpp 有 7 条 case 与可复用 fixture。但全仓检索 DecodeStatInfo / allocate_resource_rt_us / deserializeErrorDetails 在测试目录零命中。改变行为的语义全部无覆盖:EXECUTE_STAGE_FUNC 新增的 finishStage() 与 decode 阶段归因、deserializeErrorDetails 取值边界、stopStream() 三态分类、!has_error_code() → LoadErrorUnknown(既有 case 因 :42 始终...
建议: 按既有 fixture 补最小集合(均为纯 CPU 逻辑,无需 GPU):(1) DecodeStatInfo/PrefillStatInfo 阶段归因表驱动单测(含重试与提前 return);(2) deserializeErrorDetails 覆盖合法码往返、空 details、解析失败、error_code=0、error_message 为空回落 status.error_message();(3) closure 侧覆盖 response 不设 error_code、blocks_size() 为 0/超预期、Run() 与 callback_ 抛异常,断言 end() 只执行一次;(4) makeTransferRequest 在 buffer 缺失与 local_partition_count == 0 下返回 nullptr 且不崩溃;(5) notifyDone 的超发与重复回调不变量;(6) Python 侧为 load_xgrammar()(找到/未找到 .so、重复调用幂等)与 _build_backend() 各补一条——上述 P0 只需一条 _build_backend() 用例即可拦住。
Checklist: [6.1] 分布式/跨平台变更有对应覆盖
| if (!details.ParseFromString(status.error_details())) { | ||
| return false; | ||
| } | ||
| *error_info = ErrorInfo(static_cast<ErrorCode>(details.error_code()), |
There was a problem hiding this comment.
[P2] deserializeErrorDetails 缺少取值域与 NONE_ERROR 校验,可把失败降级为无错误并绕过坏连接淘汰
函数只要 details 非空且 ParseFromString 成功即返回 true(:41-52),既无 transRPCErrorCode(:54-97)那样的白名单兜底(default → UNKNOWN_ERROR),也不拒绝 error_code == 0。ErrorDetailsPB{int64 error_code=1; string error_message=2} 与 gRPC 生态标准 google.rpc.Status{int32 code=1; string message=2} 字段号/wire type 完全兼容,代理写入 grpc-status-details-bin 的 payload 也会解析成功。命中 error_code=0 时返回 true 且 *error_info 为 NONE_ERROR:该分支位于 CLIENT_GRPC_RET_IF_ERROR 最前(PrefillRpcServer.cc:127-129),命中即跳过整条 Connect Failed/`Connect...
建议: 补两道校验:details.error_code() == static_cast<int64_t>(ErrorCode::NONE_ERROR) 时返回 false(details 只承载真实错误),使其与同文件既有契约一致——PrefillRpcServer.cc:426-427 显式要求 error_code() != NONE_ERROR、PrefillBatchRpcServer.cc:142-147 要求 == PRIORITY_PREEMPTED;并把裸 static_cast<ErrorCode> 换成集中式白名单校验(复用/扩展 transRPCErrorCode 的映射表,未知取值回落 UNKNOWN_ERROR),避免非本仓 peer 的错误码原样穿透到 Python ExceptionType。随后用该函数统一替换 PrefillRpcServer.cc:425-436、PrefillGenerateContext.cc:239、PrefillBatchRpcServer.cc:142 的内联解析,避免同一 wire 契约出现第四份漂移实现(R.I.1)。若要彻底消除与 google.rpc.Status 的歧义,可在 ErrorDetailsPB 增加魔数/版本字段。
| end(true, CacheStoreErrorCode::None); | ||
| } | ||
|
|
||
| void TcpCacheStoreLoadServiceClosure::end(bool success, CacheStoreErrorCode ec) { |
There was a problem hiding this comment.
[P2] closure end() 自毁与 Run() 新增 catch 重入构成 use-after-free 与 double free 窗口
end() 首行 std::unique_ptr<TcpCacheStoreLoadServiceClosure> self(this)(:108),函数返回或因异常退栈即 delete this。但 end() 非 noexcept::109 的 collector_->markEnd(success) 在 try 之外;两个 catch 分支(:113/:118)中 CacheStoreErrorCodeToString(ec) 构造 std::string 可抛 bad_alloc。任一处抛出时对象已删除,异常传播到本 PR 新增的 Run() catch(:95/:100)后先解引用 request_block_buffer_(UAF),再调用 end(false, ...),在已释放对象上二次构造 unique_ptr self(this) 并调 markEnd(double free + 指标重复上报)。改动前 Run() 无 catch、end() 也无 `unique_pt...
建议: 两处 closure 统一加幂等与不抛保证:给 end() 标 noexcept(让 markEnd/日志异常直接终止而非退化为内存破坏),并增加 bool ended_ = false; 哨兵,end() 入口 if (ended_) return; ended_ = true;,Run() 兜底 catch 仅在 !ended_ 时调用;同时把 collector_->markEnd(success) 移入 try 内,并统一 transfer_request_ / request_block_buffer_ 的空指针假设(Run() 的 catch 做了保护、end() 内直接解引用)。请在已有的 test/TcpCacheStoreLoadServiceClosureTest.cpp fixture 上补两条 case:Run() 体内抛异常、callback_ 抛异常,断言 end() 只执行一次且无重复释放。
| return; | ||
| } | ||
|
|
||
| if (blocks.size() > static_cast<size_t>(unfinished_count_)) { |
There was a problem hiding this comment.
[P2] notifyDone 忽略 success 形参,完成不变量仅按数量校验,重复上报仍会误判成功并回 EC_SUCCESS
notifyDone(bool success, ...)(:47-91)全函数从不读取 success,只按 error_code != None 分支(:62):调用方传 (false, None) 会走成功路径并可能 runSuccess() 回 EC_SUCCESS。新增校验 blocks.size() > static_cast<size_t>(unfinished_count_)(:76)只比数量、不校验 block key 身份:以总量 4 为例,首次上报 A/B(4→2)后若同一批被重复回调,2 > 2 不成立,unfinished_count_ 归零并 runSuccess(),而 C/D 从未传输,decode 侧会读到未初始化 KV,比计数变负更难发现。另 run() 的 :30-36 在 lock() 失败走 runFailed() 时未置 finished_ = true。TransferConnection::read 的分批回调实现不在本仓视图内,重复回调是否可达无法确认,故...
建议: 把计数式判定改为按 key 去重:以 std::set<std::string>(或复用 local_blocks_ 的 key 集合)记录待完成 block,回调时逐个 erase,返回 0 即为「重复或未知 key」的不变量违反,集合为空才 runSuccess()——这样同时覆盖超发与重复上报。同时让 success 参与判定(if (!success || error_code != None) 走失败路径),或删除该未使用形参以消除误导(推荐用 RTP_LLM_CHECK 断言其与 error_code 一致);并给 :34 的 runFailed() 前补 finished_ = true,使 finished_ 成为唯一的完成态真值来源。
Checklist: [6.1] 边界 case 覆盖(空、单元素、最大值)
| error_info_ = ErrorInfo(error_code, ErrorCodeToString(error_code)); | ||
| RTP_LLM_LOG_WARNING("remote store task notify request done, some request failed, request id is %s", | ||
| request_->request_id.c_str()); | ||
| auto internal_error_code = transCacheStoreErrorCode(error_code); |
There was a problem hiding this comment.
[P2] !success 且 error_code 为 None 时 error_info_ 退化成 NONE_ERROR,失败被读成无错误
改动前该分支恒用 ErrorCode::UNKNOWN_ERROR,保证不变量「!success ⇒ error_info_.hasError()」。改为 transCacheStoreErrorCode(error_code) 后,ErrorCodeUtil.h:10 将 CacheStoreErrorCode::None 映射为 ErrorCode::NONE_ERROR,此时 all_success_ = false(:212)但 error_info_.hasError() 为 false,通过 RemoteStoreTask::getErrorInfo() 判定的消费方会把一次失败的 KV 传输读成无错误。核查全部 TransferRequest::callback(false, ...) 站点(Messager.cpp:37/44、CacheTransferServiceClosure.cpp:41-42、NormalCacheStore.cpp 多处)当前均传非 None,故暂不可达,属不变量退化而非线上活...
建议: 在赋值前补一道收敛:if (internal_error_code == ErrorCode::NONE_ERROR) { internal_error_code = ErrorCode::UNKNOWN_ERROR; },把「失败必须带错误码」固定在唯一写入点,避免后续新增 callback 站点误传 None 时静默劣化。错误码透传本身是好改进,此处只需补齐兜底;这与上一条 notifyDone 的建议属同一类不变量加固,建议一并处理。
| package_dir = Path(spec.origin).parent | ||
| binding_path = package_dir / "libxgrammar_bindings.so" | ||
| if binding_path.is_file(): | ||
| library_paths = os.environ.get("LD_LIBRARY_PATH", "").split(":") |
There was a problem hiding this comment.
[P2] 运行期改写 LD_LIBRARY_PATH 对当前进程 dlopen 不可靠,且与仓库既有预加载范式分叉
load_xgrammar() 在 :154-159 把包目录前置进 os.environ["LD_LIBRARY_PATH"],随后 :161 在同一进程内 import xgrammar as xgr。glibc 的 ld.so 在进程启动时一次性解析并缓存该变量,进程内后续修改不影响本进程 dlopen;只有 xgrammar/tvm-ffi 自己在 Python 层读该变量时才生效,而 docstring 声称的 "including from Bazel rules_python runfiles" 无任何测试覆盖。仓内 models_py/modules/dsv4/tilelang_kernels.py:54-71 处理同一问题时明确注释 bare-name dlopen "only works if the lib is on LD_LIBRARY_PATH or loaded eagerly",并改用 ctypes.CDLL(..., RTLD_GLOBAL)。该写入还是进程级持久副作用,会被 spawn 出的 ...
建议: 复用仓内既有且已验证的范式:find_spec 已拿到 binding_path 绝对路径,直接 ctypes.CDLL(str(binding_path), mode=ctypes.RTLD_GLOBAL) 预加载后再 import xgrammar(捕获 OSError 后回退裸 import),不改 os.environ——预加载后同名依赖会命中已加载镜像,在当前进程即刻生效。建议把该逻辑抽成 rtp_llm/utils 下共用 helper 供 z3 与 xgrammar 复用,避免第三份实现(R.I.1)。若确实只为 spawn 子进程服务,请把函数名与 docstring 改为表达该语义、至少改为追加而非前置以降低遮蔽风险,并用 contextlib 在 import 前后恢复原值。无论哪种方案,都需要一个在 Bazel runfiles 布局下断言 load_xgrammar() 加载成功的测试,否则本 PR 的主目标无法验证。
Checklist: [6.1] 分层边界:新概念在正确层级,不泄漏内部;[6.1] 新逻辑有聚焦单测 + 相关集成/smoke 测试
| RTP_LLM_LOG_ERROR(error_msg); | ||
| decode_context.error_status = grpc::Status(grpc::StatusCode::RESOURCE_EXHAUSTED, error_msg); | ||
| decode_context.error_info = ErrorInfo(error_code, error_msg); | ||
| RTP_LLM_LOG_ERROR("request [%s] allocate resource failed, error code [%s], error message [%s]", |
There was a problem hiding this comment.
[P3] allocateResource 失败路径重复打印同一错误并丢失 request_info
:207 的 RTP_LLM_LOG_ERROR 之后紧接着 :211 调用 serializeErrorMsg(),而后者内部(LocalRpcServer.cc:91-94)会再打一条内容几乎相同的 RTP_LLM_LOG_WARNING,同一次 allocate 失败产生两条日志;该函数由 EXECUTE_WITH_RETRY 驱动,每次重试成对翻倍。另外 :211 用的是两参重载 serializeErrorMsg(request_key, error_info),其内部传入空 RequestInfo()(LocalRpcServer.cc:83-85);但 decode_context.request_info 早在 :180 已赋值,CHECK_REQUEST_TIMEOUT/CHECK_REQUEST_CANCELLED 也都用三参版本,此处白丢了请求侧标签。
建议: 去掉 :207 的显式 ERROR 日志(错误码与消息已由 serializeErrorMsg 统一打印),或反之让 serializeErrorMsg 不再打日志、由调用方决定;同时把 :211 改为三参重载 serializeErrorMsg(decode_context.request_key, decode_context.request_info, decode_context.error_info),与同文件其它错误路径保持一致的日志标签。
| ErrorInfo loadCacheSyncForTp(DecodeGenerateContext& decode_context, LoadKVCacheContext& load_context); | ||
| BroadcastLoadRequestPB constructRemoteLoadRequest(const LoadKVCacheContext& load_context, | ||
| int index, | ||
| const std::vector<std::string>& peer_ips) const; |
There was a problem hiding this comment.
📍 实际位置 rtp_llm/cpp/model_rpc/DecodeRpcServer.h:103(不在 diff 展示范围内,就近挂载)
[P3] 删除 loadCacheSyncForTp 后 thread_pool_ 与 initThreadPool 成为完全不可达的死代码
loadCacheSyncForTp 是 thread_pool_ 的唯一使用者。全仓检索 rtp_llm/cpp/model_rpc 确认,删除该函数后 DecodeRpcServer::thread_pool_ 只剩 initThreadPool() 内的创建(DecodeRpcServer.cc:131-133)与析构里的 stop()(:138-141),再无任何 pushTask/async;而 initThreadPool() 本身也已零调用点(仅 .h:70 声明 + .cc:127 定义)。即便被调用其逻辑也已失效:if (resource_.workers.size() > 0) { return; } 意味着只有 workers 为空时才建池,而池大小恰为 resource_.workers.size() * 8 即 0。
建议: 在同一 PR 内一并删除 DecodeRpcServer::thread_pool_ 成员、initThreadPool() 声明与定义、以及析构中的 stop 分支,使死代码清理完整(已确认零消费者,无需替代实现或迁移说明);并同步清理 BUILD 中不再需要的 autil 线程池依赖(若无其他使用者)。若后续确有异步 load cache 计划,请以注释或 TODO 说明保留原因,避免读者误以为该线程池仍在服务某条路径。
Checklist: [6.1] KISS/YAGNI:无投机性抽象;[I] 删除或重命名内部 file、registry entry、model name、metric enum、op binding、plugin symbol 时,必须全仓搜索消费者,并提供替代实现、迁移说明或 smoke 覆盖;只有暴露到 HTTP/RPC/config/persisted format 时才按外部兼容性处理
| error_code = ErrorCode::LOAD_KV_CACHE_FAILED; | ||
| error_msg += std::to_string(rank) + ": " + status.error_message() + ", "; | ||
| const auto& worker_addr = resource_.grpc_workers.at(rank); | ||
| const auto peer_addr = rank < decode_context.peer_addrs.size() ? |
There was a problem hiding this comment.
[P3] TP load cache 报错日志用 rank 直接索引 peer_addrs,非对称 TP 下指向错误对端
:676-678 与 :687-689 用 rank < decode_context.peer_addrs.size() ? decode_context.peer_addrs[rank] : "<missing>" 取对端地址。越界已有保护(不会 OOB),但索引语义不对:同文件 constructRemoteLoadRequestForMla(:403-417)表明 worker→peer 的正确映射为 peer_addrs[index / part_cnt](workers 多于 peers)或 peer_addrs[index * group_num](peers 多于 workers),只有 1:1 时才等于 peer_addrs[rank]。因此在非对称 TP 部署下这条新增的 peer= 字段会稳定指向错误对端,且因 rank < size() 成立而不会退化为 <missing>,可能把排障引向无关机器。影响仅限日志文本,不改变控制流。
建议: 抽一个 peerAddrForRank(rank) 辅助函数复用 constructRemoteLoadRequest/ForMla 的映射规则(含 1:1、workers 多、peers 多、prefill_cp_size > 1 四种情形),在 :590、:676、:687 三处统一调用;无法确定映射时输出 peer=<unmapped> 而非可能错误的地址,避免新增的诊断字段反而误导定位。
Checklist: [6.1] 可观测性:日志/指标/超时可操作、非噪声
| @@ -40,10 +40,7 @@ AUTIL_LOG_SETUP(rtp_llm, RtpLLMRemoteCacheSDKMetrics); | |||
| REPORT_MUTABLE_QPS(name##_metric); \ | |||
There was a problem hiding this comment.
📍 实际位置 rtp_llm/cpp/metrics/RtpLLMMetrics.cc:38(不在 diff 展示范围内,就近挂载)
[P3] 同一文件内"是否上报"存在三套并存且不一致的约定
改动后同一文件并存三套约定:REPORT_QPS(:38-41)保留 if (collector->name) 真值判断;PrefillRecentCacheKeyMetrics::report(:135)用 has_value、RtpLLMExecutorMetrics::report(:364)用 context_batch_size != 0 做 ad-hoc 显式判断;REPORT_GAUGE(:43)改为完全无条件上报。宏名仍叫 REPORT_GAUGE,无法体现「无条件」这一新语义,维护者容易按旧直觉误以为仍有值判断。另 min_response_done_time_us 默认值为 1lu << 60(.h:59),在旧宏下即为真值恒被上报,也印证纯真值判断本身不可靠。
建议: 统一为单一约定:保留一个带显式 has/optional 判断的上报入口(如 REPORT_GAUGE_IF_SET),把无条件上报单独命名(如 REPORT_GAUGE_ALWAYS),使调用点自解释,并在宏定义处补一行注释说明两者取舍与适用场景。若短期不做统一,至少将 REPORT_GAUGE 重命名以反映「无条件上报」语义,避免与仍带真值判断的 REPORT_QPS 形成误导性对称。
Checklist: [6.1] DRY:重复非平凡逻辑被抽取或显式复用;[I] 同一功能用统一工具函数
| import xgrammar as xgr | ||
|
|
||
| return xgr | ||
|
|
There was a problem hiding this comment.
[P3] 新增与修改的行未通过 black / clang-format
grammar_tokenizer_info.py::163 为 load_xgrammar() 的 return xgr,:164 仅一个空行,:165 即顶层 def build_grammar_tokenizer_info_json(;Black/PEP 8 E302 与本文件其余顶层定义(:134-135、:145-147)均为两个空行,black --check 会报改动。Messager.cpp:80:同一个 RTP_LLM_LOG_WARNING 的实参列表中 :77-79 与 :81 为 16 空格缩进,而 :80 的 local_key.c_str(), 为 32 空格,仓库根 .clang-format(4 空格、120 列、实参对齐)不会产生这种同列表内不一致的对齐。仓库 pre-commit 已启用 black + isort + clang-format。
建议: 提交前对改动文件跑一次仓库 pre-commit(black、isort --profile=black、clang-format),使逻辑改动与格式化彻底分离;否则这些行会在后续任意提交中被自动重排,把无关格式化 churn 混入其他变更,干扰 blame 与 CI 二分定位。
Checklist: [6.1] 逻辑变更未混入无关格式化
Summary
Validation
origin/maingit diff --checkpython3 -m py_compile rtp_llm/config/grammar_tokenizer_info.py rtp_llm/dash_sc/inference/grammar_validator.py