分离式 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 内存中提供它们。

序列号#

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

如果序列号达到 UINT32_MAX，应允许其回滚，因为不太可能有足够的未处理消息等待处理，从而导致序列号重叠。

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

协议#

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

要求#

实现此协议的传输方式必须提供两项功能

  • 消息发送

    • 带分隔符的消息（如 gRPC），而不是无分隔符的流（如没有额外帧的纯 TCP）。

    • 或者，可以使用像 IPC 协议的 封装消息格式 这样的帧机制，但省略主体字节。

  • 带标签的消息发送

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

URI 规范#

当向消费者提供用于此协议的 URI (例如通过 Flight 的 Location 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 消息一起传递的 ticket 将是此消息的主体。由于它是不透明的，它可以是服务器希望使用的任何东西。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 位整数，指示消息的序列号。流中的第一条消息（必须始终是模式消息）必须具有序列号 0。随后的每条消息必须将数字递增 1。

  • Arrow IPC 头的完整 Flatbuffers 字节

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

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

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

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

数据流序列#

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

对于数据流中的每条 IPC 消息，如果该消息包含主体（即 Record Batch 或 Dictionary 消息），则必须在数据流上发送一条带标签的消息。每条消息的 标签 结构应如下所示：

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