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;
}