網(wǎng)絡(luò)編程定時(shí)器三：使用最小堆

2019-11-11 04:03:26

字體：大中小

供稿：網(wǎng)友

前面討論的定時(shí)方案都是以固定頻率調(diào)用心搏函數(shù)tick，并在其中一次檢測(cè)到期的定時(shí)器，然后執(zhí)行到期定時(shí)器上的回調(diào)函數(shù)。設(shè)計(jì)定時(shí)器的另一中思路是，將所有定器超時(shí)時(shí)間最小的一個(gè)定時(shí)器的超時(shí)值作為心搏間隔。這樣，一旦心搏函數(shù)tick執(zhí)行，超時(shí)時(shí)間最小的定時(shí)器必然到期。我們就可以從剩余定時(shí)器中選出超時(shí)時(shí)間最小的一個(gè)，并將這個(gè)時(shí)間設(shè)為下一次心搏間隔。如此反復(fù)，就實(shí)現(xiàn)了較為精確的定時(shí)。

最小堆適合這種解決方案，下面直接給出最小堆方案的代碼，并附有測(cè)試用例。不過(guò)測(cè)試用例我仍然只用了alarm來(lái)做測(cè)試。

#ifndef MIN_HEAP_H#define MIN_HEAP_H#include <iostream>#include <netinet/in.h>#include <time.h>#include <assert.h>#include <string.h>const int BUFFER_SIZE = 1024;class heap_timer;//綁定socket和定時(shí)器struct client_data { sockaddr_in addr_; int sockfd_; char buf_[BUFFER_SIZE]; heap_timer* timer_;};//定時(shí)器類class heap_timer {public: heap_timer(int delay) { 測(cè)試代碼：

