Muduo 网络编程示例之八：用 Timing wheel 踢掉空闲连接

Dear 丶 2022-09-30 07:07 142阅读 0赞

Muduo 网络编程示例之八：Timing wheel 踢掉空闲连接

陈硕 (giantchen\_AT\_gmail)

Blog.csdn.net/Solstice [t.sina.com.cn/giantchen][t.sina.com.cn_giantchen]

这是《Muduo 网络编程示例》系列的第八篇文章，原计划讲文件传输，这里插入一点计划之外的内容。

Muduo 全系列文章列表： [http://blog.csdn.net/Solstice/category/779646.aspx][http_blog.csdn.net_Solstice_category_779646.aspx]

本文介绍如何使用 timing wheel 来踢掉空闲的连接，一个连接如果若干秒没有收到数据，就认为是空闲连接。

本文的代码见 [http://code.google.com/p/muduo/source/browse/trunk/examples/idleconnection][http_code.google.com_p_muduo_source_browse_trunk_examples_idleconnection]

在严肃的网络程序中，应用层的心跳协议是必不可少的。应该用心跳消息来判断对方进程是否能正常工作，“踢掉空闲连接”只是一时权宜之计。我这里想顺便讲讲 shared\_ptr 和 weak\_ptr 的用法。

如果一个连接连续几秒钟（后文以 8s 为例）内没有收到数据，就把它断开，为此有两种简单粗暴的做法：

*  每个连接保存“最后收到数据的时间 lastReceiveTime”，然后用一个定时器，每秒钟遍历一遍所有连接，断开那些 (now - connection.lastReceiveTime) > 8s 的 connection。这种做法全局只有一个 repeated timer，不过每次 timeout 都要检查全部连接，如果连接数目比较大（几千上万），这一步可能会比较费时。
 *  每个连接设置一个 one-shot timer，超时定为 8s，在超时的时候就断开本连接。当然，每次收到数据要去更新 timer。这种做法需要很多个 one-shot timer，会频繁地更新 timers。如果连接数目比较大，可能对 reactor 的 timer queue 造成压力。

使用 timing wheel 能避免上述两种做法的缺点。timing wheel 可以翻译为“时间轮盘”或“刻度盘”，本文保留英文。

连接超时不需要精确定时，只要大致 8 秒钟超时断开就行，多一秒少一秒关系不大。处理连接超时可以用一个简单的数据结构：8 个桶组成的循环队列。第一个桶放下一秒将要超时的连接，第二个放下 2 秒将要超时的连接。每个连接一收到数据就把自己放到第 8 个桶，然后在每秒钟的 callback 里把第一个桶里的连接断开，把这个空桶挪到队尾。这样大致可以做到 8 秒钟没有数据就超时断开连接。更重要的是，每次不用检查全部的 connection，只要检查第一个桶里的 connections，相当于把任务分散了。

# Timing wheel 原理 #

《Hashed and hierarchical timing wheels: efficient data structures for implementing a timer facility》这篇论文详细比较了实现定时器的各种数据结构，并提出了层次化的 timing wheel 与 hash timing wheel 等新结构。针对本文要解决的问题的特点，我们不需要实现一个通用的定时器，只用实现 simple timing wheel 即可。

Simple timing wheel 的基本结构是一个循环队列，还有一个指向队尾的指针 (tail)，这个指针每秒钟移动一格，就像钟表上的时针，timing wheel 由此得名。

以下是某一时刻 timing wheel 的状态，格子里的数字是倒计时（与通常的 timing wheel 相反），表示这个格子（桶子）中的连接的剩余寿命。

![wheel1][]

一秒钟以后，tail 指针移动一格，原来四点钟方向的格子被清空，其中的连接已被断开。

![wheel2][]

### 连接超时被踢掉的过程 ###

假设在某个时刻，conn 1 到达，把它放到当前格子中，它的剩余寿命是 7 秒。此后 conn 1 上没有收到数据。

![wheel3][]

1 秒钟之后，tail 指向下一个格子，conn 1 的剩余寿命是 6 秒。

![wheel4][]

又过了几秒钟，tail 指向 conn 1 之前的那个格子，conn 1 即将被断开。

![wheel5][]

下一秒，tail 重新指向 conn 1 原来所在的格子，清空其中的数据，断开 conn 1 连接。

![wheel6][]

### 连接刷新 ###

如果在断开 conn 1 之前收到数据，就把它移到当前的格子里。

![wheel4][wheel4 1]

收到数据，conn 1 的寿命延长为 7 秒。

![wheel7][]

时间继续前进，conn 1 寿命递减，不过它已经比第一种情况长寿了。

![wheel8][]

### 多个连接 ###

timing wheel 中的每个格子是个 hash set，可以容纳不止一个连接。

比如一开始，conn 1 到达。

![wheel3][wheel3 1]

随后，conn 2 到达，这时候 tail 还没有移动，两个连接位于同一个格子中，具有相同的剩余寿命。（下图中画成链表，代码中是哈希表。）

![wheel9][]

几秒钟之后，conn 1 收到数据，而 conn 2 一直没有收到数据，那么 conn 1 被移到当前的格子中。这时 conn 1 的寿命比 conn 2 长。

![wheel10][]

# 代码实现与改进 #

我们用以前多次出现的 EchoServer 来说明具体如何实现 timing wheel。代码见 [http://code.google.com/p/muduo/source/browse/trunk/examples/idleconnection][http_code.google.com_p_muduo_source_browse_trunk_examples_idleconnection]

在具体实现中，格子里放的不是连接，而是一个特制的 Entry struct，每个 Entry 包含 TcpConnection 的 weak\_ptr。Entry 的析构函数会判断连接是否还存在（用 weak\_ptr），如果还存在则断开连接。

数据结构：

typedef boost::weak_ptr WeakTcpConnectionPtr;
    
      struct Entry : public muduo::copyable
      {
        Entry(const WeakTcpConnectionPtr& weakConn)
          : weakConn_(weakConn)
        {
        }
    
        ~Entry()
        {
          muduo::net::TcpConnectionPtr conn = weakConn_.lock();
          if (conn)
          {
            conn->shutdown();
          }
        }
    
        WeakTcpConnectionPtr weakConn_;
      };
      typedef boost::shared_ptr EntryPtr;
      typedef boost::weak_ptr WeakEntryPtr;
      typedef boost::unordered_set Bucket;
      typedef boost::circular_buffer WeakConnectionList;

在实现中，为了简单起见，我们不会真的把一个连接从一个格子移到另一个格子，而是采用引用计数的办法，用 shared\_ptr 来管理 Entry。如果从连接收到数据，就把对应的 EntryPtr 放到这个格子里，这样它的引用计数就递增了。当 Entry 的引用计数递减到零，说明它没有在任何一个格子里出现，那么连接超时，Entry 的析构函数会断开连接。

Timing wheel 用 boost::circular\_buffer 实现，其中每个 Bucket 元素是个 hash set of EntryPtr。

在构造函数中，注册每秒钟的回调（EventLoop::runEvery() 注册 EchoServer::onTimer() ），然后把 timing wheel 设为适当的大小。

EchoServer::EchoServer(EventLoop* loop,
                           const InetAddress& listenAddr,
                           int idleSeconds)
      : loop_(loop),
        server_(loop, listenAddr, "EchoServer"),
        connectionBuckets_(idleSeconds)
    {
      server_.setConnectionCallback(
          boost::bind(&EchoServer::onConnection, this, _1));
      server_.setMessageCallback(
          boost::bind(&EchoServer::onMessage, this, _1, _2, _3));
      loop->runEvery(1.0, boost::bind(&EchoServer::onTimer, this));
      connectionBuckets_.resize(idleSeconds);
    }

其中 EchoServer::onTimer() 的实现只有一行：往队尾添加一个空的 Bucket，这样 circular\_buffer 会自动弹出队首的 Bucket，并析构之。在析构 Bucket 的时候，会依次析构其中的 EntryPtr 对象，这样 Entry 的引用计数就不用我们去操心，C++ 的值语意会帮我们搞定一切。

void EchoServer::onTimer()
    {
      connectionBuckets_.push_back(Bucket());
    }

在连接建立时，创建一个 Entry 对象，把它放到 timing wheel 的队尾。另外，我们还需要把 Entry 的弱引用保存到 TcpConnection 的 context 里，因为在收到数据的时候还要用到 Entry。（**思考题**：如果 TcpConnection::setContext 保存的是强引用 EntryPtr，会出现什么情况？）

void EchoServer::onConnection(const TcpConnectionPtr& conn)
    {
      LOG_INFO << "EchoServer - " << conn->peerAddress().toHostPort() << " -> "
        << conn->localAddress().toHostPort() << " is "
        << (conn->connected() ? "UP" : "DOWN");
    
      if (conn->connected())
      {
        EntryPtr entry(new Entry(conn));
        connectionBuckets_.back().insert(entry);
        WeakEntryPtr weakEntry(entry);
        conn->setContext(weakEntry);
      }
      else
      {
        assert(!conn->getContext().empty());
        WeakEntryPtr weakEntry(boost::any_cast(conn->getContext()));
        LOG_DEBUG << "Entry use_count = " << weakEntry.use_count();
      }
    }

在收到消息时，从 TcpConnection 的 context 中取出 Entry 的弱引用，把它提升为强引用 EntryPtr，然后放到当前的 timing wheel 队尾。（**思考题**，为什么要把 Entry 作为 TcpConnection 的 context 保存，如果这里再创建一个新的 Entry 会有什么后果？）

void EchoServer::onMessage(const TcpConnectionPtr& conn,
                               Buffer* buf,
                               Timestamp time)
    {
      string msg(buf->retrieveAsString());
      LOG_INFO << conn->name() << " echo " << msg.size() << " bytes at " << time.toString();
      conn->send(msg);
    
      assert(!conn->getContext().empty());
      WeakEntryPtr weakEntry(boost::any_cast(conn->getContext()));
      EntryPtr entry(weakEntry.lock());
      if (entry)
      {
        connectionBuckets_.back().insert(entry);
      }
    }

然后呢？没有然后了，程序已经完成了我们想要的功能。（完整的代码会打印 circular\_buffer 变化的情况，运行一下即可理解。）

希望本文有助于您理解 shared\_ptr 和 weak\_ptr。

### 改进 ###

在现在的实现中，每次收到消息都会往队尾添加 EntryPtr （当然，hash set 会帮我们去重。）一个简单的改进措施是，在 TcpConnection 里保存“最后一次往队尾添加引用时的 tail 位置”，然后先检查 tail 是否变化，若无变化则不重复添加 EntryPtr。这样或许能提高效率。

以上改进留作练习。

[t.sina.com.cn_giantchen]: http://t.sina.com.cn/giantchen
[http_blog.csdn.net_Solstice_category_779646.aspx]: http://blog.csdn.net/Solstice/category/779646.aspx
[http_code.google.com_p_muduo_source_browse_trunk_examples_idleconnection]: http://code.google.com/p/muduo/source/browse/trunk/examples/idleconnection
[wheel1]: /images/20220708/ad13cf676ab542a29b2dcdf0aeab0aa3.png
[wheel2]: /images/20220708/d3397d948b994b6b94e6ad5009c3c6b9.png
[wheel3]: /images/20220708/7238bc5cc9d94fb19bc6a69fd06bc4ed.png
[wheel4]: /images/20220708/94cd8769571a46bcb2c58dd7d2bef41a.png
[wheel5]: /images/20220708/dbab27ee6b8e466ebbfd030cb0ba0ef3.png
[wheel6]: /images/20220708/357e017efc10439dbb275b2959cbbf5b.png
[wheel4 1]: /images/20220708/2727897d229e4145a4f03e0436b65b27.png
[wheel7]: /images/20220708/e961ed6172a04413a52b8f3aaec3e2f8.png
[wheel8]: /images/20220708/af2e316d0bd24f278c0994b0cfbe3930.png
[wheel3 1]: /images/20220708/7a391e861afa425b8698bbca62002b20.png
[wheel9]: /images/20220708/496be5a3bd8f43d6bfafc03cac5281ec.png
[wheel10]: /images/20220708/397c4884656b409a96bcc42cdbfe1793.png