如何在 C++ 中使用无锁循环缓冲区实现零拷贝 tcp答案

【问题标题】：How to implement zero-copy tcp using lock-free circular buffer in C++如何在 C++ 中使用无锁循环缓冲区实现零拷贝 tcp
【发布时间】：2012-07-02 14:16:05
【问题描述】：

我有多个线程需要使用来自 TCP 流的数据。我希望使用共享内存中的循环缓冲区/队列从 TCP 套接字读取。 TCP 接收将直接写入循环队列。消费者将从队列中读取。

此设计应启用零复制和零锁定。但是这里有两个不同的问题。

是否可以/有效地从 TCP 套接字中仅读取 1 条逻辑消息？如果没有，并且我阅读了超过 1 条消息，我将不得不将残差从 this 复制到 this->next。
真的可以实现无锁队列吗？我知道有原子操作，但这些也可能很昂贵。因为所有的 CPU 缓存都需要失效。这将影响我所有 24 个内核上的所有操作。

我对低级 TCP 有点生疏，不太清楚如何判断消息何时完成。我是在寻找 \0 还是特定于实现？

【问题讨论】：

您空间中的消息是什么？您有崇高的目标，但简单直接的解决方案的性能还不够吗？
无锁并不意味着“成本不高”，尽管在实践中无锁通常比锁定中涉及的上下文切换快几个数量级。只要您使缓冲区大小合适，就可以仅通过原子增量来实现您想要的，这是超快的，例如65536. 总而言之，我同意 Nim。
@Damon，有趣的是，我还没有看到一个单独依赖原子增量的环实现，可能你有一篇论文（链接）吗？我见过（并且已经做过）的唯一实现依赖于比较/交换操作..
如果你传递对象指针，而不是实际的批量数据，缓冲区的大小有点无关紧要——无论如何，每个缓冲区只传递 32/64 位。排队 100MB 与排队 10 个字节所用的时间相同。这实际上比一些带有狡猾的无锁指针和充斥着缓存抖动问题的讨厌的共享批量数据队列更容易管理。
请注意，首先考虑“批量数据”在概念上是错误的。尽管很多数据可能随时通过网络到达，但从来没有像“批量”这样的东西，绝对没有。如果有，您的系统将无法工作。进入的数据被放置在队列中并由消费者比新数据进入更快处理。总之，您的消费者必须比任何可能进入的数据都更快，具有良好的安全边际。否则，您的工作队列，无论您如何实现它，也无论您拥有多少 RAM , 难免溢出。

标签： c++ tcp circular-buffer lockless zero-copy

【解决方案1】：

不幸的是，TCP 不能传输消息，只能传输字节流。如果要传输消息，则必须在顶部应用协议。高性能的最佳协议是那些使用健全性可检查标头指定消息长度的协议 - 这允许您将正确数量的数据直接读取到合适的缓冲区对象中，而无需逐字节迭代数据以寻找结束 -消息字符。然后可以将缓冲区指针排队到另一个线程，并为下一条消息创建/分离一个新的缓冲区对象。这避免了批量数据的任何复制，并且对于大型消息，足够高效，使用非阻塞队列作为消息对象指针有点毫无意义。

下一个可用的优化是池化对象 *buffers 以避免持续的 new/dispose，回收消费者线程中的 *buffers 以便在网络接收线程中重新使用。使用 ConcurrentQueue 很容易做到这一点，如果池暂时清空，最好阻塞以允许流量控制而不是数据损坏或段错误/AV。

接下来，在每个 *buffer 数据成员的开头添加一个 [cacheline size] 'dead-zone'，以防止任何线程与任何其他线程虚假共享数据。

结果应该是完整消息的高带宽流到消费者线程，延迟、CPU 浪费或缓存抖动非常小。您的所有 24 个内核都可以在不同的数据上运行。

在多线程应用程序中复制大量数据是对糟糕设计和失败的承认。

跟进..

