feat(config): extends config yaml with core config and delay mode

This allows configuration of the core pin per node.

Signed-off-by: Tiago Castro <[email protected]>

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

@@ -291,7 +291,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let ms = MayastorEnvironment::new(args.clone()).init();
     start_tokio_runtime(&args);
 
-    Reactors::current().running();
+    Reactors::current().init_running();
     Reactors::current().poll_reactor();
 
     ms.fini();

io-engine/src/core/env.rs

@@ -249,11 +249,23 @@ pub struct MayastorCliArgs {
     /// LVM pools can then be created by specifying the LVM pool type.
     /// If LVM is enabled and LVM_SUPPRESS_FD_WARNINGS is not set then it will
     /// be set to 1.
-    #[clap(long = "enable-lvm", env = "ENABLE_LVM")]
+    #[clap(long = "enable-lvm", env = "ENABLE_LVM", value_parser = delay_compat)]
     pub lvm: bool,
     /// Enables experimental Snapshot Rebuild support.
-    #[clap(long = "enable-snapshot-rebuild", env = "ENABLE_SNAPSHOT_REBUILD")]
+    #[clap(long = "enable-snapshot-rebuild", env = "ENABLE_SNAPSHOT_REBUILD", value_parser = delay_compat)]
     pub snap_rebuild: bool,
+    /// Reactors sleep 1ms before each poll.
+    /// # Warning: Don't use this in production.
+    #[clap(long, env = "MAYASTOR_DELAY", hide = true, value_parser = delay_compat)]
+    pub developer_delay: bool,
+}
+
+fn delay_compat(s: &str) -> Result<bool, String> {
+    match s {
+        "1" | "true" => Ok(true),
+        "" | "0" | "false" => Ok(false),
+        _else => Err(format!("Must be one of: 1,true,0,false")),
+    }
 }
 
 /// Mayastor features.
@@ -312,6 +324,7 @@ impl Default for MayastorCliArgs {
             enable_nexus_channel_debug: false,
             lvm: false,
             snap_rebuild: false,
+            developer_delay: false,
         }
     }
 }
@@ -411,6 +424,7 @@ pub struct MayastorEnvironment {
     api_versions: Vec<ApiVersion>,
     skip_sig_handler: bool,
     enable_io_all_thrd_nexus_channels: bool,
+    developer_delay: bool,
 }
 
 impl Default for MayastorEnvironment {
@@ -458,6 +472,7 @@ impl Default for MayastorEnvironment {
             api_versions: vec![ApiVersion::V0, ApiVersion::V1],
             skip_sig_handler: false,
             enable_io_all_thrd_nexus_channels: false,
+            developer_delay: false,
         }
     }
 }
