Merge pull request #53 from johnramsden/john/locking

Reduce lock contention for promo

Author: johnramsden

oxcache/src/cache/bucket.rs

@@ -11,6 +11,21 @@ pub struct Chunk {
     pub size: Byte,
 }
 
+// // Only hash and compare based on uuid - offset/size are just read parameters
+// impl PartialEq for Chunk {
+//     fn eq(&self, other: &Self) -> bool {
+//         self.uuid == other.uuid
+//     }
+// }
+//
+// impl Eq for Chunk {}
+//
+// impl std::hash::Hash for Chunk {
+//     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+//         self.uuid.hash(state);
+//     }
+// }
+
 #[derive(Debug, PartialEq, Clone, Eq, Hash)]
 pub struct ChunkLocation {
     pub zone: Zone,

oxcache/src/device.rs

@@ -707,6 +707,17 @@ impl Device for Zoned {
                 let mut zones = zone_mtx.lock().unwrap();
                 zones.reset_zones(&zones_to_evict, &*self)?;
 
+                // Reset atomic counters for evicted zones
+                let mut policy = eviction_policy.lock().unwrap();
+                if let EvictionPolicyWrapper::Promotional(p) = &*policy {
+                    for &zone in &zones_to_evict {
+                        if (zone as usize) < p.zone_chunk_counts.len() {
+                            p.zone_chunk_counts[zone as usize].store(0, Ordering::Relaxed);
+                        }
+                    }
+                }
+                drop(policy);
+
                 Ok(())
             }
         }
@@ -1051,6 +1062,18 @@ impl Device for BlockInterface {
                 let state_mtx = Arc::clone(&self.state);
                 let mut state = state_mtx.lock().unwrap();
                 state.active_zones.reset_zones(&locations, &*self)?;
+
+                // Reset atomic counters for evicted zones
+                let mut policy = eviction_policy.lock().unwrap();
+                if let EvictionPolicyWrapper::Promotional(p) = &*policy {
+                    for &zone in &locations {
+                        if (zone as usize) < p.zone_chunk_counts.len() {
+                            p.zone_chunk_counts[zone as usize].store(0, Ordering::Relaxed);
+                        }
+                    }
+                }
+                drop(policy);
+
                 Ok(())
             }
         }

oxcache/src/eviction.rs

@@ -8,7 +8,7 @@ use lru_mem::{LruCache, MemSize};
 use nvme::types::{Chunk, Zone};
 use std::sync::{
     Arc, Mutex,
-    atomic::{AtomicBool, Ordering},
+    atomic::{AtomicBool, AtomicU64, Ordering},
 };
 use std::thread::{self, JoinHandle};
 use std::time::Duration;
