如何通过pybind11实现Python asyncio与C++线程的互操作？

一、段错误的核心原因与修复方案 #

你遇到的段错误通常由以下几个原因导致：

• Python Future 对象被GC提前回收，C++端仍在操作它
• 后台线程错误地获取/操作事件循环
• call_soon_threadsafe 捕获的变量生命周期失效
• 未捕获后台线程的异常导致崩溃

正确实现代码示例 #

C++ 绑定代码

#include <pybind11/pybind11.h>
#include <pybind11/async.h>
#include <thread>
#include <chrono>
#include <memory>

namespace py = pybind11;

// 模拟C++库的耗时任务
void background_task(py::object future, int input) {
    try {
        // 模拟耗时操作
        std::this_thread::sleep_for(std::chrono::seconds(2));
        int result = input * 3;

        // 从Future获取主线程的事件循环（必须在主线程创建Future，确保loop有效）
        py::object loop = future.attr("get_loop")();
        // 用call_soon_threadsafe提交线程安全的结果设置操作
        loop.attr("call_soon_threadsafe")(
            py::cpp_function([future, result]() {
                future.attr("set_result")(result);
            })
        );
    } catch (const std::exception& e) {
        py::object loop = future.attr("get_loop")();
        loop.attr("call_soon_threadsafe")(
            py::cpp_function([future, e]() {
                py::object exc = py::value_error(e.what());
                future.attr("set_exception")(exc);
            })
        );
    }
}

// 暴露给Python的异步接口
py::object async_process(int input) {
    // 必须在调用线程（Python主线程）创建Future和获取事件循环
    py::object asyncio = py::module_::import("asyncio");
    py::object loop = asyncio.attr("get_running_loop")();
    py::object future = loop.attr("create_future")();

    // 启动后台线程，传递Future拷贝（pybind11的object是智能指针，拷贝会增加引用计数）
    std::thread(background_task, future, input).detach();

    return future;
}

PYBIND11_MODULE(cpp_async_lib, m) {
    m.def("async_process", &async_process, py::return_value_policy::move);
}

Python 测试代码

import asyncio
import cpp_async_lib

async def main():
    print("启动异步任务...")
    result = await cpp_async_lib.async_process(15)
    print(f"任务结果: {result}")

asyncio.run(main())

关键修复点 #

1. Future的生命周期管理：必须在Python主线程创建Future，返回给Python后由Python持有引用，避免被GC回收。pybind11的py::object是智能指针，拷贝会增加引用计数，传递给后台线程时要传拷贝而非引用。
2. 事件循环的获取时机：只能在调用C++函数的Python主线程获取事件循环，后台线程不能调用get_running_loop。
3. 异常捕获：后台线程的所有异常必须捕获，通过set_exception传递给Python，否则会导致崩溃。

二、基于C++ Request类的替代实现方案 #

如果你的C++库已提供Request类（含wait()阻塞等待、test()非阻塞检查状态），可以用两种方式包装为asyncio兼容的接口：

方案1：后台线程阻塞等待+线程安全通知 #

适合不想做轮询的场景，利用wait()阻塞后台线程，完成后通过事件循环通知Python：

#include <pybind11/pybind11.h>
#include <thread>
#include <memory>

namespace py = pybind11;

// 模拟C++库的Request类
class Request {
public:
    void start(int input) {
        input_ = input;
        // 启动异步任务（模拟）
        std::thread([this]() {
            std::this_thread::sleep_for(std::chrono::seconds(2));
            result_ = input_ * 2;
            completed_ = true;
        }).detach();
    }

    bool test() const { return completed_; }
    int get_result() const { return result_; }
    void wait() const {
        while (!completed_) {
            std::this_thread::yield();
        }
    }

private:
    int input_ = 0;
    int result_ = 0;
    bool completed_ = false;
};

py::object async_request(int input) {
    py::object asyncio = py::module_::import("asyncio");
    py::object loop = asyncio.attr("get_running_loop")();
    py::object future = loop.attr("create_future")();

    // 用shared_ptr持有Request，确保后台线程访问时对象存活
    auto req = std::make_shared<Request>();
    req->start(input);

    std::thread([req, future, loop]() {
        try {
            req->wait();
            int result = req->get_result();
            loop.attr("call_soon_threadsafe")(
                py::cpp_function([future, result]() {
                    future.attr("set_result")(result);
                })
            );
        } catch (const std::exception& e) {
            loop.attr("call_soon_threadsafe")(
                py::cpp_function([future, e]() {
                    py::object exc = py::value_error(e.what());
                    future.attr("set_exception")(exc);
                })
            );
        }
    }).detach();

    return future;
}

PYBIND11_MODULE(cpp_request_lib, m) {
    m.def("async_request", &async_request, py::return_value_policy::move);
}

方案2：事件循环轮询test()方法 #

适合不想阻塞后台线程的场景，利用事件循环定期检查test()状态：

py::object async_request_poll(int input) {
    py::object asyncio = py::module_::import("asyncio");
    py::object loop = asyncio.attr("get_running_loop")();
    py::object future = loop.attr("create_future")();

    auto req = std::make_shared<Request>();
    req->start(input);

    // 定义轮询函数
    auto poll = [req, future, loop]() mutable {
        if (req->test()) {
            try {
                int result = req->get_result();
                future.attr("set_result")(result);
            } catch (const std::exception& e) {
                py::object exc = py::value_error(e.what());
                future.attr("set_exception")(exc);
            }
            return;
        }
        // 10ms后再次轮询
        loop.attr("call_later")(0.01, poll);
    };

    // 启动第一轮轮询
    loop.attr("call_soon")(poll);

    return future;
}

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