feat: add RDMA listener to Mayastor Nvmf target

io-engine/src/bin/io-engine.rs

@@ -108,7 +108,7 @@ fn start_tokio_runtime(args: &MayastorCliArgs) {
 
     if args.rdma {
         env::set_var("ENABLE_RDMA", "true");
-        warn!("RDMA is enabled for Mayastor NVMEoF target");
+        warn!("RDMA is requested to be enabled for Mayastor NVMEoF target");
     }
 
     unsafe {

io-engine/src/core/bdev.rs

@@ -11,7 +11,7 @@ use snafu::ResultExt;
 use spdk_rs::libspdk::{spdk_bdev, spdk_get_ticks_hz};
 
 use crate::{
-    bdev::bdev_event_callback,
+    bdev::{bdev_event_callback, nexus::NEXUS_MODULE_NAME},
     bdev_api::bdev_uri_eq,
     core::{
         share::{NvmfShareProps, Protocol, Share, UpdateProps},
@@ -210,6 +210,7 @@ where
     ) -> Result<Self::Output, Self::Error> {
         let me = unsafe { self.get_unchecked_mut() };
         let props = NvmfShareProps::from(props);
+        let is_nexus_bdev = me.driver() == NEXUS_MODULE_NAME;
 
         let ptpl = props.ptpl().as_ref().map(|ptpl| ptpl.path());
 
@@ -232,7 +233,7 @@ where
             .await
             .context(ShareNvmf {})?;
 
-        subsystem.start().await.context(ShareNvmf {})
+        subsystem.start(is_nexus_bdev).await.context(ShareNvmf {})
     }
 
     fn create_ptpl(&self) -> Result<Option<PtplProps>, Self::Error> {

io-engine/src/core/env.rs

@@ -834,6 +834,11 @@ impl MayastorEnvironment {
             .map(|s| s.clone())
     }
 
+    /// Check if RDMA needs to be enabled for Mayastor nvmf target.
+    pub fn rdma(&self) -> bool {
+        self.rdma
+    }
+
     /// Detects IP address for NVMF target by the interface specified in CLI
     /// arguments.
     fn detect_nvmf_tgt_iface_ip(iface: &str) -> Result<String, String> {

io-engine/src/subsys/config/opts.rs

@@ -35,6 +35,7 @@ use std::{
 };
 
 use crate::core::MayastorEnvironment;
+use strum_macros::{AsRefStr, EnumString, EnumVariantNames};
 
 pub trait GetOpts {
     fn get(&self) -> Self;
@@ -84,6 +85,30 @@ impl GetOpts for NexusOpts {
 /// Must be equal to the size of `spdk_nvmf_target_opts.crdt`.
 pub const TARGET_CRDT_LEN: usize = 3;
 
+#[derive(Clone, Default, EnumString, EnumVariantNames, AsRefStr)]
+#[strum(serialize_all = "lowercase")]
+pub enum NvmfTgtTransport {
+    Rdma,
+    #[default]
+    Tcp,
+}
+
+impl NvmfTransportOpts {
+    /// Tweak a few opts more suited for rdma.
+    fn for_rdma(mut self) -> Self {
+        self.in_capsule_data_size = try_from_env(
+            "NVMF_RDMA_IN_CAPSULE_DATA_SIZE",
+            self.in_capsule_data_size,
+        );
+        self.io_unit_size = try_from_env("NVMF_RDMA_IO_UNIT_SIZE", 8192); // SPDK_NVMF_RDMA_MIN_IO_BUFFER_SIZE
+        self.data_wr_pool_size =
+            try_from_env("NVMF_RDMA_DATA_WR_POOL_SIZE", 4095); // SPDK_NVMF_RDMA_DEFAULT_DATA_WR_POOL_SIZE
+        self.num_shared_buf =
+            try_from_env("NVMF_RDMA_NUM_SHARED_BUF", self.num_shared_buf);
+        self
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(default, deny_unknown_fields)]
 pub struct NvmfTgtConfig {
@@ -94,11 +119,13 @@ pub struct NvmfTgtConfig {
     /// NVMF target Command Retry Delay in x100 ms.
     pub crdt: [u16; TARGET_CRDT_LEN],
     /// TCP transport options
-    pub opts: NvmfTcpTransportOpts,
+    pub opts_tcp: NvmfTransportOpts,
     /// NVMF target interface (ip, mac, name or subnet).
     pub interface: Option<String>,
     /// Enable RDMA for NVMF target or not
     pub rdma: Option<bool>,
+    /// RDMA transport options.
+    pub opts_rdma: NvmfTransportOpts,
 }
 
 impl From<NvmfTgtConfig> for Box<spdk_nvmf_target_opts> {
@@ -126,9 +153,10 @@ impl Default for NvmfTgtConfig {
             name: "mayastor_target".to_string(),
             max_namespaces: 2048,
             crdt: args.nvmf_tgt_crdt,
-            opts: NvmfTcpTransportOpts::default(),
+            opts_tcp: NvmfTransportOpts::default(),
             interface: None,
             rdma: None,
+            opts_rdma: NvmfTransportOpts::default().for_rdma(),
         }
     }
 }
@@ -139,10 +167,10 @@ impl GetOpts for NvmfTgtConfig {
     }
 }
 
-/// Settings for the TCP transport
+/// Nvmf settings for the transports
 #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
 #[serde(default, deny_unknown_fields)]
-pub struct NvmfTcpTransportOpts {
+pub struct NvmfTransportOpts {
     /// max queue depth
     max_queue_depth: u16,
     /// max qpairs per controller
@@ -264,7 +292,7 @@ where
     }
 }
 
-impl Default for NvmfTcpTransportOpts {
+impl Default for NvmfTransportOpts {
     fn default() -> Self {
         Self {
             max_queue_depth: try_from_env("NVMF_TCP_MAX_QUEUE_DEPTH", 32),
@@ -283,16 +311,16 @@ impl Default for NvmfTcpTransportOpts {
             acceptor_poll_rate: try_from_env("NVMF_ACCEPTOR_POLL_RATE", 10_000),
             zcopy: try_from_env("NVMF_ZCOPY", 1) == 1,
             ack_timeout: try_from_env("NVMF_ACK_TIMEOUT", 0),
-            data_wr_pool_size: try_from_env("NVMF_DATA_WR_POOL_SIZE", 0),
+            data_wr_pool_size: 0,
         }
     }
 }
 
 /// we cannot add derives for YAML to these structs directly, so we need to
 /// copy them. The upside though, is that if the FFI structures change, we will
 /// know about it during compile time.
-impl From<NvmfTcpTransportOpts> for spdk_nvmf_transport_opts {
-    fn from(o: NvmfTcpTransportOpts) -> Self {
+impl From<NvmfTransportOpts> for spdk_nvmf_transport_opts {
+    fn from(o: NvmfTransportOpts) -> Self {
         struct_size_init!(
             Self {
                 max_queue_depth: o.max_queue_depth,

io-engine/src/subsys/nvmf/subsystem.rs

@@ -48,6 +48,7 @@ use spdk_rs::{
         spdk_nvmf_subsystem_state_change_done,
         spdk_nvmf_subsystem_stop,
         spdk_nvmf_tgt,
+        spdk_nvmf_tgt_get_transport,
         SPDK_NVME_SCT_GENERIC,
         SPDK_NVME_SC_CAPACITY_EXCEEDED,
         SPDK_NVME_SC_RESERVATION_CONFLICT,
@@ -68,8 +69,13 @@ use crate::{
     ffihelper::{cb_arg, done_cb, AsStr, FfiResult, IntoCString},
     lvs::Lvol,
     subsys::{
+        config::opts::NvmfTgtTransport,
         make_subsystem_serial,
-        nvmf::{transport::TransportId, Error, NVMF_TGT},
+        nvmf::{
+            transport::{TransportId, RDMA_TRANSPORT},
+            Error,
+            NVMF_TGT,
+        },
         Config,
     },
 };
@@ -798,7 +804,10 @@ impl NvmfSubsystem {
     }
 
     // we currently allow all listeners to the subsystem
-    async fn add_listener(&self) -> Result<(), Error> {
+    async fn add_listener(
+        &self,
+        transport: NvmfTgtTransport,
+    ) -> Result<(), Error> {
         extern "C" fn listen_cb(arg: *mut c_void, status: i32) {
             let s = unsafe { Box::from_raw(arg as *mut oneshot::Sender<i32>) };
             s.send(status).unwrap();
@@ -807,8 +816,8 @@ impl NvmfSubsystem {
         let cfg = Config::get();
 
         // dont yet enable both ports, IOW just add one transportID now
-
-        let trid_replica = TransportId::new(cfg.nexus_opts.nvmf_replica_port);
+        let trid_replica =
+            TransportId::new(cfg.nexus_opts.nvmf_replica_port, transport);
 
         let (s, r) = oneshot::channel::<i32>();
         unsafe {
@@ -907,8 +916,23 @@ impl NvmfSubsystem {
     /// start the subsystem previously created -- note that we destroy it on
     /// failure to ensure the state is not in limbo and to avoid leaking
     /// resources
-    pub async fn start(self) -> Result<String, Error> {
-        self.add_listener().await?;
+    pub async fn start(self, need_rdma: bool) -> Result<String, Error> {
+        self.add_listener(NvmfTgtTransport::Tcp).await?;
+        // Only attempt rdma listener addition for this subsystem after making
+        // sure the Mayastor nvmf tgt has rdma transport created.
+        if need_rdma && self.nvmf_tgt_has_rdma_xprt() {
+            let _ =
+                self.add_listener(NvmfTgtTransport::Rdma)
+                    .await
+                    .map_err(|e| {
+                        warn!(
+                            "NvmfSubsystem RDMA listener add failed {}. \
+                        Subsystem will be accessible over TCP only.\
+                        {:?}",
+                            e, self
+                        );
+                    });
+        }
 
         if let Err(e) = self
             .change_state("start", |ss, cb, arg| unsafe {
@@ -958,9 +982,15 @@ impl NvmfSubsystem {
     }
 
     /// get ANA state
+    /// XXX: The SPDK NVME multipath is transport protocol independent. Should
+    /// we keep returning the ana state here by default using TCP transport
+    /// as today?
     pub async fn get_ana_state(&self) -> Result<u32, Error> {
         let cfg = Config::get();
-        let trid_replica = TransportId::new(cfg.nexus_opts.nvmf_replica_port);
+        let trid_replica = TransportId::new(
+            cfg.nexus_opts.nvmf_replica_port,
+            NvmfTgtTransport::Tcp,
+        );
         let listener = unsafe {
             nvmf_subsystem_find_listener(self.0.as_ptr(), trid_replica.as_ptr())
         };
@@ -981,29 +1011,35 @@ impl NvmfSubsystem {
             let s = unsafe { Box::from_raw(arg as *mut oneshot::Sender<i32>) };
             s.send(status).unwrap();
         }
-        let cfg = Config::get();
-        let trid_replica = TransportId::new(cfg.nexus_opts.nvmf_replica_port);
 
-        let (s, r) = oneshot::channel::<i32>();
+        // setting ANA state can only be done when subsystem is shared, meaning
+        // it'll have listeners configured. So let's fetch transport ids
+        // based on active listeners instead of reading static Config.
+        let trids = self.listeners_to_vec().unwrap_or_default();
 
-        unsafe {
-            spdk_nvmf_subsystem_set_ana_state(
-                self.0.as_ptr(),
-                trid_replica.as_ptr(),
-                ana_state,
-                0,
-                Some(set_ana_state_cb),
-                cb_arg(s),
-            );
+        for trid in trids {
+            debug!("set_ana_state {ana_state}, {trid:?}");
+            let (s, r) = oneshot::channel::<i32>();
+            unsafe {
+                spdk_nvmf_subsystem_set_ana_state(
+                    self.0.as_ptr(),
+                    trid.as_ptr(),
+                    ana_state,
+                    0,
+                    Some(set_ana_state_cb),
+                    cb_arg(s),
+                );
+            }
+            r.await
+                .expect("Cancellation is not supported")
+                .to_result(|e| Error::Subsystem {
+                    source: Errno::from_i32(-e),
+                    nqn: self.get_nqn(),
+                    msg: "failed to set_ana_state of the subsystem".to_string(),
+                })?;
         }
 
-        r.await
-            .expect("Cancellation is not supported")
-            .to_result(|e| Error::Subsystem {
-                source: Errno::from_i32(-e),
-                nqn: self.get_nqn(),
-                msg: "failed to set_ana_state of the subsystem".to_string(),
-            })
+        Ok(())
     }
 
     /// destroy all subsystems associated with our target, subsystems must be in
@@ -1073,6 +1109,18 @@ impl NvmfSubsystem {
         Bdev::checked_from_ptr(unsafe { spdk_nvmf_ns_get_bdev(ns) })
     }
 
+    fn nvmf_tgt_has_rdma_xprt(&self) -> bool {
+        NVMF_TGT.with(|t| {
+            let transport = unsafe {
+                spdk_nvmf_tgt_get_transport(
+                    t.borrow().tgt.as_ptr(),
+                    RDMA_TRANSPORT.as_ptr(),
+                )
+            };
+            !transport.is_null()
+        })
+    }
+
     fn listeners_to_vec(&self) -> Option<Vec<TransportId>> {
         unsafe {
             let mut listener =