@@ -119,7 +119,8 @@ pub struct PromotionalEvictionPolicy {
     high_water: Zone,
     low_water: Zone,
     nr_zones: Zone,
-    nr_chunks_per_zone: Chunk,
+    pub nr_chunks_per_zone: Chunk,
+    pub zone_chunk_counts: Arc<Vec<AtomicU64>>,
     lru: LruCache<Zone, ()>,
     #[cfg(feature = "eviction-metrics")]
     pub metrics: Option<Arc<crate::eviction_metrics::EvictionMetrics>>,
@@ -133,11 +134,17 @@ impl PromotionalEvictionPolicy {
         nr_chunks_per_zone: Chunk,
     ) -> Self {
         let lru = LruCache::new(usize::MAX); // Effectively unbounded
+        let zone_chunk_counts = Arc::new(
+            (0..nr_zones)
+                .map(|_| AtomicU64::new(0))
+                .collect()
+        );
         Self {
             high_water,
             low_water,
             nr_zones,
             nr_chunks_per_zone,
+            zone_chunk_counts,
             lru,
             #[cfg(feature = "eviction-metrics")]
             metrics: None,
@@ -157,12 +164,9 @@ impl EvictionPolicy for PromotionalEvictionPolicy {
             metrics.record_write(&chunk);
         }
 
-        // assert!(!self.lru.contains(&chunk.zone)); // We cannot assert this because we allow out of order writes
-
-        // We only want to put it in the LRU once the zone is full
-        if chunk.index == self.nr_chunks_per_zone - 1 {
-            self.lru.insert(chunk.zone, ()).ok();
-        }
+        // This is now only called when zone is full (atomic check done outside lock)
+        // Just insert the zone into the LRU
+        self.lru.insert(chunk.zone, ()).ok();
     }
 
     fn read_update(&mut self, chunk: ChunkLocation) {
@@ -171,9 +175,8 @@ impl EvictionPolicy for PromotionalEvictionPolicy {
             metrics.record_read(&chunk);
         }
 
-        // We only want to put it in the LRU once the zone is full
-        // If it has filled before we want to update every time "promoting" it
-        // Following this, only zones that have filled prior are updated
+        // This is now only called when zone is full (atomic check done outside lock)
+        // Promote the zone in the LRU if it's already there
         if self.lru.contains(&chunk.zone) {
             self.lru.insert(chunk.zone, ()).ok();
         }
@@ -558,7 +561,10 @@ mod tests {
 
         // zone=[_,_,_,_], lru=()
         let mut order: VecDeque<Zone> = VecDeque::new();
-        policy.write_update(ChunkLocation::new(3, 0));
+        let count = policy.zone_chunk_counts[3].fetch_add(1, Ordering::Relaxed);
+        if count + 1 == policy.nr_chunks_per_zone {
+            policy.write_update(ChunkLocation::new(3, 0));
+        }
         compare_order(&mut policy.lru, &order);
         let et = policy.get_evict_targets(false);
         let expect_none: Vec<Zone> = vec![];
@@ -569,7 +575,10 @@ mod tests {
         );
 
         // zone=[_,_,_,_], lru=()
-        policy.write_update(ChunkLocation::new(3, 1));
+        let count = policy.zone_chunk_counts[3].fetch_add(1, Ordering::Relaxed);
+        if count + 1 == policy.nr_chunks_per_zone {
+            policy.write_update(ChunkLocation::new(3, 1));
+        }
         // zone=[_,_,_,3], lru=(3)
         order.push_back(3);
         compare_order(&mut policy.lru, &order);
@@ -580,12 +589,18 @@ mod tests {
             expect_none, et
         );
 
-        policy.write_update(ChunkLocation::new(1, 0));
+        let count = policy.zone_chunk_counts[1].fetch_add(1, Ordering::Relaxed);
+        if count + 1 == policy.nr_chunks_per_zone {
+            policy.write_update(ChunkLocation::new(1, 0));
+        }
         // There should be no change
         // zone=[_,_,_,3], lru=(3)
         compare_order(&mut policy.lru, &order);
 
-        policy.write_update(ChunkLocation::new(1, 1));
+        let count = policy.zone_chunk_counts[1].fetch_add(1, Ordering::Relaxed);
+        if count + 1 == policy.nr_chunks_per_zone {
+            policy.write_update(ChunkLocation::new(1, 1));
+        }
         // zone=[_,1,_,3], lru=(3, 1)
         order.push_front(1);
         compare_order(&mut policy.lru, &order);
@@ -595,15 +610,22 @@ mod tests {
             "Expected = {:?}, but got {:?}",
             expect_none, et
         );
-
-        policy.write_update(ChunkLocation::new(2, 0));
-        policy.write_update(ChunkLocation::new(2, 1));
+        let count = policy.zone_chunk_counts[2].fetch_add(1, Ordering::Relaxed);
+        if count + 1 == policy.nr_chunks_per_zone {
+            policy.write_update(ChunkLocation::new(2, 0));
+        }
+        let count = policy.zone_chunk_counts[2].fetch_add(1, Ordering::Relaxed);
+        if count + 1 == policy.nr_chunks_per_zone {
+            policy.write_update(ChunkLocation::new(2, 1));
+        }
         order.push_front(2);
         // zone=[_,1,2,3], lru=(3, 1, 2)
         compare_order(&mut policy.lru, &order);
 
         // Should update in place, and adjust order
-        policy.read_update(ChunkLocation::new(3, 1));
+        if policy.zone_chunk_counts[3].load(Ordering::Relaxed) >= policy.nr_chunks_per_zone {
+            policy.read_update(ChunkLocation::new(3, 1));
+        }
         let c = order.pop_back().unwrap();
         order.push_front(c);
         // zone=[_,1,2,3], lru=(1, 2, 3)

oxcache/src/readerpool.rs

@@ -3,6 +3,7 @@ use crate::{cache, device};
 use bytes::Bytes;
 use flume::{Receiver, Sender, unbounded};
 use std::sync::{Arc, Mutex};
+use std::sync::atomic::Ordering;
 use std::thread::{self, JoinHandle};
 use crate::metrics::{MetricType, METRICS};
 use crate::cache::bucket::PinGuard;
@@ -53,8 +54,33 @@ impl Reader {
             METRICS.update_metric_histogram_latency("device_read_latency_ms", start.elapsed(), MetricType::MsLatency);
             if result.is_ok() {
                 let mtx = Arc::clone(&self.eviction_policy);
-                let mut policy = mtx.lock().unwrap();
-                policy.read_update(msg.location);
+                let policy = mtx.lock().unwrap();
+
+                match &*policy {
+                    EvictionPolicyWrapper::Promotional(p) => {
+                        // Clone Arc references before dropping lock
+                        let zone_idx = msg.location.zone as usize;
+                        let nr_chunks = p.nr_chunks_per_zone;
+                        let counters = Arc::clone(&p.zone_chunk_counts);
+                        drop(policy);
+
+                        // Check atomically if zone is full (outside lock)
+                        if zone_idx < counters.len() {
+                            let count = counters[zone_idx].load(Ordering::Relaxed);
+
+                            // Only acquire lock if zone is full
+                            if count >= nr_chunks {
+                                let mut policy = mtx.lock().unwrap();
+                                policy.read_update(msg.location);
+                            }
+                        }
+                    }
+                    EvictionPolicyWrapper::Chunk(_) => {
+                        drop(policy);
+                        let mut policy = mtx.lock().unwrap();
+                        policy.read_update(msg.location);
+                    }
+                }
             }
             let resp = ReadResponse { data: result };
             let snd = msg.responder.send(resp);

oxcache/src/writerpool.rs

@@ -123,11 +123,35 @@ impl Writer {
         let start = std::time::Instant::now();
 
         let result = self.device.append(msg.data).inspect(|loc| {
-            let mtx = Arc::clone(&self.eviction);
-
             if msg.update_lru {
-                let mut policy = mtx.lock().unwrap();
-                policy.write_update(loc.clone());
+                let mtx = Arc::clone(&self.eviction);
+                let policy = mtx.lock().unwrap();
+
+                match &*policy {
+                    EvictionPolicyWrapper::Promotional(p) => {
+                        // Clone Arc references before dropping lock
+                        let zone_idx = loc.zone as usize;
+                        let nr_chunks = p.nr_chunks_per_zone;
+                        let counters = Arc::clone(&p.zone_chunk_counts);
+                        drop(policy);
+
+                        // Do atomic increment outside lock
+                        if zone_idx < counters.len() {
+                            let count = counters[zone_idx].fetch_add(1, Ordering::Relaxed);
+
+                            // Only re-acquire lock when zone becomes full
+                            if count + 1 == nr_chunks {
+                                let mut policy = mtx.lock().unwrap();
+                                policy.write_update(loc.clone());
+                            }
+                        }
+                    }
+                    EvictionPolicyWrapper::Chunk(_) => {
+                        drop(policy);
+                        let mut policy = mtx.lock().unwrap();
+                        policy.write_update(loc.clone());
+                    }
+                }
             }
         });
         METRICS.update_metric_histogram_latency("device_write_latency_ms", start.elapsed(), MetricType::MsLatency);
@@ -153,9 +177,33 @@ impl Writer {
 
             if let Ok(ref loc) = result {
                 let mtx = Arc::clone(&self.eviction);
-                let mut policy = mtx.lock().unwrap();
-                policy.write_update(loc.clone());
-                drop(policy);
+                let policy = mtx.lock().unwrap();
+
+                match &*policy {
+                    EvictionPolicyWrapper::Promotional(p) => {
+                        // Clone Arc references before dropping lock
+                        let zone_idx = loc.zone as usize;
+                        let nr_chunks = p.nr_chunks_per_zone;
+                        let counters = Arc::clone(&p.zone_chunk_counts);
+                        drop(policy);
+
+                        // Do atomic increment outside lock
+                        if zone_idx < counters.len() {
+                            let count = counters[zone_idx].fetch_add(1, Ordering::Relaxed);
+
+                            // Only re-acquire lock when zone becomes full
+                            if count + 1 == nr_chunks {
+                                let mut policy = mtx.lock().unwrap();
+                                policy.write_update(loc.clone());
+                            }
+                        }
+                    }
+                    EvictionPolicyWrapper::Chunk(_) => {
+                        drop(policy);
+                        let mut policy = mtx.lock().unwrap();
+                        policy.write_update(loc.clone());
+                    }
+                }
             } else {
                 tracing::error!("Failed to append: {:?}", result);
             }