UDN-企业互联网技术人气社区

板块导航

浏览  : 1445
回复  : 0

[运维] Linux内核风格的定时器实现

[复制链接]
呵呵燕的头像 楼主
  定时器,是一个比较常见的组件。单就服务端来说,框架层面需要借助定时器来做会话的超时,应用层面需要借助定时器来处理一些跟时间有关的业务逻辑。对于游戏这种大量需求定时器的业务,一个简单高效的定时器组件是必不可少的。

  定时器组件的实现可以分为两部分:

  一部分是面向使用方,抽象成什么样的概念。

  一部分是底层的定时器调度算法。

  第一部分比较简单,但是实现方式多种多样,而且基本都是跟语言相关的,因此并不是本文重点。所谓抽象成的概念其实就是指使用者如何来用。

  linux内核中的定时器模块,使用者用的时候就是在内核的timer链表中插入一个新节点,节点里面挂一个作为callback的函数指针。

  C++/Java框架里实现的定时器会更高级一点,最常见的是采用观察者模式(Obeserver),比如回调逻辑写在一个class里面,实现IObservable,然后注册到定时器管理器中。定时器管理器根据一定策略轮询,检查timeout的IObservable,回调。

  如果是在函数具有更高级抽象的语言中,就免去了IObservable这一步,直接注册一个delegate就可以了。

  第二部分虽然比起第一部分需要更多的代码量,但是实现方式很有限。

  一种是比较容易能想到的,也是最常见的,由定时器管理器维护一个最小堆,每次tick都查一下堆中top项有没有timeout,timeout了就取出来,继续等待下一次tick。

  这种模型好处就是简单,找个学过数据结构的毕业生就能写出来,不容易有bug。add的时间复杂度是n(lgn),timeout的时间复杂度也是n(lgn)。

  但是,假设我们的业务系统如果面对的是这样的需求:短期内注册了大量短时间内就要timeout的timer。很显然,最小堆的实现就有点尴尬了。

  另一种就是小说君今天要介绍的linux内核风格的定时器实现。跟最小堆实现相比,能较好地处理之前说到的情况,同时,针对普适情况也有还不错的性能表现。linux内核的代码虽然说github上都能看到,定时器相关的只有一个文件,kernel/time/timer.c,但是里面的细枝末节太多,掩盖了值得学习的精华之处。

  下面进入正文,小说君就介绍下我们在应用层如何实现一个linux内核风格的定时器。语言以C#为例。

  为了做性能对比,我们要先实现一个基于最小堆的定时器管理器,最小堆的接口如下,具体实现就不港了,毕竟是最基础的数据结构。

  
  1. public class PriorityQueue : IEnumerable

  2.   {

  3.   public PriorityQueue(IComparer comparer);

  4.   public void Push(T v);

  5.   public T Pop();

  6.   public T Top();

  7.   }
复制代码


  
  1. public interface ITimeManager

  2.   {

  3.   ITimer AddTimer(uint afterTick, OnTimerTimeout callback, params object[] userData);

  4.   FixedTick();

  5.   }
复制代码


  
  1. public class TrivialTimeManager : ITimeManager

  2.   {

  3.   // ...

  4.   }
复制代码


  然后是linux内核风格定时器的管理器实现。首先有一个设计前提:

  我们需要用tick来定义整个系统的时间精度下限。比如说对于游戏来说,10ms以下的精度不需要care,因此我们可以把tick的长度定为10ms。也就是说先挂上去的WaitFor(8ms)和后挂上去的WaitFor(5ms),有可能是前者先timeout的。一个tick为10ms,那么一个32bit的tick能表达的时间粒度就有将近500天,远超过一个服务器组不重启的时间了。

  其实这种定时器实现,就是因为这个取舍,在面对之前提到的问题时,方才具有了更佳的性能表现。每次根据tick拿到timeout链表,直接dispatch,拿到这个链表的时间是一个常数,而最小堆方法拿到这个链表需要的时间是m*lgn。

  由于空间有限,我们不可能做到每个将要timeout的tick都有对应的链表。考虑到其实80%以上的timer的时间都不会超过2.55s,我们只针对前256个tick做这种优化措施即可。

  那如何处理注册的256tick之后的timer?我们可以把时间还比较长的timer放在更粗粒度的链表中,等到还剩下的tick数小于256之后再把他们取出来重新整理一下链表就能搞定。

  如果我们保证每一次tick都严格的做到:

  未来256tick内的链表都能常数时间取到。

  新加入的256tick以及更迟的timer才会加入到粗粒度链表。

  保证这两点,就需要每个tick都对所有链表做一次整理。这样就得不偿失了,所以这里有个trade-off,就是我通过一个指针(index),来标记我当前处理的position,每过256tick是一个cycle,才进行一次整理。而整理的成本就通过均摊在256tick中,降低了实际上的单位时间成本。

  概念比较抽象,接下来贴一部分代码。

  常量定义:

public const int TimeNearShift = 8;
  1. public const int TimeNearNum = 1 << TimeNearShift;      // 256

  2. public const int TimeNearMask = TimeNearNum - 1;        // 0x000000ff

  3. public const int TimeLevelShift = 6;
  4. public const int TimeLevelNum = 1 << TimeLevelShift;    // 64

  5. public const int TimeLevelMask = TimeLevelNum - 1;      // 00 00 00 (0011 1111)
复制代码


  基础数据结构:

  1. using TimerNodes = LinkedList<TimerNode>;
  2. private readonly TimerNodes[TimeNearNum] nearTimerNodes;
  3. private readonly TimerNodes[4][TimeLevelNum] levelTimerNodes;
