diff --git a/rust/src/caches/mod.rs b/rust/src/caches/mod.rs index 80e789ad5a..f122d6f98c 100644 --- a/rust/src/caches/mod.rs +++ b/rust/src/caches/mod.rs @@ -508,6 +508,28 @@ where expired } + /// When the coldest entry was last read, or `None` if the cache is empty. + pub fn oldest_access(&self) -> Option { + self.lock().oldest_access() + } + + /// Evict up to `max` of the coldest entries, as long as they were last + /// read before `before`. Returns the evicted values, for the caller to + /// drop. + /// + /// For memory-pressure eviction, which works across every cache from the + /// coldest entry up. See [`registry::CacheRegistry::evict_coldest`]. + /// Counted as evictions for size, as the cache is too big for the memory + /// it is allowed. + pub fn evict_coldest(&self, before: SystemTime, max: usize) -> Vec { + let evicted = self.lock().evict_coldest(before, max); + + self.evicted_size + .fetch_add(evicted.len() as u64, Ordering::Relaxed); + + evicted + } + /// Rescale the cache's bound to `original * factor`, evicting from the /// cold end if it shrank. Returns the evicted values, for the caller to /// drop. @@ -769,6 +791,20 @@ where evicted } + fn oldest_access(&self) -> Option { + self.entries.peek_lru().map(|(_, entry)| entry.last_read) + } + + /// Pop up to `max` of the coldest entries read before `before`, returning + /// the values for the caller to drop. + fn evict_coldest(&mut self, before: SystemTime, max: usize) -> Vec { + let mut evicted = Vec::new(); + while evicted.len() < max && self.oldest_access().is_some_and(|read| read < before) { + evicted.push(self.pop_lru().expect("oldest_access just found one")); + } + evicted + } + /// Pop every entry not read since `cutoff`, returning the values for the /// caller to drop. fn expire_before(&mut self, cutoff: SystemTime) -> Vec { @@ -1456,4 +1492,25 @@ mod tests { cache.count_miss(); assert_eq!(cache.hit_counts(), (2, 2)); } + + #[test] + fn evict_coldest_stops_at_the_bound_and_the_count() { + let cache = cache(10); + cache.insert_at("a", 1, at(1)); + cache.insert_at("b", 2, at(2)); + cache.insert_at("c", 3, at(3)); + assert_eq!(cache.oldest_access(), Some(at(1))); + + // Only entries read strictly before the bound go. + assert_eq!(cache.evict_coldest(at(2), 10), vec![1]); + assert_eq!(cache.oldest_access(), Some(at(2))); + + // And no more than asked for. + assert_eq!(cache.evict_coldest(at(10), 1), vec![2]); + assert_eq!(cache.oldest_access(), Some(at(3))); + assert_eq!(cache.eviction_counts().0, 2); + + cache.evict_coldest(at(10), 10); + assert_eq!(cache.oldest_access(), None); + } } diff --git a/rust/src/caches/python.rs b/rust/src/caches/python.rs index ee46fe45a2..e132a33f85 100644 --- a/rust/src/caches/python.rs +++ b/rust/src/caches/python.rs @@ -118,6 +118,13 @@ pub(crate) trait CacheHandle: Send + Sync { /// Evict every entry not read since `cutoff`, returning how many went. fn expire_entries_before(&self, cutoff: SystemTime) -> usize; + + /// When the coldest entry was last read, or `None` if the cache is empty. + fn oldest_access(&self) -> Option; + + /// Evict up to `max` of the coldest entries read before `before`, + /// returning how many went. + fn evict_coldest(&self, before: SystemTime, max: usize) -> usize; } /// What `register_cache` reads off a cache at each scrape. @@ -189,6 +196,15 @@ where count } + + fn oldest_access(&self) -> Option { + Cache::oldest_access(self) + } + + fn evict_coldest(&self, before: SystemTime, max: usize) -> usize { + // Dropped at the end of the statement, with the cache lock released. + Cache::evict_coldest(self, before, max).len() + } } /// A Rust cache, as Synapse sees it. Bound on the datastore under the diff --git a/rust/src/caches/registry.rs b/rust/src/caches/registry.rs index c2a215a834..258ffed8af 100644 --- a/rust/src/caches/registry.rs +++ b/rust/src/caches/registry.rs @@ -27,9 +27,14 @@ //! Unlike Python, which threads every entry of every cache onto one global //! linked list, this expiry decomposes per cache: each cache's LRU order is //! also its age order, so each can find its own stale entries and stop at the -//! first live one. (Memory-pressure eviction, which *does* need a global -//! coldest-first order, isn't implemented yet. It will need a k-way merge over -//! the per-cache LRU tails rather than a global list.) +//! first live one. +//! +//! Memory-pressure eviction does need a global coldest-first order, and gets +//! it without a global list: each cache's LRU order is a sorted run, so +//! [`CacheRegistry::evict_coldest`] merges them by always evicting from the +//! cache whose coldest entry is coldest. Python drives it in batches from the +//! same looping call, interleaved with its own walk so that both sides go +//! coldest first together. use std::{ hash::Hash, @@ -152,6 +157,57 @@ impl CacheRegistry { .sum() } + /// Evict up to `max` entries read before `before`, coldest first across + /// every cache in this registry. + /// + /// Returns how many went, and when the coldest entry left was last read, + /// or `None` if every cache is empty. A caller interleaving this with + /// another coldest-first walk can skip calling again until that walk + /// reaches entries younger than the returned time: anything inserted or + /// read since is younger still. + pub fn evict_coldest(&self, before: SystemTime, max: usize) -> (usize, Option) { + let caches = self.live(); + let mut evicted = 0; + + loop { + // The coldest cache, and how cold the next coldest is. A linear + // scan, as there are only a handful of caches. + let mut coldest: Option<(SystemTime, &Arc)> = None; + let mut runner_up: Option = None; + for cache in &caches { + let Some(oldest) = cache.oldest_access() else { + continue; + }; + match coldest { + Some((current, _)) if oldest >= current => { + runner_up = Some(runner_up.map_or(oldest, |r| r.min(oldest))); + } + _ => { + runner_up = coldest.map(|(current, _)| current).or(runner_up); + coldest = Some((oldest, cache)); + } + } + } + + let Some((oldest, cache)) = coldest else { + return (evicted, None); + }; + if oldest >= before || evicted >= max { + return (evicted, Some(oldest)); + } + + // Take everything in this cache that is colder than every other + // cache's coldest entry, in one go. When the two tie, that is + // nothing, so take one entry to make progress. + let bound = runner_up.map_or(before, |r| r.min(before)); + let mut count = cache.evict_coldest(bound, max - evicted); + if count == 0 { + count = cache.evict_coldest(before, 1); + } + evicted += count; + } + } + /// Every cache that is still alive, pruning the ones that aren't. /// /// A copy, so that callers work outside the registry lock: expiring takes @@ -234,4 +290,40 @@ mod tests { assert_eq!(registry.expire_caches(at(10), THIRTY_SECS), 0); assert_eq!(cache.len(), 1); } + + #[test] + fn evict_coldest_merges_caches_coldest_first() { + let registry = CacheRegistry::new(); + let first: Arc> = registry.new_cache("first", 10); + let second: Arc> = registry.new_cache("second", 10); + first.insert_at("a", 1, at(1)); + second.insert_at("b", 2, at(2)); + first.insert_at("c", 3, at(3)); + second.insert_at("d", 4, at(4)); + + // Three entries are colder than the bound, but only two are asked + // for, and they are the two coldest whichever cache holds them. + assert_eq!(registry.evict_coldest(at(4), 2), (2, Some(at(3)))); + assert_eq!((first.len(), second.len()), (1, 1)); + assert!(first.get_at(&"c", at(3)).is_some()); + + // The bound is exclusive, and the coldest survivor is reported. + assert_eq!(registry.evict_coldest(at(4), 10), (1, Some(at(4)))); + assert_eq!(registry.evict_coldest(at(4), 10), (0, Some(at(4)))); + + assert_eq!(registry.evict_coldest(at(5), 10), (1, None)); + assert_eq!(first.eviction_counts().0 + second.eviction_counts().0, 4); + } + + #[test] + fn evict_coldest_makes_progress_on_ties() { + let registry = CacheRegistry::new(); + let first: Arc> = registry.new_cache("first", 10); + let second: Arc> = registry.new_cache("second", 10); + first.insert_at("a", 1, at(1)); + first.insert_at("b", 2, at(1)); + second.insert_at("c", 3, at(1)); + + assert_eq!(registry.evict_coldest(at(2), 10), (3, None)); + } } diff --git a/rust/src/runtime.rs b/rust/src/runtime.rs index 507883033a..901b9805f8 100644 --- a/rust/src/runtime.rs +++ b/rust/src/runtime.rs @@ -26,7 +26,7 @@ use std::ops::Deref; use std::sync::{Arc, Mutex, Weak}; -use std::time::Duration; +use std::time::{Duration, UNIX_EPOCH}; use anyhow::Context; use pyo3::{ @@ -259,6 +259,27 @@ impl RustRuntime { .caches .expire_caches(self.inner.clock.now(), max_age)) } + + /// Evict up to `max` entries last read before `before_secs` (seconds + /// since the Unix epoch), coldest first across every Rust cache of this + /// homeserver. For memory-pressure eviction, which Python drives. + /// + /// Returns how many went, and when the coldest entry left was last read, + /// in seconds since the epoch, or `None` if the caches are empty. See + /// [`crate::caches::registry::CacheRegistry::evict_coldest`]. + fn evict_coldest(&self, before_secs: f64, max: usize) -> PyResult<(usize, Option)> { + let before = Duration::try_from_secs_f64(before_secs.max(0.0)) + .map(|since_epoch| UNIX_EPOCH + since_epoch) + .map_err(|err| PyValueError::new_err(format!("bad time {before_secs}: {err}")))?; + + let (evicted, coldest) = self.inner.caches.evict_coldest(before, max); + let coldest = coldest.map(|time| { + time.duration_since(UNIX_EPOCH) + .unwrap_or(Duration::ZERO) + .as_secs_f64() + }); + Ok((evicted, coldest)) + } } /// The callable registered with `HomeServer.register_sync_shutdown_handler`, diff --git a/synapse/synapse_rust/runtime.pyi b/synapse/synapse_rust/runtime.pyi index 85050bb156..449449b833 100644 --- a/synapse/synapse_rust/runtime.pyi +++ b/synapse/synapse_rust/runtime.pyi @@ -30,6 +30,16 @@ class RustRuntime: Rust and Python caches expire off one tick and one set of config options.""" + def evict_coldest(self, before_secs: float, max: int) -> tuple[int, float | None]: + """Evict up to `max` entries last read before `before_secs` (seconds + since the Unix epoch), coldest first across every Rust cache of this + homeserver. + + Returns how many were evicted, and when the coldest entry left was last + read, or None if the Rust caches are empty. Nothing in them gets any + colder than that, so a caller can skip calling again until it wants to + evict entries read after it.""" + def set_virtual_time_msec(millis: int) -> None: """Pin the Rust clock to the given time. Only for tests, which run against a virtual reactor clock.""" diff --git a/synapse/util/caches/lrucache.py b/synapse/util/caches/lrucache.py index baa361f4af..a4711256f6 100644 --- a/synapse/util/caches/lrucache.py +++ b/synapse/util/caches/lrucache.py @@ -136,9 +136,10 @@ def _expire_old_entries( # # `expiry_seconds` is infinite when only `cache_autotuning` is # configured, i.e. when age-based expiry is switched off and we are only - # here to evict on memory pressure, which Rust doesn't do yet. + # here to evict on memory pressure. + rust_runtime = hs.get_rust_runtime() if math.isfinite(expiry_seconds): - dropped = hs.get_rust_runtime().expire_caches(expiry_seconds) + dropped = rust_runtime.expire_caches(expiry_seconds) if dropped: logger.info("Dropped %d items from Rust caches", dropped) @@ -166,6 +167,52 @@ def _expire_old_entries( "Unable to read allocated memory, skipping memory-based cache eviction." ) + def still_over_target() -> bool: + """Re-read the memory usage. False once it is under target, or if + it cannot be read, either of which ends memory-based eviction.""" + assert jemalloc_interface is not None + try: + jemalloc_interface.refresh_stats() + mem_usage = jemalloc_interface.get_stat("allocated") + if mem_usage < target_cache_memory_usage: + logger.info("Stop memory-based cache eviction.") + return False + return True + except Exception: + logger.warning( + "Unable to read allocated memory, this may affect memory-based cache eviction." + ) + # If we've failed to read the current memory usage then we + # should stop trying to evict based on memory usage + return False + + # Under memory pressure the Rust caches are evicted coldest first + # together with the Python ones: before evicting a Python entry, every + # Rust entry colder than it goes. `rust_coldest` is when the coldest + # Rust entry was last read, as of the last call into Rust, and nothing + # there gets colder, so Rust is only called again once the walk reaches + # entries younger than that. Minus infinity means "not asked yet". + rust_coldest: float | None = -math.inf + rust_evicted = 0 + + async def evict_rust_before(before: float) -> bool: + """Evict Rust entries read before `before` while memory is over + target. Returns whether it still is. + + Yields to the reactor every 10000 entries, as the Python walk does. + """ + nonlocal rust_coldest, rust_evicted + while rust_coldest is not None and rust_coldest < before: + evicted, rust_coldest = rust_runtime.evict_coldest(before, 100) + if (rust_evicted + evicted) // 10000 > rust_evicted // 10000: + await clock.sleep(Duration(seconds=0)) + rust_evicted += evicted + # Each batch is as many entries as the Python walk evicts + # between checks of the memory usage. + if evicted and not still_over_target(): + return False + return True + while node is not GLOBAL_ROOT: # Only the root node isn't a `_TimedListNode`. assert isinstance(node, _TimedListNode) @@ -185,6 +232,21 @@ def _expire_old_entries( ): break + if evicting_due_to_memory: + evicting_due_to_memory = await evict_rust_before( + node.last_access_ts_secs + ) + # If we yielded, this node may have been evicted meanwhile. + if node.prev_node is None: + break + # Evicting from Rust may have brought the usage under target, + # in which case this node only goes if it has expired. + if ( + not evicting_due_to_memory + and node.last_access_ts_secs > now - expiry_seconds + ): + break + cache_entry = node.get_cache_entry() next_node = node.prev_node @@ -197,19 +259,7 @@ def _expire_old_entries( # Check mem allocation periodically if we are evicting a bunch of caches if jemalloc_interface and evicting_due_to_memory and (i + 1) % 100 == 0: - try: - jemalloc_interface.refresh_stats() - mem_usage = jemalloc_interface.get_stat("allocated") - if mem_usage < target_cache_memory_usage: - evicting_due_to_memory = False - logger.info("Stop memory-based cache eviction.") - except Exception: - logger.warning( - "Unable to read allocated memory, this may affect memory-based cache eviction." - ) - # If we've failed to read the current memory usage then we - # should stop trying to evict based on memory usage - evicting_due_to_memory = False + evicting_due_to_memory = still_over_target() # If we do lots of work at once we yield to allow other stuff to happen. if (i + 1) % 10000 == 0: @@ -229,6 +279,14 @@ def _expire_old_entries( i += 1 + # The Python entries left are all too young to evict on memory, or + # there are none. If memory is still high, the Rust entries that are + # old enough go too. + if evicting_due_to_memory: + await evict_rust_before(now - min_cache_ttl) + + if rust_evicted: + logger.info("Dropped %d items from Rust caches for memory", rust_evicted) logger.info("Dropped %d items from caches", i) return hs.run_as_background_process( diff --git a/tests/util/test_lrucache.py b/tests/util/test_lrucache.py index dcbfcfa2e4..c5c40aa698 100644 --- a/tests/util/test_lrucache.py +++ b/tests/util/test_lrucache.py @@ -23,6 +23,7 @@ from unittest.mock import Mock, patch from synapse.metrics.jemalloc import JemallocStats +from synapse.rest.admin.experimental_features import ExperimentalFeature from synapse.types import JsonDict from synapse.util.caches.lrucache import LruCache, setup_expire_lru_cache_entries from synapse.util.caches.treecache import TreeCache @@ -437,6 +438,113 @@ class MemoryEvictionTestCase(unittest.HomeserverTestCase): self.assertEqual(cache.get("key1"), 1) self.assertEqual(cache.get("key2"), 2) + def _fill_rust_cache(self) -> None: + """Put one entry in a Rust cache, stamped with the current time.""" + user_id = self.get_success( + self.hs.get_registration_handler().register_user(localpart="rust") + ) + self.get_success( + self.hs.get_datastores().main.is_feature_enabled( + user_id, ExperimentalFeature.MSC3881 + ) + ) + + @override_config( + { + "caches": { + "cache_autotuning": { + "max_cache_memory_usage": "700M", + "target_cache_memory_usage": "500M", + "min_cache_ttl": "5m", + } + } + } + ) + @patch("synapse.util.caches.lrucache.get_jemalloc_stats") + def test_evict_memory_reaches_rust_caches(self, jemalloc_interface: Mock) -> None: + """Rust cache entries are evicted on memory pressure, in order of + last access together with the Python ones, and not before + `min_cache_ttl`.""" + mock_jemalloc_class = Mock(spec=JemallocStats) + jemalloc_interface.return_value = mock_jemalloc_class + mock_jemalloc_class.get_stat.return_value = 924288000 + + rust_cache = self.hs.get_datastores().main.is_feature_enabled_for_user + setup_expire_lru_cache_entries(self.hs) + cache: LruCache[str, int] = LruCache( + max_size=4, server_name="test_server", clock=self.hs.get_clock() + ) + + cache["old"] = 1 + self.reactor.advance(120) + self._fill_rust_cache() + self.assertEqual(len(rust_cache), 1) + self.reactor.advance(120) + cache["new"] = 2 + + # Past `min_cache_ttl` for the Python entry read first. The Rust entry + # is younger, so it stays. + self.reactor.advance(120) + self.assertNotIn("old", cache) + self.assertEqual(len(rust_cache), 1) + self.assertIn("new", cache) + + # Then the Rust entry... + self.reactor.advance(120) + self.assertEqual(len(rust_cache), 0) + self.assertIn("new", cache) + + # ...then the Python entry read last. + self.reactor.advance(120) + self.assertNotIn("new", cache) + + @override_config( + { + "caches": { + "cache_autotuning": { + "max_cache_memory_usage": "700M", + "target_cache_memory_usage": "500M", + "min_cache_ttl": "5m", + } + } + } + ) + @patch("synapse.util.caches.lrucache.get_jemalloc_stats") + def test_evict_memory_stops_after_rust_batch( + self, jemalloc_interface: Mock + ) -> None: + """Memory is checked after each batch of Rust evictions, and once it + is under target, younger Python entries stay.""" + mock_jemalloc_class = Mock(spec=JemallocStats) + jemalloc_interface.return_value = mock_jemalloc_class + mock_jemalloc_class.get_stat.return_value = 10000 + + rust_cache = self.hs.get_datastores().main.is_feature_enabled_for_user + setup_expire_lru_cache_entries(self.hs) + cache: LruCache[str, int] = LruCache( + max_size=4, server_name="test_server", clock=self.hs.get_clock() + ) + + cache["old"] = 1 + self.reactor.advance(60) + self._fill_rust_cache() + self.reactor.advance(60) + cache["new"] = 2 + + # Everything is old enough to evict, but memory is fine. + self.reactor.advance(600) + self.assertIn("old", cache) + self.assertEqual(len(rust_cache), 1) + self.assertIn("new", cache) + + # Memory is high on the next tick's first reading, and back under + # target once the Rust entry has gone. + mock_jemalloc_class.get_stat.side_effect = [924288000] + [10000] * 100 + self.reactor.advance(30) + self.assertNotIn("old", cache) + self.assertEqual(len(rust_cache), 0) + self.assertIn("new", cache) + class ExtraIndexLruCacheTestCase(unittest.HomeserverTestCase): def test_invalidate_simple(self) -> None: