共享内存技术方案：Rust与Python进程间视频帧通信

项目背景

本项目是一个视频增强项目，需要将视频帧通过管道传输给Python算法处理，处理完成后再传回进行编码和封装以生成新视频。

该项目主要涉及媒体处理，包含FFmpeg、Libx26x、Zlib和字体库，完全使用Rust编写。但这只是整个系统的一部分——图像处理算法才是核心。我需要利用该算法增强所有视频帧，之后才能进行编码和封装生成新的MP4/MOV文件。

理想方案是将所有内容（包括FFI）整合到Rust程序中，如同使用动态链接库（so/dylibs）一样。但实际情况是，算法模型由大量Python代码组成，这些代码之间存在相互依赖关系，涉及磁盘文件搜索和动态导入。这种拓扑结构使得用C++/Rust重写变得不可能。

第一次尝试：管道方案

实现方式

在第一个版本中，我们采用管道方式实现：使用stdin将所有YUV帧流式传输给Python程序，然后通过stdout管道接收处理结果。

管道方案并不简单，因为Python代码依赖许多第三方库，这些库可能会隐式地打印调试信息。我们需要设计类似MP4 Box或H.26x Annex-B格式的头部/分隔符，用于解析真正的YUV内容。

Rust端写入stdin：

let mut idx = self.frames % infsdrs_len;
let mut data = vec![];
data.extend_from_slice(STDIO_SEP.as_bytes());
data.extend_from_slice(&(self.frames as u32).to_be_bytes());
self.prosess_avframe(self.dec_frame, &mut data)?;

Rust端读取stdout：

let std_finder = memchr::memmem::Finder::new(&*STDIO_SEP);
let sep_len = STDIO_SEP.len();
loop {
  if let Ok(n_rst) = tokio_read_stdio!(&mut stdout, &mut buf, 200) {
      match n_rst {
          Ok(n) => {
              if n > 0 {
                  cache.extend_from_slice(&buf[0..n]);
                  if let Some(m) = std_finder.find(&cache) {
                      if cache.len() >= total_frame_size + m {
                      ...

对于不规则尺寸的视频，还需要额外的数据拷贝工作。

方案缺陷

管道方案虽然可用，但存在以下问题：

1. Python依赖问题：程序需要Python环境，导致部署不便捷、可移植性差，整个系统显得"混乱"
2. 内存开销大：管道缓冲区内存占用通常很大，且缺乏对异常情况的保证（如数据截断或损坏）
3. 异常处理复杂：需要设计非常健壮的模式来处理所有异常情况

第二次尝试：共享内存方案

问题解决思路

1. Python部署问题

尝试了多种方案，从最初的miniforge到uv工具。最终发现pyinstaller可以将算法模型打包为可执行文件。经过一些调整（将Python代码中的相对路径改为绝对路径），我们成功生成了可用的可执行文件。虽然不完美，但大大简化了部署复杂度。

2. 管道内存问题

为了避免大内存占用，考虑过使用信号量（semaphore），但面临两个问题：

• 必须跨平台，因此不能仅依赖POSIX
• 算法可能需要多个worker（如3个），需要创建至少3个信号量，增加了管理复杂度

最终采用raw_sync crate，创建超快速、低内存占用的共享内存互斥锁/读写锁标志。

3. 异常处理简化

如果解决上述两个问题，异常处理将简化为仅处理原子标志、共享内存文件处理等。相比之前的纯Python + 权重搜索 + 管道设计，使用超时和标志设计来报告异常算法worker要简单得多。

共享内存架构

这次采用了shared_memory和raw_sync。raw_sync提供了互斥锁包装：POSIX平台基于pthread实现，Windows平台使用winapi::um::synchapi。

使用shared_memory创建文件时，该crate会将文件内容映射到内存中：

                                   共享 mmap() 文件
     ┌────────────────────────────────────────────────────────────────────────────┐
     │                                                                            │
     │  +----------------+----------------------------------------------------+   │
     │  | 0 ~ 7          | AtomicU8 标志 (8 字节 / 缓存行对齐)                |   │
     │  +----------------+----------------------------------------------------+   │
     │  | 8 ~ xxx        | pthread_mutex_t (PTHREAD_PROCESS_SHARED)           |   │
     │  +----------------+----------------------------------------------------+   │
     │  | 输入缓冲区     |                                                    |   │
     │  | (Rust → Python)|   原始视频帧 / 元数据                             |   │
     │  |                |                                                    |   │
     │  +----------------+----------------------------------------------------+   │
     │  | 输出缓冲区     |                                                    |   │
     │  | (Python→ Rust) |   检测结果 / AI推理结果 / 处理后帧                |   │
     │  |                |                                                    |   │
     │  +----------------+----------------------------------------------------+   │
     │                                                                            │
     └────────────────────────────────────────────────────────────────────────────┘
              ▲                                                    ▲
              │                                                    │
              │                                                    │
      写入输入帧                                         写入AI处理结果
              │                                                    │
              │                                                    │
              │                                                    │
  ┌──────────────────────────┐                        ┌──────────────────────────┐
  │      Rust 进程           │                        │   Pyo3/Python 进程       │
  │──────────────────────────│                        │──────────────────────────│
  │ 硬件解码                 │                        │ AI / 计算机视觉算法      │
  │ 硬件视频处理             │                        │ NumPy / Torch / OpenCV   │
  │ WebRTC / 编码器          │                        │ CPU/GPU 推理             │
  └──────────────────────────┘                        └──────────────────────────┘

同步机制对比

在实现过程中发现，原子标志和pthread互斥锁功能有所重叠：

1. AtomicU8标志（自旋锁）

这是一个基于共享内存的标志，为了在Python端使用，我们构建了一个Rust的Python wheel项目并安装，提供Python SDK来操作该标志。双端都可使用该标志，这是一种进程安全的模式。

有趣的是，如果使用该标志作为互斥锁，每次循环约需10ns，相当于快速响应的自旋锁：

while flag.load(Ordering::Relaxed) != CODE_CONSUMED {
    std::hint::spin_loop();
    if write_start.elapsed().as_secs() > timeout {
        bail!("Host timeout waiting for consumer flag CODE_CONSUMED")
    }
}

2. pthread互斥锁（系统调用）

使用互斥锁同步缓冲区会引入系统调用。在POSIX上是pthread_mutex或futex。由于涉及内核操作，用户态/内核态上下文切换会带来性能开销：

let lock = self.lock
  .try_lock(raw_sync::Timeout::Val(Duration::from_secs(timeout)))?;

性能对比结果

实测表明，原子标志的性能并不优于互斥锁，但有一项重要权衡：

方案选择建议

根据不同的资源条件，选择合适方案：

1. CPU核心充足，内存充足：使用AtomicU8共享内存可获得最佳性能
2. CPU核心有限，内存充足：可考虑管道（stdin/stdout）方案
3. 需要稳定可靠的方案，内存有限：考虑使用futex方式的互斥锁