让内核来完成工作！实现 Pub/Sub 服务器的技巧

这篇文章介绍了 jetrelay 的设计，它是一个与 Bluesky 的 "jetstream" 数据流兼容的 Pub/Sub 服务器。通过使用一些相关的 Linux 内核特性，它可以避免自己执行几乎任何工作。因此，它非常高效：只需 8 个 CPU 核心即可饱和 10 Gbps 的网络连接。

2025 年 5 月

挑战：以线速广播

Bluesky 构建在 ATproto 之上，而 ATproto 的一个核心部分是 "the firehose"，这是一个表示网络状态所有变化的事件流。正如您所期望的那样，firehose 包含所有新帖子；但也包括人们喜欢的东西、删除/编辑旧帖子、关注他人等。它涵盖了整个 Bluesky，因此非常活跃。

此数据有两种风格：原始的全功能 firehose 和一个新的精简版本，称为 "jetstream"。这两个流都基于 websockets，但 jetstream 将其有效负载编码为 JSON（而不是 CBOR），并省略了仅用于身份验证的部分。此外，我认为 jetstream 仅包含事件的子集。

jetstream 上的平均消息大小约为半千字节。事件速率是可变的（我猜这取决于哪些国家/地区处于活跃状态），但似乎大约是每秒 300-500 个事件。 relay 是一种服务器，它遵循上游 feed 提供商并将数据重新广播给自己的客户端。1 粗略估计：在具有 10 千兆位网卡的机器上运行，您的 relay 应该能够同时服务 10Gbps / (0.5KiB * 400/s) = ~6000 个客户端。

好的，接受挑战！我编写了一个简单的 jetstream relay，我称之为 "jetrelay"。它只有 ~500 行代码，代码在这里，在这篇文章中，我将解释它的工作原理。

这 500 行代码中实际上很少有是专门针对 jetstream 的。jetrelay 的重点是演示下面描述的技术，这些技术应该可以转移到其他 Pub/Sub 协议的实现中。但请注意，jetrelay 是一个技术演示——更多代码 在投入实际使用之前是必需的。

另请注意：我的目标不是与官方 jetstream 服务器实现功能对等。特别是，官方服务器允许客户端按 collection 或 DID 过滤数据。我只关注 "full stream" 用例：每个客户端都获得完整的 feed，无论他们是否需要它。Bluesky 显然认为过滤是一个重要的功能，所以我认为值得一提的是这个省略。还有其他一些2差异3。

组播和回填

我们的任务是接受来自上游数据 feed 的事件并将这些事件重新广播给我们的客户端。关键的观察结果是，我们正在向所有客户端发送 完全相同的数据 。我不仅仅是指 JSON 值是相同的；所有 都是相同的，直到 websocket 帧的标头。排除初始 HTTP 握手，每个客户端都看到相同的字节通过管道传输。4

这称为 "multicast"。在本地网络上，您可以使用 UDP multicast5 并让内核/网络硬件为您处理所有事情（虽然它是 UDP，所以有一些注意事项6 7)。但是，jetstream 协议基于 websockets，而 websockets 基于 TCP。TCP 的 multicast 并不是真正存在的东西，8 所以我们将不得不自己实现它。

第二个观察结果是，虽然客户端确实看到了完全相同的事件，但他们 不 一定会在完全相同的时间看到它们。快速客户端将始终接收最新的事件，但是当 feed 特别活跃时，较慢的客户端可能会开始落后。这些慢速客户端将接收到延迟版本的 feed，直到事情平静下来并且他们有机会赶上。

然后是回填：当客户端连接到服务器时，他们通过时间戳指定他们在 feed 中的初始位置。这允许客户端在断开连接后重新上线并填充他们错过的事件。换句话说，发送几分钟甚至几小时前的数据是完全正常的。

结果是我们的 relay 不会是一个纯粹的内存系统。所有事件数据的副本都需要保存到磁盘。

技巧 #1：使用 sendfile() 绕过用户空间

当新事件到达时，我们将它们附加到一个文件中。我们将数据完全按照它在网络上传输的样子存储——websocket 帧字节和所有内容——一切准备就绪。9 对于每个客户端，我们都保留一个指向文件中某个位置的游标。如果客户端的游标没有指向文件的末尾，我们将从文件中复制缺失的字节到客户端的 socket 中。...就是这样！

