[Core] InfiniBand and RDMA support for Ray object store

[hellofinch]

Description

I use Ray in an HPC cluster. The cluster has InfiniBand which has low latency and high bandwidth. Ray is based on gRPC and data transferring uses gRPC, too. I can use IPoIB(Internet Protocol over InfiniBand) in the cluster. In this way, I can not make full use of IB's bandwidth. It has the potential to get better performance.

Use case

I want to help ray to support RDMA for object transferring.

RDMA, which can reduce CPU interruptions for network processing and increase CPU utilization, is good at transferring memory data with a better performance. With the help of InfiniBand, I think ray's performance will be improved.

Tensorflow, which is also based on gRPC, has supported many ways for distributed environments such as grpc+verbs, and grpc+MPI, etc. The gRPC is used for controlling computing. The verbs and MPI are used for data transfer. As for Ray, I think they are familiar. I hope to separate the object store part and make it use RDMA as TensorFlow does.

I also do some tests on ray and MPI with different net environments. Ray and MPI used TCP over ethernet for data transfer as a baseline. MPI can speed up 16X with IPoIB and Ray can speed up about 10X. MPI can speed up 90X with RDMA and I think ray can get familiar improvement.

I have reviewed Ray’s code and it is a sophisticated project. I try to focus on the object store part in Ray’s withepaper. It help me but I still can find out where I can start with. I lost myself in the code ocean. And the whitepaper is different from the code. I cannot figure out how the data is transferred when the ray run. Are there any up-to-date documents I can refer to?

Where should I start if I want to contribute to the Ray project? Is any import class or file I should pay attention to?

[richardliaw]

Hey awesome!

I believe the you can perhaps start at the object store: https://github.com/ray-project/ray/tree/master/src/ray/object_manager

cc @jjyao

[jiaodong]

Hi @hellofinch thanks for bringing this up ! Yes it is one of the key missing piece in Ray to better support HPC workloads and current object store only uses ethernet for object transfer that is expected to be significantly slower.

Do you have a particular workload / use case in mind that drives the need for IB support ? We have been hearing similar asks from the community but haven't reached critical mass of prioritization yet.

cc: @stephanie-wang @scv119

[jjyao]

cc @cadedaniel

[jjyao]

@hellofinch Thanks for willing to contribute! I can talk about the high level code that does the Ray object transfer:

On the data receiver side: we have pull_manager.cc that sends the object pull request to the sender side to initiate the data transfer.

On the data sender side, we have HandlePull which is the entry point for handling the pull request from the data receiver. Then if you follow the code, data sender will then push the data to the receiver via a push grpc.

Hopefully this helps you to get started with the code base. Feel free to ask if you have any questions!

[hellofinch]

Thanks for all your responses!
@jiaodong recently I've been reviewing the RLlib's paper(Scalable Reingforecement Learning Systems and their Applications). It mentioned that when training RL in distributed setting the bulk of data transfer between processes happens. In my opinion, when the number of rollout and policy gradient is growing, the shortcut will show out. So I think RDMA with IB will help.

@jjyao I try to start with the code base as you mentioned. I used to follow the program running and I tried to debug the whole ray code. I started the ray head and ray worker. I get many processes that have a relationship with ray such raylet and gcs. So when I want to know how one curtain function works, I get lost. Are there any unit tests or something else which can help me get a minimal example or demo to play with one certain function in Ray?

[jjyao]

@hellofinch what function you are referring to. For the data transfer, it happens inside raylet. If it helps, I can also do a quick video call to introduce the related code you are interested in.

[bestwooo]

It is exactly what I need! I can work on this together, if needed.

[SongGuyang]

What's the state of this issue?

Can we consider integrating brpc to support RDMA?

[tmbdev]

We have the same need for our HPC workloads.