听起来您因为协议不同而无法迭代数据：(

无虚假共享的 PDU 缓冲区对象，示例：

typedef struct{
  char deadZone[256];  // anti-false-sharing
  int dataLen;
  char data[8388608]; // 8 meg of data
} SbufferData;

class TdataBuffer: public{
private:
  TbufferPool *myPool; // reference to pool used, in case more than one
  EpduState PDUstate; // enum state variable used to decode protocol
protected:
  SbufferData netData;
public:
  virtual reInit(); // zeros dataLen, resets PDUstate etc. - call when depooling a buffer
  virtual int loadPDU(char *fromHere,int len);  // loads protocol unit
  release(); // pushes 'this' back onto 'myPool'
};

loadPDU 得到一个指向原始网络数据长度的指针。它返回 0 - 意味着它还没有完全组装一个 PDU，或者它从原始网络数据中吃掉的字节数来完全组装一个 PDU，在这种情况下，将它排队，分离另一个并调用 loadPDU()使用未使用的剩余原始数据，然后继续输入下一个原始数据。

如果需要，您可以使用不同派生缓冲区类的不同池来服务不同的协议 - TbufferPool[Eprotocols] 数组。 TbufferPool 可能只是一个 BlockingCollection 队列。管理变得几乎是微不足道的——缓冲区可以发送到系统周围的队列中，发送到 GUI 以显示统计信息，然后可能发送到记录器，只要在队列链的末尾调用 release()。

显然，一个“真实”的 PDU 对象可能会加载更多方法、数据联合/结构、迭代器和一个状态引擎来操作协议，但无论如何这是基本思想。主要是易于管理、封装，并且由于没有两个线程可以在同一个缓冲区实例上操作，因此解析/访问数据不需要锁定/同步。

哦，是的，而且由于没有队列必须保持锁定的时间超过推送/弹出一个指针所需的时间，因此实际争用的可能性非常低 - 即使是传统的阻塞队列也几乎不需要使用内核锁定。

【讨论】：

谢谢！不幸的是，我无法控制消息协议。我的代码将连接到各种公共交易所和私人供应商股票代码工厂。希望他们使用标头/长度。除非这是标准，否则我可能不得不设计独立于消息协议的 TCP 接收代码。我确定我可以为此找到设计模式，甚至可以查看 ACE 源代码？谢谢@Martin James
"接下来，在每个 *buffer 数据成员的开头添加一个 [cacheline size] 'dead-zone'" 我不太确定我是否理解这解决的问题。另外，什么是“cachline 大小”？ CPU专用？ 32/64 位？ ty
@jaybny - Google '虚假分享'
更新：初始项目将连接到两个不同的场所。 Venue1 使用 sanity-headers w 长度协议。 Venue2 使用以 null 结尾的 msgId 标头，后跟不同类型的以 null 结尾的字段。最终成为基于 msgId 的固定大小结构。不知道为什么需要空值..希望能够推断出 msgId 的长度。无论哪种方式都将专注于高性能代码的 Venue1。 Venue2s 协议太糟糕了，不值得努力。
您的防伪共享缓冲区将不必要地丢弃缓存并增加工作集。一旦你触发了一个额外的页面错误，任何关于缓存效果的想法都是无关紧要的。您的缓冲区的大小是高速缓存行的两倍（安腾和未来的 Haswell 处理器）或四倍（大多数主流芯片）。理想情况下，它应该足够大，以保证每条消息都从一个新的缓存行开始。诚然，这对于可变长度消息很难完全正确，但是像 64-sizeof(int) 这样的东西仍然会好得多。也就是说……

【解决方案2】：

如果您使用的是 Windows 8 或 Windows Server 2012，则可以使用 Registered I/O，它比常规 IOCP 为更低的 CPU 提供更高的带宽；它通过消除内核转换、零拷贝等来实现这一点

API： http://msdn.microsoft.com/en-us/library/windows/desktop/ms740642%28v=vs.85%29.aspx

背景信息： http://www.serverframework.com/asynchronousevents/rio/

【讨论】：