在开发Linux网络程序时,通常需要维护多个定时器,如维护客户端心跳时间、检查多个数据包的超时重传等。如果采用Linux的SIGALARM信号实现,则会带来较大的系统开销,且不便于管理。
本文在应用层实现了一个基于时间堆的高性能定时器,同时考虑到定时的粒度问题,由于通过alarm系统调用设置的SIGALARM信号只能以秒为单位触发,因此需要采用其它手段实现更细粒度的定时操作,当然,这里不考虑使用多线程+sleep的实现方法,理由性能太低。
通常的做法还有采用基于升序的时间链表,但升序时间链表的插入操作效率较低,需要遍历链表。因此本实现方案使用最小堆来维护多个定时器,插入O(logn)、删除O(1)、查找O(1)的效率较高。
首先是每个定时器的定义:
class heap_timer { public: heap_timer( int ms_delay ) { gettimeofday( &expire, NULL ); expire.tv_usec += ms_delay * 1000; if ( expire.tv_usec > 1000000 ) { expire.tv_sec += expire.tv_usec / 1000000; expire.tv_usec %= 1000000; } } public: struct timeval expire; void (*cb_func)( client_data* ); client_data* user_data; ~heap_timer() { delete user_data; } };
包括一个超时时间expire、超时回调函数cb_func以及一个user_data变量,user_data用于存储与定时器相关的用户数据,用户数据可以根据不同的应用场合进行修改,这里实现的是一个智能博物馆的网关,网关接收来自zigbee协调器的用户数据,并为每个用户维护一段等待时间T,在T到来之前,同一个用户的所有数据都存放到user_data的target_list中,当T到来时,根据target_list列表选择一个适当的target并发送到ip_address,同时删除定时器(有点扯远了=。=)。总之,要实现的功能就是给每个用户维护一个定时器,定时值到来时做一些操作。
class client_data { public: client_data(char *address):target_count(0) { strcpy(ip_address,address); } private: char ip_address[32]; target target_list[64]; int target_count; ...... };
以下是时间堆的类定义,包括了一些基本的堆操作:插入、删除、扩容,还包括了定时器溢出时的操作函数tick()
class time_heap { public: time_heap( int cap = 1) throw ( std::exception ) : capacity( cap ), cur_size( 0 ) { array = new heap_timer* [capacity]; if ( ! array ) { throw std::exception(); } for( int i = 0; i < capacity; ++i ) { array[i] = NULL; } } ~time_heap() { for ( int i = 0; i < cur_size; ++i ) { delete array[i]; } delete [] array; } public: int get_cursize() { return cur_size; } void add_timer( heap_timer* timer ) throw ( std::exception ) { if( !timer ) { return; } if( cur_size >= capacity ) { resize(); } int hole = cur_size++; int parent = 0; for( ; hole > 0; hole=parent ) { parent = (hole-1)/2; if ( timercmp( &(array[parent]->expire), &(timer->expire), <= ) ) { break; } array[hole] = array[parent]; } array[hole] = timer; } void del_timer( heap_timer* timer ) { if( !timer ) { return; } // lazy delelte timer->cb_func = NULL; } int top(struct timeval &time_top) const { if ( empty() ) { return 0; } time_top = array[0]->expire; return 1; } void pop_timer() { if( empty() ) { return; } if( array[0] ) { delete array[0]; array[0] = array[--cur_size]; percolate_down( 0 ); } } void tick() { heap_timer* tmp = array[0]; struct timeval cur; gettimeofday( &cur, NULL ); while( !empty() ) { if( !tmp ) { break; } if( timercmp( &cur, &(tmp->expire), < ) ) { break; } if( array[0]->cb_func ) { array[0]->cb_func( array[0]->user_data ); } pop_timer(); tmp = array[0]; } } bool empty() const { return cur_size == 0; } heap_timer** get_heap_array() { return array; } private: void percolate_down( int hole ) { heap_timer* temp = array[hole]; int child = 0; for ( ; ((hole*2+1) <= (cur_size-1)); hole=child ) { child = hole*2+1; if ( (child < (cur_size-1)) && timercmp( &(array[child+1]->expire), &(array[child]->expire), < ) ) { ++child; } if ( timercmp( &(array[child]->expire), &(temp->expire), < ) ) { array[hole] = array[child]; } else { break; } } array[hole] = temp; } void resize() throw ( std::exception ) { heap_timer** temp = new heap_timer* [2*capacity]; for( int i = 0; i < 2*capacity; ++i ) { temp[i] = NULL; } if ( ! temp ) { throw std::exception(); } capacity = 2*capacity; for ( int i = 0; i < cur_size; ++i ) { temp[i] = array[i]; } delete [] array; array = temp; } private: heap_timer** array; int capacity; int cur_size; };
如何用epoll实现多个定时器的操作是本设计的关键,我们知道,epoll_wait的最后一个参数是阻塞等待的时候,单位是毫秒。可以这样设计:
1、当时间堆中没有定时器时,epoll_wait的超时时间T设为-1,表示一直阻塞等待新用户的到来;
2、当时间堆中有定时器时,epoll_wait的超时时间T设为最小堆堆顶的超时值,这样可以保证让最近触发的定时器能得以执行;
3、在epoll_wait阻塞等待期间,若有其它的用户到来,则epoll_wait返回n>0,进行常规的处理,随后应重新设置epoll_wait为小顶堆堆顶的超时时间。
为此,本实现对epoll_wait进行了封装,名为tepoll_wait,调用接口与epoll_wait差不多,但返回值有所不同:tepoll_wait不返回n=0的情况(即超时),因为超时事件在tepoll_wait中进行处理,只有等到n>0(即在等待过程中有用户数据到来)或者n<0(出现错误)才进行返回。
废话不多说,看代码最清楚:
void timer_handler() { heap.tick(); //setalarm(); } /* tselect - select with timers */ int tepoll_wait( int epollfd, epoll_event *events, int max_event_number ) { struct timeval now; struct timeval tv; struct timeval *tvp; //tevent_t *tp; int n; for ( ;; ) { if ( gettimeofday( &now, NULL ) < 0 ) perror("gettimeofday"); struct timeval time_top; if ( heap.top(time_top) ) { tv.tv_sec = time_top.tv_sec - now.tv_sec;; tv.tv_usec = time_top.tv_usec - now.tv_usec; if ( tv.tv_usec < 0 ) { tv.tv_usec += 1000000; tv.tv_sec--; } tvp = &tv; } else tvp = NULL; if(tvp == NULL) n = epoll_wait( epollfd, events, max_event_number, -1 ); else n = epoll_wait( epollfd, events, max_event_number, tvp->tv_sec*1000 + tvp->tv_usec/1000 ); if ( n < 0 ) return -1; if ( n > 0 ) return n; timer_handler(); } }
代码一目了然,在tepoll_wait中,是个死循环,只有等到上述两种情况发生时,才进行返回,此时在调用方进行处理,处理过程跟epoll_wait一样。
while( !stop_server ) { number = tepoll_wait( epollfd, events, MAX_EVENT_NUMBER); for ( i= 0; i < number; i++ ) { int fd = events[i].data.fd; if ( (events[i].events & EPOLLIN)&& (fd == uart_fd) ) { //读取用户数据 if( (timer_id = find_exist_timer(ip_address)) != -1) { //add to the exist timer heap_timer ** heap_array = heap.get_heap_array(); heap_array[timer_id]->user_data->add_target(RSSI,target_id); continue; } //new timer heap_timer *timer = new heap_timer(200); timer->cb_func = cb_func; timer->user_data = new client_data(ip_address); timer->user_data->add_target(RSSI,target_id); heap.add_timer(timer); } else if( ( fd == pipefd[0] ) && ( events[i].events & EPOLLIN ) ) { //此处进行了统一信号源处理,通过双向管道来获取SIGTERM以及SIGINT的信号,在主循环中进行统一处理 char signals[1024]; ret = recv( pipefd[0], signals, sizeof( signals ), 0 ); if( ret == -1 ) { continue; } else if( ret == 0 ) { continue; } else { for( int i = 0; i < ret; ++i ) { switch( signals[i] ) { case SIGTERM: case SIGINT: { stop_server = true; } } } } } } }
实测效果非常不错,欢迎分享!