Bip 缓冲区 - 带扭曲的循环缓冲区

• 下载源文件 - 2.4 KB

引言

Bip-Buffer 类似于循环缓冲区，但略有不同。它不维护指向缓冲区中数据的单个头部和尾部指针，而是维护两个旋转区域，从而能够快速访问数据，而无需担心缓冲区末尾的包装问题。缓冲区分配始终保持为连续块，从而允许缓冲区以高度有效的方式与 API 调用一起使用，并且还减少了将数据放入缓冲区所需的复制量。最后，两阶段分配系统允许用户悲观地预留缓冲区空间，然后将缓冲区修剪到只提交已使用的空间。

让我们先回顾一下一点历史。如果您还不了解为什么循环缓冲区可以在硬件中实现得非常高效，或者为什么它们会成为大多数电子设备的首选缓冲区，原因如下。

遥远的过去...

很久以前，计算机非常简单。它们没有 64 位数据总线。就连 16 位寄存器也没有——尽管有时您可以让一对寄存器充当此作用。那是一个更简单的时代，真正的男人会用汇编语言编程，并嘲笑任何不懂如何利用进位标志来实现各种不正当目的的人。

在那个简单的时代，人们会想出优雅的技巧来榨取每一次可用指令周期的最大功率。以简单的终端通信程序为例。较新的 RS232 串行控制器具有诸如 RTS 和 CTS 信号线的自动处理等功能来控制数据流——但这需要付出代价。也就是说，连接会不断地停止和启动，而不是持续传输。因此，在控制器卡和系统之间，通常会发现一个 FIFO。这个简单的循环缓冲区通常只有几个字节长，但它意味着系统可以平稳运行，而无需轮询以查看数据是否已到达，也不会被来自串行控制器的不断中断所淹没。

大多数 FIFO 最初都在芯片上，但人们也在他们的代码中添加了自己的 FIFO——想法是，如果您需要对传入数据进行一些非常复杂的处理，那么最好将它们批处理成一个大块并 infrequent 地处理……为系统腾出时间来做其他事情。例如滚动控制台或解码 GIF。

正如我所说，FIFO 是一个非常简单的循环缓冲区。大多数实现也非常简单；它们通常是 2^n 字节大小，这允许指针简单地溢出回到缓冲区的另一端。FIFO 逻辑可以通过头部和尾部值是否相同来判断 FIFO 是否为空，如果头部比尾部大一，则表示 FIFO 已满。

在旧的 8 位系统上，在软件中实现这些很容易。使用一对 16 位寄存器。决定在内存中的一个位置（256 的倍数）来存储 FIFO 数据。然后，在将寄存器设置为缓冲区开头后，不要触及高位寄存器——只需增加低位寄存器。这为您提供了一个 256 字节长的缓冲区，您可以通过一个指令（在 Zilog Z80 的情况下，需要 4 个周期——这是该系统上可用的最小执行单元）来遍历它。您永远不会超出缓冲区的边界，因为低位寄存器充当索引，值为 0 到 255。当您达到索引 256 时，寄存器将溢出并回绕到零。

现代

不幸的是，如今的 Windows 程序员并没有像那个简单的旧 8 位解决方案那样优雅的解决方案。当然，您可以深入汇编语言（前提是您能弄清楚编译器如何将寄存器映射到值……我从未见过足够好的解释来让我理解它），但大多数人现在没有时间学习汇编语言了。此外，我们现在处理的是 32 位寄存器——从该寄存器内部只增加一个低序字节不再是那么地“合法”。它可能导致缓存刷新、流水线停顿、打印机起火、青蛙雨等等。

如果您不能仅仅通过时钟低位寄存器来遍历缓冲区，您就必须开始担心诸如在到达缓冲区末尾之前检查已填充多少缓冲区，确保您记得将剩余数据从缓冲区开头复制过来，以及各种其他簿记难题。

我第一次尝试实现类似的东西时，寄希望于虚拟内存系统能够被欺骗，以便以一种方式设置东西，让您可以将内存的一部分镜像设置在原始内存旁边。想法是，您仍然可以使用数据的旋转分配；复制操作可以全速进行，而无需检查您是否走出了缓冲区——因为就您的进程地址空间而言，您的缓冲区的末尾也是您缓冲区的开头。