内核为此提供了一个 syscall：sendfile()。您指定一个文件、文件中的一个字节范围以及一个用于发送字节的 socket。这不仅易于使用，而且非常便宜。您可能会认为 "从磁盘获取数据听起来很昂贵"；但由于这是我们刚刚写入的数据，它将驻留在内核的页面缓存中（即在内存中）。通过 sendfile()，数据直接从页面缓存转到网络堆栈。将其与传统的 read()/write() 方法进行比较，在这种方法中，数据将被复制到我们程序的内存中，然后再复制回来。

来自上游的新事件到达并被写入文件末尾。客户端不再被认为是 "最新的"，因为他们的游标不再指向文件的末尾。对于每个客户端，我们调用 sendfile() 来发送新数据，并在调用返回时更新客户端的游标。

关于这种 "文件和游标" 设计的 最好 的事情是：它自然地执行写入批处理。最新的客户端将一次一个地接收新消息，一旦它们准备好；但是如果有多个消息准备好发送，它们都可以作为一个块复制到 socket 中。更好的是，可以按引用传递页面大小的数据块（4 KiB = ~8 个事件）。在这种情况下，网络堆栈实际上是从页面缓存中直接读取数据，而无需进行不必要的复制！

当客户端落后时，平稳地牺牲延迟以换取吞吐量对于这种类型的应用程序来说非常重要。当负载增加时效率 降低 的程序是 SRE 恐怖故事的素材。

技巧 #2：使用 io_uring 并行处理大量客户端

一个 syscall，没有复制——你还想要什么！好吧，sendfile() 是同步的：它会阻塞当前线程，直到数据被发送。但是如果一个客户端离线并且它的发送缓冲区已满，则对该客户端的下一个 sendfile() 将无限期地阻塞！这意味着我们需要为每个客户端提供其自己的专用线程。但我真的不想生成 6000 个线程...10

幸运的是，有一个更好的解决方案！Linux 有一个名为 "io_uring" 的机制。通过它，我们可以准备一堆 sendfile() 并通过单个 syscall 将它们全部提交给内核。然后，内核在 sendfile() 完成时发回完成事件。它就像 syscall 的通道！

使用 io_uring，我们的主运行循环如下所示：

  1. 对于每个不是最新的（并且可写入）的客户端，将 sendfile() 添加到提交队列。
  2. 提交所有 sendfile() 并等待完成（带有超时）。
  3. 对于每次完成，更新关联的客户端的游标。
  4. 转到 (1)。

当所有客户端都是最新的时，该线程将休眠，定期唤醒以重新检查文件长度（感谢超时）。另一方面，如果一个客户端远远落后并且渴望数据，则该线程将快速循环，一旦先前的线程完成，就提交一个新的 sendfile()，并且客户端将快速赶上。

注意：我们的程序执行的 syscall 数量不取决于客户端的数量！可以连接大量客户端，而 jetrelay 所做的工作量几乎不会改变。当然，内核 将有更多的工作要做——但这是不可避免的。我们的工作是以尽可能高效的方式协调必要的 I/O，然后避开内核。

我忽略了一个细节：io_uring 实际上没有 sendfile 操作！但不用担心：我们可以使用两个 splice() 来模拟 sendfile()。首先，您从文件到管道进行 splice，然后从管道到 socket 进行 splice。（这实际上是内核中同步 sendfile() 的实现方式。）可以同时提交两个 splice 操作；您可以将它们作为 "linked" 条目提交，这意味着 io_uring 在第一个 splice 完成之前不会启动第二个 splice。您需要为每个客户端提供其自己的管道。

感谢出色的 rustix crate，它使实现所有这些东西变得容易！11

不是技巧：连接新客户端

在第一部分中，我告诉您关于 event writer thread 的信息，该线程从上游接收 ATproto 事件并将它们写入文件。在第二部分中，我描述了 I/O orchestration thread，该线程使客户端与文件保持同步。

Jetrelay 还有另一个我尚未提及的线程：client handshake thread。它负责监听新的客户端连接。当新的客户端连接时，它会执行 HTTP/websocket 握手并将它们添加到活动客户端集中。

