如何在Python与C++模块间共享std::mutex和std::condition_variable

方案1：在Python中创建C++同步原语并传递给模块 #

PyBind11不支持直接将Python的threading.Lock/Condition转换为C的std::mutex/std::condition_variable，但你可以直接把这两个C同步原语绑定为Python可实例化的类，让Python创建后传递给各个模块。

步骤1：编写同步工具模块的C++绑定代码 #

创建一个单独的sync_utils模块，将std::mutex、std::condition_variable以及配套的std::unique_lock暴露给Python：

#include <mutex>
#include <condition_variable>
#include <pybind11/pybind11.h>

namespace py = pybind11;

PYBIND11_MODULE(sync_utils, m) {
    // 绑定std::mutex
    py::class_<std::mutex>(m, "Mutex")
        .def(py::init<>())
        .def("lock", &std::mutex::lock)
        .def("unlock", &std::mutex::unlock)
        .def("try_lock", &std::mutex::try_lock);

    // 绑定std::condition_variable
    py::class_<std::condition_variable>(m, "ConditionVariable")
        .def(py::init<>())
        .def("wait", [](std::condition_variable& cv, std::unique_lock<std::mutex>& lock) {
            cv.wait(lock);
        })
        .def("notify_one", &std::condition_variable::notify_one)
        .def("notify_all", &std::condition_variable::notify_all);

    // 绑定std::unique_lock，用于condition_variable的wait操作
    py::class_<std::unique_lock<std::mutex>>(m, "UniqueLock")
        .def(py::init<std::mutex&>());
}

步骤2：在生产者/消费者模块中添加接收同步原语的方法 #

在相机（生产者）和数据分析（消费者）的C++模块中，添加全局指针存储同步原语，并绑定设置方法：

// 生产者模块（camera_module.cpp）
#include <mutex>
#include <condition_variable>
#include <pybind11/pybind11.h>

namespace py = pybind11;

static std::mutex* g_buffer_mutex = nullptr;
static std::condition_variable* g_buffer_cv = nullptr;

// 设置同步原语的指针
void set_sync_objects(std::mutex* mutex, std::condition_variable* cv) {
    g_buffer_mutex = mutex;
    g_buffer_cv = cv;
}

// 你的Start方法（已实现线程逻辑）
void Start() {
    // 业务逻辑中使用g_buffer_mutex和g_buffer_cv
    // 比如写入缓冲区后notify_one
}

PYBIND11_MODULE(camera_module, m) {
    m.def("set_sync_objects", &set_sync_objects);
    m.def("Start", &Start);
}

消费者模块的代码结构与生产者完全一致，只需替换业务逻辑为读取缓冲区的逻辑。

步骤3：Python侧调用逻辑 #

导入模块，创建同步原语并传递给生产者和消费者：

import sync_utils
import camera_module
import analysis_module

# 创建C++同步原语实例
buffer_mutex = sync_utils.Mutex()
buffer_cv = sync_utils.ConditionVariable()

# 传递给两个模块
camera_module.set_sync_objects(buffer_mutex, buffer_cv)
analysis_module.set_sync_objects(buffer_mutex, buffer_cv)

# 启动线程
camera_module.Start()
analysis_module.Start()

方案2：在生产者模块创建同步原语，通过Python传递给消费者 #

如果希望同步原语由C++侧创建，你可以在生产者模块中初始化全局同步原语，然后将指针暴露给Python，再传递给消费者模块。

步骤1：生产者模块暴露同步原语指针 #

修改生产者模块的C++代码，添加获取同步原语指针的方法，并设置返回策略为reference（确保Python拿到的是原对象的引用，而非拷贝）：

// camera_module.cpp
#include <mutex>
#include <condition_variable>
#include <pybind11/pybind11.h>

namespace py = pybind11;

// 全局同步原语
static std::mutex g_buffer_mutex;
static std::condition_variable g_buffer_cv;

// 返回mutex指针
std::mutex* get_buffer_mutex() {
    return &g_buffer_mutex;
}

// 返回condition_variable指针
std::condition_variable* get_buffer_cv() {
    return &g_buffer_cv;
}

// 你的Start方法
void Start() {
    // 使用g_buffer_mutex和g_buffer_cv完成同步逻辑
}

PYBIND11_MODULE(camera_module, m) {
    m.def("get_buffer_mutex", &get_buffer_mutex, py::return_value_policy::reference);
    m.def("get_buffer_cv", &get_buffer_cv, py::return_value_policy::reference);
    m.def("Start", &Start);
}

步骤2：消费者模块接收同步原语指针 #

消费者模块的代码与方案1中的完全一致，只需保留set_sync_objects方法的绑定。

步骤3：Python侧传递逻辑 #

从生产者获取同步原语指针，再传递给消费者：

import camera_module
import analysis_module

# 从生产者模块获取同步原语
buffer_mutex = camera_module.get_buffer_mutex()
buffer_cv = camera_module.get_buffer_cv()

# 传递给消费者模块
analysis_module.set_sync_objects(buffer_mutex, buffer_cv)

# 启动线程
camera_module.Start()
analysis_module.Start()

关键注意事项 #

• 生命周期管理：如果使用方案1，Python会管理Mutex和ConditionVariable对象的生命周期，需确保在生产者/消费者线程运行期间，这些对象不会被Python垃圾回收（可通过将对象赋值给全局变量避免）。
• wait方法封装：如果不想在Python层操作UniqueLock，可以在C++侧封装更上层的同步接口（比如生产者的notify_buffer_ready、消费者的wait_for_buffer），简化Python调用逻辑。

内容的提问来源于stack exchange，提问作者magtweezers