当前位置:首页 >> 编程开发 >> Visual C++ >> 内容

Muduo 网络编程示例(三)定时器

时间:2015/5/19 21:32:05 作者:平凡之路 来源:xuhantao.com 浏览:

程序中的时间

程序中对时间的处理是个大问题,我打算单独写一篇文章来全面地讨论这个问 题。文章暂定名《〈程序中的日期与时间〉第二章 计时与定时》,跟《〈程序中的日期与时间〉第一 章 日期计算》放到一个系列,这个系列预计会有四篇文章。

在这篇博客里里我先简要谈谈与编 程直接相关的内容,把更深入的内容留给上面提到的日期与时间专题文章。

在一般的服务端程序设计中,与时间有关的常见任务有:

1. 获取当前时间,计算时间间隔 ;

2. 时区转换与日期计算;把纽约当地时间转换为上海当地时间;2011-02-05 之后第 100 天 是几月几号星期几?等等

3. 定时操作,比如在预定的时间执行一项任务,或者在一段延时之后 执行一项任务。

其中第 2 项看起来复杂,其实最简单。日期计算用 Julian Day Number,时区 转换用 tz database;惟一麻烦一点的是夏令时,但也可以用 tz database 解决。这些操作都是纯函 数,很容易用一套单元测试来验证代码的正确性。需要特别注意的是,用 tzset/localtime_r 来做时 区转换在多线程环境下可能会有问题;对此我的解决办法是写一个 TimeZone class,以避免影响全局 ,将来在日期与时间专题中会讲到。以下本文不考虑时区,均为 UTC 时间。

真正麻烦的是第 1 项和第 3 项。一方面,Linux 有一大把令人眼花缭乱的与时间相关的函数和结构体,在程序中该如何 选用?另一方面,计算机中的时钟不是理想的计时器,它可能会漂移或跳变;最后,民用的 UTC 时间 与闰秒的关系也让定时任务变得复杂和微妙。当然,与系统当前时间有关的操作也让单元测试变得困难 。

Linux 时间函数

Linux 的计时函数,用于获得当前时间:

* time(2) / time_t (秒 )

* ftime(3) / struct timeb (毫秒)

* gettimeofday(2) / struct timeval (微 秒)

* clock_gettime(2) / struct timespec (纳秒)

* gmtime / localtime / timegm / mktime / strftime / struct tm (这些与当前时间无关)

定时函数,用于让程序等 待一段时间或安排计划任务:

* sleep

* alarm

* usleep

* nanosleep

* clock_nanosleep

* getitimer / setitimer

* timer_create / timer_settime / timer_gettime / timer_delete

* timerfd_create / timerfd_gettime / timerfd_settime

我的取舍如下:

* (计时)只使用 gettimeofday 来获取当前时间。

* (定时)只使用 timerfd_* 系列函数来处理定时。

gettimeofday 入选原因:(这也 是 muduo::Timestamp class 的主要设计考虑)

1. time 的精度太低,ftime 已被废弃, clock_gettime 精度最高,但是它系统调用的开销比 gettimeofday 大。

2. 在 x86-64 平台上 ,gettimeofday 不是系统调用,而是在用户态实现的(搜 vsyscall),没有上下文切换和陷入内核的 开销。

3. gettimeofday 的分辨率 (resolution) 是 1 微秒,足以满足日常计时的需要。 muduo::Timestamp 用一个 int64_t 来表示从 Epoch 到现在的微秒数,其范围可达上下 30 万年。

timerfd_* 入选的原因:

1. sleep / alarm / usleep 在实现时有可能用了信号 SIGALRM,在多线程程序中处理信号是个相当麻烦的事情,应当尽量避免。(近期我会写一篇博客仔细 讲讲“多线程、RAII、fork() 与信号”)

2. nanosleep 和 clock_nanosleep 是线 程安全的,但是在非阻塞网络编程中,绝对不能用让线程挂起的方式来等待一段时间,程序会失去响应 。正确的做法是注册一个时间回调函数。

3. getitimer 和 timer_create 也是用信号来 deliver 超时,在多线程程序中也会有麻烦。timer_create 可以指定信号的接收方是进程还是线程, 算是一个进步,不过在信号处理函数(signal handler)能做的事情实在很受限。

