分离式 IPC 协议#

警告

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

原理#

Arrow IPC 格式描述了一种将 Arrow 数据作为记录批次流进行传输的协议。该协议期望一个连续的字节流,该字节流被分成离散的消息(使用长度前缀和继续指示符)。每个离散消息包含两部分:

  • 一个 Flatbuffers 头部消息

  • 一系列字节,包含展平并打包的体缓冲区(某些消息类型,如 Schema 消息,没有此部分)——这在 IPC 格式规范中称为消息体

对于大多数情况,现有的 IPC 格式已经足够高效

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

  • IPC 文件格式可以进行内存映射,因为它与位置无关,并且文件的字节与内存中预期的一致。

然而,存在一些当前格式无法处理的使用场景

  • 构建 IPC 记录批次消息需要分配一个连续的字节块,并将所有数据缓冲区复制到其中,紧密地打包在一起。这使得将现有可直接使用的数据封装到 IPC 消息中的常见情况变得低效。

  • 即使 Arrow 数据位于可跨进程边界或传输(如 UCX)访问的内存中,也没有标准方法可以向可能利用该共享位置的消费者指定该位置。

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

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

本协议试图以高效的方式解决这些使用场景。

目标#

  • 定义一种通用的协议,用于传输 Arrow IPC 数据,不与任何特定传输绑定,并且允许利用非 CPU 设备内存、共享内存以及新的“高性能”传输(如 UCXlibfabric)。

    • 这使得可以将体中的数据保存在非 CPU 设备(如 GPU)上,而无需昂贵的设备到主机复制。

  • 允许纯粹使用 Flight RPC 进行控制流,方法是将 IPC 元数据流与 IPC 体字节分离

定义#

IPC 元数据

构成 Arrow IPC 消息头部的 Flatbuffers 消息字节

标签

一个小端序的 uint64 值,用于流控制和确定如何解释消息体。可以屏蔽特定位,仅根据标签的一部分来识别消息,剩余的位可用于控制流或其他消息元数据。某些传输(如 UCX)内置支持此类标签值,并且无论消息体是否驻留在非 CPU 设备上,都会在 CPU 内存中提供这些值。

序列号

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

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

序列号有两个作用:识别对应的元数据和带标签的体数据消息,并确保我们不依赖消息必须按顺序到达。客户端应使用序列号对到达的消息进行正确排序以便处理。

协议#

利用 libcudfUCX 的参考示例实现可以在 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 位整数,指示消息的序列号。流中的第一条消息(**必须**始终是 Schema 消息)的序列号**必须**为 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 字节,可以正常解析。

      • 如果消息有体(即 Record Batch 或 Dictionary 消息),则客户端应使用相同的序列号从数据流中检索相应的带标签消息。

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

  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