cFS与SB学习记录（二）：管道建好以后，消息为什么还收不到

SB 最容易记住的接口是 CreatePipe、Subscribe、TransmitMsg 和 ReceiveBuffer。但只记调用顺序，还是解释不了几个问题：管道已经创建，为什么没有消息？订阅究竟改了哪张表？一个管道订阅两种消息时，深度和消息限额又怎样一起生效？

这篇先把发送前需要准备的东西理清。沿着“创建一个管道，再订阅一条消息”的过程，可以同时看见 SB 的资源管理、锁的边界和路由关系。

系列目录 · 上一篇：cFS、cFE与OSAL

先摆出三个不同的对象

假设有一条温度消息 TEMP，检查任务和记录任务都要接收它：

MsgId(TEMP) → 一条路由
                  ├─ 目的地：检查管道
                  └─ 目的地：记录管道

检查管道 → OSAL 队列 A
记录管道 → OSAL 队列 B

管道是接收端拥有的入口。路由把一个 MsgId 关联到目的地列表。目的地则表示“这个 MsgId 应交给这条管道”，还带着激活状态、该消息的数量限制等信息。

所以创建管道只解决“我在哪里收”，订阅才解决“哪些消息要送到这里”。同一条管道可以挂在多个 MsgId 的目的地列表里，发送端不需要知道它的名字。

最小使用顺序如下，错误分支先压缩成一个返回：

CFE_SB_PipeId_t pipe;
CFE_Status_t status;

status = CFE_SB_CreatePipe(&pipe, 8, "TEMP_RX");
if (status != CFE_SUCCESS)
    return status;

status = CFE_SB_Subscribe(temp_msg_id, pipe);
if (status != CFE_SUCCESS)
    CFE_SB_DeletePipe(pipe);

temp_msg_id 在这里表示任务配置中已经确定的消息身份。不要因为示例需要一个数值，就随意挑一个整数当合法 MsgId；具体编码由选用的消息实现和任务配置决定。

应用调用之前，SB 先准备了什么

CFE_SB_EarlyInit() 是早期初始化入口。它清零 SB 全局状态，创建共享数据互斥量，建立消息内存池，初始化管道分配游标和 SBR，再初始化统计遥测的消息头。

清全局状态
  → 创建共享互斥量
  → 建立分级内存池与缓冲区跟踪链
  → 初始化管道表分配状态
  → 初始化 SBR 路由和映射

完整 cFE 中，模块表把 EarlyInit、TaskMain 和 Cleanup 等入口交给启动框架组织。cfe_sb_objtab.c 声明这种关系；它不是在每次创建管道时重新登记一个 SB 模块。

早期初始化也不等于 SB 管理任务已经运行。应用可用的收发机制和 SB 自己处理 NOOP、HK 等管理消息的循环，需要分开看。后面的管理任务篇会再接上这一步。

创建管道不是单独建立一个 OS 队列

在 cfe_sb_priv.h 中，管道描述符的关键字段可以缩成：

/* 省略统计和填充字段，仅展示关系 */
typedef struct {
    CFE_SB_PipeId_t   PipeId;
    CFE_ES_AppId_t    AppId;
    osal_id_t         SysQueueId;
    uint16            MaxQueueDepth;
    uint16            CurrentQueueDepth;
    CFE_SB_BufferD_t *LastBuffer;
} PipeView;

PipeId 是 SB 句柄，SysQueueId 是底下的 OSAL 队列句柄，AppId 表示管道归属。LastBuffer 暂时先记成“上一次交给接收者的缓冲区”，下一篇接收分析会用到它。

CFE_SB_CreatePipe() 先取调用者 AppId、TaskId，检查深度等参数。随后在 SB 锁内找一个空闲描述符，把它标成保留状态，再解锁建立 OSAL 队列：

OS_QueueCreate(&SysQueueId, PipeName, Depth,
               sizeof(CFE_SB_BufferD_t *), 0);

这里的单条数据大小是指针大小。队列排的是消息缓冲区描述符的地址，整条消息内容不在队列条目里。

创建队列成功后，SB 再拿锁，把队列 ID、归属应用和深度写回描述符，发布最终 PipeId。若 OSAL 创建失败，保留的描述符槽位要释放掉。这个两阶段过程既防止别人抢同一槽位，也避免在持有 SB 全局锁时完成较慢的操作系统资源创建。

因此 CreatePipe 失败要分层排查：是 SB 管道表没空位，还是 OSAL 对象不足、名称冲突，或者主机队列参数受限？它们都表现为“没建成”，原因却不同。

订阅时，才把 MsgId 与管道连起来

常用的 CFE_SB_Subscribe() 是包装入口，更完整的实现汇到 CFE_SB_SubscribeFull()。这一步在共享锁保护下检查：管道句柄是否仍有效、调用应用是否拥有它、MsgId 和作用域是否合法。

接着用 MsgId 向 SBR 查 RouteId。没有对应路由时才添加路由；已有路由时直接遍历它的目的地列表，查找是否已经包含这条 PipeId。

Subscribe(TEMP, 检查管道)
  → TEMP 没有路由：建立路由
  → 路由下没有检查管道：申请目的地，挂入链表

Subscribe(TEMP, 记录管道)
  → 复用 TEMP 路由
  → 增加记录管道这个目的地

目的地中最值得看的几个字段是：

字段	含义
PipeId	投到哪条接收管道
MsgId2PipeLim	此 MsgId 在这条管道中的消息数量限制
BuffCount	当前计入这条订阅的排队数量
Active	这条投递关系是否启用
Scope	本地/全局订阅相关属性

这里的作用域也不等于自动建立网络连接。它是订阅关系供相关机制使用的属性，具体跨节点传输还需要其他组件。

重复订阅会让接收端多收到一份吗

在本文对应的源码中，同一 MsgId 下再次发现同一 PipeId，会走重复订阅分支，记录重复订阅事件和计数，状态仍可保持成功。它不会再挂一个目的地，也没有在这里给订阅增加一个需要配对退订的引用计数。

因此重复调用 Subscribe 不能拿来实现“同一条管道收两份”。这里的重复订阅统计，与消息缓冲区的 UseCount 完全是两回事。后者是在记录同一块消息内存还被多少地方持有。

深度为 8，为什么第 3 条 TEMP 就可能被拒绝

管道总深度和单个 MsgId 的限额是两层限制。假设管道深度为 8，TEMP 的订阅限额为 2，另一个 STATUS 消息限额为 6。

当前队列：[TEMP, TEMP, STATUS]
总条数只有 3，尚未达到管道深度 8
但再投 TEMP，会先碰到 TEMP 的限额 2

反过来，多种消息加起来填满队列后，即使某种消息没有达到自己的限额，也可能无法再入队。单消息限额可以限制某类高频消息挤占整个接收入口；管道深度则约束总排队容量。

在这版代码里，发送时先预记 BuffCount 和管道深度，OSAL 入队失败再回滚；成功接收导出引用时会扣减排队计数，而消息本体可能仍由 LastBuffer 保持。因此不能把“当前排队数”当成“内存池里尚未释放的消息总数”。

普通订阅使用缺省限额；需要明确控制时看 SubscribeEx 或 SubscribeLocal。这些限制要结合生产频率和消费时长选择，单纯把队列加深只能延迟暴露慢消费者的问题。

退订与删除管道，要处理已经在路上的消息

退订会从对应路由的目的地链表摘掉这个订阅。它阻止后续发送继续向这里选取目的地，却不会把此前已经排入队列的消息凭空抹掉。接收路径因此要容忍“消息在队列里，但原订阅已经不存在”的情况。

删除整条管道则更进一步：先检查归属，从各路由移除它的目的地，再把描述符标成保留状态，避免同时出现新订阅；接着释放 LastBuffer，逐条排空队列并归还消息引用，删除 OSAL 队列，最后释放管道表槽位。

停止这条管道的新路由关系
  → 保护描述符状态
  → 清 LastBuffer 与排队消息引用
  → 删除 OSAL 队列
  → 归还管道槽位

如果只执行 OS_QueueDelete()，队列里的指针没了，背后的消息引用却没有得到对应清理。这也是我觉得看生命周期代码比只看成功创建更有价值的地方：队列和缓冲区分别属于不同层，销毁时必须把两份状态一起收好。

至此，发送前的关系已经齐了：PipeId 找到接收队列，MsgId 找到路由，路由下的目的地找到 PipeId。下一篇从 CFE_SB_TransmitMsg() 开始，看这几项怎样真正被使用。

代码对应 cfe_sb_init.c、cfe_sb_api.c及 cfe_sb_priv.h。下一篇：一条消息的发送过程。