C++ 高并发异步定时器的实现

/*** poller响应timerfd的到时事件，并处理所有到时的定时任务 ***/static void __poller_handle_timeout(const struct __poller_node *time_node, poller_t *poller)                           {                                                                                   ...
    // 锁里，把list与rbtree上时间已到的节点都从数据结构里删除，临时放到一个局部变量上                                            list_for_each_safe(pos, tmp, &poller->timeo_list)                               {       ...       node->removed = 1; // 标志位：【removed】       ...    }
    if (poller->tree_first)                                                         { ... }  
    // 锁外，设置state和error，并回调Task的handle()函数    while (!list_empty(&timeo_list))                                                {                                                                                   node = list_entry(timeo_list.next, struct __poller_node, list);                 list_del(&node->list);                                                  
        node->error = ETIMEDOUT;                                                        node->state = PR_ST_ERROR;                                                      free(node->res);                                                                poller->cb((struct poller_result *)node, poller->ctx);                      }                                                                           } 

/*** 取消一个定时任务时，从poller删除它 ***/int poller_del_timer(void *timer, poller_t *poller){    ...    // 锁内：如果这个标志位还是0，表示stop还没把它拿走，这里就可以去删除这个timer    if (!node->removed)    {        node->removed = 1; // 可以让cancel和stop互斥，保证只调一次
        if (node->in_rbtree) // 从定时器的数据结构中删掉            __poller_tree_erase(node, poller);        else            list_del(&node->list);        node->error = 0;                                                                node->state = PR_ST_DELETED;                                                        stopped = poller->stopped;                                                          if (!stopped) // 标志位【stopped】，如果当时没有进程退出，把timer事件交出去处理            write(poller->pipe_wr, &node, sizeof (void *));    }    ...
    // 锁外：标志位【stopped】如果进程要退出了，立刻处理timer事件的handle    if (stopped)        poller->cb((struct poller_result *)node, poller->ctx);
    return -!node;}

/*** 处理定时器到期，由poller调用 ***/void __WFNamedTimerTask::handle(int state, int error)    {        if (node_.task) // 由于不想先加锁再处理，所以先判断任务没有被cancel处理过    {        std::lock_guard<std::mutex> lock(__timer_map.mutex_);            if (node_.task) // 锁内再检查一下，入门技能        {            timers_->del(&node_, &__timer_map.root_); // 从map中删除            node_.task = NULL; // 标志位：表示任务生命周期结束了        }        }
    mutex_.lock();   // 这里是为了等待dispatch流程保证exchange已经结束，    mutex_.unlock(); // 否则资源被释放就不能访问成员变量了。也是非常常用的技巧！    this->__WFTimerTask::handle(state, error); // 里边会释放资源，并发起任务流下一个任务}

void __NamedTimerMap::cancel(const std::string& name, size_t max)               {    ...    // 锁内，拿出命名为name的timer队列    timers = __get_object_list<TimerList>(name, &root_, false);                     if (timers)    {        do        {            if (max == 0) // 从map中删除最多max个                return;
            // 从该名字对应的队列中删除该timer            node = list_entry(timers->head.next, struct __timer_node, list);                list_del(&node->list);
            // 标识位：exchange。如果是第二次exchange，会调到task自身的cancel()从poller中删掉它            if (node->task->flag_.exchange(true))                node->task->cancel();
            // 标志位：表示生命周期正确结束，资源已经被回收，否则timeout流程或析构函数需要做回收            node->task = NULL;                                                              max--;                                                                      } while (!timers->empty());                                             
        rb_erase(&timers->rb, &root_);        delete timers;    }}

/*** 由任务流发起、或者用户手动start起来 ***/void __WFNamedTimerTask::dispatch(){        int ret;    
    mutex_.lock();        ret = this->scheduler->sleep(this); // 先把定时任务交给poller    if (ret >= 0 && flag_.exchange(true)) // exchange一下。如果是第二个调用exchange的人，会拿到true        this->cancel(); // 说明发起之前已经有人cancel过了，立刻从poller中删除即可
    mutex_.unlock();        if (ret < 0)            this->handle(SS_STATE_ERROR, errno);}

/*** 命名定时任务的析构函数，异步任务需要注意处理各种情况的资源回收 ***/virtual ~__WFNamedTimerTask(){    if (node_.task) // 标志位：如果没有置空，说明任务没有发起过。需要从全局map中删掉这个timer    {         std::lock_guard<std::mutex> lock(__timer_map.mutex_);         if (node_.task) // 锁内再检查一次             timers_->del(&node_, &__timer_map.root_);    }}

int main(){    WFHttpServer server([](WFHttpTask * task)    {        task->get_resp()->append_output_body("<html>will response after 1s</html>");        auto timer = WFTaskFactory::create_timer_task(1, 0, nullptr);        series_of(task)->push_back(timer);    }); 
    if (server.start(1412) == 0)    {           getchar();        server.stop();    }    return 0;}