最小堆定时器的实现以及与网络编程中的多路IO复用的应用

时间：2021-10-22 21:34:37

转载地址：http://blog.csdn.net/w616589292/article/details/45694987

在开发Linux网络程序时，通常需要维护多个定时器，如维护客户端心跳时间、检查多个数据包的超时重传等。如果采用Linux的SIGALARM信号实现，则会带来较大的系统开销，且不便于管理。

本文在应用层实现了一个基于时间堆的高性能定时器，同时考虑到定时的粒度问题，由于通过alarm系统调用设置的SIGALARM信号只能以秒为单位触发，因此需要采用其它手段实现更细粒度的定时操作，当然，这里不考虑使用多线程+sleep的实现方法，理由性能太低。

通常的做法还有采用基于升序的时间链表，但升序时间链表的插入操作效率较低，需要遍历链表。因此本实现方案使用最小堆来维护多个定时器，插入O(logn)、删除O(1)、查找O(1)的效率较高。

首先是每个定时器的定义：

[cpp]  view plain copy 最小堆定时器的实现以及与网络编程中的多路IO复用的应用 最小堆定时器的实现以及与网络编程中的多路IO复用的应用
  1. class heap_timer  
  2. {  
  3. public:  
  4.     heap_timer( int ms_delay )  
  5.     {  
  6.         gettimeofday( &expire, NULL );  
  7.         expire.tv_usec += ms_delay * 1000;  
  8.         if ( expire.tv_usec > 1000000 )  
  9.         {  
  10.             expire.tv_sec += expire.tv_usec / 1000000;  
  11.             expire.tv_usec %= 1000000;  
  12.         }  
  13.     }  
  14.   
  15. public:  
  16.     struct timeval expire;  
  17.     void (*cb_func)( client_data* );  
  18.     client_data* user_data;  
  19.     ~heap_timer()  
  20.     {  
  21.         delete user_data;  
  22.     }  
  23. };  

包括一个超时时间expire、超时回调函数cb_func以及一个user_data变量，user_data用于存储与定时器相关的用户数据，用户数据可以根据不同的应用场合进行修改，这里实现的是一个智能博物馆的网关，网关接收来自zigbee协调器的用户数据，并为每个用户维护一段等待时间T，在T到来之前，同一个用户的所有数据都存放到user_data的target_list中，当T到来时，根据target_list列表选择一个适当的target并发送到ip_address，同时删除定时器（有点扯远了=。=）。总之，要实现的功能就是给每个用户维护一个定时器，定时值到来时做一些操作。

[cpp]  view plain copy 最小堆定时器的实现以及与网络编程中的多路IO复用的应用 最小堆定时器的实现以及与网络编程中的多路IO复用的应用
  1. class client_data  
  2. {  
  3. public:  
  4.     client_data(char *address):target_count(0)  
  5.     {  
  6.         strcpy(ip_address,address);  
  7.     }  
  8. private:  
  9.     char ip_address[32];  
  10.     target target_list[64];  
  11.     int target_count;  
  12.     ......  
  13. };  

以下是时间堆的类定义，包括了一些基本的堆操作：插入、删除、扩容，还包括了定时器溢出时的操作函数tick()