In addition to RDMA into CPU memory, it would also be good to support CPU-GPU RDMA and GPU-GPU RDMA (https://developer.nvidia.com/gpudirect). Ideally, tensors (e.g., Torch tensors) whose data is on the GPU would be automatically and transparently transferred by Ray using GPUdirect RDMA whenever possible.

For distributed gradient descent (DDP) in deep learning, the frameworks already take care of this with internal communications primitives.

The main immediate use case for us is high performance data transfer from storage to GPUs. Each GPU easily consumes gigabytes per second in input data, and on machines with 16 GPUs, GPUdirect RDMA is really the only way of handling those data rates (such machines have separate networking cards for each pair of GPUs capable of transferring directly into GPU memory without going through the main PCI bus).

So, it would be great to see support for RDMA transfer for tensors, and for such transfers to go either to/from CPU memory and/or to/from GPU memory.

[richardliaw]

We would love to explore this path. However, we do not have much expertise in this domain, would it be possible to explore a collaboration here?

…

On Thu, Mar 2, 2023 at 2:34 PM Tom ***@***.***> wrote: We have the same need for our HPC workloads. In addition to RDMA into CPU memory, it would also be good to support CPU-GPU RDMA and GPU-GPU RDMA (https://developer.nvidia.com/gpudirect). Ideally, tensors (e.g., Torch tensors) whose data is on the GPU would be automatically and transparently transferred by Ray using GPUdirect RDMA whenever possible. For distributed gradient descent (DDP) in deep learning, the frameworks already take care of this with internal communications primitives. The main immediate use case for us is high performance data transfer from storage to GPUs. Each GPU easily consumes gigabytes per second in input data, and on machines with 16 GPUs, GPUdirect RDMA is really the only way of handling those data rates (such machines have separate networking cards for each pair of GPUs capable of transferring directly into GPU memory without going through the main PCI bus). So, it would be great to see support for RDMA transfer for tensors, and for such transfers to go either to/from CPU memory and/or to/from GPU memory. — Reply to this email directly, view it on GitHub <#30094 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABCRZZNVIML7BSH4YISTYVDW2EN5NANCNFSM6AAAAAAR2HHANI> . You are receiving this because you commented.Message ID: ***@***.***>

[jiaodong]

Thanks for your inputs @tmbdev, this makes a lot of sense and we're seeing a few more use cases from users. I have only worked on and used GPU-GPU RDMA (ray collective using NCCL) with a few questions regarding CPU-GPU RDMA:

• Does it require any particular type of instances (ex: aws that has HPC capabilities) of GPUs ?
• In a real use case that we recently ran into from a user:
  • trainer uses p4d.24xlarge that requires at least 10GB/s data reading throughput
  • we're trying to setup our dataset workers on commodity m5 CPU workers for data IO + preprocessing + networking
  • In this case -- will we still be able to use RDMA to accelerate data movement ? Is IB/EFA required on both instance types ?

Happy to learn more and collaborate on this as after a number of pushes we seem to be back on track to have p2p/collectives again :)

[tmbdev]

@jiaodong Thanks for the feedback. RoCE (RDMA over Ethernet) is basically Infiniband encapsulated in UDP. I believe the APIs are the same for RoCE and Infiniband. I think as far as Ray is concerned, it just needs support for the RDMA mechanisms. You can probably do all the necessary development and testing with Soft-RoCE on both ends.

https://enterprise-support.nvidia.com/s/article/howto-configure-soft-roce

Whether any particular hardware and networking configuration actually gives you high performance then depends a great deal on the details of the configuration. The p4d.24xlarge seems to have Infiniband cards for the GPUs, but presumably the commodity CPU workers are on Ethernet. You can probably make that configuration work, but whether it is efficient depends on what hardware is doing the RoCE to Infiniband translation. But if there are hardware performance bottlenecks, they are going to be fixed one way or another pretty soon. They shouldn't affect the design of any Ray RDMA primitives.

I think another question is whether any such support in Ray can be based on NCCL or needs other libraries. Right now, NCCL in Ray does not support CPU-to-GPU communications, but there are plugins that may enable that. UCX or UCX-Py may also be options.

https://developer.nvidia.com/blog/high-performance-python-communication-with-ucx-py/

[pentschev]

Right now, NCCL in Ray does not support CPU-to-GPU communications, but there are plugins that may enable that. UCX or UCX-Py may also be options.

I'm one of the UCX-Py developers, and I see Ray is mostly built on C++, so it may not be a great fit there as UCX-Py mainly provides Python bindings to UCX and a Python asyncio interface. However, the timing seems perfect as we are finalizing the work on what is currently named UCXX, which is a complete rewrite of UCX-Py by means of a C++ backend. libucxx is an object-oriented interface for the UCP layer of UCX, which is hopefully more convenient to be used in Ray, and it is also the core piece of ucxx, the new Python bindings and Python asyncio interface. All transports that are currently supported by UCX should be also supported by UCXX, including InfiniBand and NVLink/NVSwitch which we actively test, other transports such as RoCE should also work out-of-the-box but are currently untested.

UCX does not support disk access via GPUDirect Storage, and neither does NCCL, for that KvikIO may be used instead.

Also NCCL is targeted more at collective communication patterns, if that is something that may be leveraged in Ray, NCCL may be a better alternative than UCX, but I do not have benchmarks to guarantee that you will indeed see better performance. The UCX community has also been working a on the Unified Collective Communication (UCC), but that is not currently supported by UCXX and I'm not much involved and do not know how it compares to NCCL currently.

If you decide to experiment with UCXX in Ray, please feel free to ping me and I'll try to assist.