Avoid invalid Send trait in evicting_map.

The evict_items was async which caused an invalid Send trait to be required
on functions that use it.  This was detected by wrapping the asyncio Mutex
with a std::sync::Mutex such that the compiler could detect the invalid usage.

There was a valid usage by manually calling drop(state) in get and size_for_key,
however, a bug in the compiler (rust-lang/rust#57478)
meant that refactoring this allows the compiler to better understand the lock.

util/BUILD

@@ -82,6 +82,7 @@ rust_library(
         "//third_party:fast_async_mutex",
         "//third_party:lru",
         "//third_party:serde",
+        "//third_party:tokio",
         "//util:common",
     ],
     proc_macro_deps = ["//third_party:async_trait"],

util/evicting_map.rs

@@ -115,7 +115,7 @@ pub struct EvictingMap<T: LenEntry + Debug, I: InstantWrapper> {
 
 impl<T, I> EvictingMap<T, I>
 where
-    T: LenEntry + Debug + Clone + Send + Sync,
+    T: LenEntry + Debug + Clone + Send + Sync + 'static,
     I: InstantWrapper,
 {
     pub fn new(config: &EvictionPolicy, anchor_time: I) -> Self {
@@ -168,7 +168,7 @@ where
             );
         }
         // Just in case we allow for some cleanup (eg: old items).
-        self.evict_items(state.deref_mut()).await;
+        self.evict_items(state.deref_mut());
     }
 
     fn should_evict(&self, lru_len: usize, peek_entry: &EvictionItem<T>, sum_store_size: u64, evicting: bool) -> bool {
@@ -184,13 +184,10 @@ where
 
         let is_over_count = self.max_count != 0 && (lru_len as u64) > self.max_count;
 
-        if is_over_size || old_item_exists || is_over_count {
-            return true;
-        }
-        false
+        is_over_size || old_item_exists || is_over_count
     }
 
-    async fn evict_items(&self, state: &mut State<T>) {
+    fn evict_items(&self, state: &mut State<T>) {
         let mut peek_entry = if let Some((_, entry)) = state.lru.peek_lru() {
             entry
         } else {
@@ -201,8 +198,13 @@ where
             evicting = true;
             let (key, eviction_item) = state.lru.pop_lru().expect("Tried to peek() then pop() but failed");
             state.sum_store_size -= eviction_item.data.len() as u64;
-            eviction_item.data.unref().await;
-            log::info!("\x1b[0;31mevicting map\x1b[0m: Evicting {:?}", key);
+
+            // Perform the eviction in a separate greenlet because this function
+            // cannot be async otherwise the lock Send trait does not hold.
+            tokio::spawn(async move {
+                eviction_item.data.unref().await;
+                log::info!("\x1b[0;31mevicting map\x1b[0m: Evicting {:?}", key);
+            });
 
             peek_entry = if let Some((_, entry)) = state.lru.peek_lru() {
                 entry
@@ -212,28 +214,31 @@ where
         }
     }
 
-    pub async fn size_for_key(&self, digest: &DigestInfo) -> Option<usize> {
+    async fn get_item_for_key(&self, digest: &DigestInfo) -> Option<T> {
         let mut state = self.state.lock().await;
         if let Some(mut entry) = state.lru.get_mut(digest) {
             entry.seconds_since_anchor = self.anchor_time.elapsed().as_secs() as i32;
-            let data = entry.data.clone();
-            drop(state);
-            data.touch().await;
-            return Some(data.len());
+            Some(entry.data.clone())
+        } else {
+            None
+        }
+    }
+
+    pub async fn size_for_key(&self, digest: &DigestInfo) -> Option<usize> {
+        match self.get(digest).await {
+            Some(data) => Some(data.len()),
+            None => None,
         }
-        None
     }
 
     pub async fn get(&self, digest: &DigestInfo) -> Option<T> {
-        let mut state = self.state.lock().await;
-        if let Some(mut entry) = state.lru.get_mut(digest) {
-            entry.seconds_since_anchor = self.anchor_time.elapsed().as_secs() as i32;
-            let data = entry.data.clone();
-            drop(state);
-            data.touch().await;
-            return Some(data);
+        match self.get_item_for_key(digest).await {
+            Some(data) => {
+                data.touch().await;
+                Some(data)
+            }
+            None => None,
         }
-        None
     }
 
     /// Returns the replaced item if any.
@@ -264,7 +269,7 @@ where
             None
         };
         state.sum_store_size += new_item_size;
-        self.evict_items(state.deref_mut()).await;
+        self.evict_items(state.deref_mut());
         maybe_old_item
     }