4. timerfd_create 把时间变成了一个文件描述符,该“文件”在定时器超时的那一刻变得可 读,这样就能很方便地融入到 select/poll 框架中,用统一的方式来处理 IO 事件和超时事件,这也 正是 Reactor 模式的长处。我在一年前发表的《Linux 新增系统调用的启示》中也谈到这个想法,现 在我把这个想法在 muduo 网络库中实现了。

5. 传统的 Reactor 利用 select/poll/epoll 的 timeout 来实现定时功能,但 poll 和 epoll 的定时精度只有毫秒,远低于 timerfd_settime 的定时 精度。

必须要说明,在 Linux 这种非实时多任务操作系统中,在用户态实现完全精确可控的计 时和定时是做不到的,因为当前任务可能会被随时切换出去,这在 CPU 负载大的时候尤为明显。但是 ,我们的程序可以尽量提高时间精度,必要的时候通过控制 CPU 负载来提高时间操作的可靠性,在程 序在 99.99% 的时候都是按预期执行的。这或许比换用实时操作系统并重新编写并测试代码要经济一些 。

关于时间的精度(accuracy)问题我留到专题博客文章中讨论,它与分辨率(resolution)不完全是一 回事儿。时间跳变和闰秒的影响与应对也不在此处展开讨论了。

Muduo 的定时器接口

Muduo EventLoop 有三个定时器函数:

  1: typedef boost::function<void()> TimerCallback;
   2: 
   3: ///
   4: /// Reactor, at most one per thread.
   5: ///
   6: /// This is an interface class, so don't expose too much details.
   7: class EventLoop : boost::noncopyable
   8: {
   9:  public:
  10:   // ...
  11: 
  12:   // timers
  13: 
  14:   ///
  15:   TimerId runAt(const Timestamp& time, const TimerCallback& cb);
  16: 
  17:   ///
  18:   /// Runs callback after @c delay seconds.
  19:   /// Safe to call from other threads.
  20:   TimerId runAfter(double delay, const TimerCallback& cb);
  21: 
  22:   ///
  23:   /// Runs callback every @c interval seconds.
  24:   /// Safe to call from other threads.
  25:   TimerId runEvery(double interval, const TimerCallback& cb);
  26: 
  27:   /// Cancels the timer.
  28:   /// Safe to call from other threads.
  29:   // void cancel(TimerId timerId);
  30: 
  31:   // ...
  32: };

* runAt 在指定的时间调用 TimerCallback

* runAfter 等一段时间调用 TimerCallback

* runEvery 以固定的间隔反复调用 TimerCallback

* cancel 取消 timer,目前未实现

回调函数在 EventLoop 对象所在的线程发生,与 onMessage() onConnection() 等网络事件函数在同一个线程。

Muduo 的 TimerQueue 采用了最简单的实现( 链表)来管理定时器,它的效率比不上常见的 binary heap 的做法,如果程序中大量(10 个以上)使 用重复触发的定时器,或许值得考虑改用更高级的实现。我目前还没有在一个程序里用过这么多定时器 ,暂时也不打算优化 TimerQueue。

Boost.Asio Timer 示例

Boost.Asio 教程里以 Timer 和 Daytime 为例介绍 asio 的基本使用,daytime 已经在前文“示例一”中介绍过 ,这里着重谈谈 Timer。Asio 有 5 个 Timer 示例,muduo 把其中四个重新实现了一遍,并扩充了第 5 个示例。

1. 阻塞式的定时,muduo 不支持这种用法,无代码。

2. 非阻塞定时,见 examples/asio/tutorial/timer2

3. 在 TimerCallback 里传递参数,见 examples/asio/tutorial/timer3

4. 以成员函数为 TimerCallback,见 examples/asio/tutorial/timer4

5. 在多线程中回调,用 mutex 保护共享变量,见 examples/asio/tutorial/timer5

6. 在多线程中回调,缩小临界区,把不需要互斥执行的代码 移出来,见 examples/asio/tutorial/timer6