复制代码


  相当于是256+4*64个timer链表。

  tick有32位,每一个tick只会timeout掉expire与index相同的timer。

  循环不变式保证near表具有这样几个性质:

  第i个链表中的所有timer的expire,(expire >> 8) == (index >> 8) 且(expire & TimeNearMask) == i。

  i小于(index & TimeNearMask)的链表,都已经AllTimeout。

  level表有4个,分别对应9到14bit,15到20bit,21到26bit,27到32bit。

  由于原理都类似,我这里拿9到14bit的表来说下循环不变式:

  表中的所有64个链表,所有timer的expire的高18个bit一定是与index的高18个bit相等的。

  第i个链表的元素的expire的9到14bit单独抽出来就是i。

  i小于(index的9到14bit单独抽出来)的链表,都已经Shift。

  有了数据结构和循环不变式,后面的代码也就容易理解了。主要列一下AddTimer的逻辑和Shift逻辑。

 
  1.  private void AddTimerNode(TimerNode node)

  2.   {

  3.   var expire = node.ExpireTick; if (expire < index)

  4.   { throw new Exception();

  5.   } // expire 与 index 的高24bit相同

  6.   if ((expire | TimeNearMask) == (index | TimeNearMask))

  7.   {

  8.   nearTimerNodes[expire & TimeNearMask].AddLast(node);

  9.   } else

  10.   {

  11.   var shift = TimeNearShift; for (int i = 0; i < 4; i++)

  12.   { // expire 与 index 的高bit相同

  13.   var lowerMask = (1 << (shift+TimeLevelShift))-1;

  14.   // (24-6*(i+1))

  15.   if ((expire | lowerMask) == (index | lowerMask))

  16.   { // 取出[(8+i*6), (14+i*6))这段bits

  17.   levelTimerNodes[i][(expire >> shift)&TimeLevelMask].AddLast(node); break;

  18.   }

  19.   shift += TimeLevelShift;

  20.   }

  21.   }

  22.   }
复制代码


 
  1.  private void TimerShift()

  2.   { // TODO index回绕到0的情况暂时不考虑

  3.   index++;

  4.   var ct = index; // mask0 : 8bit

  5.   // mask1 : 14bit

  6.   // mask2 : 20bit

  7.   // mask3 : 26bit

  8.   // mask4 : 32bit

  9.   var partialIndex = ct & TimeNearMask; if (partialIndex != 0)

  10.   { return;

  11.   }

  12.   ct >>= TimeNearShift; for (int i = 0; i < 4; i++)

  13.   {

  14.   partialIndex = ct & TimeLevelMask; if (partialIndex == 0)

  15.   {

  16.   ct >>= TimeLevelShift; continue;

  17.   }

  18.   ReAddAll(levelTimerNodes[i], partialIndex); break;

  19.   }

  20.   }
复制代码


  以上代码用c/c++重写后品尝风味更佳。

  实现大概就是这些了,接下来我们测一下到底linux内核风格定时器比最小堆实现的定时器快了多少。

  构建的测试用例和测试方法:

  1.   static IEnumerable BuildTestCases(uint first, uint second)

  2.   { var rand = new Random(); for (int i = 0; i < first; i++)

  3.   { yield return new TestCase()

  4.   {

  5.   Tick = (uint)rand.Next(256),

  6.   };

  7.   } for (int i = 0; i < 4; i++)

  8.   { var begin = 1U << (8 + 6*i); var end = 1U << (14 + 6*i); for (int j = 0; j < rand.Next((int)second * (4 - i)); j++)

  9.   { yield return new TestCase()

  10.   {

  11.   Tick = (uint)rand.Next((int)(begin+end)/2),

  12.   };

  13.   }

  14.   }

  15.   }

  16.   { var maxTick = cases.Max(c => c.Tick);

  17.   var results = new HashSet();

  18.   foreach (var c in cases)

  19.   {

  20.   TestCase c1 = c;

  21.   mgr.AddTimer(c.Tick, (timer, data) =>

  22.   {

  23.   if (mgr.FixedTicks == c1.Tick)

  24.   results.Add((uint) data[0]);

  25.   }, c.Id);

  26.   }

  27.   var begin = DateTime.Now;

  28.   for (int i = 0; i < maxTick+1; i++)

  29.   {

  30.   mgr.FixedTick();

  31.   }

  32.   var end = DateTime.Now;}
复制代码


  构建测试用例时的参数first指小于等于256tick的timer数量,second是指大于256tick的timer数量。

  first固定为一千万的测试结果:

1.webp.jpg


  加速比的波动不是特别明显,但是如果second继续增加,linux内核定时器的加速比实际上还是会由于shift频率的提升而逐步降低。

  second固定为1000的情况:

2.webp.jpg


  跟第一次测试的结论差不多,256tick以内的timer占比越高,相比最小堆定时器的优势越大。

  最终结论,linux内核定时器比起最小堆定时器的优势还是很明显的,大部分情况下都有2倍以上的性能表现,强烈建议采用。

原文作者:小说君  来源:开发者头条

相关帖子

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关于我们
联系我们
  • 电话:010-86393388
  • 邮件:udn@yonyou.com
  • 地址:北京市海淀区北清路68号
移动客户端下载
关注我们
  • 微信公众号:yonyouudn
  • 扫描右侧二维码关注我们
  • 专注企业互联网的技术社区
版权所有:用友网络科技股份有限公司82041 京ICP备05007539号-11 京公网网备安1101080209224 Powered by Discuz!
快速回复 返回列表 返回顶部