现在，这种镜像技术实际上可能有效。由于一些限制，我决定（还）不自己实现它（总有一天我一定会找到它的用途）。它的想法是，首先预留两个相邻的虚拟内存区域。然后将同一个临时文件映射到两个虚拟内存段。瞧！即时镜像，以及一个很大的缓冲区域，您可以随意从中复制数据。

不幸的是，虽然它应该（再次，我没有尝试过）确实有效，但还有另一个问题——也就是说，文件只能在 64kb 的边界（在较大的内存系统上可能更大）上映射。这意味着您的缓冲区最小必须是 64kb，并且将占用您 128kb 的虚拟地址空间。根据您的应用程序，这可能是一种有效的技术。但是，我认为编写一个拥有数千个套接字的服务器应用程序在这里不是一个可行的选择。

那么该怎么办？如果镜像不起作用，我们还能在代码中接近循环缓冲区的程度有多大？甚至，即使我们接近了，我们为什么要这样做呢？

循环缓冲区的优点

使用循环缓冲区临时存储数据有许多关键优势。

当一个人将数据放入内存块时，他也必须将其取出以加以利用。（或者您可以在原地使用它）。能够在向缓冲区追加更多数据时使用缓冲区中的数据非常有用。但是，当您释放缓冲区头部的一个空间时，该空间就不能再使用了，除非您将缓冲区中尚未使用的所有数据复制到缓冲区的开头。这会释放缓冲区末尾的空间，从而允许追加更多数据。

有几种方法可以解决这个问题；您可以简单地复制数据（这是一个相当昂贵的操作），或者您可以扩展缓冲区以允许追加更多数据（一个极其昂贵的过程）。

对于循环缓冲区，缓冲区中的可用空间始终可以追加数据；数据被复制，指针被调整，就这样。无需复制，无需重新分配，无需担心。缓冲区只分配一次，然后在其整个生命周期内保持有用。

美中不足

您可以简单地通过分配一块内存并维护指针来实现循环缓冲区。当指针超出缓冲区末尾时，指针将被调整——这个操作将反映在执行的每个操作中，无论是将数据复制到缓冲区还是从中移除。长度计算比正常情况稍微复杂一些，但也不是太过分——简单的内联函数可以轻松解决这个问题，将它扫到地毯下面。

不幸的是，大多数 API 调用都不相信循环缓冲区。您必须向它们传递一块可访问的连续内存块，并且您无法修改它们的写入行为以在它们跨越缓冲区末尾时调整指针。那么该怎么办？嗯，这就是 Bip-Buffer 的用武之地。

Bip-Buffer 登场

如果无法将循环缓冲区传递给 API 函数，那么就需要一种可行的替代方案——最好能具有循环缓冲区的许多优点。可以构建一个分为两个区域的循环缓冲区——或者说是双区的（这就是 Bip-Buffer 的“Bip”的由来）。这两个区域都沿着缓冲区移动，从左侧开始，最终到达右侧。当一个区域用完空间用于追加数据时，如果只有一个区域，就会在开头创建一个新区域（如果可能）。下面的图表更详细地展示了它是如何工作的。

缓冲区最初是空的，没有区域（图 1）。（例如，调用 AllocatedBuffer 后立即）

然后，当数据首次放入缓冲区时，会创建一个区域（“A”区域）（图 2）。（假设通过调用 Reserve 然后 Commit。）

数据被添加到该区域，向右扩展（图 3）。

这是我们第一次从缓冲区移除数据（图 4）。（请参阅下面描述的 DecommitBlock 调用）

这种情况会一直持续到该区域到达缓冲区末尾（图 4）。一旦左侧的可用空间比右侧的 A 区域多，就在该空间中创建第二个区域（滑稽地命名为“B 区域”）。选择在左侧有更多可用空间时创建新区域是为了最大化缓冲区中可用的潜在可用空间。所有这些的最终结果是，它看起来有点像图 5。

如果我们现在使用更多的缓冲区空间，我们将得到图 6，新的空间仅从 B 区域的末尾分配。如果我们最终分配了足够的数据来使用 A 和 B 区域之间的所有可用空间（图 7），那么缓冲区中将不再有可用的空间，在读取一些数据之前，将不再进行任何预订。

