feat: add RDMA listener to Mayastor Nvmf target

This adds the capability to listen for rdma connections to the Mayastor
Nvmf target if the rdma feature is enabled during installation. Any Nvmf
subsystem facing the host i.e. the nexus nvmf subsystem will now be able
to support tcp and rdma both.

Signed-off-by: Diwakar Sharma <diwakar.sharma@datacore.com>

Author: dsharma-dc

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,54 @@ 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
+    }
+}
+
+impl Default for NvmfTransportOpts {
+    fn default() -> Self {
+        Self {
+            max_queue_depth: try_from_env("NVMF_TCP_MAX_QUEUE_DEPTH", 32),
+            in_capsule_data_size: 4096,
+            max_io_size: 131_072,
+            io_unit_size: 131_072,
+            max_qpairs_per_ctrl: try_from_env(
+                "NVMF_TCP_MAX_QPAIRS_PER_CTRL",
+                32,
+            ),
+            num_shared_buf: try_from_env("NVMF_TCP_NUM_SHARED_BUF", 2047),
+            buf_cache_size: try_from_env("NVMF_TCP_BUF_CACHE_SIZE", 64),
+            dif_insert_or_strip: false,
+            max_aq_depth: 32,
+            abort_timeout_sec: 1,
+            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: 0,
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(default, deny_unknown_fields)]
 pub struct NvmfTgtConfig {
@@ -94,11 +143,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 +177,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 +191,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,35 +316,11 @@ where
     }
 }
 
-impl Default for NvmfTcpTransportOpts {
-    fn default() -> Self {
-        Self {
-            max_queue_depth: try_from_env("NVMF_TCP_MAX_QUEUE_DEPTH", 32),
-            in_capsule_data_size: 4096,
-            max_io_size: 131_072,
-            io_unit_size: 131_072,
-            max_qpairs_per_ctrl: try_from_env(
-                "NVMF_TCP_MAX_QPAIRS_PER_CTRL",
-                32,
-            ),
-            num_shared_buf: try_from_env("NVMF_TCP_NUM_SHARED_BUF", 2047),
-            buf_cache_size: try_from_env("NVMF_TCP_BUF_CACHE_SIZE", 64),
-            dif_insert_or_strip: false,
-            max_aq_depth: 32,
-            abort_timeout_sec: 1,
-            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),
-        }
-    }
-}
-
 /// 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 =