Worker Thread Based on libev

Posted on | Edited on | In | Comments:

本文将基于 libev 实现一个简单的 worker thread，支持投递任务来异步执行（有空的话会更新如何支持定时器任务已更新，详见这里）。Github 地址：worker_thread。

ev_async

ev_async 是 libev 中的一种事件 handle，可以在运行 ev_loop 之外的线程通过 ev_async_send 来唤醒事件循环，可以用作一种线程之间的 signaling 机制。

这样一来，投递任务可以先将任务塞进队列，再唤醒该线程的事件循环即可。

Main Loop

worker thread 的主循环十分简单，在构造函数初始化 ev_async handle 之后，主循环直接调用 ev_run 即可：

1
2
3
void WorkerThread::MainLoop() {
    ::ev_run(loop_, 0);
}

初始化时需绑定 handle 与我们的 worker thread 实例，并注册回调，添加进事件循环：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
WorkerThread::WorkerThread()
    : loop_(::ev_loop_new(0)),
      async_(std::make_unique<struct ev_async>()),
      running_(true) {
        ev_async_init(
            async_.get(),
            [](EV_P_ struct ev_async *w, int revents) {
                auto *wt = reinterpret_cast<WorkerThread *>(w->data);
                wt->AsyncCb();
            }
        );
        ::ev_async_start(loop_, async_.get());
        async_->data = this;
        thread_ = std::make_unique<std::thread>(&WorkerThread::MainLoop, this);
        NotifyLoop();
    }

Async Handler Callback

在回调函数中我们需要查看当前队列是否还有任务，若有则执行，否则结束进入下一轮事件循环。在这里需要用锁来保护任务队列，因为其他线程 push task 的时候也会修改该队列：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
void WorkerThread::AsyncCb() {
    std::deque<Task> burst_;
    while (true) {
        {
            std::lock_guard<std::mutex> _{lock_};
            if (!running_) {
                ::ev_async_stop(loop_, async_.get());
            }
            if (tasks_.empty()) {
                break;
            }
            burst_.swap(tasks_);
        }
        while (!burst_.empty()) {
            auto task = std::move(burst_.front());
            burst_.pop_front();
            task();
        }
    }
}

Push Tasks

这里利用了 packaged_task ，用户可以通过返回的 future 来获取任务的返回值，或用来等待任务结束。在将任务塞进队列后需要调用 ev_async_send 通知事件循环来检查队列：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
template<class F, class... Args>
decltype(auto) PushTask(F&& f, Args&&... args) {
    using result_type = std::invoke_result_t<F, Args...>;
    std::packaged_task<result_type(void)>
        task{std::bind(std::forward<F>(f), std::forward<Args>(args)...)};

    auto result = task.get_future();
    {
        std::lock_guard<std::mutex> _{lock_};
        if (running_) {
            tasks_.emplace_back(std::move(task));
        }
    }
    NotifyLoop();

    return result;
}

Push Timers

请见本文。