From 6f3361b3de8715b14ba05badc374925dc1333f2e Mon Sep 17 00:00:00 2001 From: quantumish Date: Thu, 24 Jul 2025 14:08:26 -0700 Subject: [PATCH] Make references all 'static --- libs/neon-shmem/src/hash.rs | 45 ++--- libs/neon-shmem/src/hash/core.rs | 14 +- libs/neon-shmem/src/hash/entry.rs | 18 +- .../communicator/src/backend_interface.rs | 2 +- pgxn/neon/communicator/src/init.rs | 2 +- .../neon/communicator/src/integrated_cache.rs | 164 +++++++++--------- .../src/worker_process/main_loop.rs | 2 +- 7 files changed, 127 insertions(+), 120 deletions(-) diff --git a/libs/neon-shmem/src/hash.rs b/libs/neon-shmem/src/hash.rs index 3353e39a58..b58f900029 100644 --- a/libs/neon-shmem/src/hash.rs +++ b/libs/neon-shmem/src/hash.rs @@ -49,15 +49,15 @@ pub enum HashMapShrinkError { /// If a new process is launched with fork(), the child process inherits /// this struct. #[must_use] -pub struct HashMapInit<'a, K, V, S = rustc_hash::FxBuildHasher> { +pub struct HashMapInit { shmem_handle: Option, - shared_ptr: *mut HashMapShared<'a, K, V>, + shared_ptr: *mut HashMapShared, shared_size: usize, hasher: S, num_buckets: u32, } -impl<'a, K, V, S> Debug for HashMapInit<'a, K, V, S> +impl Debug for HashMapInit where K: Debug, V: Debug, @@ -79,16 +79,16 @@ where /// /// XXX: We're not making use of it at the moment, but this struct could /// hold process-local information in the future. -pub struct HashMapAccess<'a, K, V, S = rustc_hash::FxBuildHasher> { +pub struct HashMapAccess { shmem_handle: Option, - shared_ptr: *mut HashMapShared<'a, K, V>, + shared_ptr: *mut HashMapShared, hasher: S, } -unsafe impl Sync for HashMapAccess<'_, K, V, S> {} -unsafe impl Send for HashMapAccess<'_, K, V, S> {} +unsafe impl Sync for HashMapAccess {} +unsafe impl Send for HashMapAccess {} -impl<'a, K, V, S> Debug for HashMapAccess<'a, K, V, S> +impl Debug for HashMapAccess where K: Debug, V: Debug, @@ -102,14 +102,14 @@ where } } -impl<'a, K: Clone + Hash + Eq, V, S> HashMapInit<'a, K, V, S> { +impl HashMapInit { /// Change the 'hasher' used by the hash table. /// /// NOTE: This must be called right after creating the hash table, /// before inserting any entries and before calling attach_writer/reader. /// Otherwise different accessors could be using different hash function, /// with confusing results. - pub fn with_hasher(self, hasher: T) -> HashMapInit<'a, K, V, T> { + pub fn with_hasher(self, hasher: T) -> HashMapInit { HashMapInit { hasher, shmem_handle: self.shmem_handle, @@ -177,7 +177,7 @@ impl<'a, K: Clone + Hash + Eq, V, S> HashMapInit<'a, K, V, S> { } /// Attach to a hash table for writing. - pub fn attach_writer(self) -> HashMapAccess<'a, K, V, S> { + pub fn attach_writer(self) -> HashMapAccess { HashMapAccess { shmem_handle: self.shmem_handle, shared_ptr: self.shared_ptr, @@ -189,7 +189,7 @@ impl<'a, K: Clone + Hash + Eq, V, S> HashMapInit<'a, K, V, S> { /// /// This is a holdover from a previous implementation and is being kept around for /// backwards compatibility reasons. - pub fn attach_reader(self) -> HashMapAccess<'a, K, V, S> { + pub fn attach_reader(self) -> HashMapAccess { self.attach_writer() } } @@ -206,14 +206,14 @@ impl<'a, K: Clone + Hash + Eq, V, S> HashMapInit<'a, K, V, S> { /// dictionary /// /// In between the above parts, there can be padding bytes to align the parts correctly. -type HashMapShared<'a, K, V> = RwLock>; +type HashMapShared = RwLock>; -impl<'a, K, V> HashMapInit<'a, K, V, rustc_hash::FxBuildHasher> +impl HashMapInit where - K: Clone + Hash + Eq, + K: Clone + Hash + Eq + 'static, { /// Place the hash table within a user-supplied fixed memory area. - pub fn with_fixed(num_buckets: u32, area: &'a mut [MaybeUninit]) -> Self { + pub fn with_fixed(num_buckets: u32, area: &'static mut [MaybeUninit]) -> Self { Self::new( num_buckets, None, @@ -269,9 +269,10 @@ where } } -impl<'a, K, V, S: BuildHasher> HashMapAccess<'a, K, V, S> +impl HashMapAccess where - K: Clone + Hash + Eq, + K: Clone + Hash + Eq + 'static, + V: 'static { /// Hash a key using the map's hasher. #[inline] @@ -279,7 +280,7 @@ where self.hasher.hash_one(key) } - fn entry_with_hash(&self, key: K, hash: u64) -> Entry<'a, '_, K, V> { + fn entry_with_hash(&self, key: K, hash: u64) -> Entry<'_, K, V> { let mut map = unsafe { self.shared_ptr.as_ref() }.unwrap().write(); let dict_pos = hash as usize % map.dictionary.len(); let first = map.dictionary[dict_pos]; @@ -331,7 +332,7 @@ where /// /// NB: This takes a write lock as there's no way to distinguish whether the intention /// is to use the entry for reading or for writing in advance. - pub fn entry(&self, key: K) -> Entry<'a, '_, K, V> { + pub fn entry(&self, key: K) -> Entry<'_, K, V> { let hash = self.get_hash_value(&key); self.entry_with_hash(key, hash) } @@ -365,7 +366,7 @@ where /// Has more overhead than one would intuitively expect: performs both a clone of the key /// due to the [`OccupiedEntry`] type owning the key and also a hash of the key in order /// to enable repairing the hash chain if the entry is removed. - pub fn entry_at_bucket(&self, pos: usize) -> Option> { + pub fn entry_at_bucket(&self, pos: usize) -> Option> { let map = unsafe { self.shared_ptr.as_mut() }.unwrap().write(); if pos >= map.buckets.len() { return None; @@ -430,7 +431,7 @@ where /// in the process. fn rehash_dict( &self, - inner: &mut CoreHashMap<'a, K, V>, + inner: &mut CoreHashMap, buckets_ptr: *mut core::Bucket, end_ptr: *mut u8, num_buckets: u32, diff --git a/libs/neon-shmem/src/hash/core.rs b/libs/neon-shmem/src/hash/core.rs index 0d40a1877b..d160eeebda 100644 --- a/libs/neon-shmem/src/hash/core.rs +++ b/libs/neon-shmem/src/hash/core.rs @@ -32,11 +32,11 @@ where } /// Core hash table implementation. -pub(crate) struct CoreHashMap<'a, K, V> { +pub(crate) struct CoreHashMap { /// Dictionary used to map hashes to bucket indices. - pub(crate) dictionary: &'a mut [u32], + pub(crate) dictionary: &'static mut [u32], /// Buckets containing key-value pairs. - pub(crate) buckets: &'a mut [Bucket], + pub(crate) buckets: &'static mut [Bucket], /// Head of the freelist. pub(crate) free_head: u32, /// Maximum index of a bucket allowed to be allocated. [`INVALID_POS`] if no limit. @@ -45,7 +45,7 @@ pub(crate) struct CoreHashMap<'a, K, V> { pub(crate) buckets_in_use: u32, } -impl<'a, K, V> Debug for CoreHashMap<'a, K, V> +impl Debug for CoreHashMap where K: Debug, V: Debug, @@ -65,7 +65,7 @@ where #[derive(Debug, PartialEq)] pub struct FullError; -impl<'a, K: Clone + Hash + Eq, V> CoreHashMap<'a, K, V> { +impl CoreHashMap { const FILL_FACTOR: f32 = 0.60; /// Estimate the size of data contained within the the hash map. @@ -83,8 +83,8 @@ impl<'a, K: Clone + Hash + Eq, V> CoreHashMap<'a, K, V> { } pub fn new( - buckets: &'a mut [MaybeUninit>], - dictionary: &'a mut [MaybeUninit], + buckets: &'static mut [MaybeUninit>], + dictionary: &'static mut [MaybeUninit], ) -> Self { // Initialize the buckets for i in 0..buckets.len() { diff --git a/libs/neon-shmem/src/hash/entry.rs b/libs/neon-shmem/src/hash/entry.rs index 560a20db1d..133f7023f0 100644 --- a/libs/neon-shmem/src/hash/entry.rs +++ b/libs/neon-shmem/src/hash/entry.rs @@ -6,9 +6,9 @@ use crate::sync::{RwLockWriteGuard, ValueWriteGuard}; use std::hash::Hash; use std::mem; -pub enum Entry<'a, 'b, K, V> { - Occupied(OccupiedEntry<'a, 'b, K, V>), - Vacant(VacantEntry<'a, 'b, K, V>), +pub enum Entry<'b, K: 'static, V: 'static> { + Occupied(OccupiedEntry<'b, K, V>), + Vacant(VacantEntry<'b, K, V>), } /// Enum representing the previous position within a chain. @@ -22,9 +22,9 @@ pub(crate) enum PrevPos { Unknown(u64), } -pub struct OccupiedEntry<'a, 'b, K, V> { +pub struct OccupiedEntry<'b, K: 'static, V: 'static> { /// Mutable reference to the map containing this entry. - pub(crate) map: RwLockWriteGuard<'b, CoreHashMap<'a, K, V>>, + pub(crate) map: RwLockWriteGuard<'b, CoreHashMap>, /// The key of the occupied entry pub(crate) _key: K, /// The index of the previous entry in the chain. @@ -33,7 +33,7 @@ pub struct OccupiedEntry<'a, 'b, K, V> { pub(crate) bucket_pos: u32, } -impl OccupiedEntry<'_, '_, K, V> { +impl OccupiedEntry<'_, K, V> { pub fn get(&self) -> &V { &self.map.buckets[self.bucket_pos as usize] .inner @@ -104,16 +104,16 @@ impl OccupiedEntry<'_, '_, K, V> { } /// An abstract view into a vacant entry within the map. -pub struct VacantEntry<'a, 'b, K, V> { +pub struct VacantEntry<'b, K: 'static, V: 'static> { /// Mutable reference to the map containing this entry. - pub(crate) map: RwLockWriteGuard<'b, CoreHashMap<'a, K, V>>, + pub(crate) map: RwLockWriteGuard<'b, CoreHashMap>, /// The key to be inserted into this entry. pub(crate) key: K, /// The position within the dictionary corresponding to the key's hash. pub(crate) dict_pos: u32, } -impl<'b, K: Clone + Hash + Eq, V> VacantEntry<'_, 'b, K, V> { +impl<'b, K: Clone + Hash + Eq + 'static, V: 'static> VacantEntry<'b, K, V> { /// Insert a value into the vacant entry, finding and populating an empty bucket in the process. /// /// # Errors diff --git a/pgxn/neon/communicator/src/backend_interface.rs b/pgxn/neon/communicator/src/backend_interface.rs index abc982193e..49782530a8 100644 --- a/pgxn/neon/communicator/src/backend_interface.rs +++ b/pgxn/neon/communicator/src/backend_interface.rs @@ -18,7 +18,7 @@ pub struct CommunicatorBackendStruct<'t> { pending_cache_read_op: Option>, - integrated_cache: &'t IntegratedCacheReadAccess<'t>, + integrated_cache: &'t IntegratedCacheReadAccess, } #[unsafe(no_mangle)] diff --git a/pgxn/neon/communicator/src/init.rs b/pgxn/neon/communicator/src/init.rs index f5af93cc97..811c77935d 100644 --- a/pgxn/neon/communicator/src/init.rs +++ b/pgxn/neon/communicator/src/init.rs @@ -38,7 +38,7 @@ pub struct CommunicatorInitStruct { pub neon_request_slots: &'static [NeonIORequestSlot], - pub integrated_cache_init_struct: IntegratedCacheInitStruct<'static>, + pub integrated_cache_init_struct: IntegratedCacheInitStruct, } impl std::fmt::Debug for CommunicatorInitStruct { diff --git a/pgxn/neon/communicator/src/integrated_cache.rs b/pgxn/neon/communicator/src/integrated_cache.rs index 71b2ee1645..469054a1b9 100644 --- a/pgxn/neon/communicator/src/integrated_cache.rs +++ b/pgxn/neon/communicator/src/integrated_cache.rs @@ -24,6 +24,7 @@ use std::mem::MaybeUninit; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; +use std::sync::Arc; use utils::lsn::{AtomicLsn, Lsn}; @@ -40,16 +41,16 @@ use neon_shmem::shmem::ShmemHandle; const RELSIZE_CACHE_SIZE: u32 = 64 * 1024; /// This struct is initialized at postmaster startup, and passed to all the processes via fork(). -pub struct IntegratedCacheInitStruct<'t> { - relsize_cache_handle: HashMapInit<'t, RelKey, RelEntry>, - block_map_handle: HashMapInit<'t, BlockKey, BlockEntry>, +pub struct IntegratedCacheInitStruct { + relsize_cache_handle: HashMapInit, + block_map_handle: HashMapInit, } /// Represents write-access to the integrated cache. This is used by the communicator process. #[derive(Debug)] -pub struct IntegratedCacheWriteAccess<'t> { - relsize_cache: neon_shmem::hash::HashMapAccess<'t, RelKey, RelEntry>, - block_map: neon_shmem::hash::HashMapAccess<'t, BlockKey, BlockEntry>, +pub struct IntegratedCacheWriteAccess { + relsize_cache: neon_shmem::hash::HashMapAccess, + block_map: Arc>, global_lw_lsn: AtomicU64, @@ -71,12 +72,12 @@ pub struct IntegratedCacheWriteAccess<'t> { } /// Represents read-only access to the integrated cache. Backend processes have this. -pub struct IntegratedCacheReadAccess<'t> { - relsize_cache: neon_shmem::hash::HashMapAccess<'t, RelKey, RelEntry>, - block_map: neon_shmem::hash::HashMapAccess<'t, BlockKey, BlockEntry>, +pub struct IntegratedCacheReadAccess { + relsize_cache: neon_shmem::hash::HashMapAccess, + block_map: neon_shmem::hash::HashMapAccess, } -impl<'t> IntegratedCacheInitStruct<'t> { +impl IntegratedCacheInitStruct { /// Return the desired size in bytes of the fixed-size shared memory area to reserve for the /// integrated cache. pub fn shmem_size() -> usize { @@ -88,10 +89,10 @@ impl<'t> IntegratedCacheInitStruct<'t> { /// Initialize the shared memory segment. This runs once in postmaster. Returns a struct which /// will be inherited by all processes through fork. pub fn shmem_init( - shmem_area: &'t mut [MaybeUninit], + shmem_area: &'static mut [MaybeUninit], initial_file_cache_size: u64, max_file_cache_size: u64, - ) -> IntegratedCacheInitStruct<'t> { + ) -> IntegratedCacheInitStruct { // Initialize the relsize cache in the fixed-size area let relsize_cache_handle = neon_shmem::hash::HashMapInit::with_fixed(RELSIZE_CACHE_SIZE, shmem_area); @@ -115,14 +116,14 @@ impl<'t> IntegratedCacheInitStruct<'t> { self, lsn: Lsn, file_cache: Option, - ) -> IntegratedCacheWriteAccess<'t> { + ) -> IntegratedCacheWriteAccess { let IntegratedCacheInitStruct { relsize_cache_handle, block_map_handle, } = self; IntegratedCacheWriteAccess { relsize_cache: relsize_cache_handle.attach_writer(), - block_map: block_map_handle.attach_writer(), + block_map: block_map_handle.attach_writer().into(), global_lw_lsn: AtomicU64::new(lsn.0), file_cache, clock_hand: std::sync::Mutex::new(0), @@ -164,7 +165,7 @@ impl<'t> IntegratedCacheInitStruct<'t> { } /// Initialize access to the integrated cache for a backend process - pub fn backend_init(self) -> IntegratedCacheReadAccess<'t> { + pub fn backend_init(self) -> IntegratedCacheReadAccess { let IntegratedCacheInitStruct { relsize_cache_handle, block_map_handle, @@ -253,8 +254,8 @@ pub enum CacheResult { NotFound(Lsn), } -impl<'t> IntegratedCacheWriteAccess<'t> { - pub fn get_rel_size(&'t self, rel: &RelTag) -> CacheResult { +impl IntegratedCacheWriteAccess { + pub fn get_rel_size(&self, rel: &RelTag) -> CacheResult { if let Some(nblocks) = get_rel_size(&self.relsize_cache, rel) { CacheResult::Found(nblocks) } else { @@ -264,7 +265,7 @@ impl<'t> IntegratedCacheWriteAccess<'t> { } pub async fn get_page( - &'t self, + &self, rel: &RelTag, block_number: u32, dst: impl uring_common::buf::IoBufMut + Send + Sync, @@ -299,7 +300,7 @@ impl<'t> IntegratedCacheWriteAccess<'t> { } pub async fn page_is_cached( - &'t self, + &self, rel: &RelTag, block_number: u32, ) -> Result, std::io::Error> { @@ -321,10 +322,10 @@ impl<'t> IntegratedCacheWriteAccess<'t> { } } - /// Does the relation exists? CacheResult::NotFound means that the cache doesn't contain that - /// information, i.e. we don't know if the relation exists or not. - pub fn get_rel_exists(&'t self, rel: &RelTag) -> CacheResult { - // we don't currently cache negative entries, so if the relation is in the cache, it exists + /// Does the relation exists? CacheResult::NotFound means that the cache doesn contain that + /// information, i.e. we don know if the relation exists or not. + pub fn get_rel_exists(& self, rel: &RelTag) -> CacheResult { + // we don currently cache negative entries, so if the relation is in the cache, it exists if let Some(_rel_entry) = self.relsize_cache.get(&RelKey::from(rel)) { CacheResult::Found(true) } else { @@ -333,7 +334,7 @@ impl<'t> IntegratedCacheWriteAccess<'t> { } } - pub fn get_db_size(&'t self, _db_oid: u32) -> CacheResult { + pub fn get_db_size(& self, _db_oid: u32) -> CacheResult { // TODO: it would be nice to cache database sizes too. Getting the database size // is not a very common operation, but when you do it, it's often interactive, with // e.g. psql \l+ command, so the user will feel the latency. @@ -343,7 +344,7 @@ impl<'t> IntegratedCacheWriteAccess<'t> { CacheResult::NotFound(lsn) } - pub fn remember_rel_size(&'t self, rel: &RelTag, nblocks: u32, lsn: Lsn) { + pub fn remember_rel_size(& self, rel: &RelTag, nblocks: u32, lsn: Lsn) { match self.relsize_cache.entry(RelKey::from(rel)) { Entry::Vacant(e) => { tracing::info!("inserting rel entry for {rel:?}, {nblocks} blocks"); @@ -365,7 +366,7 @@ impl<'t> IntegratedCacheWriteAccess<'t> { /// Remember the given page contents in the cache. pub async fn remember_page( - &'t self, + & self, rel: &RelTag, block_number: u32, src: impl uring_common::buf::IoBuf + Send + Sync, @@ -387,7 +388,7 @@ impl<'t> IntegratedCacheWriteAccess<'t> { let mut found_existing = false; // NOTE(quantumish): honoring original semantics here (used to be update_with_fn) - // but I don't see any reason why this has to take a write lock. + // but I don see any reason why this has to take a write lock. if let Entry::Occupied(e) = self.block_map.entry(key.clone()) { let block_entry = e.get(); found_existing = true; @@ -468,7 +469,7 @@ impl<'t> IntegratedCacheWriteAccess<'t> { } else { // !is_write // - // We can assume that it doesn't already exist, because the + // We can assume that it doesn already exist, because the // caller is assumed to have already checked it, and holds // the io-in-progress lock. (The BlockEntry might exist, but no cache block) @@ -522,7 +523,7 @@ impl<'t> IntegratedCacheWriteAccess<'t> { } /// Forget information about given relation in the cache. (For DROP TABLE and such) - pub fn forget_rel(&'t self, rel: &RelTag, _nblocks: Option, flush_lsn: Lsn) { + pub fn forget_rel(& self, rel: &RelTag, _nblocks: Option, flush_lsn: Lsn) { tracing::info!("forgetting rel entry for {rel:?}"); self.relsize_cache.remove(&RelKey::from(rel)); @@ -631,15 +632,16 @@ impl<'t> IntegratedCacheWriteAccess<'t> { } } } - // Give up if we didn't find anything + // Give up if we didn find anything None } /// Resize the local file cache. - pub fn resize_file_cache(&self, num_blocks: u32) { + pub fn resize_file_cache(&'static self, num_blocks: u32) { let old_num_blocks = self.block_map.get_num_buckets() as u32; if old_num_blocks < num_blocks { + tracing::error!("growing to {num_blocks}!"); if let Err(err) = self.block_map.grow(num_blocks) { tracing::warn!( "could not grow file cache to {} blocks (old size {}): {}", @@ -649,58 +651,62 @@ impl<'t> IntegratedCacheWriteAccess<'t> { ); } } else { - // Don't hold lock for longer than necessary. - { - let mut clock_hand = self.clock_hand.lock().unwrap(); - - // Make sure the clock hand resets properly. - // TODO(quantumish): confirm this is expected behavior? - if *clock_hand > num_blocks as usize { - *clock_hand = num_blocks as usize - 1; + let page_evictions = &self.page_evictions_counter; + let global_lw_lsn = &self.global_lw_lsn; + let block_map = self.block_map.clone(); + tokio::task::spawn_blocking(move || { + block_map.begin_shrink(num_blocks); + // Evict everything in to-be-shrinked space + // TODO(quantumish): consider moving things ahead of clock hand? + for i in num_blocks..old_num_blocks { + let Some(entry) = block_map.entry_at_bucket(i as usize) else { + continue; + }; + let old = entry.get(); + if old.pinned.load(Ordering::Relaxed) != 0 { + tracing::warn!( + "could not shrink file cache to {} blocks (old size {}): entry {} is pinned", + num_blocks, + old_num_blocks, + i + ); + continue; + } + _ = global_lw_lsn.fetch_max(old.lw_lsn.load().0, Ordering::Relaxed); + old.cache_block.store(INVALID_CACHE_BLOCK, Ordering::Relaxed); + entry.remove(); + // TODO(quantumish): is this expected behavior? + page_evictions.inc(); } - self.block_map.begin_shrink(num_blocks); - } - // Evict everything in to-be-shrinked space - // TODO(quantumish): consider moving ahead of clock hand? - for i in num_blocks..old_num_blocks { - let Some(entry) = self.block_map.entry_at_bucket(i as usize) else { - continue; - }; - let old = entry.get(); - if old.pinned.load(Ordering::Relaxed) != 0 { - tracing::warn!( - "could not shrink file cache to {} blocks (old size {}): entry {} is pinned", - num_blocks, - old_num_blocks, - i - ); - return; - } - _ = self.global_lw_lsn.fetch_max(old.lw_lsn.load().0, Ordering::Relaxed); - old.cache_block.store(INVALID_CACHE_BLOCK, Ordering::Relaxed); - entry.remove(); - // TODO(quantumish): is this expected behavior? - self.page_evictions_counter.inc(); - } - - if let Err(err) = self.block_map.finish_shrink() { - tracing::warn!( - "could not shrink file cache to {} blocks (old size {}): {}", - num_blocks, - old_num_blocks, - err - ); - } + // if let Err(err) = block_map.finish_shrink() { + // tracing::warn!( + // "could not shrink file cache to {} blocks (old size {}): {}", + // num_blocks, + // old_num_blocks, + // err + // ); + // } + // Don hold lock for longer than necessary. + // { + // let mut clock_hand = self.clock_hand.lock().unwrap(); + + // // Make sure the clock hand resets properly. + // // TODO(quantumish): confirm this is expected behavior? + // if *clock_hand > num_blocks as usize { + // *clock_hand = num_blocks as usize - 1; + // } + // } + }); } - } + } pub fn dump_map(&self, _dst: &mut dyn std::io::Write) { //FIXME self.cache_map.start_read().dump(dst); } } -impl metrics::core::Collector for IntegratedCacheWriteAccess<'_> { +impl metrics::core::Collector for IntegratedCacheWriteAccess { fn desc(&self) -> Vec<&metrics::core::Desc> { let mut descs = Vec::new(); descs.append(&mut self.page_evictions_counter.desc()); @@ -769,12 +775,12 @@ pub enum GetBucketResult { /// /// This allows backends to read pages from the cache directly, on their own, without making a /// request to the communicator process. -impl<'t> IntegratedCacheReadAccess<'t> { - pub fn get_rel_size(&'t self, rel: &RelTag) -> Option { +impl IntegratedCacheReadAccess { + pub fn get_rel_size(& self, rel: &RelTag) -> Option { get_rel_size(&self.relsize_cache, rel) } - pub fn start_read_op(&'t self) -> BackendCacheReadOp<'t> { + pub fn start_read_op(& self) -> BackendCacheReadOp { BackendCacheReadOp { read_guards: Vec::new(), map_access: self, @@ -782,7 +788,7 @@ impl<'t> IntegratedCacheReadAccess<'t> { } /// Check if the given page is present in the cache - pub fn cache_contains_page(&'t self, rel: &RelTag, block_number: u32) -> bool { + pub fn cache_contains_page(& self, rel: &RelTag, block_number: u32) -> bool { self.block_map .get(&BlockKey::from((rel, block_number))) .is_some() @@ -806,7 +812,7 @@ impl<'t> IntegratedCacheReadAccess<'t> { pub struct BackendCacheReadOp<'t> { read_guards: Vec, - map_access: &'t IntegratedCacheReadAccess<'t>, + map_access: &'t IntegratedCacheReadAccess, } impl<'e> BackendCacheReadOp<'e> { diff --git a/pgxn/neon/communicator/src/worker_process/main_loop.rs b/pgxn/neon/communicator/src/worker_process/main_loop.rs index 0b2f9da366..597d08194b 100644 --- a/pgxn/neon/communicator/src/worker_process/main_loop.rs +++ b/pgxn/neon/communicator/src/worker_process/main_loop.rs @@ -46,7 +46,7 @@ pub struct CommunicatorWorkerProcessStruct<'a> { in_progress_table: RequestInProgressTable, /// Local File Cache, relation size tracking, last-written LSN tracking - pub(crate) cache: IntegratedCacheWriteAccess<'a>, + pub(crate) cache: IntegratedCacheWriteAccess, /*** Static configuration ***/ /// Stripe size doesn't change after startup. (The shard map is not stored here, it's passed