回想一下，当客户端连接时，他们可以指定其初始位置，以时间戳的形式。为了支持这一点，我们需要一个索引，该索引将时间戳转换为文件中的字节偏移量。我正在使用 BTreeMap<Timestamp, Offset>。12

我们可以让 event-writer thread 做更多的工作：在它完成将事件写入文件后，它会解析事件的时间戳并将条目推送到索引。理论上，索引不必覆盖 每个 事件，因此我们可以仅每 N 个事件执行一次；但是它很快，所以我每次都这样做。

现在，当客户端连接时，我们要做的就是在索引中查找请求的时间戳，并使用结果初始化客户端的文件偏移量。请注意，该索引由一个线程写入，而由另一个线程读取，因此我将其放入 Mutex 中。

...就是这样！下一次 io_uring 线程唤醒时，它将看到新的客户端并开始尽可能快地将数据 sendfile() 发送给它。如上所述，这可以非常有效地发送 4 KiB 的块，因此客户端可以 快速 赶上。

本节被标记为 "不是技巧"；但您可能会说，技巧在于认识到什么时候简单的方法就足够了。如果客户端需要 100 毫秒才能连接，那又有什么关系呢？如果我们将其加速到 10 毫秒，那对我们没有任何帮助。13 我们可以变得花哨并将我们的索引保留在无锁双端队列中...但是为什么要让自己过得艰难呢？Mutex<BTreeMap<Timestamp, Offset>> 完成了这项工作。

技巧 #3：使用 FALLOC_FL_PUNCH_HOLE 丢弃旧数据

我们需要解决的最后一个问题是丢弃旧数据。使用我们当前的实现，文件将不断增长，直到我们耗尽磁盘空间。我希望我们的实现能够将数据保留固定的时间，然后再删除它。它不必持续发生——定期清理操作就可以了——但它应该在不干扰连接的客户端的情况下发生。

此时，您可能正在考虑旋转文件。这将引入额外的簿记和棘手的极端情况（例如，当客户端从一个文件跨越到下一个文件时）。有必要这么复杂吗？

不！操作系统再次为我们提供了一个简单的解决方案：fallocate()。此 syscall 允许我们取消分配文件的区域。（Linux 将其称为 "在文件中打孔"。）已取消分配的区域不占用磁盘空间，并且在从中读取时返回零。

文件中的其余数据仍可像以前一样在偏移处读取。这意味着无需修复客户端游标或任何东西：对于程序的其余部分来说，就好像什么也没发生过（只要它从不从已取消分配的区域读取）。

因此，当我们想要删除早于给定时间的数据时，我们首先从索引中删除所有直到该时间的条目，然后获取我们删除的最后一个条目的偏移量，并取消分配直到该点的文件。（这就是为什么我在上一节中使用 BTreeMap 来存储索引，而不是 HashMap：btree 支持快速范围删除。）

现在，如果您使用 ls 查看该文件，您会看到它的长度只是不断增加。但是如果您使用 du 查看它，您会看到它实际使用的磁盘空间保持在限制范围内！

$ ls -l
213M jetrelay.dat
$ du -h
8.9M jetrelay.dat

您可能会想：如果文件的明显大小不断增长，那么我们最终肯定会达到某种文件大小限制？好吧，最大文件大小取决于您的配置，但通常以拍字节为单位进行衡量。因此，我们可以每月流式传输数 TB 的数据，持续数百年，然后这才会成为一个问题。作为巧妙的 hack，我认为这个 hack 具有相对较长的保质期！

需要注意的一种极端情况是：如果客户端不从其 socket 读取数据，则其游标不会移动。因此，当我们去丢弃一些旧数据时，客户端的游标可能仍然指向我们要取消分配的区域。我们应该在此刻踢掉此类客户端，否则如果他们突然开始再次读取数据，他们会看到零。即使新客户端请求一个古老的时间戳，他们也永远不会进入已取消分配的区域，因为我们已从索引中删除了这些条目。

测试它

...针对回环

首先，让我们在本地试一试。我正在我的笔记本电脑上运行 jetrelay 并将大量客户端连接到它。客户端将通过回环接口连接到 jetrelay；这意味着没有实际的网络流量会到达我的网卡（这很好，因为我正在使用 wi-fi）。好的，让我们开始吧...