为节省篇幅,这里只列出 timer4:

 

 1: #include <muduo/net/EventLoop.h>
   2: 
   3: #include <iostream>
   4: #include <boost/bind.hpp>
   5: #include <boost/noncopyable.hpp>
   6: 
   7: class Printer : boost::noncopyable
   8: {
   9:  public:
  10:   Printer(muduo::net::EventLoop* loop)
  11:     : loop_(loop),
  12:       count_(0)
  13:   {
  14:     loop_->runAfter(1, boost::bind(&Printer::print, this));
  15:   }
  16: 
  17:   ~Printer()
  18:   {
  19:     std::cout << "Final count is " << count_ << "n";
  20:   }
  21: 
  22:   void print()
  23:   {
  24:     if (count_ < 5)
  25:     {
  26:       std::cout << count_ << "n";
  27:       ++count_;
  28: 
  29:       loop_->runAfter(1, boost::bind(&Printer::print, this));
  30:     }
  31:     else
  32:     {
  33:       loop_->quit();
  34:     }
  35:   }
  36: 
  37: private:
  38:   muduo::net::EventLoop* loop_;
  39:   int count_;
  40: };
  41: 
  42: int main()
  43: {
  44:   muduo::net::EventLoop loop;
  45:   Printer printer(&loop);
  46:   loop.loop();
  47: }

查看本栏目更多精彩内容:http://www.bianceng.cn/Programming/cplus/

最后我再强调一遍,在非阻塞服务端编程中,绝对不能用  sleep 或类似的办法来 让程序原地停留等待,这会让程序失去响应,因为主事件循环被挂起了,无法处理 IO 事件。这就像在 Windows 编程中绝对不能在消息循环里执行耗时的代码一样,会让程序界面失去响应。Reactor 模式的 网络编程确实有些类似传统的消息驱动的 Windows 编程。对于“定时”任务,就把它变成 一个特定的消息,到时候触发相应的消息处理函数就行了。

Boost.Asio 的 timer 示例只用到 了 EventLoop::runAfter,我再举一个 EventLoop::runEvery 的例子。

Java Netty 示例

Netty 是一个非常好的 Java NIO 网络库,它附带的示例程序有 echo 和 discard 两个简单网络协 议,与前文不同,Netty 版的服务端有流量统计功能,这需要用到 EventLoop::runEvery。

这 里列出 discard server 的代码,其 client 的代码类似前文的 chargen,为节省篇幅,请阅读源码 http://code.google.com/p/muduo/source/browse/trunk/examples/netty/ 。

