戰(zhàn)：包裝庫進(jìn)程、Ray Collective 與內(nèi)嵌 HTTP 服務(wù)器)
Ray Actor 帶外Out-of-Band通信實(shí)戰(zhàn)包裝庫進(jìn)程、Ray Collective 與內(nèi)嵌 HTTP 服務(wù)器【免費(fèi)下載鏈接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.項(xiàng)目地址: https://gitcode.com/gh_mirrors/ra/rayRay 的 Actor 之間默認(rèn)通過方法調(diào)用通信、通過分布式對(duì)象存儲(chǔ)共享數(shù)據(jù)但在包裝第三方分布式庫、追求高性能集合通信、以及向集群外部暴露服務(wù)等場景下繞過 Ray 控制面的帶外通信Out-of-band Communication反而更高效。本文基于 out-of-band-communication.rst 展開結(jié)合倉庫中ray.util.collective的實(shí)現(xiàn)與示例代碼講清楚三類帶外通信的適用場景、配置方法與底層機(jī)制讀完你可以在自己的 Ray 應(yīng)用中獨(dú)立實(shí)現(xiàn)與 Horovod/Spark 等庫進(jìn)程協(xié)同、CPU/GPU 上的 NCCL/GLOO 集合通信以及 actor 內(nèi)嵌 HTTP 服務(wù)。為什么需要帶外通信Ray 的常規(guī) Actor 通信模型是方法調(diào)用 對(duì)象存儲(chǔ)客戶端通過actor.method.remote()提交任務(wù)數(shù)據(jù)以 ObjectRef 的形式放入分布式對(duì)象存儲(chǔ)由 Ray 的調(diào)度器與引用計(jì)數(shù)系統(tǒng)統(tǒng)一管理。這套模型的好處是容錯(cuò)、引用管理、內(nèi)存生命周期都由 Ray 負(fù)責(zé)無需用戶操心。但存在兩類場景讓帶外通信更有價(jià)值庫進(jìn)程協(xié)同很多分布式計(jì)算庫如 Horovod、Spark早已擁有成熟、高性能的內(nèi)部通信棧NCCL、MPI、內(nèi)部 RPC 等它們把 Ray 當(dāng)作語言級(jí) Actor 調(diào)度器讓 Ray 負(fù)責(zé)進(jìn)程的創(chuàng)建與編排而實(shí)際的數(shù)據(jù)交換交給這些庫自身的通信棧完成——這就是帶外通信。性能與語義訴求集合通信allreduce、broadcast 等經(jīng)過對(duì)象存儲(chǔ)中轉(zhuǎn)會(huì)產(chǎn)生額外序列化、拷貝與調(diào)度開銷直接讓參與進(jìn)程通過 NCCL/GLOO 等在帶外互連性能更高。此外向 Ray 集群之外的客戶端暴露服務(wù)HTTP/gRPC 端點(diǎn)本質(zhì)上也是帶外通信。場景一包裝第三方庫進(jìn)程許多庫已經(jīng)具備成熟的內(nèi)部通信機(jī)制Ray 只承擔(dān)語言集成式 Actor 調(diào)度器的角色庫進(jìn)程之間的實(shí)際通信大多走既有通信棧。文檔中給出的典型例子Horovod-on-Ray使用 NCCL 或 MPI 進(jìn)行集合通信Ray 只負(fù)責(zé)拉起并管理參與訓(xùn)練的進(jìn)程RayDPRay Spark使用 Spark 自身的內(nèi)部 RPC 和對(duì)象管理器Object Manager進(jìn)行數(shù)據(jù)交換。倉庫中也有對(duì)應(yīng)的驗(yàn)證性測試?yán)?horovod_example.py 與 test_horovod.py以及 test_raydp.py可以看到 Ray 在這些組合中扮演調(diào)度與生命周期管理角色、而把通信讓位給第三方棧的協(xié)作方式。這種模式的核心收益是不必把成熟庫的通信協(xié)議翻譯成 Ray 的方法調(diào)用避免性能損耗同時(shí)仍能享受 Ray 的彈性調(diào)度、故障恢復(fù)與資源管理能力。場景二Ray Collective 集合通信庫Ray 自帶的集合通信庫ray.util.collective是文檔重點(diǎn)推薦的帶外通信方案用于在分布式 CPU 或 GPU 之間進(jìn)行高效的集合通信與點(diǎn)對(duì)點(diǎn)通信。其完整說明見 ray-collective.rst實(shí)現(xiàn)集中在 collective.py。架構(gòu)與后端從源碼看collective.pyray.util.collective采用后端注冊(cè)機(jī)制導(dǎo)入時(shí)會(huì)嘗試注冊(cè)兩個(gè)后端——NCCL對(duì)應(yīng)NCCLGroup位于nccl_collective_group.py和 GLOO對(duì)應(yīng)TorchGLOOGroup基于torch.distributed.gloo。若對(duì)應(yīng)依賴未安裝注冊(cè)會(huì)被跳過可通過nccl_available()/gloo_available()/is_backend_available()查詢可用性。支持矩陣來自 ray-collective.rst后端GLOOtorch.distributed.glooNCCL設(shè)備CPUGPUCPUGPUsend / recv????broadcast????allreduce????reduce????allgather????reduce_scatter????barrier????gather / scatter / all-to-all????支持三種張量類型torch.Tensor、numpy.ndarray、cupy.ndarray。ReduceOp枚舉定義于 types.py支持SUM、PRODUCT、MIN、MAX。安裝方面ray.util.collective已隨 Ray wheel 打包只需按后端補(bǔ)充第三方依賴——GLOO 需要torchNCCL 需要cupy-cudaxxxx 對(duì)應(yīng)環(huán)境的 CUDA 版本。創(chuàng)建集合通信組命令式與聲明式集合原語都作用在組collective group上組內(nèi)的一組進(jìn)程通常是 Ray 管理的 Actor 或 Task會(huì)一起進(jìn)入集合調(diào)用。使用前必須先靜態(tài)聲明這些進(jìn)程為組。文檔與 ray-collective.rst 給出了兩種初始化 API。命令式在每個(gè)進(jìn)程內(nèi)部初始化init_collective_group(world_size, rank, backend, group_name)參數(shù)含義見 collective.pyworld_size為組內(nèi)總進(jìn)程數(shù)rank為當(dāng)前進(jìn)程排名backend為 NCCL 或 GLOOgroup_name為組名同一進(jìn)程可屬于多個(gè)組組名是唯一標(biāo)識(shí)gloo_timeout為 GLOO 操作超時(shí)毫秒。源碼還會(huì)校驗(yàn)rank必須滿足0 rank world_size。聲明式在 driver 中聲明create_collective_group(actors, world_size, ranks, backend, group_name)把一組 Actor 一次性聲明為組。源碼collective.py會(huì)校驗(yàn)每個(gè) actor 對(duì)應(yīng)一個(gè) rank、ranks 必須是0..len(ranks)-1的排列、后端必須已注冊(cè)且可用隨后會(huì)創(chuàng)建一個(gè)名為info_group_name的 detached NamedActor 來保存組信息供各進(jìn)程后續(xù)查詢。一個(gè)完整的 NCCL allreduce 示例精簡自 nccl_allreduce_example.pyimport cupy as cp import ray import ray.util.collective as collective ray.remote(num_gpus1) class Worker: def __init__(self): self.send cp.ones((4,), dtypecp.float32) self.recv cp.zeros((4,), dtypecp.float32) def setup(self, world_size, rank): collective.init_collective_group(world_size, rank, nccl, default) return True def compute(self): collective.allreduce(self.send, default) return self.send def destroy(self): collective.destroy_group() if __name__ __main__: ray.init(num_gpus2) num_workers 2 workers, init_rets [], [] for i in range(num_workers): w Worker.remote() workers.append(w) init_rets.append(w.setup.remote(num_workers, i)) _ ray.get(init_rets) results ray.get([w.compute.remote() for w in workers]) ray.shutdown()對(duì)同一組進(jìn)程可以構(gòu)建多個(gè)集合組以group_name區(qū)分從而在進(jìn)程的不同子集之間描述更復(fù)雜的通信拓?fù)?。集合通信原語的行為特征ray.util.collective當(dāng)前提供的集合 API 是命令式的文檔明確了三個(gè)行為特征同步阻塞所有集合 API 都是阻塞調(diào)用必須全員參與每個(gè) API 只描述通信的一部分需要組內(nèi)每個(gè)參與進(jìn)程都發(fā)起調(diào)用并相互 rendezvous會(huì)合后通信才真正發(fā)生并繼續(xù)帶外執(zhí)行API 必須在參與進(jìn)程actor/task的代碼內(nèi)部使用。常用的集合原語在 collective.py 中的簽名包括原語簽名要點(diǎn)行號(hào)allreduce(tensor, group_name, op)全組歸約默認(rèn)ReduceOp.SUML312barrier(group_name)組內(nèi)同步屏障L352reduce(tensor, dst_rank, group_name, op)歸約到指定 rankL362broadcast(tensor, src_rank, group_name)從src_rank廣播L421allgather(tensor_list, tensor, group_name)全組收集L468reducescatter(tensor_list, tensor, group_name, op)歸約后再分發(fā)L511send(tensor, dst_rank, group_name)/recv(tensor, src_rank, group_name)點(diǎn)對(duì)點(diǎn)收發(fā)L567 / L624點(diǎn)對(duì)點(diǎn)通信與常見反模式P2P 的send/recv與集合函數(shù)行為一致同步阻塞成對(duì)出現(xiàn)。示例邏輯來自 ray-collective.rst# 假設(shè) A、B 已在同一個(gè)組中rank 分別為 0、1 ray.get([A.do_send.remote(target_rank1), B.do_recv.remote(src_rank0)]) # 正確 ray.get([A.do_send.remote(target_rank1)]) # 反模式會(huì)掛起文檔明確警告只調(diào)用 send 而不實(shí)例化對(duì)面的 recv 是反模式程序會(huì)一直掛起——因?yàn)?send/recv 必須成功 rendezvous 才能推進(jìn)。多 GPU 集合通信當(dāng)一臺(tái)機(jī)器有多個(gè) GPU 時(shí)可利用 NVLINK 等 GPU-GPU 帶寬顯著提升通信性能。ray.util.collective提供多 GPU 版本 API如allreduce_multigpu、send_multigpu、recv_multigpu允許一個(gè)進(jìn)程如num_gpus4的 actor管理多塊 GPU。相關(guān)限制見 ray-collective.rst僅支持 NCCL 后端參與多 GPU 調(diào)用的進(jìn)程必須擁有相同數(shù)量的 GPU輸入通常是張量列表每個(gè)張量位于調(diào)用方進(jìn)程持有的一塊不同 GPU 上。多 GPU 場景比單 GPU 原語 與 GPU 數(shù)量相同的進(jìn)程通常更具性能優(yōu)勢。倉庫在 examples 下提供了nccl_allreduce_multigpu_example.py、nccl_p2p_example_multigpu.py等可直接運(yùn)行參考的示例。環(huán)境變量const.py 定義了NCCL_USE_MULTISTREAM環(huán)境變量默認(rèn)開啟用于控制 NCCL 后端是否使用多流multistream執(zhí)行可作為調(diào)優(yōu)入口。場景三在 Actor 內(nèi)嵌 HTTP 服務(wù)器另一種常見的帶外通信是在 actor 內(nèi)部啟動(dòng)一個(gè) HTTP 服務(wù)器并對(duì)外暴露端點(diǎn)讓Ray 集群之外的用戶可以直接與 actor 通信而不必經(jīng)過 Ray 的方法調(diào)用層。文檔給出的完整示例位于 actor-http-server.pyimport ray import asyncio import requests from aiohttp import web ray.remote class Counter: async def __init__(self): self.counter 0 asyncio.get_running_loop().create_task(self.run_http_server()) async def run_http_server(self): app web.Application() app.add_routes([web.get(/, self.get)]) runner web.AppRunner(app) await runner.setup() site web.TCPSite(runner, 127.0.0.1, 25001) await site.start() async def get(self, request): return web.Response(textstr(self.counter)) async def increment(self): self.counter self.counter 1 ray.init() counter Counter.remote() [ray.get(counter.increment.remote()) for i in range(5)] r requests.get(http://127.0.0.1:25001/) assert r.text 5這個(gè)示例的精妙之處在于把兩條通信路徑打通并協(xié)同帶內(nèi)路徑driver 通過counter.increment.remote()調(diào)用 actor 方法 5 次遞增內(nèi)部計(jì)數(shù)器帶外路徑actor 的__init__中通過asyncio.get_running_loop().create_task(...)啟動(dòng)一個(gè) aiohttp 服務(wù)器監(jiān)聽127.0.0.1:25001外部客戶端示例中是requests直接發(fā)起 HTTP 請(qǐng)求讀取當(dāng)前計(jì)數(shù)值最終斷言返回5驗(yàn)證兩條路徑操作的是同一份狀態(tài)。要點(diǎn)說明actor 構(gòu)造函數(shù)是異步的async def __init__在初始化時(shí)就把 HTTP 服務(wù)任務(wù)注冊(cè)進(jìn)事件循環(huán)從而在 actor 存活期間持續(xù)服務(wù)采用asyncio.create_task而非在構(gòu)造函數(shù)內(nèi)await阻塞保證__init__能及時(shí)返回、actor 正常完成創(chuàng)建監(jiān)聽地址與端口可按需調(diào)整示例固定為127.0.0.1:25001若需要被集群外客戶端訪問應(yīng)綁定到可達(dá)的 IP 并處理好防火墻/網(wǎng)絡(luò)策略。同樣的思路也適用于暴露其他協(xié)議的服務(wù)器例如在 actor 內(nèi)啟動(dòng)gRPC 服務(wù)器文檔明確提及對(duì)外提供 gRPC 端點(diǎn)。這意味著你可以把 actor 當(dāng)作一個(gè)常駐的分布式微服務(wù)進(jìn)程用 Ray 管理其生命周期與資源同時(shí)用其自帶的服務(wù)器協(xié)議與外部系統(tǒng)互操作。限制與注意事項(xiàng)文檔在最后明確提醒帶外通信的關(guān)鍵限制Ray 不管理 actor 之間的帶外調(diào)用。方法調(diào)用/對(duì)象存儲(chǔ)路徑上 Ray 提供的分布式引用計(jì)數(shù)distributed reference counting、內(nèi)存生命周期管理等保障對(duì)帶外通信不生效不要把對(duì)象引用ObjectRef通過帶外通道傳遞。由于帶外通道繞過了 Ray 的引用計(jì)數(shù)對(duì)象引用的創(chuàng)建與釋放無法被正確追蹤可能造成對(duì)象被提前回收或泄漏引發(fā)難以排查的錯(cuò)誤。因此實(shí)踐中的紅線是帶外通道只傳普通數(shù)據(jù)張量、序列化值、業(yè)務(wù)消息等絕不傳 ObjectRef需要 Ray 參與管理的對(duì)象引用一律走常規(guī)的remote()調(diào)用與ray.get()/ray.put()路徑。小結(jié)帶外通信是 Ray Actor 編程模型的重要補(bǔ)充倉庫文檔總結(jié)的三條路徑各有定位包裝庫進(jìn)程Horovod-on-Ray、RayDP 等把通信交給成熟第三方棧NCCL/MPI、Spark RPCRay 專注調(diào)度與生命周期Ray Collectiveray.util.collective在 Ray 內(nèi)部提供 NCCL/GLOO 雙后端的集合與點(diǎn)對(duì)點(diǎn)原語覆蓋 allreduce、broadcast、send/recv、多 GPU 等場景是高性能分布式訓(xùn)練的首選內(nèi)嵌服務(wù)器在 actor 內(nèi)啟動(dòng) HTTP/gRPC 服務(wù)把 actor 變成集群外部可直接訪問的服務(wù)端點(diǎn)適用于對(duì)外提供接口或接入外部系統(tǒng)的場景。使用時(shí)要始終牢記帶外通信繞過 Ray 的管理面Ray 的分布式引用計(jì)數(shù)不適用切勿在帶外通道中傳遞對(duì)象引用。把握住這一邊界帶外通信就能成為你手中連接Ray 調(diào)度能力與高性能通信/對(duì)外服務(wù)的橋梁?！久赓M(fèi)下載鏈接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.項(xiàng)目地址: https://gitcode.com/gh_mirrors/ra/ray創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 Ray Actor 帶外（Out-of-Band）通信實(shí)戰(zhàn)：包裝庫進(jìn)程、Ray Collective 與內(nèi)嵌 HTTP 服務(wù)器
Ray Actor 帶外Out-of-Band通信實(shí)戰(zhàn)包裝庫進(jìn)程、Ray Collective 與內(nèi)嵌 HTTP 服務(wù)器【免費(fèi)下載鏈接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.項(xiàng)目地址: https://gitcode.com/gh_mirrors/ra/rayRay 的 Actor 之間默認(rèn)通過方法調(diào)用通信、通過分布式對(duì)象存儲(chǔ)共享數(shù)據(jù)但在包裝第三方分布式庫、追求高性能集合通信、以及向集群外部暴露服務(wù)等場景下繞過 Ray 控制面的帶外通信Out-of-band Communication反而更高效。本文基于 out-of-band-communication.rst 展開結(jié)合倉庫中ray.util.collective的實(shí)現(xiàn)與示例代碼講清楚三類帶外通信的適用場景、配置方法與底層機(jī)制讀完你可以在自己的 Ray 應(yīng)用中獨(dú)立實(shí)現(xiàn)與 Horovod/Spark 等庫進(jìn)程協(xié)同、CPU/GPU 上的 NCCL/GLOO 集合通信以及 actor 內(nèi)嵌 HTTP 服務(wù)。為什么需要帶外通信Ray 的常規(guī) Actor 通信模型是方法調(diào)用 對(duì)象存儲(chǔ)客戶端通過actor.method.remote()提交任務(wù)數(shù)據(jù)以 ObjectRef 的形式放入分布式對(duì)象存儲(chǔ)由 Ray 的調(diào)度器與引用計(jì)數(shù)系統(tǒng)統(tǒng)一管理。這套模型的好處是容錯(cuò)、引用管理、內(nèi)存生命周期都由 Ray 負(fù)責(zé)無需用戶操心。但存在兩類場景讓帶外通信更有價(jià)值庫進(jìn)程協(xié)同很多分布式計(jì)算庫如 Horovod、Spark早已擁有成熟、高性能的內(nèi)部通信棧NCCL、MPI、內(nèi)部 RPC 等它們把 Ray 當(dāng)作語言級(jí) Actor 調(diào)度器讓 Ray 負(fù)責(zé)進(jìn)程的創(chuàng)建與編排而實(shí)際的數(shù)據(jù)交換交給這些庫自身的通信棧完成——這就是帶外通信。性能與語義訴求集合通信allreduce、broadcast 等經(jīng)過對(duì)象存儲(chǔ)中轉(zhuǎn)會(huì)產(chǎn)生額外序列化、拷貝與調(diào)度開銷直接讓參與進(jìn)程通過 NCCL/GLOO 等在帶外互連性能更高。此外向 Ray 集群之外的客戶端暴露服務(wù)HTTP/gRPC 端點(diǎn)本質(zhì)上也是帶外通信。場景一包裝第三方庫進(jìn)程許多庫已經(jīng)具備成熟的內(nèi)部通信機(jī)制Ray 只承擔(dān)語言集成式 Actor 調(diào)度器的角色庫進(jìn)程之間的實(shí)際通信大多走既有通信棧。文檔中給出的典型例子Horovod-on-Ray使用 NCCL 或 MPI 進(jìn)行集合通信Ray 只負(fù)責(zé)拉起并管理參與訓(xùn)練的進(jìn)程RayDPRay Spark使用 Spark 自身的內(nèi)部 RPC 和對(duì)象管理器Object Manager進(jìn)行數(shù)據(jù)交換。倉庫中也有對(duì)應(yīng)的驗(yàn)證性測試?yán)?horovod_example.py 與 test_horovod.py以及 test_raydp.py可以看到 Ray 在這些組合中扮演調(diào)度與生命周期管理角色、而把通信讓位給第三方棧的協(xié)作方式。這種模式的核心收益是不必把成熟庫的通信協(xié)議翻譯成 Ray 的方法調(diào)用避免性能損耗同時(shí)仍能享受 Ray 的彈性調(diào)度、故障恢復(fù)與資源管理能力。場景二Ray Collective 集合通信庫Ray 自帶的集合通信庫ray.util.collective是文檔重點(diǎn)推薦的帶外通信方案用于在分布式 CPU 或 GPU 之間進(jìn)行高效的集合通信與點(diǎn)對(duì)點(diǎn)通信。其完整說明見 ray-collective.rst實(shí)現(xiàn)集中在 collective.py。架構(gòu)與后端從源碼看collective.pyray.util.collective采用后端注冊(cè)機(jī)制導(dǎo)入時(shí)會(huì)嘗試注冊(cè)兩個(gè)后端——NCCL對(duì)應(yīng)NCCLGroup位于nccl_collective_group.py和 GLOO對(duì)應(yīng)TorchGLOOGroup基于torch.distributed.gloo。若對(duì)應(yīng)依賴未安裝注冊(cè)會(huì)被跳過可通過nccl_available()/gloo_available()/is_backend_available()查詢可用性。支持矩陣來自 ray-collective.rst后端GLOOtorch.distributed.glooNCCL設(shè)備CPUGPUCPUGPUsend / recv????broadcast????allreduce????reduce????allgather????reduce_scatter????barrier????gather / scatter / all-to-all????支持三種張量類型torch.Tensor、numpy.ndarray、cupy.ndarray。ReduceOp枚舉定義于 types.py支持SUM、PRODUCT、MIN、MAX。安裝方面ray.util.collective已隨 Ray wheel 打包只需按后端補(bǔ)充第三方依賴——GLOO 需要torchNCCL 需要cupy-cudaxxxx 對(duì)應(yīng)環(huán)境的 CUDA 版本。創(chuàng)建集合通信組命令式與聲明式集合原語都作用在組collective group上組內(nèi)的一組進(jìn)程通常是 Ray 管理的 Actor 或 Task會(huì)一起進(jìn)入集合調(diào)用。使用前必須先靜態(tài)聲明這些進(jìn)程為組。文檔與 ray-collective.rst 給出了兩種初始化 API。命令式在每個(gè)進(jìn)程內(nèi)部初始化init_collective_group(world_size, rank, backend, group_name)參數(shù)含義見 collective.pyworld_size為組內(nèi)總進(jìn)程數(shù)rank為當(dāng)前進(jìn)程排名backend為 NCCL 或 GLOOgroup_name為組名同一進(jìn)程可屬于多個(gè)組組名是唯一標(biāo)識(shí)gloo_timeout為 GLOO 操作超時(shí)毫秒。源碼還會(huì)校驗(yàn)rank必須滿足0 rank world_size。聲明式在 driver 中聲明create_collective_group(actors, world_size, ranks, backend, group_name)把一組 Actor 一次性聲明為組。源碼collective.py會(huì)校驗(yàn)每個(gè) actor 對(duì)應(yīng)一個(gè) rank、ranks 必須是0..len(ranks)-1的排列、后端必須已注冊(cè)且可用隨后會(huì)創(chuàng)建一個(gè)名為info_group_name的 detached NamedActor 來保存組信息供各進(jìn)程后續(xù)查詢。一個(gè)完整的 NCCL allreduce 示例精簡自 nccl_allreduce_example.pyimport cupy as cp import ray import ray.util.collective as collective ray.remote(num_gpus1) class Worker: def __init__(self): self.send cp.ones((4,), dtypecp.float32) self.recv cp.zeros((4,), dtypecp.float32) def setup(self, world_size, rank): collective.init_collective_group(world_size, rank, nccl, default) return True def compute(self): collective.allreduce(self.send, default) return self.send def destroy(self): collective.destroy_group() if __name__ __main__: ray.init(num_gpus2) num_workers 2 workers, init_rets [], [] for i in range(num_workers): w Worker.remote() workers.append(w) init_rets.append(w.setup.remote(num_workers, i)) _ ray.get(init_rets) results ray.get([w.compute.remote() for w in workers]) ray.shutdown()對(duì)同一組進(jìn)程可以構(gòu)建多個(gè)集合組以group_name區(qū)分從而在進(jìn)程的不同子集之間描述更復(fù)雜的通信拓?fù)?。集合通信原語的行為特征ray.util.collective當(dāng)前提供的集合 API 是命令式的文檔明確了三個(gè)行為特征同步阻塞所有集合 API 都是阻塞調(diào)用必須全員參與每個(gè) API 只描述通信的一部分需要組內(nèi)每個(gè)參與進(jìn)程都發(fā)起調(diào)用并相互 rendezvous會(huì)合后通信才真正發(fā)生并繼續(xù)帶外執(zhí)行API 必須在參與進(jìn)程actor/task的代碼內(nèi)部使用。常用的集合原語在 collective.py 中的簽名包括原語簽名要點(diǎn)行號(hào)allreduce(tensor, group_name, op)全組歸約默認(rèn)ReduceOp.SUML312barrier(group_name)組內(nèi)同步屏障L352reduce(tensor, dst_rank, group_name, op)歸約到指定 rankL362broadcast(tensor, src_rank, group_name)從src_rank廣播L421allgather(tensor_list, tensor, group_name)全組收集L468reducescatter(tensor_list, tensor, group_name, op)歸約后再分發(fā)L511send(tensor, dst_rank, group_name)/recv(tensor, src_rank, group_name)點(diǎn)對(duì)點(diǎn)收發(fā)L567 / L624點(diǎn)對(duì)點(diǎn)通信與常見反模式P2P 的send/recv與集合函數(shù)行為一致同步阻塞成對(duì)出現(xiàn)。示例邏輯來自 ray-collective.rst# 假設(shè) A、B 已在同一個(gè)組中rank 分別為 0、1 ray.get([A.do_send.remote(target_rank1), B.do_recv.remote(src_rank0)]) # 正確 ray.get([A.do_send.remote(target_rank1)]) # 反模式會(huì)掛起文檔明確警告只調(diào)用 send 而不實(shí)例化對(duì)面的 recv 是反模式程序會(huì)一直掛起——因?yàn)?send/recv 必須成功 rendezvous 才能推進(jìn)。多 GPU 集合通信當(dāng)一臺(tái)機(jī)器有多個(gè) GPU 時(shí)可利用 NVLINK 等 GPU-GPU 帶寬顯著提升通信性能。ray.util.collective提供多 GPU 版本 API如allreduce_multigpu、send_multigpu、recv_multigpu允許一個(gè)進(jìn)程如num_gpus4的 actor管理多塊 GPU。相關(guān)限制見 ray-collective.rst僅支持 NCCL 后端參與多 GPU 調(diào)用的進(jìn)程必須擁有相同數(shù)量的 GPU輸入通常是張量列表每個(gè)張量位于調(diào)用方進(jìn)程持有的一塊不同 GPU 上。多 GPU 場景比單 GPU 原語 與 GPU 數(shù)量相同的進(jìn)程通常更具性能優(yōu)勢。倉庫在 examples 下提供了nccl_allreduce_multigpu_example.py、nccl_p2p_example_multigpu.py等可直接運(yùn)行參考的示例。環(huán)境變量const.py 定義了NCCL_USE_MULTISTREAM環(huán)境變量默認(rèn)開啟用于控制 NCCL 后端是否使用多流multistream執(zhí)行可作為調(diào)優(yōu)入口。場景三在 Actor 內(nèi)嵌 HTTP 服務(wù)器另一種常見的帶外通信是在 actor 內(nèi)部啟動(dòng)一個(gè) HTTP 服務(wù)器并對(duì)外暴露端點(diǎn)讓Ray 集群之外的用戶可以直接與 actor 通信而不必經(jīng)過 Ray 的方法調(diào)用層。文檔給出的完整示例位于 actor-http-server.pyimport ray import asyncio import requests from aiohttp import web ray.remote class Counter: async def __init__(self): self.counter 0 asyncio.get_running_loop().create_task(self.run_http_server()) async def run_http_server(self): app web.Application() app.add_routes([web.get(/, self.get)]) runner web.AppRunner(app) await runner.setup() site web.TCPSite(runner, 127.0.0.1, 25001) await site.start() async def get(self, request): return web.Response(textstr(self.counter)) async def increment(self): self.counter self.counter 1 ray.init() counter Counter.remote() [ray.get(counter.increment.remote()) for i in range(5)] r requests.get(http://127.0.0.1:25001/) assert r.text 5這個(gè)示例的精妙之處在于把兩條通信路徑打通并協(xié)同帶內(nèi)路徑driver 通過counter.increment.remote()調(diào)用 actor 方法 5 次遞增內(nèi)部計(jì)數(shù)器帶外路徑actor 的__init__中通過asyncio.get_running_loop().create_task(...)啟動(dòng)一個(gè) aiohttp 服務(wù)器監(jiān)聽127.0.0.1:25001外部客戶端示例中是requests直接發(fā)起 HTTP 請(qǐng)求讀取當(dāng)前計(jì)數(shù)值最終斷言返回5驗(yàn)證兩條路徑操作的是同一份狀態(tài)。要點(diǎn)說明actor 構(gòu)造函數(shù)是異步的async def __init__在初始化時(shí)就把 HTTP 服務(wù)任務(wù)注冊(cè)進(jìn)事件循環(huán)從而在 actor 存活期間持續(xù)服務(wù)采用asyncio.create_task而非在構(gòu)造函數(shù)內(nèi)await阻塞保證__init__能及時(shí)返回、actor 正常完成創(chuàng)建監(jiān)聽地址與端口可按需調(diào)整示例固定為127.0.0.1:25001若需要被集群外客戶端訪問應(yīng)綁定到可達(dá)的 IP 并處理好防火墻/網(wǎng)絡(luò)策略。同樣的思路也適用于暴露其他協(xié)議的服務(wù)器例如在 actor 內(nèi)啟動(dòng)gRPC 服務(wù)器文檔明確提及對(duì)外提供 gRPC 端點(diǎn)。這意味著你可以把 actor 當(dāng)作一個(gè)常駐的分布式微服務(wù)進(jìn)程用 Ray 管理其生命周期與資源同時(shí)用其自帶的服務(wù)器協(xié)議與外部系統(tǒng)互操作。限制與注意事項(xiàng)文檔在最后明確提醒帶外通信的關(guān)鍵限制Ray 不管理 actor 之間的帶外調(diào)用。方法調(diào)用/對(duì)象存儲(chǔ)路徑上 Ray 提供的分布式引用計(jì)數(shù)distributed reference counting、內(nèi)存生命周期管理等保障對(duì)帶外通信不生效不要把對(duì)象引用ObjectRef通過帶外通道傳遞。由于帶外通道繞過了 Ray 的引用計(jì)數(shù)對(duì)象引用的創(chuàng)建與釋放無法被正確追蹤可能造成對(duì)象被提前回收或泄漏引發(fā)難以排查的錯(cuò)誤。因此實(shí)踐中的紅線是帶外通道只傳普通數(shù)據(jù)張量、序列化值、業(yè)務(wù)消息等絕不傳 ObjectRef需要 Ray 參與管理的對(duì)象引用一律走常規(guī)的remote()調(diào)用與ray.get()/ray.put()路徑。小結(jié)帶外通信是 Ray Actor 編程模型的重要補(bǔ)充倉庫文檔總結(jié)的三條路徑各有定位包裝庫進(jìn)程Horovod-on-Ray、RayDP 等把通信交給成熟第三方棧NCCL/MPI、Spark RPCRay 專注調(diào)度與生命周期Ray Collectiveray.util.collective在 Ray 內(nèi)部提供 NCCL/GLOO 雙后端的集合與點(diǎn)對(duì)點(diǎn)原語覆蓋 allreduce、broadcast、send/recv、多 GPU 等場景是高性能分布式訓(xùn)練的首選內(nèi)嵌服務(wù)器在 actor 內(nèi)啟動(dòng) HTTP/gRPC 服務(wù)把 actor 變成集群外部可直接訪問的服務(wù)端點(diǎn)適用于對(duì)外提供接口或接入外部系統(tǒng)的場景。使用時(shí)要始終牢記帶外通信繞過 Ray 的管理面Ray 的分布式引用計(jì)數(shù)不適用切勿在帶外通道中傳遞對(duì)象引用。把握住這一邊界帶外通信就能成為你手中連接Ray 調(diào)度能力與高性能通信/對(duì)外服務(wù)的橋梁?！久赓M(fèi)下載鏈接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.項(xiàng)目地址: https://gitcode.com/gh_mirrors/ra/ray創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考