refactor: avoid clones for `policy::Subscriber::set`

Policy subscribers basically never store `V` (the cached value), so we
should not clone that one unconditionally. But even `K` (the cache key)
is not stored in all cases. So let's pass a reference for both and let
the policies decide if they wanna clone the data or not.

cache_system/src/backend/policy/lru.rs

@@ -635,12 +635,12 @@ where
 
     fn set(
         &mut self,
-        k: Self::K,
-        v: Self::V,
+        k: &Self::K,
+        v: &Self::V,
         now: Time,
     ) -> Vec<ChangeRequest<'static, Self::K, Self::V>> {
         // determine all attributes before getting any locks
-        let consumption = self.resource_estimator.consumption(&k, &v);
+        let consumption = self.resource_estimator.consumption(k, v);
 
         // "last used" time for new entry
         // Note: this might be updated if the entry already exists
@@ -651,13 +651,13 @@ where
 
         // check for oversized entries
         if consumption > pool.limit.v {
-            return vec![ChangeRequest::remove(k)];
+            return vec![ChangeRequest::remove(k.clone())];
         }
 
         // maybe clean from pool
         {
             let mut inner = self.inner.lock();
-            if let Some((consumption, last_used_previously)) = inner.last_used.remove(&k) {
+            if let Some((consumption, last_used_previously)) = inner.last_used.remove(k) {
                 pool.remove(consumption);
                 inner.metric_count.dec(1);
                 inner.metric_usage.dec(consumption.into());
@@ -672,7 +672,7 @@ where
 
         // add new entry to inner backend AFTER adding it to the pool, so we are never overcommitting resources.
         let mut inner = self.inner.lock();
-        inner.last_used.insert(k, consumption, last_used);
+        inner.last_used.insert(k.clone(), consumption, last_used);
         inner.metric_count.inc(1);
         inner.metric_usage.inc(consumption.into());
 

cache_system/src/backend/policy/mod.rs

@@ -139,7 +139,7 @@ macro_rules! lock_inner {
 ///     type K = &'static str;
 ///     type V = u64;
 ///
-///     fn set(&mut self, k: &'static str, v: u64, _now: Time) -> Vec<CR> {
+///     fn set(&mut self, k: &&'static str, v: &u64, _now: Time) -> Vec<CR> {
 ///       // When new key `k` is set to value `v` if `v` is odd,
 ///       // request a change to set `k` to `v+1`
 ///         if v % 2 == 1 {
@@ -422,7 +422,7 @@ fn perform_changes<K, V>(
             for subscriber in &mut inner.subscribers {
                 let requests = match &record {
                     Record::Get { k } => subscriber.get(k, now),
-                    Record::Set { k, v } => subscriber.set(k.clone(), v.clone(), now),
+                    Record::Set { k, v } => subscriber.set(k, v, now),
                     Record::Remove { k } => subscriber.remove(k, now),
                 };
 
@@ -459,8 +459,8 @@ pub trait Subscriber: Debug + Send + 'static {
     /// more consistent and performant.
     fn set(
         &mut self,
-        _k: Self::K,
-        _v: Self::V,
+        _k: &Self::K,
+        _v: &Self::V,
         _now: Time,
     ) -> Vec<ChangeRequest<'static, Self::K, Self::V>> {
         // do nothing by default
@@ -1852,13 +1852,16 @@ mod tests {
 
         fn set(
             &mut self,
-            k: Self::K,
-            v: Self::V,
+            k: &Self::K,
+            v: &Self::V,
             _now: Time,
         ) -> Vec<ChangeRequest<'static, String, usize>> {
             let step = self.steps.pop_front().expect("step left for set operation");
 
-            let expected_condition = TestBackendInteraction::Set { k, v };
+            let expected_condition = TestBackendInteraction::Set {
+                k: k.clone(),
+                v: *v,
+            };
             assert_eq!(
                 step.condition, expected_condition,
                 "Condition mismatch\n\nActual:\n{:#?}\n\nExpected:\n{:#?}",

cache_system/src/backend/policy/refresh.rs

@@ -330,11 +330,11 @@ where
 
     fn set(
         &mut self,
-        k: Self::K,
-        v: Self::V,
+        k: &Self::K,
+        v: &Self::V,
         now: Time,
     ) -> Vec<ChangeRequest<'static, Self::K, Self::V>> {
-        let d = self.refresh_duration_provider.refresh_in(&k, &v);
+        let d = self.refresh_duration_provider.refresh_in(k, v);
 
         let mut timings = self.timings.lock();
 
@@ -345,10 +345,10 @@ where
                 running_refresh: None,
             };
 
-            timings.insert(k, state);
+            timings.insert(k.clone(), state);
         } else {
             // need to remove potentially existing entry that had some refresh set
-            timings.remove(&k);
+            timings.remove(k);
 
             // the removal drops the RefreshState which triggers a cancelation for any potentially running
             // refresh operation

cache_system/src/backend/policy/ttl.rs

@@ -184,13 +184,13 @@ where
 
     fn set(
         &mut self,
-        k: Self::K,
-        v: Self::V,
+        k: &Self::K,
+        v: &Self::V,
         now: Time,
     ) -> Vec<ChangeRequest<'static, Self::K, Self::V>> {
         let mut requests = self.evict_expired(now);
 
-        if let Some(ttl) = self.ttl_provider.expires_in(&k, &v) {
+        if let Some(ttl) = self.ttl_provider.expires_in(k, v) {
             if ttl.is_zero() {
                 requests.push(ChangeRequest::remove(k.clone()));
             }
@@ -201,12 +201,12 @@ where
                 }
                 None => {
                     // Still need to ensure that any current expiration is disabled
-                    self.expiration.remove(&k);
+                    self.expiration.remove(k);
                 }
             }
         } else {
             // Still need to ensure that any current expiration is disabled
-            self.expiration.remove(&k);
+            self.expiration.remove(k);
         };
 
         requests