Make references all 'static

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<K: 'static, V: 'static, S = rustc_hash::FxBuildHasher> {
     shmem_handle: Option<ShmemHandle>,
-    shared_ptr: *mut HashMapShared<'a, K, V>,
+    shared_ptr: *mut HashMapShared<K, V>,
     shared_size: usize,
     hasher: S,
     num_buckets: u32,
 }
 
-impl<'a, K, V, S> Debug for HashMapInit<'a, K, V, S>
+impl<K, V, S> Debug for HashMapInit<K, V, S>
 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<K: 'static, V: 'static, S = rustc_hash::FxBuildHasher> {
     shmem_handle: Option<ShmemHandle>,
-    shared_ptr: *mut HashMapShared<'a, K, V>,
+    shared_ptr: *mut HashMapShared<K, V>,
     hasher: S,
 }
 
-unsafe impl<K: Sync, V: Sync, S> Sync for HashMapAccess<'_, K, V, S> {}
-unsafe impl<K: Send, V: Send, S> Send for HashMapAccess<'_, K, V, S> {}
+unsafe impl<K: Sync, V: Sync, S> Sync for HashMapAccess<K, V, S> {}
+unsafe impl<K: Send, V: Send, S> Send for HashMapAccess<K, V, S> {}
 
-impl<'a, K, V, S> Debug for HashMapAccess<'a, K, V, S>
+impl<K, V, S> Debug for HashMapAccess<K, V, S>
 where
     K: Debug,
     V: Debug,
@@ -102,14 +102,14 @@ where
     }
 }
 
-impl<'a, K: Clone + Hash + Eq, V, S> HashMapInit<'a, K, V, S> {
+impl<K: Clone + Hash + Eq + 'static, V: 'static, S> HashMapInit<K, V, S> {
     /// 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<T: BuildHasher>(self, hasher: T) -> HashMapInit<'a, K, V, T> {
+    pub fn with_hasher<T: BuildHasher>(self, hasher: T) -> HashMapInit<K, V, T> {
         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<K, V, S> {
         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<K, V, S> {
         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<CoreHashMap<'a, K, V>>;
+type HashMapShared<K, V> = RwLock<CoreHashMap<K, V>>;
 
-impl<'a, K, V> HashMapInit<'a, K, V, rustc_hash::FxBuildHasher>
+impl<K, V: 'static> HashMapInit<K, V, rustc_hash::FxBuildHasher>
 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<u8>]) -> Self {
+    pub fn with_fixed(num_buckets: u32, area: &'static mut [MaybeUninit<u8>]) -> Self {
         Self::new(
             num_buckets,
             None,
@@ -269,9 +269,10 @@ where
     }
 }
 
-impl<'a, K, V, S: BuildHasher> HashMapAccess<'a, K, V, S>
+impl<K, V, S: BuildHasher> HashMapAccess<K, V, S>
 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<OccupiedEntry<'a, '_, K, V>> {
+    pub fn entry_at_bucket(&self, pos: usize) -> Option<OccupiedEntry<'_, K, V>> {
         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<K, V>,
         buckets_ptr: *mut core::Bucket<K, V>,
         end_ptr: *mut u8,
         num_buckets: u32,

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<K: 'static, V: 'static> {
     /// 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<K, V>],
+    pub(crate) buckets: &'static mut [Bucket<K, V>],
     /// 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<K, V> Debug for CoreHashMap<K, V>
 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<K: Clone + Hash + Eq, V> CoreHashMap<K, V> {
     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<Bucket<K, V>>],
-        dictionary: &'a mut [MaybeUninit<u32>],
+        buckets: &'static mut [MaybeUninit<Bucket<K, V>>],
+        dictionary: &'static mut [MaybeUninit<u32>],
     ) -> Self {
         // Initialize the buckets
         for i in 0..buckets.len() {

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<K, V>>,
     /// 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<K, V> OccupiedEntry<'_, '_, K, V> {
+impl<K: 'static, V: 'static> OccupiedEntry<'_, K, V> {
     pub fn get(&self) -> &V {
         &self.map.buckets[self.bucket_pos as usize]
             .inner
@@ -104,16 +104,16 @@ impl<K, V> 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<K, V>>,
     /// 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

pgxn/neon/communicator/src/backend_interface.rs

@@ -18,7 +18,7 @@ pub struct CommunicatorBackendStruct<'t> {
 
     pending_cache_read_op: Option<BackendCacheReadOp<'t>>,
 
-    integrated_cache: &'t IntegratedCacheReadAccess<'t>,
+    integrated_cache: &'t IntegratedCacheReadAccess,
 }
 
 #[unsafe(no_mangle)]

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 {

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<RelKey, RelEntry>,
+    block_map_handle: HashMapInit<BlockKey, BlockEntry>,
 }
 
 /// 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<RelKey, RelEntry>,
+    block_map: Arc<neon_shmem::hash::HashMapAccess<BlockKey, BlockEntry>>,
 
     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<RelKey, RelEntry>,
+    block_map: neon_shmem::hash::HashMapAccess<BlockKey, BlockEntry>,
 }
 
-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<u8>],
+        shmem_area: &'static mut [MaybeUninit<u8>],
         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<FileCache>,
-    ) -> 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<V> {
     NotFound(Lsn),
 }
 
-impl<'t> IntegratedCacheWriteAccess<'t> {
-    pub fn get_rel_size(&'t self, rel: &RelTag) -> CacheResult<u32> {
+impl IntegratedCacheWriteAccess {
+    pub fn get_rel_size(&self, rel: &RelTag) -> CacheResult<u32> {
         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<CacheResult<()>, 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<bool> {
-        // 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<bool> {
+        // 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<u64> {
+    pub fn get_db_size(& self, _db_oid: u32) -> CacheResult<u64> {
         // 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<u32>, flush_lsn: Lsn) {
+    pub fn forget_rel(& self, rel: &RelTag, _nblocks: Option<u32>, 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<u32> {
+impl IntegratedCacheReadAccess {
+    pub fn get_rel_size(& self, rel: &RelTag) -> Option<u32> {
         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<DeferredUnpin>,
-    map_access: &'t IntegratedCacheReadAccess<'t>,
+    map_access: &'t IntegratedCacheReadAccess,
 }
 
 impl<'e> BackendCacheReadOp<'e> {

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