garbage collection

redisless/src/server/mod.rs

@@ -195,6 +195,9 @@ fn start_server<T: Storage + Send + 'static>(
         }
     };
 
+    // start garbage collection
+    thread_pool.spawn(move || do_gc(&storage));
+
     // listen incoming requests
     for stream in listener.incoming() {
         match stream {
@@ -216,6 +219,10 @@ fn start_server<T: Storage + Send + 'static>(
     }
 }
 
+fn do_gc<T: Storage /*+ Send + 'static*/>(storage: &Arc<Mutex<T>>) {
+    let storage = storage.clone();
+}
+
 fn handle_tcp_stream<T: Storage + Send + 'static>(
     tcp_stream: TcpStream,
     thread_pool: &ThreadPool,

redisless/src/storage/in_memory.rs

@@ -1,4 +1,7 @@
-use std::collections::HashMap;
+use std::{
+    collections::HashMap,
+    sync::atomic::{AtomicBool, Ordering},
+};
 
 use prost::bytes::BufMut;
 
@@ -27,6 +30,7 @@ impl Storage for InMemoryStorage {
         self.data_mapper.insert(key.to_vec(), meta);
         self.string_store.insert(key.to_vec(), value.to_vec());
     }
+
     fn extend(&mut self, key: &[u8], tail: &[u8]) -> u64 {
         match self.string_store.get_mut(key) {
             Some(v) => {
@@ -49,13 +53,10 @@ impl Storage for InMemoryStorage {
         }
     }
 
-    fn read(&mut self, key: &[u8]) -> Option<&[u8]> {
+    fn read(&self, key: &[u8]) -> Option<&[u8]> {
         if let Some(value) = self.data_mapper.get(key) {
             match value.is_expired() {
-                true => {
-                    self.remove(key);
-                    None
-                }
+                true => None,
                 false => Some(self.string_store.get(key).unwrap()),
             }
         } else {
@@ -64,7 +65,15 @@ impl Storage for InMemoryStorage {
     }
 
     fn meta(&self, key: &[u8]) -> Option<&RedisMeta> {
-        self.data_mapper.get(key)
+        if let Some(meta) = self.data_mapper.get(key) {
+            if meta.is_expired() {
+                None
+            } else {
+                Some(meta)
+            }
+        } else {
+            None
+        }
     }
 
     fn remove(&mut self, key: &[u8]) -> u32 {
@@ -135,13 +144,10 @@ impl Storage for InMemoryStorage {
             .insert(key.to_vec(), RedisHashMap::new(value));
     }
 
-    fn hread(&mut self, key: &[u8], field_key: &[u8]) -> Option<&[u8]> {
+    fn hread(&self, key: &[u8], field_key: &[u8]) -> Option<&[u8]> {
         if let Some(meta) = self.data_mapper.get(key) {
             match meta.is_expired() {
-                true => {
-                    self.remove(key);
-                    None
-                }
+                true => None,
                 // good to go
                 false => {
                     // will never panic since we already checked if the key existed in data_mapper

redisless/src/storage/mod.rs

@@ -15,12 +15,12 @@ pub trait Storage {
     fn write(&mut self, key: &[u8], value: &[u8]);
     fn extend(&mut self, key: &[u8], value: &[u8]) -> u64;
     fn expire(&mut self, key: &[u8], expiry: Expiry) -> u32;
-    fn read(&mut self, key: &[u8]) -> Option<&[u8]>;
+    fn read(&self, key: &[u8]) -> Option<&[u8]>;
     fn remove(&mut self, key: &[u8]) -> u32;
     fn contains(&mut self, key: &[u8]) -> bool;
     fn type_of(&mut self, key: &[u8]) -> &[u8];
     fn hwrite(&mut self, key: &[u8], value: HashMap<RedisString, RedisString>);
-    fn hread(&mut self, key: &[u8], field_key: &[u8]) -> Option<&[u8]>;
+    fn hread(&self, key: &[u8], field_key: &[u8]) -> Option<&[u8]>;
     fn size(&self) -> u64;
     fn meta(&self, key: &[u8]) -> Option<&RedisMeta>;
 }

redisless/src/storage/models/meta.rs

@@ -1,20 +1,34 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+
 use super::{Expiry, RedisType};
 
 pub struct RedisMeta {
     pub data_type: RedisType,
     pub expiry: Option<Expiry>,
+    pub tombstone: AtomicBool,
 }
 
 impl RedisMeta {
     pub fn new(data_type: RedisType, expiry: Option<Expiry>) -> Self {
-        Self { data_type, expiry }
+        Self {
+            data_type,
+            expiry,
+            tombstone: AtomicBool::from(false),
+        }
     }
 
     pub fn is_expired(&self) -> bool {
-        if let Some(expiry) = &self.expiry {
-            expiry.duration_left_millis() <= 0
-        } else {
-            false
+        self.tombstone.load(Ordering::SeqCst) || {
+            if let Some(expiry) = &self.expiry {
+                if expiry.duration_left_millis() <= 0 {
+                    self.tombstone.store(true, Ordering::SeqCst);
+                    true
+                } else {
+                    false
+                }
+            } else {
+                false
+            }
         }
     }
 }