Merge pull request openzfs#567 from delphix/projects/merge-upstream/m…

…aster

Merge remote-tracking branch '6.0/stage' into 'master'

cmd/zfs_object_agent/util/Cargo.toml

@@ -13,6 +13,7 @@ async-trait = "0.1.51"
 atty = "0.2"
 atomic-counter = "1.0.1"
 backtrace = "0.3"
+bit-set = "0.5.2"
 bytes = "1.0"
 bytesize = "1.1.0"
 chrono = "0.4.19"

cmd/zfs_object_agent/util/src/from64.rs

@@ -8,6 +8,7 @@ pub trait From64<A> {
 }
 
 impl From64<u64> for usize {
+    #[inline]
     fn from64(a: u64) -> usize {
         a.try_into().unwrap()
     }
@@ -18,18 +19,21 @@ pub trait AsUsize {
 }
 
 impl AsUsize for ByteSize {
+    #[inline]
     fn as_usize(&self) -> usize {
         usize::from64(self.as_u64())
     }
 }
 
 impl AsUsize for ByteSize32 {
+    #[inline]
     fn as_usize(&self) -> usize {
         self.as_u32() as usize
     }
 }
 
 impl AsUsize for u64 {
+    #[inline]
     fn as_usize(&self) -> usize {
         usize::from64(*self)
     }

cmd/zfs_object_agent/util/src/lib.rs

@@ -18,7 +18,9 @@ mod logging;
 pub mod measure;
 pub mod message;
 mod nicenum;
+pub mod range_filter;
 mod range_tree;
+pub mod semaphore_ext;
 pub mod serde;
 pub mod tunable;
 pub mod unordered;

cmd/zfs_object_agent/util/src/range_filter.rs

@@ -0,0 +1,43 @@
+use bit_set::BitSet;
+
+use crate::from64::AsUsize;
+
+/// A "range filter" allows quickly checking if a given offset might be in a set of ranges.  It
+/// is similar to a Bloom filter, but requires that keys be ordered rather than hashable.
+pub struct RangeFilter {
+    chunk_shift: u32,
+    bits: BitSet,
+}
+
+impl RangeFilter {
+    /// Create a new range filter with the given chunk size.  The range of values will be divided
+    /// into power-of-two chunks based on this.
+    pub fn new(chunk_size: u64) -> Self {
+        Self {
+            chunk_shift: chunk_size.next_power_of_two().trailing_zeros(),
+            bits: Default::default(),
+        }
+    }
+
+    /// The `size` needs to be specified even if it's always <= `chunk_size`, because the `start`
+    /// is not required to be aligned to `chunk_size` and thus the region may span multiple
+    /// chunks.  For example, in the remap code the slabs are offset from the beginning of the
+    /// disk by 4KB.
+    #[inline]
+    pub fn insert(&mut self, start: u64, size: u64) {
+        let start_index = start >> self.chunk_shift;
+        let end_index = (start + size - 1) >> self.chunk_shift;
+        for i in start_index..=end_index {
+            self.bits.insert(i.as_usize());
+        }
+    }
+
+    /// Check if the given offset might be in the set of ranges.  May return `true` even if the
+    /// offset is not present but there are nearby ranges present (within the `chunk_size`
+    /// specified to `new()`).  If this returns `false`, the offset is definitely not present.
+    #[inline]
+    pub fn maybe_present(&self, value: u64) -> bool {
+        let index = value.wrapping_shr(self.chunk_shift);
+        self.bits.contains(index.as_usize())
+    }
+}

cmd/zfs_object_agent/util/src/semaphore_ext.rs

@@ -0,0 +1,62 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use tokio::sync::AcquireError;
+use tokio::sync::Semaphore;
+
+pub struct SplittableOwnedSemaphorePermit {
+    sem: Arc<Semaphore>,
+    permits: u32,
+}
+
+#[async_trait]
+pub trait SemaphoreExt {
+    async fn acquire_splittable(
+        self: Arc<Self>,
+        n: u32,
+    ) -> Result<SplittableOwnedSemaphorePermit, AcquireError>;
+}
+
+#[async_trait]
+impl SemaphoreExt for tokio::sync::Semaphore {
+    /// Like `acquire_many_owned()`, but allows the returned guard to be split into multiple
+    /// guards, each representing a subset of the N permits.
+    async fn acquire_splittable(
+        self: Arc<Self>,
+        n: u32,
+    ) -> Result<SplittableOwnedSemaphorePermit, AcquireError> {
+        self.clone().acquire_many_owned(n).await?.forget();
+        Ok(SplittableOwnedSemaphorePermit {
+            sem: self,
+            permits: n,
+        })
+    }
+}
+
+impl SplittableOwnedSemaphorePermit {
+    /// Split off N permits from this permit to a new instance.  The new permit will have N
+    /// permits, and this permit will have N fewer permits.  This allows some of the permits to
+    /// be released separately.
+    ///
+    /// # Panics
+    /// Panics if the number of permits to release is greater than the number of permits.
+    pub fn split(&mut self, n: u32) -> Self {
+        self.permits -= n;
+        Self {
+            sem: self.sem.clone(),
+            permits: n,
+        }
+    }
+
+    /// Forgets the permit **without** releasing it back to the semaphore.  This can be used to
+    /// reduce the amount of permits available from a semaphore.
+    pub fn forget(mut self) {
+        self.permits = 0;
+    }
+}
+
+impl Drop for SplittableOwnedSemaphorePermit {
+    fn drop(&mut self) {
+        self.sem.add_permits(self.permits as usize);
+    }
+}

cmd/zfs_object_agent/util/src/vec_ext.rs

@@ -245,6 +245,10 @@ impl<K, V> VecMap<K, V>
 where
     K: From<usize> + Into<usize> + Copy,
 {
+    pub fn new() -> Self {
+        Default::default()
+    }
+
     /// Returns old value (or None if not present)
     pub fn insert(&mut self, key: K, value: V) -> Option<V> {
         let index = key.into();

cmd/zfs_object_agent/util/src/zettacache_stats.rs

@@ -394,6 +394,7 @@ pub enum CacheStatCounter {
     DemandBufferBytesAvailable,
     SpeculativeBufferBytesAvailable,
     SlabCapacity,
+    AllocatedSpace,
     AvailableSpace,
     FreeBlocksSize,
     FreeSlabsSize,
@@ -455,6 +456,7 @@ impl Sub<&Self> for &CacheStats {
                 | CacheStatCounter::DemandBufferBytesAvailable
                 | CacheStatCounter::SpeculativeBufferBytesAvailable
                 | CacheStatCounter::SlabCapacity
+                | CacheStatCounter::AllocatedSpace
                 | CacheStatCounter::AvailableSpace
                 | CacheStatCounter::FreeSlabsSize
                 | CacheStatCounter::FreeBlocksSize => {

cmd/zfs_object_agent/zcache/src/stats.rs

@@ -258,7 +258,7 @@ impl StatsDisplay {
             let slab_capacity = values.value(SlabCapacity);
             let free_blocks_size = values.value(FreeBlocksSize);
             let free_slabs_size = values.value(FreeSlabsSize);
-            let allocated_space = slab_capacity - free_blocks_size - free_slabs_size;
+            let allocated_space = values.value(AllocatedSpace);
 
             self.display_bytes(allocated_space as f64);
             self.display_bytes(available_space as f64);
@@ -276,8 +276,8 @@ impl StatsDisplay {
 
         loop {
             let response = remote.call(TYPE_ZCACHE_STATS, None).await?;
-            let stats_json = response.lookup_string("stats_json").unwrap();
-            let latest: CacheStats = serde_json::from_str(stats_json.to_str()?).unwrap();
+            let stats_json = response.lookup_string("stats_json")?;
+            let latest: CacheStats = serde_json::from_str(stats_json.to_str()?)?;
 
             // Periodically display the column headers
             if (iteration % (self.get_terminal_height() - 3) as u64) == 0 {

cmd/zfs_object_agent/zcdb/src/main.rs

@@ -68,9 +68,9 @@ enum Commands {
         #[clap(long)]
         index: bool,
 
-        /// dump rebalance log
+        /// dump legacy rebalance log
         #[clap(long)]
-        rebalance: bool,
+        legacy_rebalance: bool,
 
         /// dump atime histogram of index
         #[clap(long)]
@@ -115,20 +115,20 @@ async fn main() -> Result<(), anyhow::Error> {
     match cli.command {
         Commands::Logs {
             nodefaults,
-            spacemap: spacemaps,
-            operation: operation_log_raw,
-            index: index_log_raw,
-            rebalance: rebalance_log_raw,
+            spacemap,
+            operation,
+            index,
+            legacy_rebalance,
             atime_histogram,
         } => {
             ZettaCacheDBCommand::issue_command(
                 ZettaCacheDBCommand::DumpStructures(
                     DumpStructuresOptions::default()
                         .defaults(!nodefaults)
-                        .spacemaps(spacemaps)
-                        .operation_log_raw(operation_log_raw)
-                        .index_log_raw(index_log_raw)
-                        .rebalance_log_raw(rebalance_log_raw)
+                        .spacemaps(spacemap)
+                        .operation_log_raw(operation)
+                        .index_log_raw(index)
+                        .rebalance_log_raw(legacy_rebalance)
                         .atime_histogram(atime_histogram),
                 ),
                 CacheOpenMode::DiscoveryDirectory(cli.cache_device_dir, cli.guid),

cmd/zfs_object_agent/zettacache/src/block_access.rs

@@ -66,9 +66,9 @@ tunable! {
     static ref DISK_AGG_CHUNK: ByteSize = ByteSize::mib(1);
     static ref DISK_WRITE_QUEUE_EMPTY_DELAY: Duration = Duration::from_millis(1);
     static ref DISK_READ_QUEUE_EMPTY_DELAY: Duration = Duration::from_micros(10);
-    static ref DISK_WRITE_MAX_QUEUE_DEPTH: usize = 32;
+    pub static ref DISK_WRITE_MAX_QUEUE_DEPTH: usize = 32;
     static ref DISK_METADATA_WRITE_MAX_QUEUE_DEPTH: usize = 16;
-    pub static ref DISK_READ_MAX_QUEUE_DEPTH: usize = 64;
+    static ref DISK_READ_MAX_QUEUE_DEPTH: usize = 64;
     static ref DISK_SECTOR_SIZE_OVERRIDE: Option<ByteSize> = None;
 }