[PATCH v2 proxmox-backup 2/3] datastore: gc: optionally defer and batch chunk atime updates

[Enrico Plantulli <plantulli@gmail.com>]

Phase 1 of garbage collection updates each chunk's access time on
first encounter, resolving the chunk directly from its digest. On a
cold store every update then pays an independent, serialized metadata
lookup, issued in digest order - which is uncorrelated with the
on-disk layout of the inodes by construction, since the digest picks
the directory while the inode number follows creation order.

Instead of touching a chunk on first encounter, queue the digest in a
bounded list and flush the list sorted: each chunk directory involved
is opened and iterated once right before its chunks are touched, so
file systems that prefetch inode metadata on readdir serve the
following atime updates from cache, and the updates go through
utimensat() relative to the open directory file descriptor, avoiding
repeated path traversal. This supersedes the whole-store readdir pass
of the first version of this series: measured on a production ZFS
datastore with 72M chunks, a single upfront pass did not survive in
the ARC until a multi-hour phase 1 consumed it (the GC worker was
observed blocked on zio_wait re-reading ZAP leaves and dnode blocks
the pass had loaded hours earlier), while batched warming reads the
metadata seconds before it is used.

The batch is only used for filesystem backed datastores; the S3
backend keeps the immediate path, since its in-use markers need the
per-chunk handling anyway. A chunk missing at flush time is handled
as before (touch the .bad companions, log a warning), but the warning
can no longer name the referencing index file, only the digest.

Signed-off-by: Enrico Plantulli <plantulli@gmail.com>
---
 docs/storage.rst                 |  20 ++++++
 pbs-datastore/src/chunk_store.rs | 102 +++++++++++++++++++++++++++++++
 pbs-datastore/src/datastore.rs   |  93 +++++++++++++++++++++++++---
 3 files changed, 206 insertions(+), 9 deletions(-)

diff --git a/docs/storage.rst b/docs/storage.rst
index 2c6ba0a..c190aeb 100644
--- a/docs/storage.rst
+++ b/docs/storage.rst
@@ -686,6 +686,26 @@ There are some tuning related options for the datastore that are more advanced:
   cache slots, 1048576 (= 1024 * 1024) being the default, 8388608 (= 8192 *
   1024) the maximum value.
 
+* ``gc-chunk-metadata-prefetch``: Defer and batch chunk atime updates in GC phase 1:
+  Phase 1 normally updates each chunk's access time on first encounter, reaching the
+  chunk directly by digest. On a cold store every update then pays an independent,
+  serialized metadata lookup, in an order uncorrelated with the on-disk layout. With
+  this option the updates are instead queued and flushed in sorted batches: each chunk
+  directory involved is opened and iterated once right before its chunks are touched,
+  so file systems that prefetch inode metadata on readdir (such as ZFS) serve the
+  updates from cache, and the updates use the open directory handle, avoiding repeated
+  path lookups. This can reduce phase 1 runtime substantially on large datastores on
+  local rotational disks, at the cost of the extra directory reads. Storage backends
+  that do not prefetch on readdir (for example some network attached storages) may see
+  no benefit or a regression, which is why the option is disabled by default. It has no
+  effect on S3 backed datastores.
+
+* ``gc-prefetch-batch-size``: Batch size for the deferred atime updates:
+  Number of queued chunk atime updates per sorted flush, default 1048576. Larger
+  batches group more chunks per directory read: a flush of N entries spans up to
+  min(N, 65536) directories, so N/65536 chunks are served per directory read on
+  average. Mostly useful for benchmarking the option above.
+
 * ``default-verification-workers`` and ``default-verification-readers``:
   Define the default number of threads used for verification and reading of chunks,
   respectively. By default, 4 threads are used for verification and 1 thread is used
diff --git a/pbs-datastore/src/chunk_store.rs b/pbs-datastore/src/chunk_store.rs
index 6d2ffdb..70ee49f 100644
--- a/pbs-datastore/src/chunk_store.rs
+++ b/pbs-datastore/src/chunk_store.rs
@@ -302,6 +302,108 @@ impl ChunkStore {
         Ok(is_bad)
     }
 
