Open
feat: 增加有界 PCIe 消息发送与写入结果回传 #14
liuxingsenAdd on 13 days ago
feat: 增加有界 PCIe 消息发送与写入结果回传 #14
Open
liuxingsenAdd on 13 days ago
Total: 18 files changed+960-35
@@ -1,5 +1,8 @@
1# Changelog1# Changelog
2All notable changes to this project will be documented in this file.2All notable changes to this project will be documented in this file.
3+## [1.130.19] - 2026-10-27
4+增加有界PCIe消息队列、Control/requester保留容量及驱动写入结果回传;拒绝后不再绕过调度直接写设备。
5+ 
3## [1.130.15] - 2026-08-256## [1.130.15] - 2026-08-25
4更新版本号,触发流水线推包7更新版本号,触发流水线推包
5 8 
@@ -0,0 +1,15 @@
1+# PCIe消息发送与完成语义
2+ 
3+现有requester、Control及后续responder复用同进程的唯一发送队列。完整消息在EOM后才交给写服务,禁止只发出因容量限制而被拒绝消息的前缀。所有System共用64项/256KiB上限,其中Control预留8项/16KiB、requester预留16项/64KiB、普通响应40项/176KiB;单消息最多128帧/16KiB。字节预算只包含帧数据,Lua表与IPC副本须计入最终内存验收。
4+ 
5+写服务单次实际写入后返回结果;不隐式重试、不因另一通道失败重置全部实例。失败停止同消息后续帧。请求失败只唤醒原wait对象;`on_tx_complete`通知驱动写入结果,旧Send返回只表示提交/准入,不表示对端已收到。无法撤回已交给驱动的单帧,驱动阻塞耗时仍需实测,不能将入队或write成功称为PCIe端到端完成。
6+ 
7+100ms为本轮TX等待上界候选,由队列`deadline_ms`选项集中定义,使用单调时钟逐帧检查。唯一10ms检查器回收未完成的分片和过期排队项;驱动调用过期则通知一次失败、暂停新准入,实际调用返回前仍保留其额度,不再生成替代调用。驱动恢复返回后停止旧消息尾片并恢复准入;不能通过Lua定时器中断内核write,永久挂起仍需修复驱动或人工恢复服务。完整业务期限需在responder/V2后续接入时传递。
8+ 
9+复位、重新配置、本地及owner地址变化取消排队项;peer对象/地址改变也会在写前失效。写入成功/失败计数在原debug dump中输出,失败日志按1、2、4、8…次记录原因。NVMe命令槽在同步异常时也释放。旧Send计数仍指准入,不等于驱动写入。完整跨接口及路由代次仍由后续改动补齐,该PR不宣告阶段1整体或阶段2完成。
10+ 
11+底层拒绝的修复:原`channel_protocol::push`收到回调false仍调用`write_to`;现在存在发送回调时直接向上传递其结果。用C语言类比,该回调相当于结构体里的发送函数指针,false代表调用方已拒绝,不能再次走默认发送函数。无回调的旧直写路径保留,但写失败返回false。
12+ 
13+分片上下文的修复:`mctp_ctx fragment_ctx = local_ctx`等价于复制C结构体的成员;`set_prev(&ctx)`保留当前调用的原上下文链,SOM/EOM只写副本。它是函数栈上的临时对象,在一次同步下传返回后失效。Lua队列保存帧字符串、数值和另行复制的地址字段,不保留这个临时对象地址;调用者原上下文不随分片改变。没有改变线上MCTP头格式,CPU架构差异仍由既有硬件编码层处理。
14+ 
15+验证覆盖完整聚合、消息/字节配额、Control/requester保留、有效对象不被淘汰、deadline/地址往返/取消、活跃IPC仍占用额度、驱动失败停止尾片、迟到失败与复用wait隔离、真实管道拒绝后不直写。该PR拆分基于main,需和共享传输修复及PLDM模块共同验收,不能以独立分支测试替代最终镜像验收。
@@ -2,7 +2,7 @@
2 "name": "mctpd",2 "name": "mctpd",
3 "type": "application",3 "type": "application",
4 "deployConfig": "framework.service",4 "deployConfig": "framework.service",
5- "version": "1.130.15",5+ "version": "1.130.19",
6 "author": "Huawei",6 "author": "Huawei",
7 "license": "Mulan PSL v2",7 "license": "Mulan PSL v2",
8 "options": {8 "options": {
@@ -198,6 +198,8 @@ static void register_ctxs(lua_State *L, luawrap::stack_table &t)
198 cls.def_readwrite("dest", &mctp_ctx::dest);198 cls.def_readwrite("dest", &mctp_ctx::dest);
199 cls.def_readwrite("tag_owner", &mctp_ctx::tag_owner);199 cls.def_readwrite("tag_owner", &mctp_ctx::tag_owner);
200 cls.def_readwrite("tag", &mctp_ctx::tag);200 cls.def_readwrite("tag", &mctp_ctx::tag);
201+ cls.def_readwrite("som", &mctp_ctx::som);
202+ cls.def_readwrite("eom", &mctp_ctx::eom);
201 t.set("mctp_ctx", cls);203 t.set("mctp_ctx", cls);
202 });204 });
203 register_ctx<pcie_vdm_ctx, protocol_ctx>(L, [L, &t](auto &cls) {205 register_ctx<pcie_vdm_ctx, protocol_ctx>(L, [L, &t](auto &cls) {
@@ -85,13 +85,16 @@ int channel_protocol::rx()
85 85 
86bool channel_protocol::push(const protocol_ctx &ctx, const slice_t &data, const uint32_t medium_type)86bool channel_protocol::push(const protocol_ctx &ctx, const slice_t &data, const uint32_t medium_type)
87{87{
88- if (protocol::push(ctx, data, medium_type)) {88+ const bool accepted = protocol::push(ctx, data, medium_type);
89- return true;89+ // 显式调度器拒绝必须向上传播,不能绕过背压直接写设备。
90+ if (accepted || m_tx_handler) {
91+ return accepted;
90 }92 }
91 93 
92 // 下层协议不处理自己发出去94 // 下层协议不处理自己发出去
93 if (write_to(m_fd, data) != E_OK) {95 if (write_to(m_fd, data) != E_OK) {
94 debug_log(DLOG_INFO, "channel_protocol: write failed, fd = %d, %s", m_fd, strerror(errno));96 debug_log(DLOG_INFO, "channel_protocol: write failed, fd = %d, %s", m_fd, strerror(errno));
97+ return false;
95 }98 }
96 99 
97 return true;100 return true;
@@ -306,7 +306,11 @@ bool mctp_protocol::push(const protocol_ctx &ctx, const slice_t &data, const uin
306 hdr.u.seq = i & MCTP_HDR_SEQ_MASK;306 hdr.u.seq = i & MCTP_HDR_SEQ_MASK;
307 307 
308 m_tx_buf.insert(m_tx_buf.end(), it, next_it);308 m_tx_buf.insert(m_tx_buf.end(), it, next_it);
309- if (!protocol::push(ctx, slice(m_tx_buf), msg_key)) {309+ mctp_ctx fragment_ctx = local_ctx;
310+ fragment_ctx.set_prev(&ctx);
311+ fragment_ctx.som = hdr.u.som;
312+ fragment_ctx.eom = hdr.u.eom;
313+ if (!protocol::push(fragment_ctx, slice(m_tx_buf), msg_key)) {
310 return false;314 return false;
311 }315 }
312 316 
@@ -68,6 +68,8 @@ public:
68 mctp_eid_t dest;68 mctp_eid_t dest;
69 bool tag_owner;69 bool tag_owner;
70 uint8_t tag;70 uint8_t tag;
71+ bool som = true; // 仅发送边界元数据,不改变线上头格式
72+ bool eom = true;
71};73};
72 74 
73enum class mctp_route_policy {75enum class mctp_route_policy {
@@ -150,10 +150,12 @@ function c_mctp_discovery:process_prepare_for_ep_discovery_req(req)
150 local mctp_ctx, ctx = self.ep:get_ctx_by_medium(self.ep.medium_type)150 local mctp_ctx, ctx = self.ep:get_ctx_by_medium(self.ep.medium_type)
151 local eid = mctp_ctx.src151 local eid = mctp_ctx.src
152 local phy_addr = ctx.remote_phy_addr152 local phy_addr = ctx.remote_phy_addr
153+ -- 复位先使旧事务失效;Control endpoint 及其接收上下文由 reset 保留。
154+ -- 然后恢复请求来源并在新代次提交应答,避免把本次成功响应一并取消。
155+ self:set_discovered(false)
atomgit-bot
atomgit-botatomgit-bot13 days ago

🟡 Medium Priority

改动位置:src/lualib/mctp_discovery.lua:155,把 self:set_discovered(false) 从"应答之后"移到"应答之前"。

失败链:set_discovered(false) 同步触发 on_discovery_changed:emit(false)(mctp_discovery.lua:82-89;signal 为同步派发,mctp_app.lua:145 需显式 next_tick 才能延后,可佐证),监听者调用 self.mctp_engine:reset();c_mctp_engine:reset()(mctp_engine.lua:311-321)先执行 self.owner_eid = 0; self.bmc_eid = 0; self.bmc_phy_addr = 0,之后才轮到第 158 行的 self.ep:response(...)。

问题:第 156 行 set_owner_eid(eid, phy_addr) 只恢复了 owner,bmc_eid/bmc_phy_addr 没有恢复。而 c_ep_mctp_ctrl:response(ep_mctp_ctrl.lua:57-59)用 mctp_ctx.src = self.mctp_engine.bmc_eid、dest = owner_eid 组帧,add_pcie_vdm_ctx 又写 pcie_vdm_ctx.bmc_phy_addr = mctp_engine.bmc_phy_addr。于是在重发现(discovered 原为 true)场景下,本条 PrepareForEndpointDiscovery 应答的源 EID 变成 0、bmc_phy_addr 变成 0;改动前应答在 reset 之前组帧,携带的是上一轮已分配的 bmc_eid/bmc_phy_addr。对端若按 src EID/phy_addr 校验或路由,握手应答可能被丢弃,重发现停在握手阶段。

测试 test_rediscovery_cancels_old_work_but_sends_control_response(test/unit/test_write_service.lua:363-386)只断言 #f.writes == 1,未校验写出帧头的 src/phy_addr,覆盖不到该差异。

建议:在 set_discovered(false) 触发 reset 前保存 bmc_eid/bmc_phy_addr,reset 后用请求中的值恢复;或让 control 应答改用请求携带的 src/phy_addr 组帧,避免应答帧 src EID=0、bmc_phy_addr=0。

likedislike
Incorrect?
liuxingsen
13 days ago Comments:
153 self.mctp_engine:set_owner_eid(eid, phy_addr)156 self.mctp_engine:set_owner_eid(eid, phy_addr)
154 log:debug('[System%s]mctp_discovery: prepare for endpoint discovery req', self.system_id)157 log:debug('[System%s]mctp_discovery: prepare for endpoint discovery req', self.system_id)
155 self.ep:response(req, MCTP_CTRL_CC_SUCCESS, ROUTE_TO_ROOT)158 self.ep:response(req, MCTP_CTRL_CC_SUCCESS, ROUTE_TO_ROOT)
156- self:set_discovered(false)
157 return true159 return true
158end160end
159 161 
@@ -116,8 +116,13 @@ function c_ep_mctp:send(cb)
116 -- 通知类发送不等待响应,清空rsp key,避免发送时携带上一个请求的旧key116 -- 通知类发送不等待响应,清空rsp key,避免发送时携带上一个请求的旧key
117 self.cur_rsp_key = nil117 self.cur_rsp_key = nil
118 self:prepare_mctp_request()118 self:prepare_mctp_request()
119+ self.tx_admission_failed = false
119 cb()120 cb()
120- self.req_success_count = self.req_success_count + 1121+ if self.tx_admission_failed then
O
Oopenubmc-ci-bot13 days ago

💡 建议:send 的计数与"延迟写入失败"语义需对齐

send 仅在 tx_admission_failed(准入/容量拒绝,同步返回)时递增 not_send_count 并抛错;但消息已准入、后续 drain 阶段驱动写入失败的场景(on_tx_complete 回调、wait.err)发生在返回之后,send 仍会走 req_success_count = req_success_count + 1,把该次发送计为成功。

PR 说明已声明"旧 Send 返回只表示提交/准入,不表示对端已收到",语义上是合理的;但两个计数口径(准入 vs 写入)混在一起,实机上排查"已提交但未送达"时容易误读。 修复建议:在注释中明确 req_success_count 为"准入成功"而非"写入成功",或在 on_tx_complete 上单独维护写入失败计数。

likedislike
liuxingsen
13 days ago Comments:
122+ self.not_send_count = self.not_send_count + 1
123+ error('MCTP transmit queue full')
124+ end
atomgit-bot
atomgit-botatomgit-bot13 days ago

🟡 Medium Priority

改动位置:src/lualib/mctp_endpoints/ep_mctp.lua:121-124,c_ep_mctp:send 在 cb() 后若 tx_admission_failed 为真则 error('MCTP transmit queue full')。

契约变化:send 之前从不出错,现在会抛异常。所有子类端点都是 super.send(self, function() ... end) 之后紧跟收尾动作,异常会让收尾动作整段跳过。

失败模式(具体):ep_nvme.lua:134-137 为 c_nvme_endpoint.super.send(self, function() ... end) self:release_cmd_slot(slot) 一旦 PCIe 队列准入失败(control 8 / request 16 / response 40 配额或 16KiB 字节上限打满),super.send 抛出 MCTP transmit queue full,release_cmd_slot(slot) 不再执行,NVMe 命令槽被永久占用;多次准入失败后该消息类型可用槽耗尽,后续发送全部失败(发送路径不可恢复)。同理,ep_spdm/ep_pldm/ep_ncsi 的调用方(mctp_mdb_mgmt.lua:120 return ep:send(ctx, data, req_ctx) 与 mctp_app 的 ImplEndpointPCIeEndpointSend)都没有 pcall 包裹,异常会直接冒泡到业务侧,把"队列满"变成一次异常回包而不是可判断的返回值。

建议:把 admission 失败改为返回值或让调用方保证清理:send 内不要直接 error,改为 return false, 'MCTP transmit queue full';或让所有子类 send 用 pcall 包裹 super.send,确保 release_cmd_slot 等收尾动作在异常路径下也执行。

likedislike
Incorrect?
liuxingsen
13 days ago Comments:
125+ self.req_success_count = self.req_success_count + 1 -- Send仅表示准入;实际写入结果由engine单独统计。
121end126end
122 127 
123function c_ep_mctp:prepare_mctp_request(id)128function c_ep_mctp:prepare_mctp_request(id)
@@ -105,11 +105,14 @@ function c_nvme_endpoint:request(ctx, data, timeout_ms, req_ctx, rsp_ctx)
105 end105 end
106 end106 end
107 107 
108- local rsp = c_nvme_endpoint.super.request(self, slot, function(key)108+ local ok, rsp = pcall(c_nvme_endpoint.super.request, self, slot, function(key)
109 self.wait_info[key] = { count = 0, is_replay = is_replay }109 self.wait_info[key] = { count = 0, is_replay = is_replay }
110 self.mctp_engine.nvme_proto:push(payload, self.nvme_ctx, msg_key)110 self.mctp_engine.nvme_proto:push(payload, self.nvme_ctx, msg_key)
111 end, timeout_ms, more_process_cb)111 end, timeout_ms, more_process_cb)
112 self:release_cmd_slot(slot)112 self:release_cmd_slot(slot)
113+ if not ok then
114+ error(rsp)
115+ end
113 116 
114 local index = slot + 1117 local index = slot + 1
115 local nvme_ctx = self.nvme_ctx118 local nvme_ctx = self.nvme_ctx
@@ -131,10 +134,13 @@ function c_nvme_endpoint:send(_, data, req_ctx)
131 self:pcie_ctx_set()134 self:pcie_ctx_set()
132 end135 end
133 local payload, slot, _, msg_key = self:prepare_nvme_request(req_ctx, nil, data)136 local payload, slot, _, msg_key = self:prepare_nvme_request(req_ctx, nil, data)
134- c_nvme_endpoint.super.send(self, function()137+ local ok, err = pcall(c_nvme_endpoint.super.send, self, function()
135 self.mctp_engine.nvme_proto:push(payload, self.nvme_ctx, msg_key)138 self.mctp_engine.nvme_proto:push(payload, self.nvme_ctx, msg_key)
136 end)139 end)
137 self:release_cmd_slot(slot)140 self:release_cmd_slot(slot)
141+ if not ok then
142+ error(err)
143+ end
138end144end
139 145 
140local function check_ctx_value(val)146local function check_ctx_value(val)
@@ -26,6 +26,7 @@ local stop_timer = c_timer.stop
26local mctp_utils = require('mctp_utils')26local mctp_utils = require('mctp_utils')
27local mctp_def = require('mctp_def')27local mctp_def = require('mctp_def')
28local print_msg = mctp_utils.print_msg28local print_msg = mctp_utils.print_msg
29+local shared_tx = require('mctp_tx').new()
29 30 
30local MCTP_DEV_NAME = '/dev/mctp'31local MCTP_DEV_NAME = '/dev/mctp'
31local MCTP1_DEV_NAME = '/dev/mctp1'32local MCTP1_DEV_NAME = '/dev/mctp1'
@@ -84,6 +85,9 @@ function c_mctp_engine:ctor(bus, system_id)
84 self.module_name = 'MCTP'85 self.module_name = 'MCTP'
85 self.bus = bus86 self.bus = bus
86 self.system_id = system_id87 self.system_id = system_id
88+ self.tx_queue, self.tx_epoch = shared_tx, 0
89+ self.on_tx_complete = signal.new()
atomgit-bot
atomgit-botatomgit-bot13 days ago

🔵 Low Priority

变更行:89 行新增 self.on_tx_complete = signal.new(),237 行每次完成都 emit。

影响:在整个 src 中检索 on_tx_complete 只有 88-89 行的定义与 237 行的 emit,没有任何生产代码订阅(仅 test/unit/test_write_service.lua 的用例订阅)。因此"写入失败/取消/过期"这些新增回传除了置 wait.err 之外没有任何可观测出口:例如 Control 应答(Prepare for Discovery)被容量拒绝时,既无监听者统计也无上层处理,信号实际为死代码。

可选改进(非必须):要么删除该信号,要么在 ep_mctp/ep_mctp_ctrl 的发送侧订阅并把失败计入已有的 not_send_count/req_error_count 类计数,让这项回传真正被消费。

建议:确认该信号是否真有消费方:若无,删除 on_tx_complete 定义与 237 行的 emit;若有,补上订阅点(例如在端点发送侧统计写入失败次数),并在本 PR 内一并提交,避免新增无人使用的对外接口。

likedislike
Incorrect?
liuxingsen
13 days ago Comments:
90+ self.tx_written_count, self.tx_failed_count = 0, 0
87 91 
88 self.mctp_req_timeout_count = 092 self.mctp_req_timeout_count = 0
89 self.on_mctp_req_timeout = signal.new()93 self.on_mctp_req_timeout = signal.new()
@@ -96,7 +100,14 @@ function c_mctp_engine:ctor(bus, system_id)
96end100end
97 101 
98function c_mctp_engine:dump_eid_info()102function c_mctp_engine:dump_eid_info()
99- return self.eid_table:dump_eid_table(self.endpoints)103+ local info = self.eid_table:dump_eid_table(self.endpoints)
104+ info[#info + 1] = string.format(
105+ 'PCIe TX written=%d failed=%d blocked=%s\r\n',
106+ self.tx_written_count,
107+ self.tx_failed_count,
108+ tostring(self.tx_queue.blocked == true)
109+ )
110+ return info
100end111end
101 112 
102function c_mctp_engine:get_eid_table()113function c_mctp_engine:get_eid_table()
@@ -185,6 +196,98 @@ local function get_mctp_dev_name(system_id)
185 end196 end
186end197end
187 198 
199+function c_mctp_engine:invalidate_pcie_tx()
200+ self.tx_epoch = self.tx_epoch + 1
201+ self.tx_queue:cancel(self)
202+end
203+ 
204+function c_mctp_engine:configure_pcie_tx(fd)
205+ self:invalidate_pcie_tx()
206+ local transport, physical = mctp.mctp_ctx.new(), mctp.pcie_vdm_ctx.new()
207+ self.hw_pcie_proto:set_tx_handler(function(ctx, data, msg_key)
208+ local msg_type, eid, medium = self.get_ep_details(msg_key)
209+ local ep = self.endpoints:get_endpoint(eid, msg_type, medium)
210+ transport:copy_from(ctx)
211+ physical:copy_from(ctx)
atomgit-bot
atomgit-botatomgit-bot13 days ago

🔵 Low Priority

变更行:201-204 行在每帧(每个分片)回调里都 mctp.mctp_ctx.new() + mctp.pcie_vdm_ctx.new() 并从 ctx 拷贝,另外 210 行每帧 table.concat 生成 meta.key。

影响:这是 PCIe 出口最热的路径(PLDM/NCSI/SPDM 业务每一帧都会经过),每条消息的每个分片都构造两个 C++ ctx 对象并做两次 copy_from;同文件 542 行 rx 明确用模块级单例 mctp_ctx 复用来规避这种开销,两边做法不一致。

建议:续帧(som == false)不需要重新拷贝上下文:或只在 som 帧构造 transport/physical 并在同一消息内复用,或把 valid 需要的标量(remote_phy_addr、bmc_phy_addr、src、dest、tag)提取为数字/字符串存入 meta,避免每帧两个 ctx 对象分配。

likedislike
Incorrect?
liuxingsen
13 days ago Comments:
212+ local remote_addr, local_addr = physical.remote_phy_addr, physical.bmc_phy_addr
213+ -- responder 可服务已纳管的其他类型端点;不要求业务先注册 PLDM requester。
214+ if not ep and not transport.tag_owner then
215+ for _, peer in pairs(self.endpoints.endpoints) do
216+ if peer.eid == eid and peer.medium_type == medium and peer.phy_addr == remote_addr then
217+ ep = peer
218+ break
219+ end
220+ end
221+ end
222+ local peer_type = ep and ep:get_msg_type()
223+ local epoch, bmc_eid, bmc_addr = self.tx_epoch, self.bmc_eid, self.bmc_phy_addr
224+ local peer_eid, peer_addr = ep and ep.eid, ep and ep.phy_addr
225+ local key = transport.tag_owner and ep and ep.cur_rsp_key
226+ local wait = key and self.rsp_wait_list[key]
227+ local meta = {
228+ key = table.concat({ msg_key, transport.src, transport.dest, transport.tag }, ':'),
229+ fd = fd,
230+ som = transport.som,
231+ eom = transport.eom,
232+ kind = msg_type == 0 and 'control' or (transport.tag_owner and 'request' or 'response'),
233+ deadline = self.tx_queue.now() + self.tx_queue.deadline_ms,
234+ valid = function()
235+ return self.tx_epoch == epoch
236+ and self.bmc_eid == bmc_eid
237+ and self.bmc_phy_addr == bmc_addr
238+ and (
239+ msg_type == 0
240+ or (
241+ ep ~= nil
242+ and ep.eid == peer_eid
243+ and ep.phy_addr == peer_addr
244+ and self.endpoints:get_endpoint(eid, peer_type, medium) == ep
245+ and peer_addr == remote_addr
246+ and local_addr == bmc_addr
247+ )
248+ )
249+ end,
250+ done = function(ok, reason)
251+ if ok then
252+ self.tx_written_count = self.tx_written_count + 1
253+ else
254+ self.tx_failed_count = self.tx_failed_count + 1
255+ local n = self.tx_failed_count
256+ if n & (n - 1) == 0 then
257+ log:notice(
258+ '[System%s]MCTP TX failed: key=%s reason=%s count=%d',
259+ self.system_id,
260+ msg_key,
261+ reason,
262+ n
263+ )
264+ end
265+ end
266+ if not ok and wait and self.rsp_wait_list[key] == wait then
267+ wait.err = errors.invalid_mctp_request()
268+ if wait.co ~= coroutine.running() then
269+ skynet.wakeup(wait.co)
270+ end
271+ end
atomgit-bot
atomgit-botatomgit-bot13 days ago

🟡 Medium Priority

变更行:done 在失败且命中原始 wait 时执行 wait.err = errors.invalid_mctp_request(); skynet.wakeup(wait.co)(232-236 行)。

受影响契约:request() 里 wait_info 在 pcall(cb, key) 之前就登记进 rsp_wait_list,而 cb 内部(ep_mctp/ep_mctp_ctrl → mctp_proto:push → 本 tx handler)在容量不足时通过 243-247 行同步回调 meta.done(false, 'capacity')。本 diff 新增的 492 行 if not ok or wait_info.err 正是为了处理这种"cb 未返回错误但等待对象已被置错"的同步场景,说明该路径确实存在。

失效模式:此同步路径下 wait.co 就是 coroutine.running()(该协程还没走到 510 行的 skynet.wait()),对正在运行/尚未挂起的协程调用 skynet.wakeup 属于非法唤醒,skynet 侧会出现 cannot resume non-suspended coroutine 或误唤醒调用者;触发条件正是本 PR 的目标场景——PCIe requester 消息在队列水位满/单条消息字节超限时被拒。此外多分片消息被拒时,后续每个分片(entry 为 nil 的分支)都会再回调一次 done,对同一 wait 反复唤醒。修复只需在该场景下不唤醒(492 行已经能返回 E_SEND_FAILED),或改为延迟唤醒。

建议:在登记 wait 时记录其所属协程/挂起状态(如 wait.co == coroutine.running() 时不 wakeup),或在 done 中判断 coroutine.status(wait.co) == 'suspended' 后再唤醒,并依赖 492 行的 wait_info.err 检查让同步路径直接返回 E_SEND_FAILED;也可将唤醒改为 skynet.fork 延迟执行,避免在 push 调用栈内唤醒正在运行的请求协程。

Suggestions
271
+ done = function(ok, reason)
272
+ if not ok and wait and self.rsp_wait_list[key] == wait then
273
+ wait.err = errors.invalid_mctp_request()
274
+ if wait.co ~= coroutine.running() and coroutine.status(wait.co) == 'suspended' then
275
+ skynet.wakeup(wait.co)
271
- end
276
+ end
277
+ end
278
+ self.on_tx_complete:emit(ok, reason, msg_key)
279
+ end,
Apply suggestion
likedislike
Incorrect?
liuxingsen
13 days ago Comments:
272+ self.on_tx_complete:emit(ok, reason, msg_key)
273+ end,
274+ }
275+ if ep and ep.is_trace == 'start' then
276+ print_msg('SEND', data, medium)
277+ end
278+ local admitted, notified = self.tx_queue:enqueue(self, meta, data)
279+ if not admitted then
280+ if not notified then
281+ pcall(meta.done, false, 'capacity')
282+ end
283+ if ep then
284+ ep.tx_admission_failed = true
285+ end
286+ end
atomgit-bot
atomgit-botatomgit-bot13 days ago

🟡 Medium Priority

变更行:243-247 行在 enqueue 返回未准入且未通知时直接调用 meta.done(false, 'capacity')。

受影响契约:队列内部的完成通知统一走 notify() 的 pcall(entry.meta.done, ok, reason)(mctp_tx.lua:51-55),因此使用者回调抛错被隔离;引擎自己在 246 行却无保护地直调同一个 done。

建议:将 246 行的直接回调改为受保护调用(例如 pcall(meta.done, false, 'capacity')),与 mctp_tx 内部 notify 的隔离方式保持一致,避免回传逻辑(错误模块缺失、非法唤醒、on_tx_complete 订阅者异常)影响协议栈的发送流程。

Suggestions
286
+ if not admitted then
287
+ if not notified then
288
+ pcall(meta.done, false, 'capacity')
289
+ end
290
+ if ep then
291
+ ep.tx_admission_failed = true
292
+ end
286
293
  end
Apply suggestion
likedislike
Incorrect?
liuxingsen
13 days ago Comments:
287+ return admitted
288+ end)
289+end
290+ 
188function c_mctp_engine:start()291function c_mctp_engine:start()
189 local fd = self.hw_pcie_proto:open_path(get_mctp_dev_name(self.system_id), O_RDWR)292 local fd = self.hw_pcie_proto:open_path(get_mctp_dev_name(self.system_id), O_RDWR)
190 if fd < 0 then293 if fd < 0 then
@@ -192,15 +295,7 @@ function c_mctp_engine:start()
192 return295 return
193 end296 end
194 297 
195- self.hw_pcie_proto:set_tx_handler(function(_, data, msg_key)298+ self:configure_pcie_tx(fd)
196- local msg_type, eid, medium_type = self.get_ep_details(msg_key)
197- local ep = self.endpoints:get_endpoint(eid, msg_type, medium_type)
198- if ep and ep.is_trace == 'start' then
199- print_msg('SEND', data, medium_type)
200- end
201- skynet.send('.mctp_write_service', 'lua', fd, data)
202- return true
203- end)
204 299 
205 self.smbus_proto:set_tx_handler(function(_, data, msg_key)300 self.smbus_proto:set_tx_handler(function(_, data, msg_key)
206 local msg_type, eid, medium_type = self.get_ep_details(msg_key)301 local msg_type, eid, medium_type = self.get_ep_details(msg_key)
@@ -239,6 +334,7 @@ function c_mctp_engine:start()
239end334end
240 335 
241function c_mctp_engine:stop()336function c_mctp_engine:stop()
337+ self:invalidate_pcie_tx()
242 self.pcie_read_work:stop()338 self.pcie_read_work:stop()
243 self.pcie_read_work:join()339 self.pcie_read_work:join()
244 self.pcie_read_work = nil340 self.pcie_read_work = nil
@@ -260,6 +356,7 @@ function c_mctp_engine:reset()
260end356end
261 357 
262function c_mctp_engine:reinit()358function c_mctp_engine:reinit()
359+ self:invalidate_pcie_tx()
263 log:notice('[System%s]mctp_engine: vdm engine reinit', self.system_id)360 log:notice('[System%s]mctp_engine: vdm engine reinit', self.system_id)
264 self.hw_pcie_proto:vdm_engine_reinit()361 self.hw_pcie_proto:vdm_engine_reinit()
265end362end
@@ -317,6 +414,7 @@ end
317 414 
318function c_mctp_engine:set_owner_eid(eid, phy_addr)415function c_mctp_engine:set_owner_eid(eid, phy_addr)
319 if self.owner_eid ~= eid or self.owner_phy_addr ~= phy_addr then416 if self.owner_eid ~= eid or self.owner_phy_addr ~= phy_addr then
417+ self:invalidate_pcie_tx()
320 self.owner_eid = eid418 self.owner_eid = eid
321 self.owner_phy_addr = phy_addr419 self.owner_phy_addr = phy_addr
322 log:notice('[System%s]mctp_engine: set owner_eid=%d, owner_phy_addr=%d', self.system_id, eid, phy_addr)420 log:notice('[System%s]mctp_engine: set owner_eid=%d, owner_phy_addr=%d', self.system_id, eid, phy_addr)
@@ -325,6 +423,7 @@ end
325 423 
326function c_mctp_engine:set_bmc_eid(eid, phy_addr)424function c_mctp_engine:set_bmc_eid(eid, phy_addr)
327 if self.bmc_eid ~= eid or self.bmc_phy_addr ~= phy_addr then425 if self.bmc_eid ~= eid or self.bmc_phy_addr ~= phy_addr then
426+ self:invalidate_pcie_tx()
328 self.bmc_eid = eid427 self.bmc_eid = eid
329 self.mctp_proto.pcie_eid = eid428 self.mctp_proto.pcie_eid = eid
330 self.bmc_phy_addr = phy_addr429 self.bmc_phy_addr = phy_addr
@@ -426,7 +525,8 @@ function c_mctp_engine:request(key, cb, timeout_ms, more_process_cb)
426 wait_info = { co = coroutine.running() }525 wait_info = { co = coroutine.running() }
427 self.rsp_wait_list[key] = wait_info526 self.rsp_wait_list[key] = wait_info
428 local ok, err = pcall(cb, key)527 local ok, err = pcall(cb, key)
429- if not ok then528+ if not ok or wait_info.err then
529+ err = wait_info.err or err
430 self.rsp_wait_list[key] = nil530 self.rsp_wait_list[key] = nil
431 log:error('[System%s]mctp_engine: request cb failed, %s', self.system_id, err)531 log:error('[System%s]mctp_engine: request cb failed, %s', self.system_id, err)
432 return E_SEND_FAILED, err532 return E_SEND_FAILED, err
atomgit-bot
atomgit-botatomgit-bot13 days ago

🔵 Low Priority

变更行:492-496 行新增 if not ok or wait_info.err 分支,是本次"写入结果回传"落到请求方的核心分支。

失效模式:test/unit/test_write_service.lua 里的引擎用例只用手工构造的 wait = { co = ... } 直接断言 wait.err(如 test_engine_batches_real_encoded_fragments_and_reports_write_failure),从未真正走 c_mctp_engine:request()。因此"cb 成功但 wait 已被置错时返回 E_SEND_FAILED"、"正常完成时返回 E_OK"、"置错后 rsp_wait_list[key] 被清理"这些行为都没有回归保护:例如有人把 wait_info.err 的判断挪到 skynet.wait() 之后、或改成返回 E_ERROR,测试仍然全绿,而现场表现是队列满时请求一直等到 15s 默认超时而不是快速失败。

建议:补一条针对 request() 的用例(可把 skynet.wait/skynet.wakeup 打桩):cb 内同步触发一次发送失败,断言 request() 返回 E_SEND_FAILED 且 rsp_wait_list[key] 已清空、未创建定时器;再补一条 cb 成功且被准入时返回 E_OK 的正向用例。

likedislike
Incorrect?
liuxingsen
13 days ago Comments:
@@ -0,0 +1,233 @@
1+-- Copyright (c) 2026 Huawei Technologies Co., Ltd.
2+-- openUBMC is licensed under Mulan PSL v2.
3+-- You can use this software according to the terms and conditions of the Mulan PSL v2.
4+-- You may obtain a copy of Mulan PSL v2 at:
5+-- http://license.coscl.org.cn/MulanPSL2
6+-- THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
7+-- EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
8+-- MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
9+-- See the Mulan PSL v2 for more details.
10+ 
11+-- 从 write_service 抽离消息调度;只把有界帧交给唯一写服务。
12+local class = require('mc.class')
13+local skynet = require('skynet')
14+local clock = require('utils.core')
15+local Queue = class()
16+local LIMITS = { control = { 8, 16384 }, request = { 16, 65536 }, response = { 40, 180224 } }
17+local MAX_FRAMES, MAX_BYTES = 128, 16384 -- 同 docs/PCIe发送队列.md,包含线缆帧头。
18+local DEFAULT_DEADLINE_MS = 100
19+ 
20+function Queue:ctor(options)
21+ options = options or {}
22+ self.write = options.write
23+ or function(fd, data)
24+ return skynet.call('.mctp_write_service', 'lua', fd, data)
O
Oopenubmc-ci-bot13 days ago

⚠️ 警告:阻塞式 skynet.call 写驱动可能导致整个发送队列永久卡死

默认 self.write 使用 skynet.call('.mctp_write_service', ...)(阻塞等待响应),而 drain() 只在单个 fork 出的协程中运行,且 add_frame 用 if not self.running 避免重复启动 drain(见 line 168-173)。

风险:若底层 mctp.write 因驱动繁忙/挂起长时间不返回,write_service 不会 skynet.ret,则该 drain 协程被永久挂起,self.running 保持 true,此后所有新排队消息都无法被调度,发送通路整体不可用。

并且 100ms deadline 的 reason 检查只发生在两次写之间(drain 内 for 循环中逐帧后检查),无法在一帧阻塞式写执行过程中打断它,所以 deadline 并不能兜底驱动挂死。

修复建议:

  • 若驱动阻塞是真实风险,建议使用非阻塞/带超时的写入(或在 write_service 侧对 mctp.write 增加异步超时/看门狗),并让 drain 在写入被卡住时仍能通过其他协程主动退出;
  • 或者明确约定驱动 write 必须非阻塞,并在文档/注释中记录该前提。
likedislike
liuxingsen
13 days ago Comments:
25+ end
26+ self.defer = options.defer or skynet.fork
27+ self.timeout = options.timeout or skynet.timeout
28+ self.deadline_ms = options.deadline_ms or DEFAULT_DEADLINE_MS
29+ self.now = options.now
30+ or function()
31+ local t = clock.monotonic_time()
32+ return t.tv_sec * 1000 + t.tv_nsec // 1000000
33+ end
34+ self.ready, self.partial, self.usage = {}, {}, {}
35+ for kind in pairs(LIMITS) do
36+ self.usage[kind] = { 0, 0 }
37+ end
38+ self.count, self.bytes, self.dropped = 0, 0, 0
39+end
40+ 
41+function Queue:release(entry)
42+ if entry.finished then
43+ return
44+ end
45+ entry.finished = true
46+ local used = self.usage[entry.meta.kind]
47+ used[1], used[2] = used[1] - 1, used[2] - entry.bytes
48+ self.count, self.bytes = self.count - 1, self.bytes - entry.bytes
49+ if self.partial[entry.owner] == entry then
50+ self.partial[entry.owner] = nil
51+ end
52+ return true
53+end
54+ 
55+local function notify(entry, ok, reason)
56+ if entry.notified then
57+ return
58+ end
59+ entry.notified = true
60+ if entry.meta.done then
61+ pcall(entry.meta.done, ok, reason)
62+ end
63+end
64+ 
65+function Queue:finish(entry, ok, reason)
66+ if self:release(entry) then
67+ notify(entry, ok, reason)
68+ end
69+end
70+ 
71+function Queue:cancel(owner)
72+ -- 先完成全部摘除和计数,再运行可能重入队列的使用者回调。
73+ local removed = {}
74+ local partial = self.partial[owner]
75+ if partial then
76+ self:release(partial)
77+ removed[#removed + 1] = partial
78+ end
79+ if self.active and self.active.owner == owner then
80+ self.active.cancelled = true
81+ end
82+ for i = #self.ready, 1, -1 do
83+ local entry = self.ready[i]
84+ if entry.owner == owner then
85+ table.remove(self.ready, i)
86+ self:release(entry)
87+ removed[#removed + 1] = entry
88+ end
89+ end
90+ for _, entry in ipairs(removed) do
91+ notify(entry, false, 'cancelled')
92+ end
93+end
94+ 
95+function Queue:reason(entry)
96+ if entry.cancelled then
97+ return 'cancelled'
98+ end
99+ local timed, now = pcall(self.now)
100+ if not timed or type(now) ~= 'number' or type(entry.meta.deadline) ~= 'number' then
101+ return 'clock_failed'
102+ end
103+ if now >= entry.meta.deadline then
104+ return 'expired'
105+ end
106+ local ok, valid = pcall(entry.meta.valid)
107+ if not ok or not valid then
108+ return 'stale'
109+ end
110+end
111+ 
112+function Queue:watch()
113+ if self.watching then
114+ return
115+ end
116+ self.watching = true
117+ self.timeout(1, function() -- 单个10ms定时器,不为每次提交创建独立timer。
118+ self.watching = false
119+ local removed = {}
120+ for _, entry in pairs(self.partial) do
121+ if self:reason(entry) then
122+ removed[#removed + 1] = entry
123+ end
124+ end
125+ for i = #self.ready, 1, -1 do
126+ if self:reason(self.ready[i]) then
127+ removed[#removed + 1] = table.remove(self.ready, i)
128+ end
129+ end
130+ for _, entry in ipairs(removed) do
131+ self:release(entry)
132+ end
133+ local active = self.active
134+ local reason = active and self:reason(active)
135+ if reason then
136+ self.blocked, active.cancelled = true, true
137+ end
138+ for _, entry in ipairs(removed) do
139+ notify(entry, false, self:reason(entry))
140+ end
141+ if reason then
142+ notify(active, false, reason)
143+ end
144+ -- 卡住的驱动调用仍占一项额度;不启动替代调用,也不继续准入。
145+ if self.count > 0 then
146+ self:watch()
147+ end
148+ end)
149+end
150+ 
151+function Queue:drain()
152+ while #self.ready > 0 do
153+ local entry = table.remove(self.ready, 1)
154+ self.active = entry
155+ local reason = self:reason(entry)
156+ for _, frame in ipairs(entry.frames) do
157+ if reason then
158+ break
159+ end
160+ local ok, written = pcall(self.write, entry.meta.fd, frame)
161+ reason = self:reason(entry)
162+ if not ok or not written then
163+ reason = reason or 'write_failed'
164+ end
165+ end
166+ self.active, self.blocked = nil, false
167+ self:finish(entry, reason == nil, reason or 'written')
168+ end
169+ self.running = false
170+end
atomgit-bot
atomgit-botatomgit-bot13 days ago

🔵 Low Priority

改动位置:src/lualib/mctp_tx.lua:100-119(Queue:drain),以及 133-150(add_frame 里 som 建立 partial 后需等 eom 才进入 ready)。

问题 1(drain 无异常兜底):self.running = false 只在 while 循环正常退出时执行,self.active 也只在每轮末尾清空。循环体内 local reason = self:reason(entry) 未被 pcall 保护,而 reason 内部会调用 self.now()(默认实现 clock.monotonic_time(),见第 25-29 行)并读取 entry.meta.deadline。一旦 now()/时钟实现异常或 deadline 缺失(例如外部直接构造 meta 而漏填),drain 会带着 running=true、active=非nil 退出,此后 add_frame 第 167 行 if not self.running 永远为假,再也不会 fork 新的 drain —— 整条 PCIe 发送通道永久停摆且无任何日志。触发条件是一次异常;后果是静默全局卡死,属于比"单条消息失败"严重得多的故障模式。

问题 2(partial 无回收路径):som 帧被接纳后容量计入 used[1]/self.count,但只有 eom 到达才会移入 ready 并被 drain 回收。若该消息中途终止(后续帧因 #frame == 0/类型不符在 136-138 行被拒,或上层 push 中断),self.partial[owner] 会一直持有 1 个消息配额与相应字节数;100ms deadline 只作用于 ready 中的条目,partial 没有超时扫描,只能等同 owner 的下一条 som 覆盖或 cancel。触发后 control(8)/request(16) 配额被长期占用,可能让该 System 的请求类报文持续 admission 失败。

建议:drain 主体用 pcall/xpcall 包裹,并在无论成功失败都清理 self.active 与 self.running(finally 语义);同时为 partial 中的半成品消息增加 deadline 回收,避免配额长期泄漏导致后续 admission 一直失败。

likedislike
Incorrect?
liuxingsen
13 days ago Comments:
171+ 
172+function Queue:enqueue(owner, meta, frame)
173+ local replaced = meta.som and self.partial[owner]
174+ if replaced then
175+ self:release(replaced)
176+ end
177+ local admitted, notified = self:add_frame(owner, meta, frame)
178+ if replaced then
179+ notify(replaced, false, 'incomplete')
180+ end
181+ return admitted, notified
182+end
183+ 
184+function Queue:add_frame(owner, meta, frame)
185+ local entry = self.partial[owner]
186+ local limits, used = LIMITS[meta.kind], self.usage[meta.kind]
187+ if self.blocked or not limits or type(frame) ~= 'string' or #frame == 0 then
188+ if entry then
189+ self:finish(entry, false, 'invalid_frame')
190+ return false, true
191+ end
192+ return false
193+ end
194+ if meta.som then
195+ if used[1] >= limits[1] then
196+ self.dropped = self.dropped + 1
197+ return false
198+ end
199+ entry = { owner = owner, meta = meta, frames = {}, bytes = 0 }
200+ self.partial[owner] = entry
201+ used[1], self.count = used[1] + 1, self.count + 1
202+ end
203+ if not entry then
204+ return false
205+ end
206+ if
207+ entry.meta.key ~= meta.key
208+ or entry.meta.kind ~= meta.kind
209+ or entry.bytes + #frame > MAX_BYTES
210+ or #entry.frames >= MAX_FRAMES
211+ or used[2] + #frame > limits[2]
O
Oopenubmc-ci-bot13 days ago

💡 建议:配额魔数在多处重复,建议单一数据源

每条消息上限 128 帧 / 16384 字节(此处 #entry.frames >= 128、entry.bytes + #frame > 16384)以及 LIMITS 表中的control=8 项/16KiB、request=16 项/64KiB、response=40 项/176KiB,与 docs/PCIe发送队列.md、test/unit/test_write_service.lua 中大量重复。

配额改动时容易遗漏同步,建议统一从 mctp_tx 单点导出常量供文档/测试引用(或至少在代码处加一行注释说明与文档的对应关系)。

likedislike
liuxingsen
13 days ago Comments:
212+ then
213+ self.dropped = self.dropped + 1
214+ self:finish(entry, false, 'capacity')
215+ return false, true
216+ end
atomgit-bot
atomgit-botatomgit-bot13 days ago

🔵 Low Priority

改动位置:src/lualib/mctp_tx.lua:151-161,消息级容量判定硬编码 entry.bytes + #frame > 16384 与 #entry.frames >= 128。

问题:配置声明的池容量是 response 180224 字节(176KiB),但单条消息无论属于哪一类都被限制在 16KiB 且最多 128 个分片。128 帧上限与 mctp_protocol.max_size 相乘才是真实可发送的消息长度上限(例如 max_size=64 时仅 8KiB,max_size=250 时约 31KiB,仍被 16KiB 字节上限截断)。触发条件:任一条报文在给定 max_size 下需要超过 128 个分片或超过 16KiB。

失败模式:该条件命中后会走 self:finish(entry, false, 'capacity') 直接把整条消息丢弃(已收下的前若干分片也不会写出),并让上层 request 返回 E_SEND_FAILED。对大报文(例如大块 PLDM/PDR 或 FW 数据)这会表现为"报文永远发不出去",而日志里只有 capacity 原因,难以定位是硬编码上限所致。目前测试(test_write_service.lua:102-120)只验证 16KiB 边界,未覆盖 128 帧边界与真实 max_size 的乘积关系。

建议:把 16384/128 提为具名常量并与 LIMITS、mctp_proto.max_size 对齐(或由 max_size 推导帧数上限),并在超限时记录被丢弃消息的 key/长度,便于定位大报文发送失败。

likedislike
Incorrect?
liuxingsen
13 days ago Comments:
217+ entry.frames[#entry.frames + 1] = frame
218+ entry.bytes, used[2], self.bytes = entry.bytes + #frame, used[2] + #frame, self.bytes + #frame
219+ self:watch()
220+ if meta.eom then
221+ self.partial[owner] = nil
222+ self.ready[#self.ready + 1] = entry
223+ if not self.running then
224+ self.running = true
225+ self.defer(function()
226+ self:drain()
227+ end)
228+ end
229+ end
230+ return true
231+end
232+ 
233+return Queue
@@ -12,26 +12,21 @@ local skynet = require('skynet')
12require('skynet.manager')12require('skynet.manager')
13local mctp = require('mctp')13local mctp = require('mctp')
14local logging = require('mc.logging')14local logging = require('mc.logging')
15- 15+local failures = 0
16-local MAX_RETRY_COUNT = 50
17-local retry_count = 0
18 16 
19skynet.start(function()17skynet.start(function()
20 skynet.register('.mctp_write_service')18 skynet.register('.mctp_write_service')
21- skynet.dispatch('lua', function(_, _, fd, data)19+ skynet.dispatch('lua', function(session, _, fd, data)
22- local ok, err20+ local ok, err = pcall(mctp.write, fd, data)
23- for i = 1, 10 do21+ if not ok then
24- ok, err = pcall(mctp.write, fd, data)22+ failures = failures + 1
25- if ok then23+ if failures & (failures - 1) == 0 then
26- retry_count = 024+ logging:notice('mctp_write_service: failed count=%d error=%s', failures, tostring(err))
27- break
28 end25 end
29- retry_count = retry_count + 1
30- logging:notice('mctp_write_service failed, count: %s, err: %s', i, err)
31- skynet.sleep(50)
32 end26 end
33- if retry_count > MAX_RETRY_COUNT then27+ -- 每次仅写一帧;上层有界调度器收到结果后才继续该消息。
34- skynet.send('.mctpd', 'lua', 'reset')28+ if session ~= 0 then
29+ skynet.ret(skynet.pack(ok))
35 end30 end
atomgit-bot
atomgit-botatomgit-bot13 days ago

🔵 Low Priority

改动位置:src/service/write_service.lua:19-26,删除 MAX_RETRY_COUNT/每帧 10 次重试与 skynet.send('.mctpd','lua','reset') 的失败升级逻辑,改为 local ok = pcall(mctp.write, fd, data) 后只做 logging:debug。

失败链:驱动写失败现在只产生一条 debug 日志;上层 mctp_tx 的 done 回调只在"存在对应 rsp_wait_list 等待者"时才唤醒请求并置 err(mctp_engine.lua:232-238),其余情况仅 on_tx_complete:emit,而 on_tx_complete 在仓库内没有订阅者。结果是:驱动持续写失败时既没有任何 notice/error 日志,也没有原先的 mctpd reset 兜底路径(该服务此前正是靠这条通道从设备异常中恢复)。

建议:写失败时保留可观测性:使用 notice/error 级日志并累计失败计数;对连续失败提供升级路径(例如通过 engine.on_tx_complete 的消费方触发 reset),避免驱动异常被静默吞掉。

likedislike
Incorrect?
liuxingsen
13 days ago Comments:
36 end)31 end)
37end)32end)
@@ -451,6 +451,18 @@ local function test_mctp_dump(bus)
451 lu.assertNotIsNil(file)451 lu.assertNotIsNil(file)
452 local data = file:read('a')452 local data = file:read('a')
453 file:close()453 file:close()
454+ -- 后台Control发送数量受轮询时序影响;逐实例验证新增诊断,原库存/请求计数继续精确比对。
455+ local tx_lines
456+ data, tx_lines = data:gsub(
457+ 'PCIe TX written=(%d+) failed=(%d+) blocked=(%a+)\r?\n',
458+ function(written, failed, blocked)
459+ lu.assertTrue(tonumber(written) > 0)
460+ lu.assertEquals(tonumber(failed), 0)
461+ lu.assertEquals(blocked, 'false')
462+ return ''
463+ end
464+ )
465+ lu.assertEquals(tx_lines, 2)
454 local pid = utils_core.getpid()466 local pid = utils_core.getpid()
455 local expert_str = string.format(str, pid, pid, pid, pid, pid)467 local expert_str = string.format(str, pid, pid, pid, pid, pid)
456 lu.assertEquals(data, expert_str)468 lu.assertEquals(data, expert_str)
@@ -114,6 +114,7 @@ function TestMctpDiscovery:test_mctp_discovery_success()
114 end)114 end)
115 115 
116 self.mctp_discovery:smbus_set_eid_to_endpoint(116 self.mctp_discovery:smbus_set_eid_to_endpoint(
117+ 'discovery-test',
117 target_eid,118 target_eid,
118 ref_chip,119 ref_chip,
119 medium_type,120 medium_type,
@@ -158,6 +159,7 @@ function TestMctpDiscovery:test_mctp_discovery_failed()
158 local ref_chip = { path = path }159 local ref_chip = { path = path }
159 160 
160 self.mctp_discovery:smbus_set_eid_to_endpoint(161 self.mctp_discovery:smbus_set_eid_to_endpoint(
162+ 'discovery-test',
161 target_eid,163 target_eid,
162 ref_chip,164 ref_chip,
163 medium_type,165 medium_type,
@@ -250,7 +250,7 @@ function TestMctpEngine:test_start()
250 end250 end
251 251 
252 mctp_engine.hw_pcie_proto.set_tx_handler = function(_, func)252 mctp_engine.hw_pcie_proto.set_tx_handler = function(_, func)
253- func(_, '123456', 123456)253+ lu.assertEquals(type(func), 'function') -- 真实分片入队由TestWriteService覆盖
254 end254 end
255 255 
256 mctp_engine.smbus_proto.set_tx_handler = function(_, func)256 mctp_engine.smbus_proto.set_tx_handler = function(_, func)
@@ -301,3 +301,18 @@ function TestMctpSeqValidation:test_tag_owner_preserved()
301 lu.assertEquals(self.rx_ctx.tag_owner, true)301 lu.assertEquals(self.rx_ctx.tag_owner, true)
302 lu.assertEquals(self.rx_ctx.tag, 5)302 lu.assertEquals(self.rx_ctx.tag, 5)
303end303end
304+ 
305+function TestMctpSeqValidation:test_tx_context_preserves_message_boundaries_without_mutating_caller()
306+ local proto, ctx, seen = mctp.mctp_protocol.new(), mctp.mctp_ctx.new(), {}
307+ ctx.src, ctx.dest, ctx.tag, ctx.tag_owner = 10, 9, 3, true
308+ proto.max_size = 64
309+ proto:set_tx_handler(function(incoming, data)
310+ local copy = mctp.mctp_ctx.new()
311+ copy:copy_from(incoming)
312+ seen[#seen + 1] = { copy.som, copy.eom, data:byte(4) }
313+ return true
314+ end)
315+ proto:push(string.rep('x', 130), ctx, mctp_def.PHY_MEDIUM_PCI)
316+ lu.assertEquals(seen, { { true, false, 0x8b }, { false, false, 0x1b }, { false, true, 0x6b } })
317+ lu.assertEquals({ ctx.som, ctx.eom }, { true, true })
318+end
@@ -20,3 +20,529 @@ function TestWriteService:test_config()
20 local write_service = require('service.write_service')20 local write_service = require('service.write_service')
21 lu.assertNotIsNil(write_service)21 lu.assertNotIsNil(write_service)
22end22end
23+ 
24+function TestWriteService:fixture()
25+ local ok, Tx = pcall(require, 'mctp_tx')
26+ lu.assertTrue(ok, '需要消息级有界发送队列')
27+ local f = { tasks = {}, timers = {}, writes = {}, results = {}, now = 100, owner = {} }
28+ f.queue = Tx.new({
29+ defer = function(fn)
30+ f.tasks[#f.tasks + 1] = fn
31+ end,
32+ timeout = function(_, fn)
33+ f.timers[#f.timers + 1] = fn
34+ end,
35+ now = function()
36+ return f.now
37+ end,
38+ write = function(fd, data)
39+ f.writes[#f.writes + 1] = { fd, data }
40+ if f.on_write then
41+ return f.on_write(fd, data)
42+ end
43+ return true
44+ end,
45+ })
46+ function f:add(data, som, eom, kind)
47+ return self.queue:enqueue(self.owner, {
48+ key = 'peer:tag',
49+ fd = 9,
50+ som = som,
51+ eom = eom,
52+ kind = kind or 'response',
53+ deadline = 200,
54+ valid = function()
55+ return not self.invalid
56+ end,
57+ done = function(ok, reason)
58+ self.results[#self.results + 1] = { ok, reason }
59+ end,
60+ }, data)
61+ end
62+ function f:run()
63+ while #self.tasks > 0 do
64+ table.remove(self.tasks, 1)()
65+ end
66+ end
67+ return f
68+end
69+ 
70+function TestWriteService:test_message_is_admitted_before_any_fragment_is_written()
71+ local f = self:fixture()
72+ lu.assertTrue(f:add('first', true, false))
73+ lu.assertEquals(#f.tasks, 0)
74+ lu.assertTrue(f:add('last', false, true))
75+ lu.assertEquals(#f.tasks, 1)
76+ lu.assertEquals(#f.writes, 0)
77+ f:run()
78+ lu.assertEquals(f.writes, { { 9, 'first' }, { 9, 'last' } })
79+ lu.assertEquals(f.results, { { true, 'written' } })
80+ lu.assertEquals({ f.queue.count, f.queue.bytes }, { 0, 0 })
81+end
82+ 
83+function TestWriteService:test_full_response_pool_preserves_requester_and_control_capacity()
84+ local f = self:fixture()
85+ for _ = 1, 40 do
86+ lu.assertTrue(f:add('r', true, true))
87+ end
88+ lu.assertFalse(f:add('overflow', true, true))
89+ for _ = 1, 16 do
90+ lu.assertTrue(f:add('q', true, true, 'request'))
91+ end
92+ for _ = 1, 8 do
93+ lu.assertTrue(f:add('c', true, true, 'control'))
94+ end
95+ lu.assertFalse(f:add('overflow', true, true, 'control'))
96+ lu.assertEquals(f.queue.count, 64)
97+ lu.assertEquals(#f.tasks, 1)
98+ f:run()
99+ lu.assertEquals(#f.writes, 64)
100+ lu.assertTrue(f:add('again', true, true))
101+ f:run()
102+ lu.assertEquals(#f.writes, 65)
103+end
104+ 
105+function TestWriteService:test_byte_budget_and_failed_partial_message_do_not_send_prefix()
106+ local f = self:fixture()
107+ for _ = 1, 11 do
108+ lu.assertTrue(f:add(string.rep('x', 16384), true, true))
109+ end
110+ lu.assertFalse(f:add('x', true, true))
111+ for _ = 1, 4 do
112+ lu.assertTrue(f:add(string.rep('q', 16384), true, true, 'request'))
113+ end
114+ lu.assertTrue(f:add(string.rep('c', 16384), true, true, 'control'))
115+ lu.assertFalse(f:add('x', true, true))
116+ lu.assertEquals(f.queue.bytes, 262144)
117+ f:run()
118+ lu.assertTrue(f:add('first', true, false))
119+ lu.assertFalse(f:add(string.rep('x', 16384), false, true))