从并发析构函数中停止 boost::asio::io_service::run()

Question

谁能解释一下为什么这个程序没有终止（见 cmets）？

#include <boost/asio/io_service.hpp>
#include <boost/asio.hpp>
#include <memory>
#include <cstdio>
#include <iostream>
#include <future>

class Service {
public:
    ~Service() {
        std::cout << "Destroying...\n";
        io_service.post([this]() {
            std::cout << "clean and stop\n"; // does not get called
            // do some cleanup
            // ...
            io_service.stop();
            std::cout << "Bye!\n";
        });
        std::cout << "...destroyed\n"; // last printed line, blocks
    }

    void operator()() {
        io_service.run();
        std::cout << "run completed\n";
    }

private:
    boost::asio::io_service io_service;
    boost::asio::io_service::work work{io_service};
};

struct Test {
    void start() {
        f = std::async(std::launch::async, [this]() { service(); std::cout << "exiting thread\n";});
    }
    std::future<void> f;
    Service service;
};

int main(int argc, char* argv[]) {
    {
        Test test;
        test.start();

        std::string exit;
        std::cin >> exit;
    }

    std::cout << "exiting program\n"; // never printed
}

Answer 1

真正的问题是io_service 的销毁（显然）不是线程安全的。

只需重置工作并加入线程。或者，设置一个标志，以便您的 IO 操作知道正在关闭。

您的测试和服务类试图分担 IO 服务的责任，这是行不通的。这里简化了很多，合并了类并删除了未使用的未来。

Live On Coliru

诀窍是让work 对象optional<>：

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

struct Service {
    ~Service() {
        std::cout << "clean and stop\n";
        io_service.post([this]() {
            work.reset(); // let io_service run out of work
        });

        if (worker.joinable())
            worker.join();
    }

    void start() {
        assert(!worker.joinable());
        worker = std::thread([this] { io_service.run(); std::cout << "exiting thread\n";});
    }

private:
    boost::asio::io_service io_service;
    std::thread worker;
    boost::optional<boost::asio::io_service::work> work{io_service};
};

int main() {
    {
        Service test;
        test.start();

        std::cin.ignore(1024, '\n');
        std::cout << "Start shutdown\n";
    }

    std::cout << "exiting program\n"; // never printed
}

打印

Start shutdown
clean and stop
exiting thread
exiting program

Answer 2

请看这里：boost::asio hangs in resolver service destructor after throwing out of io_service::run()

我认为这里的技巧是在调用io_service.stop() 之前销毁worker（work 成员）。 IE。在这种情况下，work 可以是unique_ptr，并在停止服务之前显式调用reset()。

编辑：在我的情况下，上述内容帮助了我一段时间，ioservice::stop 并没有停止，而是在等待一些从未发生过的调度事件。

但是，我在我的机器上重现了您遇到的问题，这似乎是 ioservice 内部的竞争条件，ioservice::post() 和 ioservice 破坏代码 (shutdown_service) 之间的竞争。特别是，如果在post() 通知唤醒另一个线程之前触发了shutdown_service()，shutdown_service() 代码会从队列中删除操作（并“销毁”它而不是调用它），因此 lambda 永远不会然后调用。

目前在我看来，您需要直接在析构函数中调用 io_service.stop()，而不是通过 post() 推迟，因为由于比赛，这显然在这里不起作用。

Answer 3

我能够通过像这样重写您的代码来解决问题：

class Service {
public:
    ~Service() {
        std::cout << "Destroying...\n";
        work.reset();
        std::cout << "...destroyed\n"; // last printed line, blocks
    }

    void operator()() {
        io_service.run();
        std::cout << "run completed\n";
    }

private:
    boost::asio::io_service io_service;
    std::unique_ptr<boost::asio::io_service::work> work = std::make_unique<boost::asio::io_service::work>(io_service);
};

但是，这在很大程度上是一种创可贴的解决方案。

问题在于您的设计理念；具体来说，选择不将执行线程的生命周期直接绑定到io_service 对象：

struct Test {
    void start() {
        f = std::async(std::launch::async, [this]() { service(); std::cout << "exiting thread\n";});
    }
    std::future<void> f; //Constructed First, deleted last
    Service service; //Constructed second, deleted first
};

在这种特殊情况下，线程将继续尝试在 io_service 对象本身的生命周期之后执行 io_service.run()。如果在服务上执行的不仅仅是基本的work 对象，您很快就会开始通过调用已删除对象的成员函数来处理未定义的行为。

你可以颠倒Test中成员对象的顺序：

struct Test {
    void start() {
        f = std::async(std::launch::async, [this]() { service(); std::cout << "exiting thread\n";});
    }
    Service service;
    std::future<void> f;
};

但它仍然代表着一个重大的设计缺陷。

我通常实现任何使用io_service 的方法是将其生命周期与实际将在其上执行的线程联系起来。

class Service {
public:
    Service(size_t num_of_threads = 1) :
        work(std::make_unique<boost::asio::io_service::work>(io_service))
    {
        for (size_t thread_index = 0; thread_index < num_of_threads; thread_index++) {
            threads.emplace_back([this] {io_service.run(); });
        }
    }

    ~Service() {
        work.reset();
        for (std::thread & thread : threads) 
            thread.join();
    }
private:
    boost::asio::io_service io_service;
    std::unique_ptr<boost::asio::io_service::work> work;
    std::vector<std::thread> threads;
};

现在，如果您在这些线程中的任何一个上激活了无限循环，您仍然需要确保正确清理它们，但至少该io_service 操作的特定代码已正确清理.