Discard server 注册了一个间隔为 3 秒的定时器,调用 DiscardServer::printThroughput 打印 出吞吐量。注意这段代码用了整数的原子操作 AtomicInt64 来记录收到的字节数和消息数,乍看之下 似乎没有必要,其实 DiscardServer 可以配置成多线程服务器,muduo TcpServer 有一个内置的多线 程模型,可以通过 setThreadNum() 来开启。这个话题留到以后再细说。

  1: #include <muduo/net/TcpServer.h>
   2: 
   3: #include <muduo/base/Atomic.h>
   4: #include <muduo/base/Logging.h>
   5: #include <muduo/base/Thread.h>
   6: #include <muduo/net/EventLoop.h>
   7: #include <muduo/net/InetAddress.h>
   8: 
   9: #include <boost/bind.hpp>
  10: 
  11: #include <utility>
  12: 
  13: #include <stdio.h>
  14: #include <unistd.h>
  15: 
  16: using namespace muduo;
  17: using namespace muduo::net;
  18: 
  19: int numThreads = 0;
  20: 
  21: class DiscardServer
  22: {
  23:  public:
  24:   DiscardServer(EventLoop* loop, const InetAddress& listenAddr)
  25:     : loop_(loop),
  26:       server_(loop, listenAddr, "DiscardServer"),
  27:       oldCounter_(0),
  28:       startTime_(Timestamp::now())
  29:   {
  30:     server_.setConnectionCallback(
  31:         boost::bind(&DiscardServer::onConnection, this, _1));
  32:     server_.setMessageCallback(
  33:         boost::bind(&DiscardServer::onMessage, this, _1, _2, _3));
  34:     server_.setThreadNum(numThreads);
  35:     loop->runEvery(3.0, boost::bind(&DiscardServer::printThroughput, this));
  36:   }
  37: 
  38:   void start()

  39:   {
  40:     LOG_INFO << "starting " << numThreads << " threads.";
  41:     server_.start();
  42:   }
  43: 
  44:  private:
  45:   void onConnection(const TcpConnectionPtr& conn)
  46:   {
  47:     LOG_TRACE << conn->peerAddress().toHostPort() << " -> "
  48:         << conn->localAddress().toHostPort() << " is "
  49:         << (conn->connected() ? "UP" : "DOWN");
  50:   }
  51: 
  52:   void onMessage(const TcpConnectionPtr& conn, Buffer* buf, Timestamp)
  53:   {
  54:     size_t len = buf->readableBytes();
  55:     transferred_.add(len);
  56:     receivedMessages_.incrementAndGet();
  57:     buf->retrieveAll();
  58:   }
  59: 
  60:   void printThroughput()
  61:   {
  62:     Timestamp endTime = Timestamp::now();
  63:     int64_t newCounter = transferred_.get();
  64:     int64_t bytes = newCounter - oldCounter_;
  65:     int64_t msgs = receivedMessages_.getAndSet(0);
  66:     double time = timeDifference(endTime, startTime_);
  67:     printf("%4.3f MiB/s %4.3f Ki Msgs/s %6.2f bytes per msgn",
  68:         static_cast<double>(bytes)/time/1024/1024,
  69:         static_cast<double>(msgs)/time/1024,
  70:         static_cast<double>(bytes)/static_cast<double>(msgs));
  71: 
  72:     oldCounter_ = newCounter;
  73:     startTime_ = endTime;
  74:   }
  75: 
  76:   EventLoop* loop_;
  77:   TcpServer server_;
  78: 
  79:   AtomicInt64 transferred_;
  80:   AtomicInt64 receivedMessages_;
  81:   int64_t oldCounter_;
  82:   Timestamp startTime_;
  83: };
  84: 
  85: int main(int argc, char* argv[])
  86: {
  87:   LOG_INFO << "pid = " << getpid() << ", tid = " << CurrentThread::tid();
  88:   if (argc > 1)
  89:   {
  90:     numThreads = atoi(argv[1]);
  91:   }
  92:   EventLoop loop;
  93:   InetAddress listenAddr(2009);
  94:   DiscardServer server(&loop, listenAddr);
  95: 
  96:   server.start();
  97: 
  98:   loop.loop();
  99: }
运行方法,在同一台机器的两个命令行窗口分别运行:
$ bin/netty_discard_server
$ bin/netty_discard_client 127.0.0.1 256

第一个窗口显示吞吐量:

41.001 

MiB/s 73.387 Ki Msgs/s 572.10 bytes per msg
72.441 MiB/s 129.593 Ki Msgs/s 572.40 bytes per msg
77.724 MiB/s 137.251 Ki Msgs/s 579.88 bytes per msg

改变第二个命令的最后一个参数( 上面的 256),可以观察不同的消息大小对吞吐量的影响。

练习 1:把二者的关系绘制成函数 曲线,看看有什么规律,想想为什么。

练习 2:在局域网的两台机器上运行客户端和服务端, 找出让吞吐量达到最大的消息长度。这个数字与练习 1 中的相比是大还是小?为什么?

有兴趣 的读者可以对比一下 Netty 的吞吐量,muduo 应该能轻松取胜。

discard client/server 测试 的是单向吞吐量,echo client/server 测试的是双向吞吐量。这两个服务端都支持多个并发连接,两 个客户端都是单连接的。本系列第 6 篇文章将会实现一个 pingpong 协议,用来测试 muduo 在多线程 大量连接情况下的表现。

(待续)

相关文章
  • 没有相关文章
共有评论 0相关评论
发表我的评论
  • 大名:
  • 内容:
  • 徐汉涛(www.xuhantao.com) © 2024 版权所有 All Rights Reserved.
  • 部分内容来自网络,如有侵权请联系站长尽快处理 站长QQ:965898558(广告及站内业务受理) 网站备案号:蒙ICP备15000590号-1