Error: Too many open files (os error 24)

...对。每个客户端都会消耗一个文件描述符用于其 socket，另外两个用于我们用于实现 sendfile() 的管道。默认情况下，一个程序一次只能有 1024 个文件描述符——这只有 340 个客户端！我们需要提高 fd 限制。传统的方法是使用 ulimit -n，但是由于我无论如何都在使用 systemd 运行 jetrelay，因此我可以在单元文件中设置 LimitNOFILE=65535。14

修复此问题后，我将 2 万个客户端连接到 jetrelay，并且他们可以毫无问题地跟上 feed。总吞吐量超过 24 Gbps。太棒了！

...针对真实网络

回环非常好，但我想看到它在真实的网络上真实地工作。因此，我从 Linode 租用了一个 VM。我选择了具有 10 Gbps 输出网络连接的最便宜的实例类型。

从同一数据中心中的另一台机器上，我打开了 6000 个并发客户端连接。所有 6000 个客户端都可以轻松跟上 feed：

6000 clients @ T-0s (319 evs, 158 KiB)
Total: 1_914_000 evs, 930 MiB, 7440 Mbps

当我进行这些测试时，jetstream feed 恰好相当安静（大约每秒只有 300 个事件），因此即使有 6k 个客户端，数据速率也只有 7.4 Gbps。

我不断增加客户端的数量。在 8.5k 时，它仍然可以为他们所有人提供 feed。最后，在 9k 时，我超过了限制。有一段时间，它实际上能够使所有 9000 个客户端保持最新：

9000 clients @ T-0s (279 evs, 141 KiB)
Total: 2_511_000 evs, 1245 MiB, 9960 Mbps

但是很快，一些客户端开始落后：

 130 clients @ T-1s (392 evs, 162 KiB)
8870 clients @ T-0s (269 evs, 144 KiB)
Total: 2_396_614 evs, 1255 MiB, 10040 Mbps

正如预期的那样，当所需的数据速率超过 10 Gbps 时，就会发生这种情况。我们可以看到 jetrelay 确实已使完整的 10 Gbps 饱和。

找到限制

因此，jetrelay 可以管理价值 10 Gbps 的客户端 - 太棒了。但是，如果使用功能更强大的网卡，它可以提供 更多 服务吗？

大概在 9k 客户端测试中，网卡是瓶颈；但另一方面，jetrelay 并不缺乏处理能力：这台机器有 50 个 CPU 核心！ （是的，这是我能获得的 最便宜的 具有 10 千兆位网络的机器。）

我想找到输出受到 jetrelay 限制而不是网卡限制的点。我无法增加网络带宽...但是我可以 减少 CPU 核心的数量！我的策略是逐渐减少 jetrelay 的 CPU 配额，直到它无法再管理 10 Gbps。（我将使用 systemctl set-property jetrelay.service CPUQuota= 来执行此操作。）

这是结果：

CPU 配额 | 最大客户端数量 | 吞吐量 ---|---|--- 50 个 CPU | 8.5k | 9.7 Gbps 9 个 CPU | 8.5k | 9.7 Gbps 8 个 CPU | 8.0k | 9.1 Gbps 7 个 CPU | 7.8k | 8.3 Gbps 6 个 CPU | 6.6k | 6.8 Gbps 5 个 CPU | 5.4k | 5.8 Gbps ...而且我已经用完了我的每月传输配额。

吞吐量会每秒波动。这些数字是大约一分钟的平均值。我认为在 9 个 CPU 上仍然受到网络限制。使用 8 个或更少的核心，显然受到 CPU 限制。

它与官方服务器相比如何？

官方 jetstream 服务器在架构上与 jetrelay 非常不同。每个客户端都有一个 "outbox"，用于缓冲该客户端的传出事件。当新事件到达时，它会立即复制到所有 "live" 客户端的 outbox 中。每个客户端都有一个关联的 goroutine，该 goroutine 通过 write() 将客户端的 outbox 排入其 socket 中。

另一个区别是，事件数据存储在 LSM 树中。我认为这也是考虑到过滤用例而完成的。但是，我怀疑它在某种程度上降低了性能上限。数据以 JSON 格式预先序列化存储，但是 websocket 标头每次发送事件时都会全新生成。

