分离式 IPC 协议#

警告

实验性：分离式 IPC 协议目前处于实验阶段。根据反馈和使用情况，协议定义可能会在完全标准化之前发生变化。

原理#

Arrow IPC 格式 描述了一种将 Arrow 数据作为记录批流传输的协议。该协议期望一个连续的字节流，分为离散的消息（使用长度前缀和延续指示符）。每个离散消息由两部分组成

  • 一个 Flatbuffers 头消息

  • 一系列字节，由扁平化和打包的主体缓冲区组成（某些消息类型，如 Schema 消息，没有此部分） - 这在 IPC 格式规范中称为 *消息体*。

在大多数情况下，现有 IPC 格式的效率已经足够

  • 以 IPC 格式接收数据允许零拷贝利用主体缓冲区字节，无需反序列化即可形成 Arrow 数组

  • IPC 文件格式可以进行内存映射，因为它与位置无关，并且文件中的字节与内存中预期的字节完全相同。

但是，有一些用例无法处理

  • 构建 IPC 记录批消息需要分配一个连续的字节块，并将所有数据缓冲区复制到其中，并首尾相连地打包在一起。这会使将现有的、可直接使用的数据包装到 IPC 消息中的常见情况变得糟糕。

  • 即使 Arrow 数据位于跨进程边界或传输（例如 UCX）可访问的内存中，也没有标准方法可以将该共享位置指定给可以利用它的使用者。

  • 位于非 CPU 设备（例如 GPU）上的 Arrow 数据无法使用 Arrow IPC 发送，除非将数据复制回主机设备或将 Flatbuffers 元数据字节复制到设备内存中。

    • 同样，将 IPC 消息接收设备内存需要将 Flatbuffers 元数据复制回主机 CPU 设备。这是因为 IPC 流在单个流中交错数据和元数据。

此协议尝试以有效的方式解决这些用例。

目标#

  • 定义一个用于传递 Arrow IPC 数据的通用协议，不与任何特定传输绑定，还可以利用非 CPU 设备内存、共享内存和更新的“高性能”传输，例如 UCX 或 libfabric。

    • 这允许将主体中的数据保留在非 CPU 设备（如 GPU）上，而无需进行昂贵的设备到主机复制。

  • 通过将 IPC 元数据流与 IPC 主体字节分离，允许仅将 Flight RPC 用于控制流

定义#

IPC 元数据

包含 Arrow IPC 消息头的 Flatbuffers 消息字节

标签

一个用于流控制的小端 uint64 值，用于确定如何解释消息体。可以屏蔽特定位，以便仅通过标签的一部分来识别消息，其余位用于控制流或其他消息元数据。某些传输（例如 UCX）内置支持此类标签值，并且无论消息体是否驻留在非 CPU 设备上，都将在 CPU 内存中提供它们。

序列号

流的起始值为 0 的小端 4 字节无符号整数，指示消息的序列顺序。它还用于识别特定消息，将 IPC 元数据头与其对应的主体绑定，因为元数据和主体可以通过单独的管道/流/传输发送。

如果序列号达到 UINT32_MAX，则应允许其回绕，因为不太可能存在足够多的未处理消息等待处理会导致序列号重叠。

序列号有两个用途：识别相应的元数据和标记的主体数据消息，并确保我们不依赖于必须按顺序到达的消息。客户端应使用序列号在消息到达时对其进行正确排序以进行处理。

协议#

可以在 arrow-experiments 仓库 中找到利用 libcudf 和 UCX 的参考示例实现。

要求#

实现此协议的传输**必须**提供两个功能

  • 消息发送

    • 带分隔符的消息（如 gRPC），而不是不带分隔符的流（如没有进一步框架的普通 TCP）。

    • 或者，可以使用类似于 IPC 协议的 封装消息格式 的框架机制，同时省略主体字节。

  • 带标签的消息发送

    • 发送带有附加的小端无符号 64 位整型标签的消息，用于控制流。这样的标签允许在消息体位于非 CPU 设备上的情况下对消息进行控制流操作，而无需将消息本身复制到设备之外。

URI 规范#

当向使用者提供 URI 以便与此协议一起使用时（例如通过 Flight 的位置 URI），URI 应指定易于识别的方案，如 *ucx:* 或 *fabric:*。此外，URI 应编码以下 URI 查询参数

注意

随着此协议的成熟，本文档将更新与之一起使用的常用传输方案。

  • want_data - **必需** - uint64 整数值

    • 此值应用于标记发送到服务器的初始消息，以启动数据传输。初始消息的主体应该是所请求数据流的不透明二进制标识符（如 Flight RPC 协议中的 Ticket）

  • free_data - **可选** - uint64 整数值

    • 如果服务器可能使用偏移量/地址发送消息以进行远程内存访问或共享内存位置，则 URI 应包含此参数。此值用于标记从客户端发送到数据服务器的消息，其中包含客户端不再需要的特定偏移量/地址（即，任何直接引用这些内存位置的操作，例如将远程数据复制到本地内存，均已完成）。

  • remote_handle - **可选** - base64 编码的字符串

    • 使用共享内存或远程内存时，此值指示访问内存所需的任何句柄或标识符。

      • 使用 UCX，这将是 *rkey* 值

      • 使用 CUDA IPC，这将是基本 GPU 指针或内存句柄的值，后续地址将是相对于此基本指针的偏移量。

处理反压#

*目前*此提议未指定任何方法来管理消息的反压以限制内存和带宽原因。目前，这将是**传输定义的**，而不是锁定到次优的东西。

随着不同传输和库的使用量的增长，将会出现通用模式，允许以通用但有效的方式处理不同用例的反压。

注意

虽然协议本身与传输无关，但目前的使用和示例仅使用 UCX 和 libfabric 传输进行了测试，仅此而已。

协议描述#

可能发生两种情况

  1. 元数据流和主体数据流通过单独的连接发送

%% Licensed to the Apache Software Foundation (ASF) under one %% or more contributor license agreements. See the NOTICE file %% distributed with this work for additional information %% regarding copyright ownership. The ASF licenses this file %% to you under the Apache License, Version 2.0 (the %% "License"); you may not use this file except in compliance %% with the License. You may obtain a copy of the License at %% %% https://apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, %% software distributed under the License is distributed on an %% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY %% KIND, either express or implied. See the License for the %% specific language governing permissions and limitations %% under the License. sequenceDiagram participant D as Data Stream participant C as Client participant M as Metadata Stream activate C C-->>+M: TaggedMessage(server.want_data, bytes=ID_of_desired_data) C-->>+D: TaggedMessage(server.want_data, bytes=ID_of_desired_data) M-->>C: Message(bytes([1]) + le_bytes(sequence_number) + schema_metadata) loop each batch par M-->>C: Message(bytes([1]) + le_bytes(sequence_number) + batch_metadata) and alt D-->>C: TaggedMessage((bytes[0] << 55) | le_bytes(sequence_number),<br/>bytes=batch_data) else D-->>C: TaggedMessage((bytes[1] << 55) | le_bytes(sequence_number),<br/>bytes=uint64_pairs) end end end M-->>C: Message(bytes([0]) + le_bytes(sequence_number)) deactivate M loop C-->>D: TaggedMessage(server.free_data, bytes=uint64_list) end deactivate D deactivate C

  1. 元数据流和主体数据流通过同一连接同时发送

%% Licensed to the Apache Software Foundation (ASF) under one %% or more contributor license agreements. See the NOTICE file %% distributed with this work for additional information %% regarding copyright ownership. The ASF licenses this file %% to you under the Apache License, Version 2.0 (the %% "License"); you may not use this file except in compliance %% with the License. You may obtain a copy of the License at %% %% https://apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, %% software distributed under the License is distributed on an %% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY %% KIND, either express or implied. See the License for the %% specific language governing permissions and limitations %% under the License. sequenceDiagram participant C as Client participant S as Server activate C C-->>+S: TaggedMessage(server.want_data, bytes=ID_of_desired_data) S-->>C: Message(bytes([1]) + le_bytes(sequence_number) + schema_metadata) par loop each chunk S-->>C: Message(bytes([1]) + le_bytes(sequence_number) + batch_metadata) end S-->>C: Message(bytes([0]) + le_bytes(sequence_number)) and loop each chunk alt S-->>C: TaggedMessage((bytes[0] << 55) | le_bytes(sequence_number),<br/>bytes=batch_data) else S-->>C: TaggedMessage((bytes[1] << 55) | le_bytes(sequence_number),<br/>bytes=uint64_pairs) end end end loop C-->>S: TaggedMessage(server.free_data, bytes=uint64_list) end deactivate S deactivate C

服务器序列#

可以有一个服务器同时处理 IPC 元数据流和主体数据流，或者有单独的服务器分别处理 IPC 元数据和主体数据。如果需要，这允许通过单个传输管道或两个管道进行数据流传输。

元数据流序列#

服务器的常态是等待带有特定 <want_data> 标签值的**标记**消息以启动传输。此 <want_data> 值由服务器定义，并通过提供给客户端的 URI 传播给任何客户端。此协议未规定任何特定值，因此它不会干扰任何其他依赖标签值的现有协议。该消息的主体将包含一个不透明的二进制标识符，以指示要发送的特定数据集/数据流。

注意

例如，使用 *FlightInfo* 消息传递的**票证**将是此消息的主体。因为它是不透明的，所以它可以是服务器想要使用的任何东西。URI 和标识符不需要通过 Flight RPC 提供给客户端，但可以通过任何所需的传输或协议提供。

收到 <want_data> 请求后，服务器*应*通过发送由以下内容组成的消息流来响应该请求

block-beta columns 8 block:P["\n\n\n\nPrefix"]:5 T["Message type\nByte 0"] S["Sequence number\nBytes 1-4"] end H["Flatbuffer bytes\nRest of the message"]:3

  • 5 字节前缀

    • 消息的第一个字节指示消息类型，当前只允许两种消息类型（将来可能会添加更多类型）

      1. 流结束

      2. Flatbuffers IPC 元数据消息

    • 接下来的 4 个字节是一个小端无符号 32 位整数，指示消息的序列号。流中的第一条消息（**必须**始终是 schema 消息）**必须**具有序列号 0。每个后续消息**必须**将数字递增 1。

  • Arrow IPC 头部的完整 Flatbuffers 字节

根据 Arrow IPC 格式的定义，每个元数据消息可以表示数据块或字典，供数据流使用。

发送最后一个元数据消息后，服务器**必须**通过发送一个**正好** 5 个字节的消息来指示流的结束

  • 第一个字节是 0，表示**流结束**消息

  • 最后 4 个字节是序列号（4 字节，小端字节序的无符号整数）

数据流序列#

如果单个服务器同时处理数据流和元数据流，则**应该**与元数据消息并行地开始向客户端发送数据消息。否则，与元数据序列一样，服务器的待机状态是等待带有 <want_data> 标签值的**标记**消息，其主体指示要发送到客户端的数据集/数据流。

对于数据流中的每个 IPC 消息，如果该消息具有主体（即记录批次或字典消息），则**必须**在数据流上发送**标记**消息。每个消息的标签应按如下结构

block-beta columns 8 S["Sequence number\nBytes 0-3"]:4 U["Unused (Reserved)\nBytes 4-6"]:3 T["Message type\nByte 7"]:1

  • 标签的**最低有效** 4 字节（位 0 - 31）应该是消息的无符号 32 位小端字节序序列号。

  • 标签的**最高有效**字节（位 56 - 63）将消息正文**类型**指示为 8 位无符号整数。目前仅指定了两种消息类型，但可以根据需要添加更多消息类型以扩展协议

    1. 主体包含作为打包缓冲区的原始主体缓冲区字节（即标准 IPC 格式主体字节）

    2. 主体包含一系列无符号小端 64 位整数对，用于表示共享或远程内存，其结构如下

      • 前两个整数（例如，前 16 个字节）表示所有缓冲区的**总**大小（以字节为单位）以及此消息中缓冲区的数量（因此是后续 uint64 对的数量）

      • 每对后续的 uint64 值是一个地址/偏移量，后跟该特定缓冲区的长度。

  • 标签的所有未指定位（位 32 - 55）**保留**供此协议的潜在更新将来使用。目前，它们**必须**为 0。

注意

跨网络发送的任何共享/远程内存地址**必须**由服务器保持活动状态，直到收到相应的标记 <free_data> 消息。如果客户端在发送任何 <free_data> 消息之前断开连接，则如果服务器需要，可以假定清理内存是安全的。

发送最后一个标记的 IPC 正文消息后，服务器应保持连接并等待标记的 <free_data> 消息。这些 <free_data> 消息的结构很简单：一个或多个无符号小端 64 位整数，指示可以释放的地址/偏移量。

一旦没有更多未完成的地址需要释放，此流的工作就完成了。

客户端序列#

此协议的客户端需要同时处理可能来自同一服务器或不同服务器的消息的数据流和元数据流。下图显示了客户端如何处理元数据和数据流

