boost::asio::deadline_timer::async_wait 不触发回调

Question

我io_service在一个线程中运行了一个 boost ，我想在某个客户端发生某个事件后 6 秒在该线程中触发一个回调，如果该客户端已经在运行，则重置该客户端的计时器。

我unordered_map<string, shared_ptr<deadline_timer>>为每个客户维护一个计时器。

但是，在设置时async_wait，我的回调在分配的时间量（io_serviceIS 运行）后不会触发，当我重置指针（应该调用现有计时器的析构函数，导致它）时，它也不会触发（带有错误代码）发布到服务）。我怎样才能解决这个问题？

这是我的代码的相关部分：

auto it = timersByClientId.find(clientId);
if (it == timersByClientId.end())
{
    onLogonChangeCallback(clientId, true);
    timersByClientId[clientId].reset(
        new boost::asio::deadline_timer(replyService, boost::posix_time::seconds(6))
    );
    it = timersByClientId.find(clientId);
}
else
{
    // Cancel current wait operation (should fire the callback with an error code)
    it->second.reset(
        new boost::asio::deadline_timer(replyService, boost::posix_time::seconds(6))
    );
}
it->second->async_wait([this, clientId](const boost::system::error_code& err) {
    if (!err)
    {
        onLogonChangeCallback(clientId, false);
    }
});

如果它有任何改变，我将在 Visual C++ 2010 和 boost 1.47.0 下运行。

Answer 1

你的代码/看起来/还行。

我不确定您是如何得出结论，即您的完成处理程序在我重置指针时不会“[...] 触发（带有错误代码）”。您忽略了这种情况（elselambda 中没有分支）。

把逻辑写得更清楚怎么样？

void foo(int clientId) {
    shared_timer& timer = timersByClientId[clientId];

    if (!timer) 
        onLogonChangeCallback(clientId, true);

    timer = make_timer(); // reset

    timer->async_wait([this, clientId](const boost::system::error_code& err) {
        if (!err)
            onLogonChangeCallback(clientId, false);
    });
}

这是该else分支的完整演示，让您了解发生了什么。我假设有 1 个服务线程。

在 Coliru 上看到它。

测试负载是在约 0.5 秒内对 16 个帐户进行 100 次会话活动。总运行时间约为 1.5 秒，因为我已将 Coliru 的会话过期时间从 6 秒减少到 1 秒。

如果不想让 LogonManager 的析构函数等待所有 session 过期，那么在加入后台线程之前清除 session 表：

~LogonMonitor() {
    work = boost::none;
    timersByClientId.clear();
    background.join();
}

完整列表

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

struct LogonMonitor {
    LogonMonitor() 
        : work(io_service::work(replyService)), background([this]{ replyService.run(); })
    { }

    ~LogonMonitor() {
        work = boost::none;
        // timersByClientId.clear();
        background.join();
    }

    void foo(int clientId) {
        shared_timer& timer = timersByClientId[clientId];

        if (!timer) 
            onLogonChangeCallback(clientId, true);

        timer = make_timer(); // reset

        timer->async_wait([this, clientId](const boost::system::error_code& err) {
            if (!err)
                onLogonChangeCallback(clientId, false);
            else
                std::cout << "(cancel " << clientId << " timer)" << std::endl;
        });
    }

  private:
    using io_service   = boost::asio::io_service;
    using timer        = boost::asio::deadline_timer;
    using shared_timer = boost::shared_ptr<timer>;

    io_service replyService;
    boost::optional<io_service::work> work;
    boost::thread background;

    std::map<int, shared_timer> timersByClientId;

    shared_timer make_timer() { 
        return boost::make_shared<timer>(replyService, boost::posix_time::seconds(/*6*/1)); 
    }

    void onLogonChangeCallback(int clientId, bool newLogon) 
    {
        std::cout << __FUNCTION__ << "(" << clientId << ", " << newLogon << ")" << std::endl;
    }
};

int main()
{
    LogonMonitor instance;
    for (int i = 0; i < 100; ++i)
    {
        instance.foo(rand() % 16);
        boost::this_thread::sleep_for(boost::chrono::milliseconds(rand() % 10));
    }
}