[cpp]  view plain copy 最小堆定时器的实现以及与网络编程中的多路IO复用的应用 最小堆定时器的实现以及与网络编程中的多路IO复用的应用
  1. class time_heap  
  2. {  
  3. public:  
  4.     time_heap( int cap  = 1) throw ( std::exception )  
  5.         : capacity( cap ), cur_size( 0 )  
  6.     {  
  7.         array = new heap_timer* [capacity];  
  8.         if ( ! array )  
  9.         {  
  10.             throw std::exception();  
  11.         }  
  12.         for( int i = 0; i < capacity; ++i )  
  13.         {  
  14.             array[i] = NULL;  
  15.         }  
  16.     }  
  17.   
  18.     ~time_heap()  
  19.     {  
  20.         for ( int i =  0; i < cur_size; ++i )  
  21.         {  
  22.             delete array[i];  
  23.         }  
  24.         delete [] array;  
  25.     }  
  26.   
  27. public:  
  28.     int get_cursize()  
  29.     {  
  30.         return cur_size;  
  31.     }  
  32.   
  33.     void add_timer( heap_timer* timer ) throw ( std::exception )  
  34.     {  
  35.         if( !timer )  
  36.         {  
  37.             return;  
  38.         }  
  39.         if( cur_size >= capacity )  
  40.         {  
  41.             resize();  
  42.         }  
  43.         int hole = cur_size++;  
  44.         int parent = 0;  
  45.         for( ; hole > 0; hole=parent )  
  46.         {  
  47.             parent = (hole-1)/2;  
  48.             if ( timercmp( &(array[parent]->expire), &(timer->expire), <= ) )  
  49.             {  
  50.                 break;  
  51.             }  
  52.             array[hole] = array[parent];  
  53.         }  
  54.         array[hole] = timer;  
  55.     }  
  56.     void del_timer( heap_timer* timer )  
  57.     {  
  58.         if( !timer )  
  59.         {  
  60.             return;  
  61.         }  
  62.         // lazy delelte  
  63.         timer->cb_func = NULL;  
  64.     }  
  65.     int top(struct timeval &time_top) const  
  66.     {  
  67.         if ( empty() )  
  68.         {  
  69.             return 0;  
  70.         }  
  71.         time_top = array[0]->expire;  
  72.         return 1;  
  73.     }  
  74.     void pop_timer()  
  75.     {  
  76.         if( empty() )  
  77.         {  
  78.             return;  
  79.         }  
  80.         if( array[0] )  
  81.         {  
  82.             delete array[0];  
  83.             array[0] = array[--cur_size];  
  84.             percolate_down( 0 );  
  85.         }  
  86.     }  
  87.     void tick()  
  88.     {  
  89.         heap_timer* tmp = array[0];  
  90.         struct timeval cur;  
  91.         gettimeofday( &cur, NULL );  
  92.         while( !empty() )  
  93.         {  
  94.             if( !tmp )  
  95.             {  
  96.                 break;  
  97.             }  
  98.             if( timercmp( &cur, &(tmp->expire), < ) )  
  99.             {  
  100.                 break;  
  101.             }  
  102.             if( array[0]->cb_func )  
  103.             {  
  104.                 array[0]->cb_func( array[0]->user_data );  
  105.             }  
  106.             pop_timer();  
  107.             tmp = array[0];  
  108.         }  
  109.     }  
  110.     bool empty() const  
  111.     {  
  112.         return cur_size == 0;  
  113.     }  
  114.     heap_timer** get_heap_array()  
  115.     {  
  116.         return array;  
  117.     }  
  118.   
  119. private:  
  120.     void percolate_down( int hole )  
  121.     {  
  122.         heap_timer* temp = array[hole];  
  123.         int child = 0;  
  124.         for ( ; ((hole*2+1) <= (cur_size-1)); hole=child )  
  125.         {  
  126.             child = hole*2+1;  
  127.             if ( (child < (cur_size-1)) && timercmp( &(array[child+1]->expire), &(array[child]->expire), < ) )  
  128.             {  
  129.                 ++child;  
  130.             }  
  131.             if ( timercmp( &(array[child]->expire), &(temp->expire), < ) )  
  132.             {  
  133.                 array[hole] = array[child];  
  134.             }  
  135.             else  
  136.             {  
  137.                 break;  
  138.             }  
  139.         }  
  140.         array[hole] = temp;  
  141.     }  
  142.     void resize() throw ( std::exception )  
  143.     {  
  144.         heap_timer** temp = new heap_timer* [2*capacity];  
  145.         for( int i = 0; i < 2*capacity; ++i )  
  146.         {  
  147.             temp[i] = NULL;  
  148.         }  
  149.         if ( ! temp )  
  150.         {  
  151.             throw std::exception();  
  152.         }  
  153.         capacity = 2*capacity;  
  154.         for ( int i = 0; i < cur_size; ++i )  
  155.         {  
  156.             temp[i] = array[i];  
  157.         }  
  158.         delete [] array;  
  159.         array = temp;  
  160.     }  
  161.   
  162.   
  163. private:  
  164.     heap_timer** array;  
  165.     int capacity;  
  166.     int cur_size;  
  167. };  

如何用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（出现错误）才进行返回。

废话不多说，看代码最清楚：