如果我们从缓冲区读取更多数据（比如 A 区域的剩余全部内容），我们将完全耗尽它。此时，由于 A 区域完全为空，我们不再需要跟踪两个单独的区域，B 区域的所有内部数据将被复制到 A 区域的内部数据之上，并且 B 区域将被完全清空。（图 8）

如果我们从缓冲区读取更多数据，我们将得到一个非常类似于图 4 的结果，循环继续。

Bip-Buffer 的特性

所有这些的最终结果是，平均而言，缓冲区始终拥有最大可用空间，而无需任何数据复制或重新分配来释放缓冲区末尾的空间。

从实现的角度来看，与常规循环缓冲区相比，Bip Buffer 的最大区别在于它只返回连续块。对于循环缓冲区，您需要担心缓冲区区域末尾的包装问题——这就是为什么例如如果您查看 Larry Antram 的 Fast Ring Buffer 实现，您会看到您将数据作为指针和长度传递到缓冲区，然后将数据逐字节复制到缓冲区中以考虑边缘的包装。

在公告板上提出的另一种可能性（提出它的人将保持匿名，只是因为他们……嗯……是匿名的）是简单地将调用分割到包装之间。好吧，这是一种绕过包装问题的方法，但它有一个不幸的副作用，即随着您的缓冲区填满，您传递给任何调用的可用空间量始终会减少到至少 1 字节——即使您的缓冲区开头有另一个 128kb 的可用空间，在末尾，您仍然需要处理不断缩小的块大小。Bip-Buffer 通过如果请求的大小大于缓冲区末尾剩余空间，就将其保留下来，从而巧妙地规避了这个问题。在编写网络代码时，这非常有用；您总是希望尝试接收尽可能多的数据，但您永远无法保证您会收到多少。（为了获得最佳结果，我建议分配一个缓冲区，它是您的 MTU 大小的倍数）。

是的，您将丢失一些本应是缓冲区末尾的可用空间。为了与 API 友好配合，这是一个小小的代价。

使用此缓冲区确实需要检查两次以查看缓冲区是否已清空；因为您必须处理当前有两个区域在使用的可能性。然而，灵活性和性能的提升抵消了这种小小的麻烦。

BipBuffer 类（完整的源代码在链接中提供）具有以下签名

