diff --git a/docs/notes/2.34.x.md b/docs/notes/2.34.x.md index c913a74115b..cfb05af1dba 100644 --- a/docs/notes/2.34.x.md +++ b/docs/notes/2.34.x.md @@ -57,6 +57,8 @@ Fixed pantsd logging remote Authentication bearer tokens in plaintext when `remo Pants now logs a warning when it cannot delete a local process execution sandbox. Cleanup failures were previously ignored, so sandboxes containing read-only directories leaked in the temporary directory without any message. +An idle `pantsd` no longer re-expands and rewrites the lease of every store entry its graph references every 72 seconds. A lease with more than half of its time left is no longer rewritten, and a digest leased within the last quarter of the lease time (30 minutes) is not expanded again, which removes most of the CPU time and disk writes an idle daemon spent extending leases ([#23694](https://github.com/pantsbuild/pants/issues/23694)). + ### Goals ### Backends diff --git a/src/rust/fs/store/src/lib.rs b/src/rust/fs/store/src/lib.rs index a4fb0fa39ed..e59281ec4a0 100644 --- a/src/rust/fs/store/src/lib.rs +++ b/src/rust/fs/store/src/lib.rs @@ -1088,11 +1088,21 @@ impl Store { } } + /// + /// Leases `digests` and everything below them that exists locally. + /// + /// A digest this method leased within the last quarter of the lease time is skipped, along with + /// everything below it: those leases are still long enough to wait for a later call, and + /// expanding and re-leasing every digest a long-lived graph holds is most of the cost of pantsd's + /// periodic lease extension. + /// pub async fn lease_all_recursively<'a, Ds: Iterator>( &self, digests: Ds, ) -> Result<(), StoreError> { - let reachable_digests_and_types = self.expand_local_digests(digests).await?; + let started = Instant::now(); + let digests = self.local.digests_to_renew(digests.copied(), started); + let reachable_digests_and_types = self.expand_local_digests(digests.iter()).await?; // Lease all Digests which existed (ignoring any that didn't). self.local .lease_all( @@ -1101,6 +1111,7 @@ impl Store { .flat_map(|(digest, maybe_type)| maybe_type.map(|t| (digest, t))), ) .await?; + self.local.record_renewed(&digests, started); Ok(()) } diff --git a/src/rust/fs/store/src/local.rs b/src/rust/fs/store/src/local.rs index 3e6aeb4c2e6..fae44e12c56 100644 --- a/src/rust/fs/store/src/local.rs +++ b/src/rust/fs/store/src/local.rs @@ -363,8 +363,16 @@ impl UnderlyingByteStore for ShardedFSDB { async fn lease(&self, fingerprint: Fingerprint) -> Result<(), String> { let path = self.get_path(fingerprint); + // As in `ShardedLmdb::lease`: a lease with more than half its time left is not rewritten. + let renew_before = SystemTime::now() - self.lease_time / 2; self.executor .spawn_blocking(move || { + if std::fs::metadata(&path) + .and_then(|metadata| metadata.modified()) + .is_ok_and(|mtime| mtime > renew_before) + { + return Ok(()); + } fs_set_times::set_mtime(&path, fs_set_times::SystemTimeSpec::SymbolicNow) .map_err(|e| format!("Failed to extend mtime of {path:?}: {e}")) }) @@ -558,6 +566,8 @@ struct InnerStore { directory_lmdb: Result, String>, file_fsdb: ShardedFSDB, file_source: FileSource, + lease_time: Duration, + renewed_at: Mutex>, } impl ByteStore { @@ -615,6 +625,8 @@ impl ByteStore { file_source: FileSource { open_files: Semaphore::new(1024), }, + lease_time: options.lease_time, + renewed_at: Mutex::default(), }), }) } @@ -650,6 +662,34 @@ impl ByteStore { } } + /// + /// The digests among `digests` that were not passed to `record_renewed` within the last + /// quarter of the lease time. + /// + /// `lease_all` leaves every lease it is given with at least half of the lease time left (see + /// `ShardedLmdb::lease`), so a digest renewed more recently than that, and everything below it, + /// still has at least a quarter of the lease time left: enough to wait for a later pass. + /// + pub(crate) fn digests_to_renew( + &self, + digests: impl Iterator, + now: Instant, + ) -> Vec { + let renew_after = self.inner.lease_time / 4; + let mut renewed_at = self.inner.renewed_at.lock(); + renewed_at.retain(|_, at| now.duration_since(*at) < renew_after); + digests + .filter(|digest| !renewed_at.contains_key(digest)) + .collect() + } + + pub(crate) fn record_renewed(&self, digests: &[Digest], at: Instant) { + self.inner + .renewed_at + .lock() + .extend(digests.iter().map(|digest| (*digest, at))); + } + pub async fn lease_all( &self, digests: impl Iterator, diff --git a/src/rust/fs/store/src/local_tests.rs b/src/rust/fs/store/src/local_tests.rs index 565d8037a1e..67277f96289 100644 --- a/src/rust/fs/store/src/local_tests.rs +++ b/src/rust/fs/store/src/local_tests.rs @@ -6,7 +6,7 @@ use crate::{EntryType, LocalOptions, ShrinkBehavior}; use std::collections::HashSet; use std::io::Write; use std::path::Path; -use std::time::Duration; +use std::time::{Duration, SystemTime}; use bytes::{BufMut, Bytes, BytesMut}; use hashing::{Digest, Fingerprint}; @@ -750,6 +750,40 @@ async fn get_missing_digests() { ) } +async fn large_file_mtime_after_lease(file_age: Duration) -> (SystemTime, SystemTime) { + let dir = TempDir::new().unwrap(); + let store = new_store(dir.path()); + let large_testdata = TestData::new("123456789".repeat(1000 * 512).as_str()); + let digest = prime_store_with_file_bytes(&store, large_testdata.bytes()).await; + let path = WalkDir::new(dir.path()) + .into_iter() + .map(|entry| entry.unwrap().into_path()) + .find(|path| path.file_name() == Some(digest.hash.to_hex().as_ref())) + .expect("large file should be stored in the fsdb"); + let aged_mtime = SystemTime::now() - file_age; + fs_set_times::set_mtime(&path, fs_set_times::SystemTimeSpec::Absolute(aged_mtime)).unwrap(); + + store + .lease_all(vec![(digest, EntryType::File)].into_iter()) + .await + .expect("Error leasing"); + + let mtime = std::fs::metadata(&path).unwrap().modified().unwrap(); + (aged_mtime, mtime) +} + +#[tokio::test] +async fn lease_renews_large_file_with_less_than_half_left() { + let (aged_mtime, mtime) = large_file_mtime_after_lease(Duration::from_secs(90 * 60)).await; + assert!(mtime > aged_mtime); +} + +#[tokio::test] +async fn lease_keeps_large_file_with_more_than_half_left() { + let (aged_mtime, mtime) = large_file_mtime_after_lease(Duration::from_secs(30 * 60)).await; + assert_eq!(mtime, aged_mtime); +} + pub fn new_store>(dir: P) -> ByteStore { ByteStore::new(task_executor::Executor::new(), dir).unwrap() } diff --git a/src/rust/fs/store/src/tests.rs b/src/rust/fs/store/src/tests.rs index 772ea3ab959..c9ce2ace347 100644 --- a/src/rust/fs/store/src/tests.rs +++ b/src/rust/fs/store/src/tests.rs @@ -4,7 +4,7 @@ use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::io::Read; use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; -use std::time::Duration; +use std::time::{Duration, SystemTime}; use tempfile::TempDir; use testutil::data::{TestData, TestDirectory}; @@ -22,8 +22,8 @@ use workunit_store::WorkunitStore; use crate::local::ByteStore; use crate::{ - EntryType, FileContent, MEGABYTES, RemoteProvider, RemoteStoreOptions, Snapshot, Store, - StoreError, StoreFileByDigest, UploadSummary, + EntryType, FileContent, LocalOptions, MEGABYTES, RemoteProvider, RemoteStoreOptions, Snapshot, + Store, StoreError, StoreFileByDigest, UploadSummary, }; pub(crate) const STORE_BATCH_API_SIZE_LIMIT: usize = 4 * 1024 * 1024; @@ -1843,3 +1843,64 @@ async fn big_file_mutable_root_dir() { assert_path(&materialize_dir.path().join("nested").join("file")); } } + +async fn store_large_file(store: &Store, dir: &Path) -> (Digest, PathBuf) { + let large_testdata = TestData::new("123456789".repeat(1000 * 512).as_str()); + let digest = store + .store_file_bytes(large_testdata.bytes(), true) + .await + .unwrap(); + let path = walkdir::WalkDir::new(dir) + .into_iter() + .map(|entry| entry.unwrap().into_path()) + .find(|path| path.file_name() == Some(digest.hash.to_hex().as_ref())) + .expect("large file should be stored in the fsdb"); + (digest, path) +} + +fn set_mtime(path: &Path, mtime: SystemTime) { + fs_set_times::set_mtime(path, fs_set_times::SystemTimeSpec::Absolute(mtime)).unwrap(); +} + +fn mtime(path: &Path) -> SystemTime { + std::fs::metadata(path).unwrap().modified().unwrap() +} + +#[tokio::test] +async fn lease_all_recursively_skips_recently_renewed_digest() { + let dir = TempDir::new().unwrap(); + let store = new_local_store(dir.path()); + let (digest, path) = store_large_file(&store, dir.path()).await; + store.lease_all_recursively([digest].iter()).await.unwrap(); + + let aged_mtime = SystemTime::now() - Duration::from_secs(90 * 60); + set_mtime(&path, aged_mtime); + store.lease_all_recursively([digest].iter()).await.unwrap(); + + assert_eq!(mtime(&path), aged_mtime); +} + +#[tokio::test] +async fn lease_all_recursively_renews_digest_after_a_quarter_of_the_lease_time() { + let dir = TempDir::new().unwrap(); + let lease_time = Duration::from_secs(4); + let store = Store::local_only_with_options( + task_executor::Executor::new(), + dir.path(), + dir.path(), + LocalOptions { + lease_time, + ..LocalOptions::default() + }, + ) + .unwrap(); + let (digest, path) = store_large_file(&store, dir.path()).await; + store.lease_all_recursively([digest].iter()).await.unwrap(); + + let aged_mtime = SystemTime::now() - lease_time; + set_mtime(&path, aged_mtime); + tokio::time::sleep(lease_time / 4 + Duration::from_millis(100)).await; + store.lease_all_recursively([digest].iter()).await.unwrap(); + + assert!(mtime(&path) > aged_mtime); +} diff --git a/src/rust/sharded_lmdb/src/lib.rs b/src/rust/sharded_lmdb/src/lib.rs index 325cde3d998..862815ba5a7 100644 --- a/src/rust/sharded_lmdb/src/lib.rs +++ b/src/rust/sharded_lmdb/src/lib.rs @@ -30,6 +30,12 @@ pub const DEFAULT_LEASE_TIME: Duration = Duration::from_secs(2 * 60 * 60); const VERSIONED_FINGERPRINT_SIZE: usize = FINGERPRINT_SIZE + 1; +fn lease_from_bytes(bytes: &[u8]) -> u64 { + let mut array = [0_u8; 8]; + array.copy_from_slice(bytes); + u64::from_le_bytes(array) +} + /// VersionedFingerprint is a byte buffer one longer than the number of bytes stored in a /// Fingerprint. It is just the byte pattern of a Fingerprint with the version number concatenated /// onto the end of it. @@ -395,11 +401,7 @@ impl ShardedLmdb { // when we delete from lmdb to track how much we've freed). let lease_until_unix_timestamp = txn .get(*lease_database, &key) - .map(|b| { - let mut array = [0_u8; 8]; - array.copy_from_slice(b); - u64::from_le_bytes(array) - }) + .map(lease_from_bytes) .unwrap_or_else(|e| match e { lmdb::Error::NotFound => 0, e => panic!("Error reading lease, probable lmdb corruption: {e:?}"), @@ -613,17 +615,36 @@ impl ShardedLmdb { .flatten() } + /// + /// Extends the lease on `fingerprint` to `lease_time` from now. + /// + /// A lease with more than half of `lease_time` left is not rewritten. pantsd renews the leases + /// of everything its graph references more often than that, so the entry is still renewed + /// before it could expire, and an idle pantsd does not rewrite every lease it holds on every + /// pass. + /// pub async fn lease(&self, fingerprint: Fingerprint) -> Result<(), String> { let store = self.clone(); self.executor .spawn_blocking(move || { let until_secs_since_epoch: u64 = store.lease_until_secs_since_epoch(); let (env, _, lease_database) = store.get(&fingerprint); - env.begin_rw_txn() - .and_then(|mut txn| { + let key = VersionedFingerprint::new(fingerprint, ShardedLmdb::SCHEMA_VERSION); + store + .lease_outlasts( + &env, + lease_database, + &key, + until_secs_since_epoch - store.lease_time.as_secs() / 2, + ) + .and_then(|fresh| { + if fresh { + return Ok(()); + } + let mut txn = env.begin_rw_txn()?; store.lease_inner( lease_database, - &VersionedFingerprint::new(fingerprint, ShardedLmdb::SCHEMA_VERSION), + &key, until_secs_since_epoch, &mut txn, )?; @@ -636,6 +657,21 @@ impl ShardedLmdb { .flatten() } + fn lease_outlasts( + &self, + env: &Environment, + lease_database: Database, + versioned_fingerprint: &VersionedFingerprint, + secs_since_epoch: u64, + ) -> Result { + let txn = env.begin_ro_txn()?; + match txn.get(lease_database, &versioned_fingerprint.as_ref()) { + Ok(bytes) => Ok(lease_from_bytes(bytes) > secs_since_epoch), + Err(lmdb::Error::NotFound) => Ok(false), + Err(e) => Err(e), + } + } + fn lease_inner( &self, database: Database, diff --git a/src/rust/sharded_lmdb/src/tests.rs b/src/rust/sharded_lmdb/src/tests.rs index 68cd601a3fe..5fd35bf1af9 100644 --- a/src/rust/sharded_lmdb/src/tests.rs +++ b/src/rust/sharded_lmdb/src/tests.rs @@ -1,14 +1,16 @@ // Copyright 2022 Pants project contributors (see CONTRIBUTORS.md). // Licensed under the Apache License, Version 2.0 (see LICENSE). use std::collections::HashMap; +use std::time::{SystemTime, UNIX_EPOCH}; use bytes::{Buf, Bytes}; -use hashing::Digest; +use hashing::{Digest, Fingerprint}; +use lmdb::Transaction; use parking_lot::Mutex; use task_executor::Executor; use tempfile::TempDir; -use crate::{DEFAULT_LEASE_TIME, ShardedLmdb}; +use crate::{DEFAULT_LEASE_TIME, ShardedLmdb, VersionedFingerprint, lease_from_bytes}; fn new_store(shard_count: u8) -> (ShardedLmdb, TempDir) { let tempdir = TempDir::new().unwrap(); @@ -95,3 +97,75 @@ async fn store_failure() { fn bytes(content: u8) -> Bytes { Bytes::from(vec![content; 100]) } + +fn now_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() +} + +fn versioned(fingerprint: Fingerprint) -> VersionedFingerprint { + VersionedFingerprint::new(fingerprint, ShardedLmdb::SCHEMA_VERSION) +} + +fn set_lease(s: &ShardedLmdb, fingerprint: Fingerprint, until_secs_since_epoch: u64) { + let (env, _, lease_database) = s.get(&fingerprint); + let mut txn = env.begin_rw_txn().unwrap(); + s.lease_inner( + lease_database, + &versioned(fingerprint), + until_secs_since_epoch, + &mut txn, + ) + .unwrap(); + txn.commit().unwrap(); +} + +fn stored_lease(s: &ShardedLmdb, fingerprint: Fingerprint) -> u64 { + let (env, _, lease_database) = s.get(&fingerprint); + let txn = env.begin_ro_txn().unwrap(); + lease_from_bytes( + txn.get(lease_database, &versioned(fingerprint).as_ref()) + .unwrap(), + ) +} + +#[tokio::test] +async fn lease_creates_missing_lease() { + let (s, _tempdir) = new_store(1); + let fingerprint = Digest::of_bytes(&bytes(0)).hash; + + let before = now_secs(); + s.lease(fingerprint).await.unwrap(); + + assert!(stored_lease(&s, fingerprint) >= before + DEFAULT_LEASE_TIME.as_secs()); +} + +#[tokio::test] +async fn lease_renews_lease_with_less_than_half_left() { + let (s, _tempdir) = new_store(1); + let fingerprint = Digest::of_bytes(&bytes(0)).hash; + set_lease( + &s, + fingerprint, + now_secs() + DEFAULT_LEASE_TIME.as_secs() / 2 - 60, + ); + + let before = now_secs(); + s.lease(fingerprint).await.unwrap(); + + assert!(stored_lease(&s, fingerprint) >= before + DEFAULT_LEASE_TIME.as_secs()); +} + +#[tokio::test] +async fn lease_keeps_lease_with_more_than_half_left() { + let (s, _tempdir) = new_store(1); + let fingerprint = Digest::of_bytes(&bytes(0)).hash; + let until = now_secs() + DEFAULT_LEASE_TIME.as_secs() / 2 + 60; + set_lease(&s, fingerprint, until); + + s.lease(fingerprint).await.unwrap(); + + assert_eq!(stored_lease(&s, fingerprint), until); +}