[cpp]  view plain copy 最小堆定时器的实现以及与网络编程中的多路IO复用的应用 最小堆定时器的实现以及与网络编程中的多路IO复用的应用
  1. void timer_handler()  
  2. {  
  3.     heap.tick();  
  4.     //setalarm();  
  5. }  
  6.   
  7. /* tselect - select with timers */  
  8. int tepoll_wait( int epollfd, epoll_event *events, int max_event_number )  
  9. {  
  10.     struct timeval now;  
  11.     struct timeval tv;  
  12.     struct timeval *tvp;  
  13.     //tevent_t *tp;  
  14.     int n;  
  15.   
  16.     for ( ;; )  
  17.     {  
  18.         if ( gettimeofday( &now, NULL ) < 0 )  
  19.             perror("gettimeofday");  
  20.         struct timeval time_top;  
  21.         if ( heap.top(time_top) )  
  22.         {  
  23.             tv.tv_sec = time_top.tv_sec - now.tv_sec;;  
  24.             tv.tv_usec = time_top.tv_usec - now.tv_usec;  
  25.             if ( tv.tv_usec < 0 )  
  26.             {  
  27.                 tv.tv_usec += 1000000;  
  28.                 tv.tv_sec--;  
  29.             }  
  30.             tvp = &tv;  
  31.         }  
  32.         else  
  33.             tvp = NULL;  
  34.   
  35.         if(tvp == NULL)  
  36.             n = epoll_wait( epollfd, events, max_event_number, -1 );  
  37.         else  
  38.             n = epoll_wait( epollfd, events, max_event_number, tvp->tv_sec*1000 + tvp->tv_usec/1000 );  
  39.         if ( n < 0 )  
  40.             return -1;  
  41.         if ( n > 0 )  
  42.             return n;  
  43.   
  44.         timer_handler();  
  45.     }  
  46. }  
代码一目了然，在tepoll_wait中，是个死循环，只有等到上述两种情况发生时，才进行返回，此时在调用方进行处理，处理过程跟epoll_wait一样。

[cpp]  view plain copy 最小堆定时器的实现以及与网络编程中的多路IO复用的应用 最小堆定时器的实现以及与网络编程中的多路IO复用的应用
  1. while( !stop_server )  
  2.     {  
  3.         number = tepoll_wait( epollfd, events, MAX_EVENT_NUMBER);  
  4.         for ( i= 0; i < number; i++ )  
  5.         {  
  6.             int fd = events[i].data.fd;  
  7.             if ( (events[i].events & EPOLLIN)&& (fd == uart_fd) )  
  8.             {  
  9.                //读取用户数据  
  10.                 if( (timer_id = find_exist_timer(ip_address)) != -1)  
  11.                 {  
  12.                     //add to the exist timer  
  13.                     heap_timer ** heap_array = heap.get_heap_array();  
  14.                     heap_array[timer_id]->user_data->add_target(RSSI,target_id);  
  15.                     continue;  
  16.                 }  
  17. <span style="white-space:pre">      </span>//new timer  
  18.                 heap_timer *timer = new heap_timer(200);  
  19.                 timer->cb_func = cb_func;  
  20.                 timer->user_data = new client_data(ip_address);  
  21.                 timer->user_data->add_target(RSSI,target_id);  
  22.                 heap.add_timer(timer);  
  23.             }  
  24.             else if( ( fd == pipefd[0] ) && ( events[i].events & EPOLLIN ) )  
  25.             {  
  26.                 //此处进行了统一信号源处理，通过双向管道来获取SIGTERM以及SIGINT的信号，在主循环中进行统一处理  
  27. <span style="white-space:pre">      </span>char signals[1024];  
  28.                 ret = recv( pipefd[0], signals, sizeof( signals ), 0 );  
  29.                 if( ret == -1 )  
  30.                 {  
  31.                     continue;  
  32.                 }  
  33.                 else if( ret == 0 )  
  34.                 {  
  35.                     continue;  
  36.                 }  
  37.                 else  
  38.                 {  
  39.                     for( int i = 0; i < ret; ++i )  
  40.                     {  
  41.                         switch( signals[i] )  
  42.                         {  
  43.                         case SIGTERM:  
  44.                         case SIGINT:  
  45.                         {  
  46.                             stop_server = true;  
  47.                         }  
  48.   
  49.                         }  
  50.                     }  
  51.                 }  
  52.             }  
  53.         }  
  54.     }  
顶
1
踩
0

标签：

相关文章