class BipBuffer
{
private:
    BYTE* pBuffer;
    int ixa, sza, ixb, szb, buflen, ixResrv, szResrv;

public:
    BipBuffer();

构造函数初始化用于跟踪区域和内存指针的内部变量为 null；它不为缓冲区分配任何内存，以防万一您需要在不支持异常处理的环境中使用该类。

~BipBuffer();

析构函数仅释放已分配给缓冲区的任何内存。

bool AllocateBuffer(int buffersize = 4096);

AllocateBuffer 从虚拟内存中分配一个缓冲区。缓冲区的大小将向上舍入到最近的完整页面大小。如果成功，该函数返回 true，否则返回 false。

void FreeBuffer();

FreeBuffer 释放 AllocateBuffer 调用分配给缓冲区的任何内存，并释放 Bip-Buffer 中分配的任何区域。

bool IsInitialized() const;

IsInitialized 如果缓冲区已分配内存（通过调用 AllocateBuffer），则返回 true，否则返回 false。

int GetBufferSize() const;

GetBufferSize 返回缓冲区（以字节为单位）的总大小。如果传递给 AllocateBuffer 的值不是系统页面大小的倍数，则此值可能大于该值。

void Clear();

Clear…嗯……清除缓冲区。它不释放分配给缓冲区的任何内存；它只是将区域指针重置为 null，使整个缓冲区可用于新数据。

BYTE* Reserve(int size, OUT int& reserved);

现在进入关键部分。在 Bip-Buffer 中分配数据是一个两阶段操作。首先，通过调用 Reserve 函数预留一个区域；然后，通过调用 Commit 函数“提交”该区域。这允许您例如为 IO 调用预留内存，并在该 IO 调用失败时，假装它从未发生过。或者，在调用重叠的 WSARecv() 函数时，它允许您声明网络堆栈可用的内存量，用于传入数据，然后根据实际读取的数据量调整使用的空间量（可能少于请求量）。

要使用 Reserve，请传入请求的块的大小。该函数将在您传递的 reserved 参数中返回小于或等于 size 的最大可用空闲块的大小。它还将返回一个 BYTE* 指针指向您已预留的缓冲区区域。

在缓冲区没有可用空间的情况下，Reserve 将返回一个 NULL 指针，并且 reserved 将设置为零。

注意：您不能嵌套调用 Reserve 和 Commit；在调用 Reserve 之后，您必须先调用 Commit，然后才能再次调用 Reserve。

void Commit(int size);

这是分配的另一半。Commit 接受一个 size 参数，即您实际使用并希望保留在缓冲区中的字节数（从 Reserve 返回的 BYTE* 开始）。如果为此大小传递零，则预留将被完全释放，就好像您从未预留过任何空间一样。或者，在调试版本中，如果您传递的值大于原始预留大小，则会触发一个断言。（在发布版本中，将使用原始预留大小，没人会知道）。将数据提交到缓冲区使其可供从缓冲区取回数据的例程使用。

上图显示了 Reserve 和 Commit 的工作方式。当您调用 Reserve 时，它将返回一个指向上面灰色区域开头的指针（图 1）。假设您只使用了该缓冲区的一部分，例如蓝色区域（图 2）。将这部分区域分配出去并浪费掉是可惜的，因此您可以调用 Commit，只传入您使用的那么多数据，这就得到了图 3——即，提交的空间延伸到只填满您需要的部分，其余部分保持自由。

int GetReservationSize() const;

如果您随时需要了解您是否有待处理的预留，或者需要了解该预留的大小，您可以调用 GetReservationSize 来获取预留的量。没有预留？您将获得零。

BYTE* GetContiguousBlock(OUT int& size);

好吧，在投入了所有这些工作来将东西放入缓冲区之后，我们最好有一个办法能把它们取出来。

首先，如果您需要计算缓冲区中总共有多少数据可供读取，该怎么办？

int GetCommittedSize() const;

一种方法是调用 GetCommittedSize，它将返回缓冲区中数据的总长度——这是两个区域的总大小。我不建议依赖这个数字，因为如果您这样做，很容易忘记 Bip-Buffer 中有两个区域。那将是一件坏事（正如我数周痛苦的调试经验所证明的那样）。作为替代，您可以调用

BYTE* GetContiguousBlock(OUT int& size);

……这将返回一个 BYTE* 指针，指向缓冲区中的第一个（按 FIFO 顺序，不是最左边）连续已提交数据区域。size 参数也会用该块的长度进行更新。如果没有可用数据，该函数将返回 NULL（并且 size 参数设置为零）。

为了完全清空缓冲区，您可能希望循环调用 GetContiguousBlock 直到它返回 NULL。如果您吝啬，可以只调用它两次。但是，我建议前者；这意味着您可以忘记有两个区域，而只记住有一个以上。

void DecommitBlock(int size);

那么，在消耗了缓冲区中的数据之后，您该怎么做呢？嗯，为了与前面提到的 Reserve 和 Commit 调用精神一致，然后调用 DecommitBlock 来释放数据。数据按 FIFO 顺序释放，仅从第一个连续块释放——因此，如果您要调用 DecommitBlock，应该在调用 GetContiguousBlock 之后不久进行。如果您传递的大小大于连续块的长度，则整个块将被释放——但其他块（如果存在）都不会被释放。这是一个故意设计的选择，旨在提醒您存在多个块，您应该相应地采取行动。（如果您确实需要能够丢弃尚未读取的数据块中的数据，则复制 DecommitBlock 函数并修改它以使其在两个块上运行并不难；只需取消 if 语句的嵌套，并在第一个子句后调整大小参数。这个实现的实现被留给您知道的那个可怕的东西）。

这就是 Bip-Buffer 的实现。下面提供了一个简短的使用示例。

#include "BipBuffer.h"

BipBuffer buffer;
SOCKET s;
bool read_EOF;

bool StartUp
{
    // Allocate a buffer 8192 bytes in length
    if (!buffer.AllocateBuffer(8192)) return false;
    readEOF = false;

    s = socket(...

    ... do something else ...
}

void Foo()
{
    _ASSERTE(buffer.IsValid());

    // Reserve as much space as possible in the buffer:
    int space;
    BYTE* pData = buffer.Reserve(GetBufferSize(), space);

    // We now have *space* amount of room to play with.

    if (pData == NULL) return;

    // Obviously we've not emptied the buffer recently
    // because there isn't any room in it if we return.

    // Let's use the buffer!
    int recvcount = recv(s, (char*)pData, space, 0);

    if (recvcount == SOCKET_ERROR) return;
    // heh... that's some kind of error handling...

    // We now have data in the buffer (or, if the
    // connection was gracefully closed, we don't have any)

    buffer.Commit(recvcount);

    if (recvcount == 0) read_EOF = true;
}

void Bar()
{
    _ASSERTE(buffer.IsValid());

    // Let's empty the buffer.

    int allocated;
    BYTE* pData;

    while (pData = buffer.GetContiguousBlock(allocated)
           != NULL)
    {
        // Let's do something with the data.

        fwrite(pData, allocated, 1, outputfile);

        // (again, lousy error handling)

        buffer.DecommitBlock(allocated);
    }
}

告别

因此，我们到达了终点。我希望您发现这段代码和这种数据管理方式很有用；我肯定发现它在编写网络代码方面非常方便。如果您确实发现它很有用，或者在您的任何代码中使用它，我所要求的只是您给我发一封电子邮件，让我知道代码是如何使用的（什么项目类型、什么公司等）。如果 NDA 会妨碍，请保持含糊不清；很高兴知道它就在那里，活着，并且正在做很酷的事情。

结语（2020 年 7 月）

我不得不写这个，但结果是，有很多人很奇怪，他们似乎乐于贬低别人，最近其中一些人冒了出来。

首先，据我所知，我首次 提出了此处所示 的算法，而且是独立于任何其他人提出的（我需要它来加速我正在进行的 UDP 网络通信，用于我当时正在帮助原型设计质谱仪的控制程序）。我没有展示任何人的发明。而且，在公开之后，凭着事后诸葛亮的眼光，一切都显而易见了。

API 设计背后的原则绝对 是我的——我从未见过其他类似的东西。写入的 ReserveMax -> WriteN ->CommitN 模式，以及读取的 GetMax -> ReadN -> DecommitN 模式，是一个了不起的顿悟，使得它的使用比我预期的要好得多。我喜欢它的优雅（那是一个伟大的顿悟）。从那时起，我在处理其他 API 时一直牢记类似的想法（包括一些 Xbox 的较新 API，我在那里是 API 设计审查委员会的成员）。

看起来我比发明虚拟内存“完美”环形缓冲区晚了大约 2 年（这归功于 Phil Howard 在 2001 年 - 见这里），但我后来试图在 Xbox One 上推广它的使用，在我给出的一些技术讲座中，并且在我还在微软的 Windows Base 团队时，能够与内部人员合作，使它的创建在 Win32 和 UWP 方面都更加健壮。（如果我没记错的话，基于 PowerPC 的 Xbox 360 不能使用这个技巧，因为它的缓存基于虚拟地址而不是物理地址，这意味着缓存一致性可能成为一个问题，使得这个技巧不可行）。

自从这篇文章首次写成以来，还有许多其他人在这各种不同的地方使用了这段代码。它后来出现在帝国时代：在线、PuTTy 等项目的代码库中。（如果我获得机会，我会在某个时候整理一份完整的列表）。

Andrea Latuda 和 James Munns 在将我 2003 年潦草的笔记转化为 BipBuffer 的无锁实现方面做得非常出色，适用于高吞吐量多线程场景。您可以在 这里 和 这里 找到他们的版本。

Kevin Raowulf 还有一个 Rust 版本 在这里，但如果您仔细查找，还有一些其他的，例如 这个。

（显然，Rust 是我应该学习的语言——这就是为什么我“办公室”里有几本关于它的书等着我。）

它还被引用在一些专利中（均由 Sanford 公司）。

• US20110320733A1 - 缓存管理和存储介质加速
• US9323659B2 - 缓存管理包括固态设备虚拟化

（您是专业的软件或电子工程师吗？请勿阅读它们——那是有害的信息。留给律师去操心吧。）

版本历史

• 2020 年 7 月 20 日：添加了关于该想法起源的注释
• 2003 年 5 月 10 日：更新了源代码以修复 PeterChen 注意到的错误，修正了拼写，添加了摘要和更多基于评论的信息。更改了图。只是进行了一般性的整理。
• 2003 年 1 月 6 日：首次发布