所以无论如何，它表现如何？该服务器包含一些按 IP 进行速率限制的逻辑，因此为了进行压力测试，我首先必须通过手术移除该代码。

完成后，我开始以与上述相同的方式对其进行测试。但是，即使有所有 50 个核心可用，我也无法使其超过 2 Gbps；并且典型的吞吐量更接近 1 Gbps。我尝试增加工作进程计数和每个客户端的速率，但没有观察到任何差异。

应该注意的是，官方服务器显然针对 "大量过滤" 情况进行了优化。我们正在对 "无过滤" 情况进行压力测试。这始终是对它的一个艰难考验。

但是，我注意到了一些奇怪的事情：客户端没有落后。尽管吞吐量很低，但所有客户端都继续观察到非常近期的事件。发生了什么？

事实证明，当服务器过载时，它会开始跳过事件！我不确定这是否是错误或有意为之。有趣的是，看起来所有客户端都丢弃了 相同的 事件。

所以...我不太确定官方服务器的最大吞吐量是多少。如果此事件删除是由于错误引起的，那么也许在修复该错误后可以获得超过 2 Gbps 的吞吐量。我真的试图根据它们的 架构 可以支持的吞吐量来比较这些 relay，而不是特定的实现。

总结

现在您了解了 Jetrelay 的工作原理，为什么不查看代码？这是 创建 sendfile() 请求的函数，而这是 在客户端完成时更新客户端游标的函数。这是 将事件写入文件的函数，而这是 在文件过大时删除旧数据的函数。

这是我第一次写博客文章。这是一项令人惊讶的工作！感谢 Jasper 和 Francesco 给我一些建议。

附录：其他 90%

Jetrelay 是一个技术演示。这是一个非详尽的列表，列出了在实际运行它之前你想要的东西：

运营相关：

安全相关（也许 nginx 可以提供其中一些？）：

缺少的功能：

最后但并非最不重要的一点是：更多测试，包括模糊测试。添加这些东西中的大多数应该没有问题；这只是需要工作。

在实际部署中，您可能希望在 nginx 后面运行 jetrelay，以实现虚拟主机和 TLS 等。Nginx 无法处理 jetrelay 可以处理的连接数和流量，因此在这种设置中，nginx 将成为瓶颈。

附录：关于 ATproto 和 "基于推送的互联网" 的思考

OG 互联网是基于拉取的：客户端请求一个资源，服务器将其提供给他们。但是在某些情况下，您希望在不必询问的情况下将东西发送给您。经典的例子是电子邮件：如果您只有拉取，那么您将被迫定期询问服务器 "有什么新东西吗？"；但理想情况下，服务器应该主动向您推送 "新电子邮件" 通知（此时您进行拉取以获取内容）。

“推送”的问题是您需要保持打开的 TCP 连接数。您需要与您想要接收通知的每个服务器建立一个打开的连接，这无法很好地扩展。与您的电子邮件提供商保持持久连接是一回事；但考虑一下微博：如果您关注的人跨越了一百个不同的服务器，您将需要保持一百个长期存在的 TCP 连接——哎哟！

因此，与拉取不同，推送需要一个中间人才能实际可行。这个中间人汇总所有通知，然后将它们扇出——现在您只需要订阅中间人！想想 Apple 的推送通知系统：您的手机连接到 Apple 的一台服务器，然后该服务器代表 WhatsApp、Uber 等中继推送通知。

您可以进行的另一个比较是与 RSS。RSS 的工作方式是每个发布者都有一个条目列表 ("feed.xml")，并且他们可以向列表顶部添加条目。Feed 聚合器定期轮询这些列表，并且当出现新条目时，他们会将通知推送到订阅者。非常明智。

ATproto 也很相似！但它做了一些改进。首先：数据结构。RSS 为您提供了一个仅附加列表。ATproto 为您提供了一个从路径到记录的映射。这意味着，除了创建新条目之外，您还可以编辑或删除旧条目。最终得到的是一个类似于文件系统的文件树......就像 HTTP！15 模型与 OG 互联网匹配这一事实意味着您可以为宁愿拉取的人通过 HTTP 公开相同的数据。为每个记录提供身份（以路径的形式）本身也非常有用。

