monarch_rdma/backend/ibverbs/
mlx_queue_pair.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! mlx5 queue pair built on the mlx5dv extended verbs.
10
11use std::io::Error;
12use std::result::Result;
13
14use super::IbvBuffer;
15use super::domain::IbvDomain;
16use super::domain::IbvDomainImpl;
17use super::primitives::GidScope;
18use super::primitives::GidType;
19use super::primitives::IbvConfig;
20use super::primitives::IbvCq;
21use super::primitives::IbvQp;
22use super::primitives::IbvQpInfo;
23use super::primitives::IbvWc;
24use super::queue_pair::IbvQueuePair;
25use super::queue_pair::PollCompletionError;
26use super::queue_pair::PollTarget;
27use super::queue_pair::RCQueuePair;
28use super::queue_pair::WorkRequestError;
29
30/// An mlx5 RC queue pair created through `mlx5dv_create_qp`, so it carries the
31/// mlx5dv send-ops flags that arm a direct-WQE/doorbell data path.
32///
33/// Operations delegate to an inner [`RCQueuePair`], driving the QP
34/// through the standard `ibv_post_send`/`ibv_poll_cq` verbs; the mlx5dv
35/// send-ops flags are not exercised by that path.
36#[derive(Debug)]
37pub struct MlxQueuePair(RCQueuePair);
38
39impl MlxQueuePair {
40    /// Creates the `mlx5dv` RC QP and the two completion queues backing it,
41    /// against `domain`'s context and PD, returned as an [`IbvQp`] that owns the
42    /// CQs and PD. The QP carries the mlx5dv send-ops flags that arm its
43    /// extended work-request builder. A null context or PD on `domain` yields
44    /// `Err`.
45    ///
46    /// # Safety
47    ///
48    /// `domain`'s context and PD, if non-null, must be live.
49    pub(super) unsafe fn create_ibv_qp<I: IbvDomainImpl>(
50        domain: &IbvDomain<I>,
51        config: &IbvConfig,
52    ) -> Result<IbvQp, anyhow::Error> {
53        let context = domain.context().as_ptr();
54        let pd = domain.as_ptr();
55        if pd.is_null() {
56            anyhow::bail!("cannot create an MlxQueuePair on a null protection domain");
57        }
58
59        // Separate send/recv completion queues, each owning a clone of the
60        // device context. Each `IbvCq` destroys its queue on drop, so an early
61        // return below cleans them up.
62        // SAFETY: `domain`'s context is live (an `IbvDomain` holds a null-or-live
63        // context); `IbvCq::create` rejects a null context.
64        let send_cq = unsafe { IbvCq::create(domain.context().clone(), config.cq_entries) }?;
65        // SAFETY: as for `send_cq` above.
66        let recv_cq = unsafe { IbvCq::create(domain.context().clone(), config.cq_entries) }?;
67
68        // An mlx5dv extended RC QP with the caps from `config`. The
69        // `SEND_OPS_FLAGS` enable the mlx5dv extended work-request builder; the
70        // standard verbs this QP runs on ignore them.
71        let mut init_attr = rdmaxcel_sys::ibv_qp_init_attr_ex {
72            send_cq: send_cq.as_ptr(),
73            recv_cq: recv_cq.as_ptr(),
74            cap: rdmaxcel_sys::ibv_qp_cap {
75                max_send_wr: config.max_send_wr,
76                max_recv_wr: config.max_recv_wr,
77                max_send_sge: config.max_send_sge,
78                max_recv_sge: config.max_recv_sge,
79                max_inline_data: 0,
80            },
81            qp_type: rdmaxcel_sys::ibv_qp_type::IBV_QPT_RC,
82            sq_sig_all: 0,
83            pd,
84            comp_mask: rdmaxcel_sys::IBV_QP_INIT_ATTR_PD
85                | rdmaxcel_sys::IBV_QP_INIT_ATTR_SEND_OPS_FLAGS,
86            send_ops_flags: (rdmaxcel_sys::IBV_QP_EX_WITH_RDMA_WRITE
87                | rdmaxcel_sys::IBV_QP_EX_WITH_RDMA_READ
88                | rdmaxcel_sys::IBV_QP_EX_WITH_SEND) as u64,
89            ..Default::default()
90        };
91        let mut mlx5dv_attr = rdmaxcel_sys::mlx5dv_qp_init_attr {
92            comp_mask: rdmaxcel_sys::MLX5DV_QP_INIT_ATTR_MASK_SEND_OPS_FLAGS as u64,
93            send_ops_flags: (rdmaxcel_sys::MLX5DV_QP_EX_WITH_MKEY_CONFIGURE
94                | rdmaxcel_sys::MLX5DV_QP_EX_WITH_MR_LIST) as u64,
95            ..Default::default()
96        };
97        // SAFETY: `context` and `pd` are non-null (checked above) and live (an
98        // `IbvDomain` holds null-or-live pointers); both attr structs are fully
99        // initialized and outlive the call, and their CQ pointers came from the
100        // freshly created `send_cq`/`recv_cq`. `mlx5dv_create_qp` returns null on
101        // failure.
102        let qp =
103            unsafe { rdmaxcel_sys::mlx5dv_create_qp(context, &mut init_attr, &mut mlx5dv_attr) };
104        if qp.is_null() {
105            // `send_cq`/`recv_cq` drop here, destroying the CQs.
106            anyhow::bail!(
107                "failed to create mlx5dv queue pair (QP): {}",
108                Error::last_os_error()
109            );
110        }
111        // SAFETY: `qp` is a live RC QP just created against `pd` with
112        // `send_cq`/`recv_cq`; `IbvQp` takes ownership of all of them plus the PD
113        // and destroys them in order on drop.
114        Ok(unsafe { IbvQp::from_raw(qp, send_cq, recv_cq, domain.pd().clone()) })
115    }
116}
117
118impl IbvQueuePair for MlxQueuePair {
119    unsafe fn new<I: IbvDomainImpl<QueuePair = Self>>(
120        domain: &IbvDomain<I>,
121        config: IbvConfig,
122    ) -> Result<Self, anyhow::Error> {
123        tracing::debug!("creating an MlxQueuePair from config {}", config);
124        let gid = domain.device_info().select_gid(
125            config.port_num,
126            Some(GidScope::Global),
127            Some(GidType::RoCEv2),
128        )?;
129        // SAFETY: an `IbvDomain` holds a null-or-live context and PD, which is
130        // `create_ibv_qp`'s contract.
131        let qp = unsafe { Self::create_ibv_qp(domain, &config) }?;
132        let access_flags = domain.access_flags();
133        // SAFETY: `create_ibv_qp` returns a live, fully non-null `IbvQp` (it
134        // bails on any null handle).
135        Ok(MlxQueuePair(unsafe {
136            RCQueuePair::from_qp(qp, config, gid, access_flags)
137        }))
138    }
139
140    fn connect(&mut self, info: &IbvQpInfo) -> Result<(), anyhow::Error> {
141        self.0.connect(info)
142    }
143
144    fn get_qp_info(&mut self) -> Result<IbvQpInfo, anyhow::Error> {
145        self.0.get_qp_info()
146    }
147
148    fn state(&mut self) -> Result<u32, anyhow::Error> {
149        self.0.state()
150    }
151
152    fn put(
153        &mut self,
154        remote_dst: IbvBuffer,
155        local_src: IbvBuffer,
156    ) -> Result<Vec<u64>, anyhow::Error> {
157        self.0.put(remote_dst, local_src)
158    }
159
160    fn get(
161        &mut self,
162        local_dst: IbvBuffer,
163        remote_src: IbvBuffer,
164    ) -> Result<Vec<u64>, anyhow::Error> {
165        self.0.get(local_dst, remote_src)
166    }
167
168    fn poll_completion(
169        &mut self,
170        target: PollTarget,
171    ) -> Result<Option<Result<IbvWc, WorkRequestError>>, PollCompletionError> {
172        self.0.poll_completion(target)
173    }
174}
monarch_rdma/backend/ibverbs/mlx_queue_pair.rs

monarch_rdma/backend/ibverbs/
mlx_queue_pair.rs
