fix bugs

storage-node/src/cache/data_store_cache/memdisk/mod.rs

@@ -146,6 +146,7 @@ impl<R: DataStoreReplacer<MemDiskStoreReplacerKey, MemDiskStoreReplacerValue>> D
                 let data_store_key = replacer_value.as_value();
                 match mem_store.read().await.read_data(data_store_key) {
                     Ok(Some(rx)) => {
+                        // TODO(lanlou): actually we should unpin after all the data are consumed...
                         if !mem_replacer.unpin(&remote_location) {
                             warn!(
                                 "Failed to unpin the key ({}) in memory replacer.",
@@ -388,8 +389,11 @@ impl<R: DataStoreReplacer<MemDiskStoreReplacerKey, MemDiskStoreReplacerValue>> D
         let ((status, size), notify) = status_of_keys.get_mut(&remote_location).unwrap();
         *status = Status::Completed;
         *size = data_size;
-        for _ in 0..*notify.1.lock().await {
-            self.disk_replacer.lock().await.pin(&remote_location);
+        {
+            let mut disk_replacer = self.disk_replacer.lock().await;
+            for _ in 0..*notify.1.lock().await {
+                disk_replacer.pin(&remote_location);
+            }
         }
         notify.0.notify_waiters();
         Ok(data_size)

storage-node/src/cache/replacer/lru.rs

@@ -59,10 +59,6 @@ impl<K: ReplacerKey, V: ReplacerValue> LruReplacer<K, V> {
         while (self.size + value.size()) > self.max_capacity {
             // Previous code guarantees that there is at least 1 element in the cache_map, so we
             // can safely unwrap here.
-            println!(
-                "-------- To Evicting Key: {:?} --------",
-                self.cache_map.front().unwrap().1 .1
-            );
             if self.cache_map.front().unwrap().1 .1 > 0 {
                 // TODO(lanlou): Actually we should look next to evict, but for simplicity, just
                 // return None here, it is temporarily okay since we will not pin an element for

storage-node/src/cache/replacer/lru_k.rs

@@ -134,9 +134,8 @@ impl<K: ReplacerKey, V: ReplacerValue> LruKReplacer<K, V> {
         let mut evicted_keys = Vec::new();
         while (self.size + updated_size) > self.max_capacity {
             let key_to_evict = self.evict(&key);
-            if key_to_evict.is_none() {
-                return None;
-            }
+            // If key_to_evict is none, return none
+            key_to_evict.as_ref()?;
             if let Some(evicted_key) = key_to_evict {
                 evicted_keys.push(evicted_key);
             }

storage-node/src/disk/disk_manager.rs

@@ -120,6 +120,7 @@ impl DiskManager {
         if let Some(mut stream) = stream {
             let bytes_cur = stream.next().await;
             if bytes_cur.is_none() {
+                file.flush().await?;
                 return Ok(bytes_written);
             }
             let mut bytes_cur = bytes_cur.unwrap()?;