blocks

Author: TheSafo

lading/src/generator/common.rs

@@ -36,7 +36,7 @@ use lading_payload::block;
 
 use super::General;
 use crate::generator::common::{
-    BytesThrottleConfig, MetricsBuilder, ThrottleConversionError, create_throttle,
+    BlockThrottle, MetricsBuilder, ThrottleConfig, ThrottleConversionError, create_throttle,
 };
 
 /// An enum to allow us to determine what operation caused an IO errror as the
@@ -147,8 +147,9 @@ pub struct Config {
     /// Whether to use a fixed or streaming block cache
     #[serde(default = "lading_payload::block::default_cache_method")]
     block_cache_method: block::CacheMethod,
-    /// The load throttle configuration
-    pub throttle: Option<BytesThrottleConfig>,
+    /// Throughput profile controlling emission rate (bytes or blocks).
+    #[serde(default)]
+    pub throttle: Option<ThrottleConfig>,
 }
 
 #[derive(Debug)]
@@ -214,7 +215,7 @@ impl Server {
         let mut handles = Vec::new();
 
         for idx in 0..config.concurrent_logs {
-            let throttle =
+            let throughput_throttle =
                 create_throttle(config.throttle.as_ref(), config.bytes_per_second.as_ref())?;
 
             let mut dir_path = config.root.clone();
@@ -234,8 +235,9 @@ impl Server {
                 &basename,
                 config.total_rotations,
                 maximum_bytes_per_log,
+                maximum_block_size.get(),
                 Arc::clone(&block_cache),
-                throttle,
+                throughput_throttle,
                 shutdown.clone(),
                 child_labels,
             );
@@ -283,8 +285,9 @@ struct Child {
     names: Vec<PathBuf>,
     // The soft limit bytes per file that will trigger a rotation.
     maximum_bytes_per_log: NonZeroU32,
+    maximum_block_size: u32,
     block_cache: Arc<block::Cache>,
-    throttle: lading_throttle::Throttle,
+    throttle: BlockThrottle,
     shutdown: lading_signal::Watcher,
     labels: Vec<(String, String)>,
 }
@@ -295,8 +298,9 @@ impl Child {
         basename: &Path,
         total_rotations: u8,
         maximum_bytes_per_log: NonZeroU32,
+        maximum_block_size: u32,
         block_cache: Arc<block::Cache>,
-        throttle: lading_throttle::Throttle,
+        throttle: BlockThrottle,
         shutdown: lading_signal::Watcher,
         labels: Vec<(String, String)>,
     ) -> Self {
@@ -316,6 +320,7 @@ impl Child {
         Self {
             names,
             maximum_bytes_per_log,
+            maximum_block_size,
             block_cache,
             throttle,
             shutdown,
@@ -324,7 +329,10 @@ impl Child {
     }
 
     async fn spin(mut self) -> Result<(), Error> {
-        let buffer_capacity = self.throttle.maximum_capacity() as usize;
+        let mut handle = self.block_cache.handle();
+        let buffer_capacity = self
+            .throttle
+            .maximum_capacity_bytes(self.maximum_block_size);
         let mut total_bytes_written: u64 = 0;
         let maximum_bytes_per_log: u64 = u64::from(self.maximum_bytes_per_log.get());
 
@@ -357,21 +365,16 @@ impl Child {
                 })?,
         );
 
-        let mut handle = self.block_cache.handle();
-
         let shutdown_wait = self.shutdown.recv();
         tokio::pin!(shutdown_wait);
         loop {
             // SAFETY: By construction the block cache will never be empty
             // except in the event of a catastrophic failure.
-            let total_bytes = self.block_cache.peek_next_size(&handle);
-
             tokio::select! {
-                result = self.throttle.wait_for(total_bytes) => {
+                result = self.throttle.wait_for_block(&self.block_cache, &handle) => {
                     match result {
                         Ok(()) => {
-                            let block = self.block_cache.advance(&mut handle);
-                            write_bytes(block,
+                            write_bytes(self.block_cache.advance(&mut handle),
                                     &mut fp,
                                     &mut total_bytes_written,
                                     buffer_capacity,

lading/src/generator/file_gen/logrotate.rs

@@ -5,6 +5,7 @@
 #![allow(clippy::cast_possible_wrap)]
 
 use crate::generator;
+use crate::generator::common::{RateSpec, ThrottleConversionError, ThrottleMode};
 use fuser::{
     BackgroundSession, FileAttr, Filesystem, MountOption, ReplyAttr, ReplyData, ReplyDirectory,
     ReplyEntry, Request, spawn_mount2,
@@ -55,41 +56,74 @@ pub struct Config {
     maximum_block_size: byte_unit::Byte,
     /// The mount-point for this filesystem
     mount_point: PathBuf,
-    /// The load profile, controlling bytes per second as a function of time.
+    /// The load profile, controlling bytes or blocks per second as a function of time.
     load_profile: LoadProfile,
 }
 
 /// Profile for load in this filesystem.
 #[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq)]
 #[serde(rename_all = "snake_case")]
 pub enum LoadProfile {
-    /// Constant bytes per second
-    Constant(byte_unit::Byte),
-    /// Linear growth of bytes per second
+    /// Constant rate (bytes or blocks per second).
+    Constant(RateSpec),
+    /// Linear growth of rate (bytes or blocks per second).
     Linear {
-        /// Starting point for bytes per second
-        initial_bytes_per_second: byte_unit::Byte,
-        /// Amount to increase per second
-        rate: byte_unit::Byte,
+        /// Starting point for the rate.
+        #[serde(alias = "initial_bytes_per_second")]
+        initial: RateSpec,
+        /// Amount to increase per second.
+        rate: RateSpec,
     },
 }
 
 impl LoadProfile {
-    fn to_model(self) -> model::LoadProfile {
+    fn to_model(self) -> Result<model::LoadProfile, ThrottleConversionError> {
         // For now, one tick is one second.
         match self {
-            LoadProfile::Constant(bpt) => model::LoadProfile::Constant(bpt.as_u128() as u64),
-            LoadProfile::Linear {
-                initial_bytes_per_second,
-                rate,
-            } => model::LoadProfile::Linear {
-                start: initial_bytes_per_second.as_u128() as u64,
-                rate: rate.as_u128() as u64,
-            },
+            LoadProfile::Constant(rate) => {
+                let (mode, cap) = resolve_rate(&rate)?;
+                match mode {
+                    ThrottleMode::Bytes => Ok(model::LoadProfile::Constant(u64::from(cap.get()))),
+                    ThrottleMode::Blocks => Ok(model::LoadProfile::Blocks {
+                        blocks_per_tick: u64::from(cap.get()),
+                    }),
+                }
+            }
+            LoadProfile::Linear { initial, rate } => {
+                let (m1, init) = resolve_rate(&initial)?;
+                let (m2, rate) = resolve_rate(&rate)?;
+                if m1 != m2 {
+                    return Err(ThrottleConversionError::MixedModes);
+                }
+                match m1 {
+                    ThrottleMode::Bytes => Ok(model::LoadProfile::Linear {
+                        start: u64::from(init.get()),
+                        rate: u64::from(rate.get()),
+                    }),
+                    ThrottleMode::Blocks => Ok(model::LoadProfile::BlocksLinear {
+                        start: u64::from(init.get()),
+                        rate: u64::from(rate.get()),
+                    }),
+                }
+            }
         }
     }
 }
 
+fn resolve_rate(rate: &RateSpec) -> Result<(ThrottleMode, NonZeroU32), ThrottleConversionError> {
+    match rate {
+        RateSpec::Bytes { bytes_per_second } => {
+            let val = bytes_per_second.as_u128();
+            let val = u32::try_from(val)
+                .map_err(|_| ThrottleConversionError::ValueTooLarge(*bytes_per_second))?;
+            NonZeroU32::new(val)
+                .map(|n| (ThrottleMode::Bytes, n))
+                .ok_or(ThrottleConversionError::Zero)
+        }
+        RateSpec::Blocks { blocks_per_second } => Ok((ThrottleMode::Blocks, *blocks_per_second)),
+    }
+}
+
 #[derive(thiserror::Error, Debug)]
 /// Error for `LogrotateFs`
 pub enum Error {
@@ -99,6 +133,9 @@ pub enum Error {
     /// Creation of payload blocks failed.
     #[error("Block creation error: {0}")]
     Block(#[from] block::Error),
+    /// Throttle conversion error
+    #[error("Throttle configuration error: {0}")]
+    ThrottleConversion(#[from] ThrottleConversionError),
     /// Failed to convert, value is 0
     #[error("Value provided must not be zero")]
     Zero,
@@ -154,10 +191,18 @@ impl Server {
             // divvy this up in the future.
             total_bytes.get() as usize,
         )?;
+        let load_profile = config.load_profile.to_model()?;
 
         let start_time = Instant::now();
         let start_time_system = SystemTime::now();
 
+        let block_cache_size = block_cache.total_size();
+        info!(
+            "LogrotateFS block cache initialized: requested={}, actual={} bytes, blocks={}",
+            config.maximum_prebuild_cache_size_bytes,
+            block_cache_size,
+            block_cache.len()
+        );
         let state = model::State::new(
             &mut rng,
             start_time.elapsed().as_secs(),
@@ -166,7 +211,7 @@ impl Server {
             block_cache,
             config.max_depth,
             config.concurrent_logs,
-            config.load_profile.to_model(),
+            load_profile,
         );
 
         info!(
@@ -481,6 +526,7 @@ impl Filesystem for LogrotateFS {
 #[cfg(test)]
 mod tests {
     use super::LoadProfile;
+    use crate::generator::common::RateSpec;
     use serde::Deserialize;
 
     #[derive(Debug, Deserialize)]
@@ -503,8 +549,28 @@ mod tests {
         "#;
         let w = parse_wrapper(yaml);
         assert!(matches!(w.load_profile, LoadProfile::Constant(..)));
-        if let LoadProfile::Constant(bytes) = w.load_profile {
-            assert_eq!(bytes.as_u64(), 5 * 1024 * 1024);
+        if let LoadProfile::Constant(rate) = w.load_profile {
+            assert!(matches!(rate, RateSpec::Bytes { .. }));
+            if let RateSpec::Bytes { bytes_per_second } = rate {
+                assert_eq!(bytes_per_second.as_u64(), 5 * 1024 * 1024);
+            }
+        }
+    }
+
+    #[test]
+    fn load_profile_constant_blocks_per_second() {
+        let yaml = r#"
+                load_profile:
+                    constant:
+                        blocks_per_second: 100
+            "#;
+        let w = parse_wrapper(yaml);
+        assert!(matches!(w.load_profile, LoadProfile::Constant(_)));
+        if let LoadProfile::Constant(rate) = w.load_profile {
+            assert!(matches!(rate, RateSpec::Blocks { .. }));
+            if let RateSpec::Blocks { blocks_per_second } = rate {
+                assert_eq!(blocks_per_second.get(), 100);
+            }
         }
     }
 
@@ -519,13 +585,61 @@ mod tests {
 
         let w = parse_wrapper(yaml);
         assert!(matches!(w.load_profile, LoadProfile::Linear { .. }));
-        if let LoadProfile::Linear {
-            initial_bytes_per_second,
-            rate,
-        } = w.load_profile
-        {
-            assert_eq!(initial_bytes_per_second.as_u64(), 10 * 1024 * 1024);
-            assert_eq!(rate.as_u64(), 1 * 1024 * 1024);
+        if let LoadProfile::Linear { initial, rate } = w.load_profile {
+            assert!(matches!(initial, RateSpec::Bytes { .. }));
+            if let RateSpec::Bytes { bytes_per_second } = initial {
+                assert_eq!(bytes_per_second.as_u64(), 10 * 1024 * 1024);
+            }
+            assert!(matches!(rate, RateSpec::Bytes { .. }));
+            if let RateSpec::Bytes { bytes_per_second } = rate {
+                assert_eq!(bytes_per_second.as_u64(), 1 * 1024 * 1024);
+            }
+        }
+    }
+
+    #[test]
+    fn load_profile_linear_new_format_flattened_bytes() {
+        let yaml = r#"
+                load_profile:
+                    linear:
+                        initial: "1 MiB"
+                        rate: "100 KiB"
+            "#;
+        let w = parse_wrapper(yaml);
+        assert!(matches!(w.load_profile, LoadProfile::Linear { .. }));
+        if let LoadProfile::Linear { initial, rate } = w.load_profile {
+            assert!(matches!(initial, RateSpec::Bytes { .. }));
+            if let RateSpec::Bytes { bytes_per_second } = initial {
+                assert_eq!(bytes_per_second.as_u64(), 1 * 1024 * 1024);
+            }
+            assert!(matches!(rate, RateSpec::Bytes { .. }));
+            if let RateSpec::Bytes { bytes_per_second } = rate {
+                assert_eq!(bytes_per_second.as_u64(), 100 * 1024);
+            }
+        }
+    }
+
+    #[test]
+    fn load_profile_linear_blocks_per_second() {
+        let yaml = r#"
+                load_profile:
+                    linear:
+                        initial:
+                            blocks_per_second: 100
+                        rate:
+                            blocks_per_second: 10
+            "#;
+        let w = parse_wrapper(yaml);
+        assert!(matches!(w.load_profile, LoadProfile::Linear { .. }));
+        if let LoadProfile::Linear { initial, rate } = w.load_profile {
+            assert!(matches!(initial, RateSpec::Blocks { .. }));
+            if let RateSpec::Blocks { blocks_per_second } = initial {
+                assert_eq!(blocks_per_second.get(), 100);
+            }
+            assert!(matches!(rate, RateSpec::Blocks { .. }));
+            if let RateSpec::Blocks { blocks_per_second } = rate {
+                assert_eq!(blocks_per_second.get(), 10);
+            }
         }
     }
 }