#include <sys/types.h>#include <sys/socket.h>#include <netinet/in.h>#include <arpa/inet.h>#include <assert.h>#include <stdio.h>#include <signal.h>#include <unistd.h>#include <errno.h>#include <string.h>#include <fcntl.h>#include <stdlib.h>#include <sys/epoll.h>#include <pthread.h>#include <memory>#include <vector>#include "min_heap.h"const int FD_LIMIT = 65535;const int MAX_EVENT_NUMBER = 1024;const int TIME_SLOT = 5;const int INIT_HEAP_SIZE = 2;static int p ipefd[2];static int epollfd = 0;static std::shared_ptr<time_heap> timer_lst;int setnonblocking( int fd ){ int old_option = fcntl( fd, F_GETFL ); int new_option = old_option | O_NONBLOCK; fcntl( fd, F_SETFL, new_option ); return old_option;}void addfd(int fd ){ epoll_event event; event.data.fd = fd; event.events = EPOLLIN | EPOLLET; epoll_ctl( epollfd, EPOLL_CTL_ADD, fd, &event ); setnonblocking( fd );}void sig_handler( int sig ){ int save_errno = errno; int msg = sig; send( pipefd[1], ( char* )&msg, 1, 0 ); errno = save_errno;}void addsig( int sig ){ struct sigaction sa; memset( &sa, '/0', sizeof( sa ) ); sa.sa_handler = sig_handler; sa.sa_flags |= SA_RESTART; sigfillset( &sa.sa_mask ); assert( sigaction( sig, &sa, NULL ) != -1 );}void timer_handler(){ timer_lst->tick(); alarm( TIME_SLOT );}void cb_func( client_data* user_data ){ epoll_ctl( epollfd, EPOLL_CTL_DEL, user_data->sockfd_, 0 ); assert( user_data ); close( user_data->sockfd_ ); printf( "close fd %d/n", user_data->sockfd_ );}int main( int argc, char* argv[] ){ if( argc <= 2 ) { printf( "usage: %s ip_address port_number/n", basename( argv[0] ) ); return 1; } const char* ip = argv[1]; int port = atoi( argv[2] ); int ret = 0; struct sockaddr_in address; bzero( &address, sizeof( address ) ); address.sin_family = AF_INET; inet_pton( AF_INET, ip, &address.sin_addr ); address.sin_port = htons( port ); int listenfd = socket( PF_INET, SOCK_STREAM, 0 ); assert( listenfd >= 0 ); int on = 1; ret = setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)); assert(ret != -1); ret = bind( listenfd, ( struct sockaddr* )&address, sizeof( address ) ); assert( ret != -1 ); ret = listen( listenfd, 5 ); assert( ret != -1 ); epoll_event events[ MAX_EVENT_NUMBER ]; epollfd = epoll_create( 5 ); assert( epollfd != -1 ); addfd(listenfd ); ret = socketpair( PF_UNIX, SOCK_STREAM, 0, pipefd ); assert( ret != -1 ); setnonblocking( pipefd[1] ); addfd(pipefd[0] ); // add all the interesting signals here addsig( SIGALRM ); addsig( SIGTERM ); bool stop_server = false; std::vector<client_data> users(FD_LIMIT); bool timeout = false; alarm( TIME_SLOT ); ////////////////////////////////////////////////////// timer_lst.reset(new time_heap(INIT_HEAP_SIZE)); ////////////////////////////////////////////////////// while( !stop_server ) { int number = epoll_wait( epollfd, events, MAX_EVENT_NUMBER, -1 ); if ( ( number < 0 ) && ( errno != EINTR ) ) { printf( "epoll failure/n" ); break; } for ( int i = 0; i < number; i++ ) { int sockfd = events[i].data.fd; if( sockfd == listenfd ) { struct sockaddr_in client_address; socklen_t client_addrlength = sizeof( client_address ); int connfd = accept( listenfd, ( struct sockaddr* )&client_address, &client_addrlength ); addfd(connfd); users[connfd].addr_ = client_address; users[connfd].sockfd_ = connfd; printf("user: %d/n", connfd); heap_timer *timer = new heap_timer(3 * TIME_SLOT); timer_lst->add_timer(timer); timer->user_data_ = &users[connfd]; timer->timeout_callback_ = cb_func; users[connfd].timer_ = timer; } else if( ( sockfd == pipefd[0] ) && ( events[i].events & EPOLLIN ) ) { int sig; char signals[1024]; ret = recv( pipefd[0], signals, sizeof( signals ), 0 ); if( ret == -1 ) { // handle the error continue; } else if( ret == 0 ) { continue; } else { for( int i = 0; i < ret; ++i ) { switch( signals[i] ) { case SIGALRM: { timeout = true; break; } case SIGTERM: { stop_server = true; } } } } } else if( events[i].events & EPOLLIN ) { memset( users[sockfd].buf_, '/0', BUFFER_SIZE ); ret = recv( sockfd, users[sockfd].buf_, BUFFER_SIZE-1, 0 ); printf( "get %d bytes of client data %s from %d/n", ret, users[sockfd].buf_, sockfd ); heap_timer* timer = users[sockfd].timer_; if( ret < 0 ) { if( errno != EAGAIN ) { cb_func( &users[sockfd] ); if( timer ) { timer_lst->del_timer( timer ); } } } else if( ret == 0 ) { cb_func( &users[sockfd] ); if( timer ) { timer_lst->del_timer( timer ); } } else { //send( sockfd, users[sockfd].buf, BUFFER_SIZE-1, 0 ); if( timer ) { printf("user: %d/n", sockfd); heap_timer* new_timer = new heap_timer(3*TIME_SLOT); timer_lst->adjust_timer(timer, new_timer); new_timer->user_data_ = &users[sockfd]; new_timer->timeout_callback_ = cb_func; users[sockfd].timer_ = new_timer; } } } else { // others } } if( timeout ) { timer_handler(); timeout = false; } } close( listenfd ); close( epollfd ); close( pipefd[1] ); close( pipefd[0] ); return 0;}

對(duì)時(shí)間堆而言，添加一個(gè)定時(shí)器的時(shí)間復(fù)雜度是O(lgn)，刪除一個(gè)定時(shí)器的時(shí)間復(fù)雜度是O(1)（采用了延遲刪除），執(zhí)行一個(gè)定時(shí)器的時(shí)間復(fù)雜度是O(1)。因此，時(shí)間堆的效率是很高的。

上一篇：【GDKOI2017模擬1.21】Equation

下一篇：13.3