+    /// Update the atime of a sorted batch of chunks, warming each chunk directory right
+    /// before its chunks are touched.
+    ///
+    /// The digests are sorted in place, so consecutive entries share their chunk directory.
+    /// Every time a new directory is entered it is opened and its entries are iterated once,
+    /// discarding the results: on file systems that prefetch inode metadata on readdir this
+    /// faults the chunk metadata in right before the atime updates need it. The updates then
+    /// go through utimensat() relative to the open directory file descriptor, avoiding the
+    /// full path lookup. The chunk store mutex is taken per touch, mirroring
+    /// cond_touch_chunk(); the readdir warming happens outside the lock.
+    ///
+    /// Clears the batch and returns the digests of chunks that did not exist, for the caller
+    /// to handle.
+    pub(super) fn touch_chunk_batch(
+        &self,
+        digests: &mut Vec<[u8; 32]>,
+        worker: &dyn WorkerTaskContext,
+    ) -> Result<Vec<[u8; 32]>, Error> {
+        // unwrap: only `None` in unit tests
+        assert!(self.locker.is_some());
+
+        use nix::dir::Dir;
+        use nix::fcntl::OFlag;
+        use nix::sys::stat::Mode;
+
+        let times: [libc::timespec; 2] = [
+            // access time -> update to now
+            libc::timespec {
+                tv_sec: 0,
+                tv_nsec: libc::UTIME_NOW,
+            },
+            // modification time -> keep as is
+            libc::timespec {
+                tv_sec: 0,
+                tv_nsec: libc::UTIME_OMIT,
+            },
+        ];
+
+        digests.sort_unstable();
+
+        let mut missing = Vec::new();
+        let mut current_prefix = std::path::PathBuf::new();
+        let mut current_dir: Option<Dir> = None;
+        let mut first = true;
+
+        for digest in digests.iter() {
+            worker.check_abort()?;
+            worker.fail_on_shutdown()?;
+
+            let prefix = digest_to_prefix(digest);
+            if first || prefix != current_prefix {
+                first = false;
+                let mut path = self.chunk_dir.clone();
+                path.push(&prefix);
+                current_dir = match Dir::open(&path, OFlag::O_RDONLY, Mode::empty()) {
+                    Ok(mut dir) => {
+                        // warm the directory: iterate the entries once, discarding them
+                        for entry in dir.iter() {
+                            if entry.is_err() {
+                                break;
+                            }
+                        }
+                        Some(dir)
+                    }
+                    Err(nix::errno::Errno::ENOENT) => None,
+                    Err(err) => bail!("unable to open chunk directory {path:?} - {err}"),
+                };
+                current_prefix = prefix;
+            }
+
+            let dir = match current_dir.as_ref() {
+                Some(dir) => dir,
+                None => {
+                    missing.push(*digest);
+                    continue;
+                }
+            };
+
+            let digest_str = hex::encode(digest);
+            let filename = std::ffi::CString::new(digest_str)?;
+            let res = {
+                let _lock = self.mutex.lock().unwrap();
+                unsafe {
+                    nix::errno::Errno::result(libc::utimensat(
+                        dir.as_raw_fd(),
+                        filename.as_ptr(),
+                        &times[0],
+                        libc::AT_SYMLINK_NOFOLLOW,
+                    ))
+                }
+            };
+            match res {
+                Ok(_) => (),
+                Err(nix::errno::Errno::ENOENT) => missing.push(*digest),
+                Err(err) => bail!("update atime failed for chunk {filename:?} - {err}"),
+            }
+        }
+
+        digests.clear();
+        Ok(missing)
+    }
+
     fn get_chunk_store_iterator(
         &self,
     ) -> Result<
diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index c3db522..6b3601a 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -84,6 +84,25 @@ const S3_CLIENT_RATE_LIMITER_BASE_PATH: &str = pbs_buildcfg::rundir!("/s3/shmem/
 const NAMESPACE_MARKER_FILENAME: &str = ".namespace";
 // s3 put request times out after upload_size / 1 Kib/s, so about 2.3 hours for 8 MiB
 const CHUNK_LOCK_TIMEOUT: Duration = Duration::from_secs(3 * 60 * 60);
+
+/// Deferred chunk access time updates for garbage collection phase 1.
+///
+/// On a miss of the chunk digest LRU cache the digest is queued here instead of being
+/// touched right away; full batches are sorted and flushed via
+/// [`ChunkStore::touch_chunk_batch`], which warms each chunk directory via readdir right
+/// before its chunks are touched.
+struct GcTouchBatch {
+    digests: Vec<[u8; 32]>,
+    capacity: usize,
+}
+
+/// Per-run state of garbage collection phase 1 chunk marking.
+struct GcMarkState {
+    /// avoid multiple expensive atime updates for the same chunk
+    chunk_lru_cache: Option<LruCache<[u8; 32], ()>>,
+    /// deferred, batched atime updates (see [`GcTouchBatch`])
+    touch_batch: Option<GcTouchBatch>,
+}
 // s3 deletion batch size to avoid 1024 open files soft limit
 const S3_DELETE_BATCH_LIMIT: usize = 100;
 // max defer time for s3 batch deletions
@@ -2110,11 +2129,34 @@ impl DataStore {
     }
 
     // mark chunks  used by ``index`` as used
+    fn gc_touch_batch_flush(
+        &self,
+        batch: &mut GcTouchBatch,
+        worker: &dyn WorkerTaskContext,
+    ) -> Result<(), Error> {
+        if batch.digests.is_empty() {
+            return Ok(());
+        }
+        let missing = self
+            .inner
+            .chunk_store
+            .touch_chunk_batch(&mut batch.digests, worker)?;
+        for digest in missing {
+            // touch any corresponding .bad files to keep them around, mirroring the direct
+            // path below; the referencing index file is no longer known at this point, so
+            // the warning only carries the digest
+            self.inner.chunk_store.cond_touch_bad_chunks(&digest)?;
+            let hex = hex::encode(digest);
+            warn!("warning: unable to access non-existent chunk {hex} (deferred atime update)");
+        }
+        Ok(())
+    }
+
     fn index_mark_used_chunks(
         &self,
         index: Box<dyn IndexFile>,
         file_name: &Path, // only used for error reporting
-        chunk_lru_cache: &mut Option<LruCache<[u8; 32], ()>>,
+        mark_state: &mut GcMarkState,
         status: &mut GarbageCollectionStatus,
         worker: &dyn WorkerTaskContext,
         s3_client: Option<Arc<S3Client>>,
@@ -2128,7 +2170,7 @@ impl DataStore {
             let digest = index.index_digest(pos).unwrap();
 
             // Avoid multiple expensive atime updates by utimensat
-            if let Some(chunk_lru_cache) = chunk_lru_cache {
+            if let Some(chunk_lru_cache) = &mut mark_state.chunk_lru_cache {
                 if chunk_lru_cache.insert(*digest, (), |_| Ok(()))? {
                     if let Some(cache_stats) = status.cache_stats.as_mut() {
                         cache_stats.hits += 1;
@@ -2140,6 +2182,14 @@ impl DataStore {
                 }
             }
 
+            if let Some(batch) = &mut mark_state.touch_batch {
+                batch.digests.push(*digest);
+                if batch.digests.len() >= batch.capacity {
+                    self.gc_touch_batch_flush(batch, worker)?;
+                }
+                continue;
+            }
+
             if !self.inner.chunk_store.cond_touch_chunk(digest, false)? {
                 // touch any corresponding .bad files to keep them around, meaning if a chunk is
                 // rewritten correctly they will be removed automatically, as well as if no index
@@ -2174,6 +2224,7 @@ impl DataStore {
         status: &mut GarbageCollectionStatus,
         worker: &dyn WorkerTaskContext,
         cache_capacity: usize,
+        touch_batch: Option<GcTouchBatch>,
         s3_client: Option<Arc<S3Client>>,
     ) -> Result<(), Error> {
         // Iterate twice over the datastore to fetch index files, even if this comes with an
@@ -2190,10 +2241,13 @@ impl DataStore {
         let mut unprocessed_index_list = self.list_index_files()?;
         let mut index_count = unprocessed_index_list.len();
 
-        let mut chunk_lru_cache = if cache_capacity > 0 {
-            Some(LruCache::new(cache_capacity))
-        } else {
-            None
+        let mut mark_state = GcMarkState {
+            chunk_lru_cache: if cache_capacity > 0 {
+                Some(LruCache::new(cache_capacity))
+            } else {
+                None
+            },
+            touch_batch,
         };
         let mut processed_index_files = 0;
         let mut last_percentage: usize = 0;
@@ -2265,7 +2319,7 @@ impl DataStore {
                             self.index_mark_used_chunks(
                                 index,
                                 &path,
-                                &mut chunk_lru_cache,
+                                &mut mark_state,
                                 status,
                                 worker,
                                 s3_client.as_ref().cloned(),
@@ -2308,7 +2362,7 @@ impl DataStore {
             self.index_mark_used_chunks(
                 index,
                 &path,
-                &mut chunk_lru_cache,
+                &mut mark_state,
                 status,
                 worker,
                 s3_client.as_ref().cloned(),
@@ -2330,13 +2384,17 @@ impl DataStore {
             self.index_mark_used_chunks(
                 index,
                 path,
-                &mut chunk_lru_cache,
+                &mut mark_state,
                 status,
                 worker,
                 s3_client.as_ref().cloned(),
             )
         })?;
 
+        if let Some(batch) = mark_state.touch_batch.as_mut() {
+            self.gc_touch_batch_flush(batch, worker)?;
+        }
+
         Ok(())
     }
 
@@ -2459,12 +2517,29 @@ impl DataStore {
             1024 * 1024
         };
 
+        let touch_batch = if tuning.gc_chunk_metadata_prefetch.unwrap_or(false) {
+            if s3_client.is_some() {
+                info!("Deferred chunk atime updates not supported for S3 backed datastore.");
+                None
+            } else {
+                let capacity = tuning.gc_prefetch_batch_size.unwrap_or(1024 * 1024);
+                info!("Using deferred chunk atime updates with batch size {capacity}.");
+                Some(GcTouchBatch {
+                    digests: Vec::with_capacity(capacity),
+                    capacity,
+                })
+            }
+        } else {
+            None
+        };
+
         info!("Start GC phase1 (mark used chunks)");
 
         self.mark_used_chunks(
             &mut gc_status,
             worker,
             gc_cache_capacity,
+            touch_batch,
             s3_client.as_ref().cloned(),
         )
         .context("marking used chunks failed")?;

base-commit: 5d95fc20eeae6df1936216b1ab1ab6952472fdcf
-- 
2.47.3