Memcached server

Cargo.toml

@@ -7,7 +7,8 @@ members = [
     "http-load-tester",
     "huffman-compression",
     "json-parser",
-    "load-balancer",
+    "load-balancer", 
+    "memcached",
     "redis-server",
     "shell",
     "sort-tool",

memcached/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "memcached"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow = "1.0.81"
+bytes = "1.6.0"
+crossbeam = "0.8"
+itertools = "0.12.1"
+linked-hash-map = "0.5.6"
+multipeek = "0.1.2"
+tokio = { version = "1.37.0", features = ["full"] }

memcached/README.md

@@ -0,0 +1,32 @@
+## Running the server
+
+```bash
+cargo run
+```
+
+You can also set the cache size. ie: The max number of bytes the server can hold. The default is `1000 bytes` if the `CACHE_SIZE` flag is not provided.
+
+```bash
+CACHE_SIZE=2000 cargo run
+```
+
+In a new terminal, connect to the server via telnet:
+
+```bash
+telnet localhost 11211
+```
+
+Passing some commands:
+
+set a value
+
+```bash
+set test 0 0 4
+1234
+```
+
+get a value
+
+```bash
+get test
+```

memcached/src/commands/add.rs

@@ -0,0 +1,33 @@
+use crate::{
+    db::{Content, Db},
+    response::Response,
+};
+
+use super::{extractors::ExtractedData, Parser};
+
+pub struct AddCommand {
+    data: ExtractedData,
+}
+
+impl AddCommand {
+    pub fn parse(parser: Parser) -> anyhow::Result<Self> {
+        let data = ExtractedData::parse(parser)?;
+
+        Ok(Self { data })
+    }
+
+    pub fn execute(self, db: &Db) -> Response {
+        db.with_data_mut(|data| {
+            if data.contains_key(&self.data.key) {
+                Response::NotStored
+            } else {
+                data.insert(self.data.key.clone(), Content::from(&self.data));
+                if self.data.noreply {
+                    Response::NoReply
+                } else {
+                    Response::Stored
+                }
+            }
+        })
+    }
+}

memcached/src/commands/append.rs

@@ -0,0 +1,33 @@
+use crate::{
+    db::{Content, Db},
+    response::Response,
+};
+
+use super::{extractors::ExtractedData, Parser};
+
+pub struct AppendCommand {
+    data: ExtractedData,
+}
+
+impl AppendCommand {
+    pub fn parse(parser: Parser) -> anyhow::Result<Self> {
+        let data = ExtractedData::parse(parser)?;
+
+        Ok(Self { data })
+    }
+
+    pub fn execute(self, db: &Db) -> Response {
+        db.with_data_mut(|data| {
+            let appended = data.append(&self.data.key, Content::from(&self.data));
+            if appended {
+                if self.data.noreply {
+                    Response::NoReply
+                } else {
+                    Response::Stored
+                }
+            } else {
+                Response::NotStored
+            }
+        })
+    }
+}

memcached/src/commands/extractors/mod.rs

@@ -0,0 +1,86 @@
+use anyhow::Context;
+
+use crate::{anyhow, db::Content};
+use std::cmp::Ordering;
+use std::time::Duration;
+
+use super::Parser;
+#[derive(Debug)]
+pub struct ExtractedData {
+    pub key: String,
+    pub flags: u32,
+    pub exptime: Option<Duration>,
+    pub bytes: usize,
+    pub noreply: bool,
+    pub content: Vec<u8>,
+}
+
+impl ExtractedData {
+    pub fn parse(mut parser: Parser) -> anyhow::Result<Self> {
+        let key = parser.next_string().ok_or(anyhow!("Expected a key"))?;
+
+        let flags = parser
+            .next_string()
+            .ok_or(anyhow!("Expected a flag"))?
+            .parse()
+            .context("Failed to parse flags")?;
+
+        let exptime_in_sec = parser
+            .next_string()
+            .ok_or(anyhow!("Expected expiry time"))?
+            .parse::<i64>()
+            .context("Failed to parse exptime")?;
+
+        let exptime = match exptime_in_sec.cmp(&0) {
+            Ordering::Equal => None,
+            // expires immediately
+            Ordering::Less => Some(std::time::Duration::from_secs(0)),
+            Ordering::Greater => {
+                let exptime = std::time::SystemTime::now()
+                    .duration_since(std::time::UNIX_EPOCH)
+                    .unwrap()
+                    + std::time::Duration::from_secs(exptime_in_sec as u64);
+                Some(exptime)
+            }
+        };
+
+        let bytes = parser
+            .next_string()
+            .ok_or(anyhow!("Expected bytes count"))?
+            .parse()
+            .context("Failed to parse number of bytes")?;
+
+        let maybe_noreply = parser
+            .peek_next_string()
+            .ok_or(anyhow!("Expected to get noreply or bytes"))?;
+
+        let noreply = if maybe_noreply == "noreply" {
+            let _ = parser.next_string();
+            true
+        } else {
+            false
+        };
+
+        let content = parser.next_bytes().ok_or(anyhow!("Expected bytes"))?;
+
+        Ok(Self {
+            key,
+            flags,
+            exptime,
+            bytes,
+            noreply,
+            content,
+        })
+    }
+}
+
+impl From<&ExtractedData> for Content {
+    fn from(value: &ExtractedData) -> Self {
+        Self {
+            data: value.content.clone(),
+            byte_count: value.bytes,
+            flags: value.flags,
+            exp_duration: value.exptime,
+        }
+    }
+}

memcached/src/commands/get.rs

@@ -0,0 +1,23 @@
+use crate::{db::Db, response::Response};
+
+use super::Parser;
+use anyhow::anyhow;
+
+pub struct GetCommand {
+    key: String,
+}
+
+impl GetCommand {
+    pub fn parse(mut parser: Parser) -> anyhow::Result<Self> {
+        let key = parser.next_string().ok_or(anyhow!("Expected a key"))?;
+        Ok(Self { key })
+    }
+
+    pub fn execute(self, db: &Db) -> Response {
+        let content = db.get(&self.key);
+        content
+            .as_ref()
+            .map(|content| Response::Value((content, self.key).into()))
+            .unwrap_or(Response::End)
+    }
+}