@@ -562,7 +577,8 @@ struct SubsystemCtx {
 
 static MAYASTOR_FEATURES: OnceCell<MayastorFeatures> = OnceCell::new();
 
-static MAYASTOR_DEFAULT_ENV: OnceCell<MayastorEnvironment> = OnceCell::new();
+static MAYASTOR_DEFAULT_ENV: OnceCell<parking_lot::Mutex<MayastorEnvironment>> =
+    OnceCell::new();
 
 impl MayastorEnvironment {
     pub fn new(args: MayastorCliArgs) -> Self {
@@ -597,6 +613,7 @@ impl MayastorEnvironment {
             nvmf_tgt_crdt: args.nvmf_tgt_crdt,
             api_versions: args.api_versions,
             skip_sig_handler: args.skip_sig_handler,
+            developer_delay: args.developer_delay,
             enable_io_all_thrd_nexus_channels: args
                 .enable_io_all_thrd_nexus_channels,
             ..Default::default()
@@ -610,15 +627,23 @@ impl MayastorEnvironment {
     }
 
     fn setup_static(self) -> Self {
-        MAYASTOR_DEFAULT_ENV.get_or_init(|| self.clone());
+        match MAYASTOR_DEFAULT_ENV.get() {
+            None => {
+                MAYASTOR_DEFAULT_ENV
+                    .get_or_init(|| parking_lot::Mutex::new(self.clone()));
+            }
+            Some(some) => {
+                *some.lock() = self.clone();
+            }
+        }
         self
     }
 
     /// Get the global environment (first created on new)
     /// or otherwise the default one (used by the tests)
     pub fn global_or_default() -> Self {
         match MAYASTOR_DEFAULT_ENV.get() {
-            Some(env) => env.clone(),
+            Some(env) => env.lock().clone(),
             None => MayastorEnvironment::default(),
         }
     }
@@ -913,9 +938,9 @@ impl MayastorEnvironment {
         }
     }
 
-    /// load the config and apply it before any subsystems have started.
+    /// Load the config and apply it before any subsystems have started.
     /// there is currently no run time check that enforces this.
-    fn load_yaml_config(&self) {
+    fn load_yaml_config(&mut self) {
         let cfg = if let Some(yaml) = &self.mayastor_config {
             info!("loading mayastor config YAML file {}", yaml);
             Config::get_or_init(|| {
@@ -930,6 +955,19 @@ impl MayastorEnvironment {
             Config::get_or_init(Config::default)
         };
         cfg.apply();
+        if let Some(mask) = cfg.eal_opts.reactor_mask.as_ref() {
+            self.reactor_mask = mask.clone();
+        }
+        if cfg.eal_opts.core_list.is_some() {
+            self.core_list = cfg.eal_opts.core_list.clone();
+        }
+        if let Some(delay) = cfg.eal_opts.developer_delay {
+            self.developer_delay = delay;
+        }
+        if let Some(interface) = &cfg.nvmf_tcp_tgt_conf.interface {
+            self.nvmf_tgt_interface = Some(interface.clone());
+        }
+        self.clone().setup_static();
     }
 
     /// load the pool config file.
@@ -991,7 +1029,7 @@ impl MayastorEnvironment {
         }
 
         // allocate a Reactor per core
-        Reactors::init();
+        Reactors::init(self.developer_delay);
 
         // launch the remote cores if any. note that during init these have to
         // be running as during setup cross call will take place.

io-engine/src/core/reactor.rs

@@ -117,6 +117,9 @@ pub struct Reactor {
     incoming: crossbeam::queue::SegQueue<spdk_rs::Thread>,
     /// the logical core this reactor is created on
     lcore: u32,
+    /// Sleeps for 1ms before each poll.
+    /// # Warning: for development only.
+    developer_delay: bool,
     /// represents the state of the reactor
     flags: Cell<ReactorState>,
     /// Unique identifier of the thread on which reactor is running.
@@ -134,7 +137,7 @@ thread_local! {
 
 impl Reactors {
     /// initialize the reactor subsystem for each core assigned to us
-    pub fn init() {
+    pub fn init(developer_delay: bool) {
         REACTOR_LIST.get_or_init(|| {
             let rc = unsafe {
                 spdk_thread_lib_init_ext(
@@ -149,7 +152,7 @@ impl Reactors {
             Reactors(
                 Cores::count()
                     .into_iter()
-                    .map(Reactor::new)
+                    .map(|core| Reactor::new(core, developer_delay))
                     .collect::<Vec<_>>(),
             )
         });
@@ -282,7 +285,7 @@ impl<'a> IntoIterator for &'a Reactors {
 
 impl Reactor {
     /// create a new ['Reactor'] instance
-    fn new(core: u32) -> Self {
+    fn new(core: u32, developer_delay: bool) -> Self {
         // create a channel to receive futures on
         let (sx, rx) =
             unbounded::<Pin<Box<dyn Future<Output = ()> + 'static>>>();
@@ -291,6 +294,7 @@ impl Reactor {
             threads: RefCell::new(VecDeque::new()),
             incoming: crossbeam::queue::SegQueue::new(),
             lcore: core,
+            developer_delay,
             flags: Cell::new(ReactorState::Init),
             tid: Cell::new(0),
             sx,
@@ -308,11 +312,7 @@ impl Reactor {
             warn!("calling poll on a reactor who is not in the INIT state");
         }
 
-        if std::env::var("MAYASTOR_DELAY").is_ok() {
-            reactor.developer_delayed();
-        } else {
-            reactor.running();
-        }
+        reactor.init_running();
         // loops
         reactor.poll_reactor();
         0
@@ -417,9 +417,18 @@ impl Reactor {
         self.set_state(ReactorState::Running)
     }
 
+    /// Init the reactor to running or developer delayed state.
+    pub fn init_running(&self) {
+        if self.developer_delay {
+            self.developer_delayed();
+        } else {
+            self.running();
+        }
+    }
+
     /// set the reactor to sleep each iteration
     pub fn developer_delayed(&self) {
-        info!("core {} set to developer delayed poll mode", self.lcore);
+        warn!("core {} set to developer delayed poll mode", self.lcore);
         self.set_state(ReactorState::Delayed);
     }
 
@@ -667,11 +676,7 @@ impl Future for &'static Reactor {
                 Poll::Pending
             }
             ReactorState::Init => {
-                if std::env::var("MAYASTOR_DELAY").is_ok() {
-                    self.developer_delayed();
-                } else {
-                    self.running();
-                }
+                self.init_running();
                 cx.waker().wake_by_ref();
                 Poll::Pending
             }

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

@@ -114,6 +114,19 @@ impl ConfigSubsystem {
     }
 }
 
+/// Various EAL (Environment Abstraction Layer) configuration options.
+#[derive(Default, Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct EalOpts {
+    /// The reactor mask to be used for starting up the instance.
+    pub reactor_mask: Option<String>,
+    /// List of cores to run on instead of using the core mask. When specified
+    /// it supersedes the reactor mask.
+    pub core_list: Option<String>,
+    /// Delay core polling by 1ms.
+    pub developer_delay: Option<bool>,
+}
+
 /// Main config structure of Mayastor. This structure can be persisted to disk.
 #[derive(Default, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(default, deny_unknown_fields)]
@@ -133,6 +146,8 @@ pub struct Config {
     pub socket_opts: PosixSocketOpts,
     /// iobuf specific options
     pub iobuf_opts: IoBufOpts,
+    /// Environment Abstraction Layer options.
+    pub eal_opts: EalOpts,
 }
 
 impl Config {
@@ -196,6 +211,7 @@ impl Config {
             nexus_opts: self.nexus_opts.get(),
             socket_opts: self.socket_opts.get(),
             iobuf_opts: self.iobuf_opts.get(),
+            eal_opts: self.eal_opts.clone(),
         }
     }
 

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

@@ -92,6 +92,8 @@ pub struct NvmfTgtConfig {
     pub crdt: [u16; TARGET_CRDT_LEN],
     /// TCP transport options
     pub opts: NvmfTcpTransportOpts,
+    /// NVMF target interface (ip, mac, name or subnet).
+    pub interface: Option<String>,
 }
 
 impl From<NvmfTgtConfig> for Box<spdk_nvmf_target_opts> {
@@ -112,6 +114,7 @@ impl Default for NvmfTgtConfig {
             max_namespaces: 2048,
             crdt: args.nvmf_tgt_crdt,
             opts: NvmfTcpTransportOpts::default(),
+            interface: None,
         }
     }
 }