4. RPMsg消息发送机制：从应用层到virtqueue的完整数据流与内存管理
RPMsg的消息发送，说白了就是一条从应用层到硬件的流水线。我刚开始接触这个协议栈时，总觉得它就是个简单的消息传递，后来在项目里踩了不少坑才明白——真正的复杂度全藏在数据流和内存管理里。
这一章，我们就沿着消息的路径走一遍。从用户空间的write()调用，一路追到virtqueue的buffer提交。每一步我都会结合项目经验，告诉你哪里容易出问题。
4.1 整体数据流概览
先看一张整体流程图，把脉络理清楚：
这条路径一共五层。每一层都有自己的职责，也有自己的坑。我们一层一层拆开看。
4.2 应用层入口：rpmsg_send() 的调用链
用户空间的程序通常通过字符设备节点（比如 /dev/rpmsg0）来发送消息。调用 write() 后，内核会走到 rpmsg_send()。
这里有个细节我特别想强调：rpmsg_send() 是同步的还是异步的？
答案是：默认是同步的。它会等待buffer被放入virtqueue并完成一次kick，才会返回。但如果你用了 rpmsg_send_offchannel() 的非阻塞版本，情况就不一样了。
关键点：应用层的write()调用，最终会映射到 rpmsg_send_offchannel()。这个函数会做三件事：
- 检查通道是否有效
- 分配一个rpmsg_hdr结构体
- 将用户数据拷贝到内核buffer中
我在项目中遇到过一个问题：用户层write()返回成功，但远端根本没收到消息。查了半天，发现是buffer拷贝时发生了缺页，导致数据没完整拷贝过去。嗯，这里要注意——copy_from_user() 可能会睡眠，如果你在原子上下文中调用，那就等着死锁吧。
4.3 内存管理的核心：rpmsg_hdr 与 buffer 池
RPMsg的消息格式，核心就是 rpmsg_hdr 结构体。它长这样：
struct rpmsg_hdr {
__u32 src; // 源地址
__u32 dst; // 目的地址
__u32 len; // 数据长度
__u16 flags; // 标志位
__u16 reserved; // 保留
// 后面紧跟着实际数据
} __packed;这个结构体只有16字节。但真正麻烦的是它后面的数据buffer怎么管理。
RPMsg使用了一种预分配buffer池的机制。说白了，就是在初始化时一次性申请一大块内存，切成固定大小的buffer。每个buffer的大小通常是 RPMSG_BUF_SIZE（一般是512字节）。
| 参数 | 典型值 | 说明 |
|---|---|---|
| RPMSG_BUF_SIZE | 512 | 每个buffer的大小，包含hdr |
| RPMSG_NUM_BUFS | 512 | buffer池中的buffer数量 |
| 总内存占用 | 512 × 512 = 256KB | 每个buffer池的总大小 |
为什么要用预分配？我个人的理解是：避免运行时动态分配带来的不确定性。在核间通信的场景下，你根本不知道远端什么时候会发消息过来。如果每次都要临时分配，万一内存紧张，消息就丢了。
我的经验：buffer池的大小要根据实际业务来调。我曾经在一个项目中，默认512个buffer不够用，消息频繁丢包。后来改成1024，问题就解决了。但也不能太大，否则内存浪费严重。建议通过压测来确定最优值。
4.4 virtqueue 的 buffer 提交机制
到了virtqueue这一层，事情就变得有意思了。virtqueue本质上是一个共享的环形队列，主端和远端都能访问它。
提交一个buffer到virtqueue，核心函数是 virtqueue_add_sgs()。它接受一个scatter-gather列表，把多个不连续的内存片段组合成一个消息。
你想想看，为什么需要scatter-gather？因为RPMsg的消息头和数据可能不在同一个物理页上。比如hdr在buffer池里，数据在用户空间的某个页里。virtqueue需要把它们拼在一起，一次性提交给远端。
// 伪代码：virtqueue_add_sgs() 的典型用法
struct scatterlist sg[2];
sg_init_table(sg, 2);
sg_set_buf(&sg[0], hdr, sizeof(*hdr)); // 第一个sg：消息头
sg_set_buf(&sg[1], data, len); // 第二个sg：消息体
// 提交到virtqueue
err = virtqueue_add_sgs(vq, sg, 1, 0, cookie, GFP_KERNEL);
if (err)
// 处理错误这里有个容易忽略的点：virtqueue_add_sgs() 的第三个参数是 out_num，表示输出buffer的数量。上面代码传的是1，意思是只有1个sg是输出（从主端发往远端）。但实际上，RPMsg通常需要两个sg：一个用于hdr，一个用于数据。所以正确的做法是传2。
避坑指南：我曾经在调试时发现，消息总是少发几个字节。查了两天才发现，是sg的数量传错了。如果你只传了1个sg，virtqueue只会把hdr发出去，数据根本没带上。记住：out_num 是sg的数量，不是buffer的数量。
4.5 kick 机制：通知远端取数据
buffer提交到virtqueue后，远端并不会立刻知道。你需要kick一下，也就是触发一个硬件中断，告诉远端：有消息来了，快来取。
kick的实现方式取决于底层硬件。在virtio-mmio上，就是写一个寄存器；在virtio-pci上，就是写一个MMIO地址。但不管哪种方式，核心逻辑都一样：
- 调用
virtqueue_kick() - 内部调用
virtqueue_notify() - 最终触发硬件中断
这里有个性能优化的点：批量kick。如果你一次要发很多小消息，可以先把所有buffer都提交到virtqueue，最后统一kick一次。这样可以减少中断次数，提升吞吐量。
我在项目中做过一个测试：单条消息逐个kick，吞吐量只有200MB/s；改成批量kick后，直接飙到800MB/s。差距就是这么明显。
4.6 完整数据流总结
把上面所有环节串起来，完整的发送流程就是：
| 步骤 | 函数/操作 | 关键点 |
|---|---|---|
| 1 | write() / rpmsg_send() | 用户态到内核态，copy_from_user |
| 2 | rpmsg_send_offchannel() | 检查通道，分配hdr |
| 3 | virtio_rpmsg_send() | 从buffer池取一个空闲buffer |
| 4 | virtqueue_add_sgs() | 构造sg列表，提交到virtqueue |
| 5 | virtqueue_kick() | 触发硬件中断，通知远端 |
每一步都可能出问题。我见过最多的情况是：buffer池耗尽。当远端处理速度跟不上主端的发送速度时，buffer池里的buffer会被占满，新的消息就发不出去了。这时候 rpmsg_send() 会返回 -ENOMEM。
调试建议：如果你遇到消息发送失败，先检查 /sys/kernel/debug/rpmsg/ 下的统计信息。看看 tx_dropped 和 tx_errors 的计数。如果 tx_dropped 在增长，大概率是buffer池不够用了。
好了，这一章的内容就到这里。数据流和内存管理是RPMsg的基石，搞懂了这些，后面的故障定位就会轻松很多。