与 RSS 的第二个改进是 ATproto 记录已签名。这使得 relay 无法将虚假更新归因于他人。理论上，RSS 聚合器可能会执行这种攻击（尽管我从未听说过）。请注意，相反的攻击（relay 选择性地删除事件）在 ATproto 中仍然是可能的。

从 relay 中删除信任是很好的，因为它意味着您可以使用物理上最接近的任何 relay，而不必担心操作员是谁。16

基于拉取的互联网有 HTTP。基于推送的互联网已经长期使用 RSS。一种新的、功能更强大的标准可能是一个伟大的发展。

脚注

  1. 通过将 relay 链接在一起，您可以快速将 feed 扇出到大量客户端。如果您将您的 relay 放置在战略位置，您也可以在世界各地分发 feed，而无需产生太多流量。↩︎
  2. 官方 jetstream 服务器将全功能 firehose 用作其上游数据源，但为了简单起见，jetrelay 使用另一个 jetstream 服务器作为其上游数据源。JSON 输入，JSON 输出。↩︎
  3. 官方 jetstream 服务器提供的另一个功能是 feed 的 zstd 压缩版本。我没有添加对此的支持，但是添加它将是微不足道的。↩︎
  4. 您可能想知道，“加密呢？” 对于 TLS 加密的 (wss://) websockets，网络上的位对于所有客户端来说 不 相同。（如果它们相同，那么它将不是一种非常安全的加密方法！）因此，此技巧不适用于加密流。但是！像 jetrelay 这样的服务器通常无论如何都不会以原生方式支持 TLS。这是因为您通常会在反向代理（nginx 或类似代理）后面运行它，以支持虚拟主机等；在这种情况下，您不妨让 nginx 为您处理 TLS。因此，无论 socket 直接指向我们的客户端还是指向 nginx，我们的工作都是尽快将未加密的 jetstream 数据放入这些 socket 中。↩︎
  5. 您创建一个 组播组，这是一种特殊的 socket，您可以将订阅者添加到其中，并且您发送到组 socket 的任何数据包都会镜像到所有订阅者。这非常方便！↩︎
  6. 消息必须适合网络定义的特定大小限制。如果您超过 508 字节，您可能会发现每次消息都被删除。↩︎
  7. 交付是不可靠的，因此客户端需要一种重新请求丢失的数据包的方法。↩︎
  8. 虽然各种可靠组播协议确实存在，但据我所知，没有一种非常流行。↩︎
  9. 这是我从 Kafka 学到的技巧。↩︎
  10. 每个客户端的线程架构 可以 在此规模下工作：6000 个线程并不疯狂。每个线程分配 8MiB 的堆栈空间，但这不是问题：此内存实际上在写入之前不会提交。主要问题是调度程序的压力。数千个线程不断唤醒和休眠，系统的其余部分将变得非常无响应。我确定这是一个可以解决的问题，只需要一些 cgroups wizardry......但我没有探索到足以知道的程度。io_uring 只是比使用大量线程更容易。↩︎
  11. Rustix 是一个 crate，它以一个漂亮的 rusty 包装器提供了 Linux 的用户空间 API。它有点像 libc，为 rust 重新构想（尽管总的来说，它比 libc "更薄"）。它比 libstd 更灵活，但同样用户友好。太棒了！↩︎
  12. 为什么是 BTreeMap 而不是 HashMap？您将在下一节中找到答案！↩︎
  13. 对于客户端来说，连接是一次性的事情。除非他们正在连接、读取一两个事件，然后在一个紧密的循环中断开连接……但这并不是我希望鼓励的行为！↩︎
  14. 为什么使用 2**16？我实际上不知道！但是这是我总是看到人们使用的限制。也许这只是一个 meme。↩︎
  15. ...除了路径使用点作为分隔符，而不是斜杠...但不是最后一个分隔符，那 是 一个斜杠。🤷↩︎
  16. 我认为能够理论上验证消息就足够了。大多数客户端实际上不会费心这样做，但这很好。只要有一些客户端正在检查 relay（并且只要 relay 不知道哪些客户端正在检查），任何偷偷摸摸的商业行为都有被抓住的风险。如果 relay 运营商与其用户签订了合同，这可能会导致声誉损害，甚至财务损害。↩︎