%% Licensed to the Apache Software Foundation (ASF) under one %% or more contributor license agreements. See the NOTICE file %% distributed with this work for additional information %% regarding copyright ownership. The ASF licenses this file %% to you under the Apache License, Version 2.0 (the %% "License"); you may not use this file except in compliance %% with the License. You may obtain a copy of the License at %% https://apache.org/licenses/LICENSE-2.0 %% Unless required by applicable law or agreed to in writing, %% software distributed under the License is distributed on an %% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY %% KIND, either express or implied. See the License for the %% specific language governing permissions and limitations %% under the License. graph LR client((Client))-->c1{{Send #60;want_data#gt; Msg}} subgraph meta [Meta Message] direction LR m1[/Msg Type #40;byte 0#41;<br/>Seq Num #40;bytes 1-5#41;/]-- type 1 -->m2[[Process IPC Header]] m2-- IPC has body -->m3[Get Corresponding<br/>Tagged Msg] m2-- Schema Msg -->m4[/Store Schema/] m1-- type 0 -->e[Indicate End of Stream] end subgraph data [Data Stream] direction LR d1[Request Msg<br/>for Seq Num]-->d2{Most Significant<br/>Byte} d2-- 0 -->d3[Construct from<br/>Metadata and Body] d2-- 1 -->d4[Get shared/remote<br/>buffers] d4 -->d5[Construct from<br/>Metadata and buffers] d3 & d5 -->e2[Output Batch] end client -- recv untagged msg --> meta client -- get tagged msg --> data

  1. 首先，客户端发送一个标记消息，使用在 URI 中提供的 <want_data> 值作为标记，并将不透明 ID 作为正文。

    • 如果元数据服务器和数据服务器是分开的，则需要分别向每个服务器发送 <want_data> 消息。

    • 在任一情况下，元数据流和数据流都可以根据传输的性质并发和/或异步处理。

  2. 对于客户端在元数据流中接收到的每个**未标记**消息

    • 消息的第一个字节指示它是**流结束**消息（值 0）还是元数据消息（值 1）。

    • 接下来的 4 个字节是消息的序列号，一个无符号 32 位小端字节序整数。

    • 如果它**不是** *流结束* 消息，则其余字节是 IPC Flatbuffer 字节，可以正常解释。

      • 如果消息具有正文（即记录批次或字典消息），则客户端应使用相同的序列号从数据流中检索标记消息。

    • 如果它是**流结束**消息，则如果接收到的序列号中没有间隙，则可以安全地关闭元数据连接。

  3. 当收到需要正文的元数据消息时，**应该**将 0x00000000FFFFFFFF 的标签掩码与序列号一起使用，以匹配消息，而不管高位字节如何（例如，我们只关心将低 4 个字节与序列号匹配）

    • 收到后，最高有效字节的值决定客户端如何处理正文数据

      • 如果最高有效字节为 0：则消息的正文是原始 IPC 打包的正文缓冲区，允许使用相应的元数据头字节轻松处理它。

      • 如果最高有效字节为 1：消息的正文将包含一系列成对的无符号 64 位小端字节序整数。

        • 前两个整数表示 *1)* 所有正文缓冲区的总大小，以便在需要中间缓冲区时轻松分配，以及 *2)* 正在发送的缓冲区数量（nbuf）。

        • 消息的其余部分将是 nbuf 对整数，每个缓冲区一对。每对是 *1)* 缓冲区的地址/偏移量和 *2)* 该缓冲区的长度。然后可以通过基于底层传输的共享或远程内存例程来检索内存。**必须**保留这些地址/偏移量，以便以后可以在 <free_data> 消息中发送回它们，向服务器指示客户端不再需要共享内存。

  4. 收到*流结束*消息后，客户端应处理任何剩余的未处理 IPC 元数据消息。

  5. 在远程服务器能够释放单个内存地址/偏移量之后（在它发送这些地址/偏移量而不是完整正文字节的情况下），客户端应向服务器发送相应的 <free_data> 消息。

    • 单个 <free_data> 消息包含任意数量的无符号 64 位整数值，表示可以释放的地址/偏移量。它是*任意数量*的原因是允许客户端选择是发送多条消息来释放多个地址，还是将多个地址合并到更少的消息中来释放（从而使协议在需要时不那么“健谈”）

持续开发#

如果您决定在您自己的环境和系统中尝试此协议，我们非常欢迎您提供反馈并了解您的用例。由于这目前是一个**实验性**协议，我们需要实际使用它才能促进改进它并找到合适的泛化方法以在各种传输中实现标准化。

请使用 Arrow 开发人员邮件列表参与讨论： https://arrow.apache.org/community/#mailing-lists