基于单个io_context并行运行带超时多进程的代码修复求助

我在修改一个示例，要让进程并行运行——我的场景里进程多数时间处于空闲状态，并行运行能更高效利用CPU资源，但遇到了两个问题：

问题1：使用boost::thread_group时偶发崩溃 #

运行时会意外崩溃，错误信息如下：

a.out(17512,0x1dac25c40) malloc: *** error for object 0x600003d60000: pointer being freed was not allocated
a.out(17512,0x1dac25c40) malloc: *** set a breakpoint in malloc_error_break to debug

libc++abi: terminating due to uncaught exception of type std::__1::future_error: The associated promise has been destructed prior to the associated state becoming ready.

问题2：使用boost::asio::post到io_context时死锁 #

改用boost::asio::post(ioc, [&]() {将任务提交到io_context后，出现死锁：std::future阻塞线程，导致进程无法运行。我知道可以给io_context添加更多线程，但更希望使用可让出的异步机制而非阻塞等待。

当前代码 #

#include <boost/asio.hpp>
#include <boost/process.hpp>
#include <iostream>
#include <boost/thread.hpp>


using duration = std::chrono::system_clock::duration;
namespace asio = boost::asio;
using namespace std::chrono_literals;

std::string ExecuteProcess(boost::filesystem::path  exe,
                           std::vector<std::string> args, //
                           duration                 time, //
                           std::error_code&         ec,   //
                           asio::io_context&        ioc) {
    namespace bp = boost::process;
    std::future<std::string> data, err_output;

    auto const deadline = std::chrono::steady_clock::now() + time;

    bp::group  g;
    ec.clear();
    bp::child child(exe, args, ioc, g, bp::error(ec), bp::std_in.null(), bp::std_out > data,
                    bp::std_err > err_output);

    if (ec) {
        return {};
    }

    if (data.wait_until(deadline) == std::future_status::ready) {
        return data.get();
    }

    if (std::error_code ignore; child.running(ignore)) {
        g.terminate(ignore);
    }

    ec = make_error_code(asio::error::timed_out); // TODO FIXME
    return {};
}

int main() {
    constexpr duration              timeout = 20s;
    [[maybe_unused]] constexpr auto script1 = "/usr/bin/curl http://httpbin.org/ip -m 5";
    [[maybe_unused]] constexpr auto script2 = R"(delay="0.5"; sleep "$delay"; echo -n "sanity restored after $delay")";

    asio::io_context ioc;

    auto        work = make_work_guard(ioc); // prevent running out of work
    std::thread io_thread([&ioc] { ioc.run(); });


// Option 1 : use thread group

//    boost::thread_group worker_threads;
//    for (int i = 0; i < 20; i++) {
//      worker_threads.create_thread([&]() {
//          std::error_code ec;
//          auto s = ExecuteProcess("/bin/bash", {"-c", script2}, timeout, ec, ioc);
//          std::cout << "got " << ec.message() << ": " << s << std::endl;
//      });
//    }
//
//    work.reset(); // allow running out of work
//    io_thread.join();
//    worker_threads.join_all();



// Option 2 : use post-io_context
    for (int i = 0; i < 20; i++) {
    boost::asio::post(ioc, [&]() {
            std::error_code ec;
            auto s = ExecuteProcess("/bin/bash", {"-c", script2}, timeout, ec, ioc);
            std::cout << "got " << ec.message() << ": " << s << std::endl;
      });
    }
    work.reset(); // allow running out of work
    io_thread.join();

}

测试编译命令 #

g++ -std=c++20 -g -O3 -Wall -pedantic -pthread -lboost_{thread,coroutine,context} ~/main.cpp -I<path_to_boost_headers> -L<path_to_boost_libs>

代码修复建议 #

问题根源 #

1. thread_group崩溃：ExecuteProcess中std::future绑定的内部promise生命周期与子进程异步操作不匹配，子进程销毁时promise可能提前释放，触发future_error；多线程共享io_context时缺乏足够的工作线程，导致资源访问冲突引发malloc错误。
2. io_context死锁：单线程运行io_context时，std::future.wait_until阻塞唯一工作线程，导致io_context无法处理子进程输出完成的异步事件，形成死锁。

修复方案 #

改用异步回调替代阻塞等待，确保资源生命周期覆盖异步操作，并启动多线程运行io_context：

#include <boost/asio.hpp>
#include <boost/process.hpp>
#include <iostream>
#include <boost/thread.hpp>
#include <memory>
#include <functional>

using duration = std::chrono::system_clock::duration;
namespace asio = boost::asio;
using namespace std::chrono_literals;
namespace bp = boost::process;

void ExecuteProcessAsync(boost::filesystem::path exe,
                         std::vector<std::string> args,
                         duration timeout,
                         asio::io_context& ioc,
                         std::function<void(std::error_code, std::string)> callback) {
    // 用shared_ptr管理所有异步操作相关资源，确保生命周期覆盖整个异步流程
    auto state = std::make_shared<struct {
        bp::group g;
        bp::child child;
        std::promise<std::string> out_promise;
        asio::steady_timer timer;
    }>();

    state->timer = asio::steady_timer(ioc, timeout);
    std::error_code ec;

    // 启动子进程
    state->child = bp::child(
        std::move(exe), std::move(args), ioc, state->g, bp::error(ec),
        bp::std_in.null(), bp::std_out > state->out_promise, bp::std_err > bp::null
    );

    if (ec) {
        callback(ec, {});
        return;
    }

    // 设置超时处理
    state->timer.async_wait([state, callback](const std::error_code& timer_ec) {
        if (!timer_ec) { // 超时触发
            std::error_code ignore;
            if (state->child.running(ignore)) {
                state->g.terminate(ignore);
            }
            callback(make_error_code(asio::error::timed_out), {});
        }
    });

    // 异步等待进程输出完成
    auto out_future = state->out_promise.get_future();
    asio::post(ioc, [state, callback, out_future = std::move(out_future)]() mutable {
        try {
            std::string output = out_future.get();
            state->timer.cancel(); // 完成后取消超时定时器
            callback({}, output);
        } catch (...) {
            callback(std::make_error_code(std::errc::broken_pipe), {});
        }
    });
}

int main() {
    constexpr duration timeout = 20s;
    [[maybe_unused]] constexpr auto script1 = "/usr/bin/curl http://httpbin.org/ip -m 5";
    [[maybe_unused]] constexpr auto script2 = R"(delay="0.5"; sleep "$delay"; echo -n "sanity restored after $delay")";

    asio::io_context ioc;
    auto work = make_work_guard(ioc);

    // 启动与CPU核心数匹配的工作线程，避免单线程阻塞
    boost::thread_group threads;
    for (size_t i = 0; i < std::thread::hardware_concurrency(); ++i) {
        threads.create_thread([&ioc] { ioc.run(); });
    }

    // 提交20个异步任务
    for (int i = 0; i < 20; ++i) {
        ExecuteProcessAsync("/bin/bash", {"-c", script2}, timeout, ioc,
            [](std::error_code ec, std::string s) {
                std::cout << "got " << ec.message() << ": " << s << std::endl;
            });
    }

    work.reset();
    threads.join_all();
}

修复要点 #

• 用异步回调替代阻塞的std::future.wait_until，让io_context线程可以处理其他任务，避免死锁。
• 用std::shared_ptr管理子进程、定时器、promise等资源，确保它们在异步操作完成前不会被销毁。
• 启动多线程运行io_context，利用多核CPU，同时避免单线程被阻塞时无法处理事件。
• 进程输出完成后及时取消超时定时器，避免误终止已完成的进程。

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