Creating a background thread for key expiration

Cargo.toml

@@ -16,4 +16,5 @@ path = "src/client/client.rs"
 [dependencies]
 rayon = "1.10"
 env_logger = "0.11.3"
-log = "0.4"
+log = "0.4"
+priority-queue = "2.0.2"

src/client/client.rs

@@ -13,7 +13,7 @@ fn main() {
     // let port = &args[2];
     // let mut stream = TcpStream::connect(format!("{ip}:{port}"))?;
 
-    let mut stream = TcpStream::connect("127.0.0.1:7878").expect("Failed to connect to server");
+    let stream = TcpStream::connect("127.0.0.1:7878").expect("Failed to connect to server");
     let stream_clone = stream.try_clone().unwrap();
     let mut reader = BufReader::new(stream);
     let mut writer = BufWriter::new(stream_clone);

src/main.rs

@@ -7,6 +7,8 @@ mod server {
     pub mod control_plane;
 
     pub mod commands;
+
+    pub mod evictor;
 }
 
 

src/server/cache.rs

@@ -1,25 +1,61 @@
+use std::cmp::Reverse;
 use std::collections::HashMap;
+use std::ops::Add;
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::{Duration, SystemTime};
+use log::info;
+use priority_queue::PriorityQueue;
 
 
 pub struct Cache {
-    hash_map: HashMap<String, String>,
+    hash_map: Arc<Mutex<HashMap<String, String>>>,
+    ttl_queue: Arc<Mutex<PriorityQueue<String, Reverse<SystemTime>>>>,
 }
 
 impl Cache {
     pub fn new() -> Cache {
-        return Cache {
-            hash_map: HashMap::new(),
+        let hash_map: Arc<Mutex<HashMap<String, String>>> = Arc::new(Mutex::new(HashMap::new()));
+        let ttl_queue: Arc<Mutex<PriorityQueue<String, Reverse<SystemTime>>>> = Arc::new(Mutex::new(PriorityQueue::new()));
+
+        let hash_map_clone = hash_map.clone();
+        let ttl_queue_clone = ttl_queue.clone();
+
+        thread::spawn(move || {
+            loop {
+                thread::sleep(Duration::from_secs(2));
+
+                let cur_time = SystemTime::now();
+                while let Some((key, expiration_time)) = ttl_queue_clone.lock().unwrap().peek() {
+                    if expiration_time.0 >= cur_time {
+                        info!("It's not yet time to expire {key}");
+                        break;
+                    }
+                    info!("{key} expired, removing");
+                    hash_map_clone.lock().unwrap().remove(key);
+                    ttl_queue_clone.lock().unwrap().pop();
+                }
+            }
+        });
+
+
+        let cache = Cache {
+            hash_map,
+            ttl_queue,
         };
+        return cache;
     }
     pub fn put(&mut self, key: &String, value: &String) {
-        self.hash_map.insert(key.to_string(), value.to_string());
+        self.hash_map.lock().unwrap().insert(key.to_string(), value.to_string());
+        let expiration_time = SystemTime::now().add(Duration::from_secs(15));
+        self.ttl_queue.lock().unwrap().push(key.to_string(), Reverse(expiration_time));
     }
 
-    pub fn get(&self, key: &String) -> Option<&String> {
-        return self.hash_map.get(key);
+    pub fn get(&self, key: &String) -> Option<String> {
+        return self.hash_map.lock().unwrap().get(key).map(|s| s.clone());
     }
 
     pub fn exists(&self, key: &String) -> bool {
-        return self.hash_map.contains_key(key);
+        return self.hash_map.lock().unwrap().contains_key(key);
     }
 }

src/server/control_plane.rs

@@ -1,4 +1,4 @@
-use log::warn;
+use log::{info, warn};
 use crate::server::cache::Cache;
 use crate::server::commands;
 use crate::server::commands::CommandEnum;