Linux下定时器的实现方式分析.doc

Linux 下定时器的实现方式分析定时器属于基本的基础组件，不管是用户空间的程序开发，还是内核空间的程序开发，很多时候都需要有定时器作为基础组件的支持，但使用场景的不同，对定时器的实现考虑也不尽相同，本文讨论了在Linux环境下，应用层和内核层的定时器的各种实现方法，并分析了各种实现方法的利弊以及适宜的使用环境。首先，给出一个基本模型，定时器的实现，需要具备以下几个行为，这也是在后面评判各种定时器实现的一个基本模型1：StartTimer(Interval,TimerId,ExpiryAction)注册一个时间间隔为Interval后执行ExpiryAction的定时器实例，其中，返回TimerId以区分在定时器系统中的其他定时器实例。StopTimer(TimerId)根据TimerId找到注册的定时器实例并执行Stop。PerTickBookkeeping()在一个Tick内，定时器系统需要执行的动作，它最主要的行为，就是检查定时器系统中，是否有定时器实例已经到期。注意，这里的Tick实际上已经隐含了一个时间粒度(granularity)的概念。ExpiryProcessing()在定时器实例到期之后，执行预先注册好的ExpiryAction行为。上面说了基本的定时器模型，但是针对实际的使用情况，又有以下2种基本行为的定时器：Single-Shot Timer这种定时器，从注册到终止，仅仅只执行一次。Repeating Timer这种定时器，在每次终止之后，会自动重新开始。本质上，可以认为Repeating Timer是在Single-Shot Timer终止之后，再次注册到定时器系统里的Single-Shot Timer，因此，在支持Single-Shot Timer的基础上支持Repeating Timer并不算特别的复杂。在2.4的内核中，并没有提供POSIX timer2的支持，要在进程环境中支持多个定时器，只能自己来实现，好在Linux提供了setitimer(2)的接口。它是一个具有间隔功能的定时器(interval timer)，但如果想在进程环境中支持多个计时器，不得不自己来管理所有的计时器。setitimer(2)的定义如下：清单1.setitimer的原型#include sys/time.h int setitimer(int which,const struct itimerval*new_value,struct itimerval*old_value)；setitimer能够在Timer到期之后，自动再次启动自己，因此，用它来解决Single-Shot Timer和Repeating Timer的问题显得很简单。该函数可以工作于3种模式：ITIMER_REAL以实时时间(real time)递减，在到期之后发送SIGALRM信号ITIMER_VIRTUAL仅进程在用户空间执行时递减，在到期之后发送SIGVTALRM信号ITIMER_PROF进程在用户空间执行以及内核为该进程服务时(典型如完成一个系统调用)都会递减，与ITIMER_VIRTUAL共用时可度量该应用在内核空间和用户空间的时间消耗情况，在到期之后发送SIGPROF信号定时器的值由下面的结构定义：struct itimervalstruct timeval it_interval；/*next value*/struct timeval it_value；/*current value*/；struct timevallong tv_sec；/*seconds*/long tv_usec；/*microseconds*/；setitimer()以new_value设置特定的定时器，如果old_value非空，则它返回which类型时间间隔定时器的前一个值。定时器从it_value递减到零，然后产生一个信号，并重新设置为it_interval，如果此时it_interval为零，则该定时器停止。任何时候，只要it_value设置为零，该定时器就会停止。由于setitimer()不支持在同一进程中同时使用多次以支持多个定时器，因此，如果需要同时支持多个定时实例的话，需要由实现者来管理所有的实例。用setitimer()和链表，可以构造一个在进程环境下支持多个定时器实例的Timer，在一般的实现中的PerTickBookkeeping时，会递增每个定时器的elapse值，直到该值递增到最初设定的interval则表示定时器到期。基于链表实现的定时器可以定义为：typedef int timer_id；/*The type of callback function to be called by timer scheduler when atimer*has expired.*param id The timer id.*param user_data The user data.*$param len The length of user data.*/typedef int timer_expiry(timer_id id,void*user_data,int len)；/*The type of the timer*/struct timerLIST_ENTRY(timer)entries；/*list entry*/timer_id id；/*timer id*/int interval；/*timer interval(second)*/int elapse；/*0-interval*/timer_expiry*cb；/*call if expiry*/void*user_data；/*callback arg*/int len；/*user_data length*/；定时器的时间间隔以interval表示，而elapse则在PerTickBookkeeping()时递增，直到interval表示定时器中止，此时调用回调函数cb来执行相关的行为，而user_data和len为用户可以传递给回调函数的参数。所有的定时器实例以链表来管理：/*The timer list*/struct timer_listLIST_HEAD(listheader,timer)header；/*list header*/int num；/*timer entry number*/int max_num；/*max entry number*/void(*old_sigfunc)(int)；/*save previous signal handler*/void(*new_sigfunc)(int)；/*our signal handler*/struct itimerval ovalue；/*old timer value*/struct itimerval value；/*our internal timer value*/；这里关于链表的实现使用了BSD风格关于链表的一组宏，避免了再造轮子；该结构中，old_sigfunc在init_timer初始定时器链表时候用来保存系统对SIGALRM的处理函数，在定时器系统destory时用来恢复到之前的处理函数；ovalue的用途与此类似。/*Create atimer list.*param count The maximum number of timer entries to be supported initially.*return 0means ok,the other means fail.*/int init_timer(int count)int ret=0；if(count=0|count MAX_TIMER_NUM)printf(the timer max number MUST less than%d.n,MAX_TIMER_NUM)；return-1；memset(&timer_list,0,sizeof(struct timer_list)；LIST_INIT(&timer_list.header)；timer_list.max_num=count；/*Register our internal signal handler and store old signal handler*/if(timer_list.old_sigfunc=signal(SIGALRM,sig_func)=SIG_ERR)return-1；timer_list.new_sigfunc=sig_func；/*Setting our interval timer for driver our mutil-timer and store old timer value*/timer_list.value.it_value.tv_sec=TIMER_START；timer_list.value.it_value.tv_usec=0；timer_list.value.it_interval.tv_sec=TIMER_TICK；timer_list.value.it_interval.tv_usec=0；ret=setitimer(ITIMER_REAL,&timer_list.value,&timer_list.ovalue)；return ret；/*Destroy the timer list.*return 0means ok,the other means fail.*/int destroy_timer(void)struct timer*node=NULL；if(signal(SIGALRM,timer_list.old_sigfunc)=SIG_ERR)return-1；if(setitimer(ITIMER_REAL,&timer_list.ovalue,&timer_list.value)0)return-1；while(！LIST_EMPTY(&timer_list.header)/*Delete.*/node=LIST_FIRST(&timer_list.header)；LIST_REMOVE(node,entries)；/*Free node*/printf(Remove id%dn,node-id)；free(node-user_data)；free(node)；memset(&timer_list,0,sizeof(struct timer_list)；return 0；添加定时器的动作非常的简单，本质只是一个链表的插入而已：/*Add atimer to timer list.*param interval The timer interval(second).*param cb When cb！=NULL and timer expiry,call it.*param user_data Callbacks param.*param len The length of the user_data.*return The timer ID,if=INVALID_TIMER_ID,add timer fail.*/timer_id add_timer(int interval,timer_expiry*cb,void*user_data,int len)struct timer*node=NULL；if(cb=NULL|interval=0)return INVALID_TIMER_ID；if(timer_list.num timer_list.max_num)timer_list.num+；elsereturn INVALID_TIMER_ID；if(node=malloc(sizeof(struct timer)=NULL)return INVALID_TIMER_ID；if(user_data！=NULL|len！=0)node-user_data=malloc(len)；memcpy(node-user_data,user_data,len)；node-len=len；node-cb=cb；node-interval=interval；node-elapse=0；node-id=timer_list.num；LIST_INSERT_HEAD(&timer_list.header,node,entries)；return node-id；注册的信号处理函数则用来驱动定时器系统：/*Tick Bookkeeping*/static void sig_func(int signo)struct timer*node=timer_list.header.lh_first；for(；node！=NULL；node=node-entries.le_next)node-elapse+；if(node-elapse=node-interval)node-elapse=0；node-cb(node-id,node-user_data,node-len)；它主要是在每次收到SIGALRM信号时，执行定时器链表中的每个定时器elapse的自增操作，并与interval相比较，如果相等，代表注册的定时器已经超时，这时则调用注册的回调函数。上面的实现，有很多可以优化的地方：考虑另外一种思路，在定时器系统内部将维护的相对interval转换成绝对时间，这样，在每PerTickBookkeeping时，只需将当前时间与定时器的绝对时间相比较，就可以知道是否该定时器是否到期。这种方法，把递增操作变为了比较操作。并且上面的实现方式，效率也不高，在执行StartTimer，StopTimer，PerTickBookkeeping时，算法复杂度分别为O(1)，O(n)，O(n)，可以对上面的实现做一个简单的改进，在StartTimer时，即在添加Timer实例时，对链表进行排序，这样的改进，可以使得在执行StartTimer，StopTimer，PerTickBookkeeping时，算法复杂度分别为O(n)，O(1)，O(1)。改进后的定时器系统如下图1：图1.基于排序链表的定时器Linux自2.6开始，已经开始支持POSIX timer2所定义的定时器，它主要由下面的接口构成：#include signal.h#include time.h int timer_create(clockid_t clockid,struct sigevent*evp,timer_t*timerid)；int timer_settime(timer_t timerid,int flags,const struct itimerspec*new_value,struct itimerspec*old_value)；int timer_gettime(timer_t timerid,struct itimerspec*curr_value)；int timer_getoverrun(timer_t timerid)；int timer_delete(timer_t timerid)；这套接口是为了让操作系统对实时有更好的支持，在链接时需要指定-lrt。timer_create(2)：创建了一个定时器。timer_settime(2)：启动或者停止一个定时器。timer_gettime(2)：返回到下一次到期的剩余时间值和定时器定义的时间间隔。出现该接口的原因是，如果用户定义了一个1ms的定时器，可能当时系统负荷很重，导致该定时器实际山10ms后才超时，这种情况下，overrun=9ms。timer_getoverrun(2)：返回上次定时器到期时超限值。timer_delete(2)：停止并删除一个定时器。上面最重要的接口是timer_create(2)，其中，clockid表明了要使用的时钟类型，在POSIX中要求必须实现CLOCK_REALTIME类型的时钟。evp参数指明了在定时到期后，调用者被通知的方式。该结构体定义如下：union sigvalint sival_int；void*sival_ptr；struct sigeventint sigev_notify；/*Notification method*/int sigev_signo；/*Timer expiration signal*/union sigval sigev_value；/*Value accompanying signal or passed to thread function*/void(*sigev_notify_function)(union sigval)；/*Function used for thread notifications(SIGEV_THREAD)*/void*sigev_notify_attributes；/*Attributes for notification thread(SIGEV_THREAD)*/pid_t sigev_notify_thread_id；/*ID of thread to signal(SIGEV_THREAD_ID)*/；其中，sigev_notify指明了通知的方式：SIGEV_NONE当定时器到期时，不发送异步通知，但该定时器的运行进度可以使用timer_gettime(2)监测。SIGEV_SIGNAL当定时器到期时，发送sigev_signo指定的信号。SIGEV_THREAD当定时器到期时，以sigev_notify_function开始一个新的线程。该函数使用sigev_value作为其参数，当sigev_notify_attributes非空，则制定该线程的属性。注意，由于Linux上线程的特殊性，这个功能实际上是由glibc和内核一起实现的。SIGEV_THREAD_ID(Linux-specific)仅推荐在实现线程库时候使用。如果evp为空的话，则该函数的行为等效于：sigev_notify=SIGEV_SIGNAL，sigev_signo=SIGVTALRM，sigev_value.sival_int=timer ID。由于POSIX timer2接口支持在一个进程中同时拥有多个定时器实例，所以在上面的基于setitimer()和链表的PerTickBookkeeping动作就交由Linux内核来维护，这大大减轻了实现定时器的负担。由于POSIX timer2接口在定时器到期时，有更多的控制能力，因此，可以使用实时信号避免信号的丢失问题，并将sigev_value.sival_int值指定为timer ID，这样，就可以将多个定时器一起管理了。需要注意的是，POSIX timer2接口只在进程环境下才有意义(fork(2)和exec(2)也需要特殊对待)，并不适合多线程环境。与此相类似的，Linux提供了基于文件描述符的相关定时器接口：#include sys/timerfd.h int timerfd_create(int clockid,int flags)；int timerfd_settime(int fd,int flags,const struct itimerspec*new_value,struct itimerspec*old_value)；int timerfd_gettime(int fd,struct itimerspec*curr_value)；这样，由于基于文件描述符，使得该接口可以支持select(2)，poll(2)等异步接口，使得定时器的实现和使用更加的方便，更重要的是，支持fork(2)，exec(2)这样多进程的语义，因此，可以用在多线程环境之中，它们的使用比POSIX timer2更加的灵活，其根本原因在于定时器的管理统一到了unix/linux基本哲学之一-一切皆文件之下。最小堆指的是满足除了根节点以外的每个节点都不小于其父节点的堆。这样，堆中的最小值就存放在根节点中，并且在以某个结点为根的子树中，各节点的值都不小于该子树根节点的值。一个最小堆的例子如下图2：图2.最小堆一个最小堆，一般支持以下几种操作：Insert(TimerHeap,Timer)：在堆中插入一个值，并保持最小堆性质，具体对应于定时器的实现，则是把定时器插入到定时器堆中。根据最小堆的插入算法分析，可以知道该操作的时间复杂度为O(lgn)。Minimum(TimerHeap)：获取最小堆的中最小值；在定时器系统中，则是返回定时器堆中最先可能终止的定时器。由于是最小堆，只需返回堆的root即可。此时的算法复杂度为O(1)。ExtractMin(TimerHeap)：在定时器到期后，执行相关的动作，它的算法复杂度为O(1)。最小堆本质上是一种最小优先级队列(min-priority queue)。定时可以作为最小优先级队列的一个应用，该优先级队列把定时器的时间间隔值转化为一个绝对时间来处理，ExtractMin操则是在所有等待的定时器中，找出最先超时的定时器。在任何时候，一个新的定时器实例都可通过Insert操作加入到定时器队列中去。在pjsip项目的基础库pjlib中，有基于最小堆实现的定时器，它主要提供了以下的几个接口：/*Create atimer heap.*/PJ_DECL(pj_status_t)pj_timer_heap_create(pj_pool_t*pool,pj_size_t count,pj_timer_heap_t*ht)；/*Destroy the timer heap.*/PJ_DECL(void)pj_timer_heap_destroy(pj_timer_heap_t*ht)；/*Initialize atimer entry.Application should call this function at least*once before scheduling the entry to the timer heap,to properly initialize*the timer entry.*/PJ_DECL(pj_timer_entry*)pj_timer_entry_init(pj_timer_entry*entry,int id,void*user_data,pj_timer_heap_callback*cb)；/*Schedule atimer entry which will expire AFTER the specified delay.*/PJ_DECL(pj_status_t)pj_timer_heap_schedule(pj_timer_heap_t*ht,pj_timer_entry*entry,const pj_time_val*delay)；/*Cancel apreviously registered timer.*/PJ_DECL(int)pj_timer_heap_cancel(pj_timer_heap_t*ht,pj_timer_entry*entry)；/*Poll the timer heap,check for expired timers and call the callback for*each of the expired timers.*/PJ_DECL(unsigned)pj_timer_heap_poll(pj_timer_heap_t*ht,pj_time_val*next_delay)；pjlib中的定时器在内部使用数组的方式实现堆，这样对于内存空间的使用将更加的紧凑；它的实现还可在定时器的数量超过预先设定的最大数量时会自己增加最大定时器数量。文件pjlib/src/pjlib-test/timer.c是它的一个单元测试。与基于链表方式的实现相比较，明显它的时间复杂度要低一些，这样可以支持更多的定时器实例。时间轮(Timing-Wheel)算法类似于一以恒定速度旋转的左轮手枪，枪的撞针则撞击枪膛，如果枪膛中有子弹，则会被击发；与之相对应的是：对于PerTickBookkeeping，其最本质的工作在于以Tick为单位增加时钟，如果发现有任何定时器到期，则调用相应的ExpiryProcessing。设定一个循环为N个Tick单元，当前时间是在S个循环之后指向元素i(i=0 and i=N-1)，则当前时间(Current Time)Tc可以表示为：Tc=S*N+i；如果此时插入一个时间间隔(Time Interval)为Ti的定时器，设定它将会放入元素n(Next)中，则n=(Tc+Ti)mod N=(S*N+i+Ti)mod N=(i+Ti)mod N。如果我们的N足够的大，显然StartTimer，StopTimer，PerTickBookkeeping时，算法复杂度分别为O(1)，O(1)，O(1)。在5中，给出了一个简单定时器轮实现的定时。下图3是一个简单的时间轮定时器：图3.简单时间轮如果需要支持的定时器范围非常的大，上面的实现方式则不能满足这样的需求。因为这样将消耗非常可观的内存，假设需要表示的定时器范围为：0 23-1ticks，则简单时间轮需要232个元素空间，这对于内存空间的使用将非常的庞大。也许可以降低定时器的精度，使得每个Tick表示的时间更长一些，但这样的代价是定时器的精度将大打折扣。现在的问题是，度量定时器的粒度，只能使用唯一粒度吗?想想日常生活中常遇到的水表，如下图4：图4.水表在上面的水表中，为了表示度量范围，分成了不同的单位，比如1000，100，10等等，相似的，表示一个32bits的范围，也不需要232个元素的数组。实际上，Linux的内核把定时器分为5组，每组的粒度分别表示为：1 jiffies，256 jiffies，256*64 jiffies，256*64*64 jiffies，256*64*64*64 jiffies，每组中桶的数量分别为：256，64，64，64，64，这样，在256+64+64+64+64=512个桶中，表示的范围为232。有了这样的实现，驱动内核定时器的机