get, set, echo done

Author: Shiva953

.gitattributes

@@ -0,0 +1 @@
+* text=auto

.gitignore

@@ -0,0 +1,8 @@
+debug/
+target/
+
+**/*.rs.bk
+
+*.pdb
+dump.rdb
+

Cargo.lock

@@ -0,0 +1,25 @@
+# DON'T EDIT THIS!
+#
+# Codecrafters relies on this file being intact to run tests successfully. Any changes
+# here will not reflect when CodeCrafters tests your code, and might even cause build
+# failures.
+#
+# DON'T EDIT THIS!
+[package]
+name = "redis-starter-rust"
+version = "0.1.0"
+authors = ["Codecrafters <hello@codecrafters.io>"]
+edition = "2021"
+
+# DON'T EDIT THIS!
+#
+# Codecrafters relies on this file being intact to run tests successfully. Any changes
+# here will not reflect when CodeCrafters tests your code, and might even cause build
+# failures.
+#
+# DON'T EDIT THIS!
+[dependencies]
+anyhow = "1.0.59"                                   # error handling
+bytes = "1.3.0"                                     # helps manage buffers
+thiserror = "1.0.32"                                # error handling
+tokio = { version = "1.23.0", features = ["full"] } # async networking

Cargo.toml

@@ -0,0 +1 @@
+Minimal Redis Implementation in Rust.

README.md

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+redis-cli ping &
+redis-cli ping &
+redis-cli ping

concurrent.sh

@@ -0,0 +1,7 @@
+#!/bin/sh
+exec cargo run \
+    --quiet \
+    --release \
+    --target-dir=/tmp/codecrafters-redis-target \
+    --manifest-path $(dirname $0)/Cargo.toml \
+    -- "$@"

spawn_redis_server.sh

@@ -0,0 +1,106 @@
+use std::{
+    sync::{mpsc, Arc, Mutex},
+    thread,
+};
+
+
+pub struct ThreadPool {
+    workers: Vec<Worker>,
+    sender: Option<mpsc::Sender<Job>>,
+}
+
+type Job = Box<dyn FnOnce() + Send + 'static>;
+
+impl ThreadPool {
+    /// Create a new ThreadPool.
+    ///
+    /// The size is the number of threads in the pool.
+    ///
+    /// # Panics
+    ///
+    /// The `new` function will panic if the size is zero.
+    #[tokio::main]
+    pub async fn new(size: usize) -> ThreadPool {
+        assert!(size > 0);
+
+        let (sender, receiver) = mpsc::channel();
+
+        let receiver = Arc::new(Mutex::new(receiver));
+
+        let mut workers = Vec::with_capacity(size);
+
+        for id in 0..size {
+            workers.push(Worker::new(id, Arc::clone(&receiver)));
+        }
+
+        ThreadPool {
+            workers,
+            sender: Some(sender),
+        }
+    }
+    #[tokio::main]
+    pub async fn execute<F>(&self, f: F)
+    where
+        F: FnOnce() + Send + 'static,
+    {
+        let job = Box::new(f);
+
+        self.sender.as_ref().unwrap().send(job).unwrap();
+    }
+
+    pub async fn async_execute<F>(&self, f: F)
+    where
+        F: FnOnce() + Send + 'static,
+    {
+        let job = Box::new(f);
+
+        self.sender.as_ref().unwrap().send(job).unwrap();
+    }
+}
+
+//for each thread to finish the requests they're working on before closing
+impl Drop for ThreadPool {
+    #[tokio::main]
+    async fn drop(&mut self) {
+        drop(self.sender.take());
+
+        for worker in &mut self.workers {
+            println!("Shutting down worker {}", worker.id);
+
+            if let Some(thread) = worker.thread.take() {
+                thread.join().unwrap();
+            }
+        }
+    }
+}
+
+struct Worker {
+    id: usize,
+    thread: Option<thread::JoinHandle<()>>,
+}
+
+impl Worker {
+    #[tokio::main]
+    async fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
+        let thread = thread::spawn(move || loop {
+            let message = receiver.lock().unwrap().recv();
+
+            match message {
+                Ok(job) => {
+                    println!("Worker {id} got a job; executing.");
+
+                    job();
+                }
+                Err(_) => {
+                    println!("Worker {id} disconnected; shutting down.");
+                    break;
+                }
+            }
+        });
+
+        Worker {
+            id,
+            thread: Some(thread),
+        }
+    }
+}

src/lib.rs

@@ -0,0 +1,99 @@
+// use std::{
+//     io::{Read, Write},
+// };
+use redis_starter_rust::ThreadPool;
+use std::collections::HashMap;
+use tokio::net::{TcpListener, TcpStream};
+use anyhow::Result;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use std::sync::{Arc, Mutex};
+use resp::Value;
+
+mod resp;
+
+#[tokio::main]
+async fn main() {
+    // listening for connections
+    let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();
+    //creating a global k-v store, which gets updated each time a client(prev/new) adds a new (k,v)
+    //Arc is used 
+    let kv_store: Arc<Mutex<HashMap<String, String>>> = Arc::new(Mutex::new(HashMap::<String,String>::new()));
+    // for stream in listener.incoming() { 
+        //HANDLING CONCURRENT CLIENTS, NEW THREAD FOR EACH CLIENTi/SERVER stream
+        loop{ //INSTEAD OF USING for stream in listener.incoming() and synchronously iterating over each stream, we are asynchronously iterating over each stream till the data from the buffer ends
+        let stream = listener.accept().await; // listener.accept().await ASYNCHRONOUSLY WAITS FOR A NEW CONNECTION, INSTEAD OF AN SYNCHRONOUS ITERATOR LIKE listener.incoming() which takes each new connection and puts inside it
+        let mut kv_store = Arc::clone(&kv_store);
+        match stream { 
+            Ok((stream, _)) => {
+                //SPAWNING A NEW THREAD FOR EACH CLIENT REQ->S
+                //tried using threadpool and pool.execute, turns out each thread in it was unable to handle ASYNC read/write tasks
+                //the below spawns a new ASYNC THREAD for each new client request to the redis server
+                tokio::spawn(async move{ 
+                    handle_conn(stream, &mut kv_store).await;
+                });
+
+                //ECHO print command
+            }
+            Err(e) => {
+                println!("{e}");
+            }
+        }
+    }
+}
+
+async fn handle_conn(stream: TcpStream, kv_store: &mut Arc<std::sync::Mutex<HashMap<String, String>>>) {
+    let mut handler = resp::RespHandler::new(stream);
+    // let mut kv_store = HashMap::<String,String>::new();
+    loop{
+        let value = handler.read_value().await.unwrap(); //ALL PARSING HAPPENS IN THS FUNCTION 
+
+        let res = if let Some(v) = value {
+            let (command, args) = extract_command(v).unwrap();
+            match command.as_str().to_lowercase().as_str() {
+                "ping" => Value::SimpleString("PONG".to_string()),
+                "echo" => args.first().unwrap().clone(),
+                "set" => {store_key_value(unpack_bulk_str(args[0].clone()).unwrap(), unpack_bulk_str(args[1].clone()).unwrap(), kv_store)},
+                "get" => {get_value_from_key(unpack_bulk_str(args[0].clone()).unwrap(), kv_store)}, //by default, consider a input string as bulk string
+                c => panic!("Cannot handle command {}", c),
+            }
+        } else {
+            break;
+        };
+        handler.write_value(res).await;
+    }
+}
+//makes sense to store in a global shared hashmap
+
+fn store_key_value(key: String, value: String, kv_store: &mut Arc<std::sync::Mutex<HashMap<String, String>>>) -> Value{
+    kv_store.lock().unwrap().insert(key, value);
+    println!("{:?}", kv_store.lock().unwrap());
+    Value::SimpleString("OK".to_string())
+}
+
+fn get_value_from_key(key: String, kv_store: &mut Arc<std::sync::Mutex<HashMap<String, String>>>) -> Value{
+    println!("{:?}", kv_store);
+    match kv_store.lock().unwrap().get(&key) {
+        Some(v) => Value::BulkString(v.to_string()),
+        None => Value::SimpleString("(null)".to_string())
+    }
+}
+
+//extracting the command used after redis-cli, along with the args after the command[redis-cli <command> [..args]]
+// returning (command, [..args])
+fn extract_command(value: Value) -> Result<(String, Vec<Value>)> {
+    match value {
+        Value::Array(a) => { //[command, ..arguments]
+            Ok((
+                unpack_bulk_str(a.first().unwrap().clone())?, //command 
+                a.into_iter().skip(1).collect(), //[..arguments]
+            ))
+        },
+        _ => Err(anyhow::anyhow!("Unexpected command format")),
+    }
+}
+fn unpack_bulk_str(value: Value) -> Result<String> {
+    match value {
+        Value::BulkString(s) => Ok(s),
+        _ => Err(anyhow::anyhow!("Expected command to be a bulk string"))
+    }
+}

src/main.rs

@@ -0,0 +1,106 @@
+use tokio::{io::{AsyncReadExt, AsyncWriteExt}, net::TcpStream, stream};
+use bytes::{Buf, BytesMut};
+use anyhow::{Error, Ok, Result};
+
+#[derive(Clone, Debug)]
+pub enum Value {
+    SimpleString(String),
+    BulkString(String),
+    Array(Vec<Value>),
+}
+impl Value {
+    pub fn serialize(self) -> String {
+        match self {
+            Value::SimpleString(s) => format!("+{}\r\n", s),
+            Value::BulkString(s) => format!("${}\r\n{}\r\n", s.chars().count(), s),
+            _ => panic!("Unsupported value for serialize"),
+        }
+    }
+}
+
+pub struct RespHandler {
+    stream: TcpStream,
+    buffer: BytesMut,
+}
+
+impl RespHandler{
+    pub fn new(stream: TcpStream) -> Self{
+        RespHandler {
+            stream,
+            buffer: BytesMut::with_capacity(512),
+        }
+    }
+
+    pub async fn read_value(&mut self) -> Result<Option<Value>>{
+        let mut buf = [0; 512];
+        let read_count = self.stream.read_buf(&mut self.buffer).await?; //async READ using tokio
+        if read_count == 0 {
+            return Ok(None);
+        }
+        // convert [buffer] => [command, ..args]
+        let (v, _) = parse(self.buffer.split())?; 
+        Ok(Some(v))
+    }
+
+    pub async fn write_value(&mut self, response: Value){
+        self.stream.write(response.serialize().as_bytes()).await.unwrap();
+    }
+}
+
+fn parse(buffer: BytesMut) -> Result<(Value, usize)> {
+    match buffer[0] as char {
+        '+' => parse_simple_string(buffer),
+        '*' => parse_array(buffer),
+        '$' => parse_bulk_string(buffer),
+        _ => Err(anyhow::anyhow!("Not a known value type {:?}", buffer)),
+    }
+}
+
+fn parse_simple_string(buffer: BytesMut) -> Result<(Value, usize)> {
+    if let Some((line, len)) = read_until_crlf(&buffer[1..]) {
+        let string = String::from_utf8(line.to_vec()).unwrap();
+        return Ok((Value::SimpleString(string), len + 1))
+    }
+    return Err(anyhow::anyhow!("Invalid string {:?}", buffer));
+}
+
+fn parse_array(buffer: BytesMut) -> Result<(Value, usize)> {
+    let (array_length, mut bytes_consumed) = if let Some((line, len)) = read_until_crlf(&buffer[1..]) {
+        let array_length = parse_int(line)?;
+        (array_length, len + 1)
+    } else {
+        return Err(anyhow::anyhow!("Invalid array format {:?}", buffer));
+    };
+    let mut items = vec![];
+    for _ in 0..array_length {
+        let (array_item, len) = parse(BytesMut::from(&buffer[bytes_consumed..]))?;
+        items.push(array_item);
+        bytes_consumed += len;
+    }
+    return Ok((Value::Array(items), bytes_consumed))
+}
+
+fn parse_bulk_string(buffer: BytesMut) -> Result<(Value, usize)> {
+    let (bulk_str_len, bytes_consumed) = if let Some((line, len)) = read_until_crlf(&buffer[1..]) {
+        let bulk_str_len = parse_int(line)?;
+        (bulk_str_len, len + 1)
+    } else {
+        return Err(anyhow::anyhow!("Invalid array format {:?}", buffer));
+    };
+    let end_of_bulk_str = bytes_consumed + bulk_str_len as usize;
+    let total_parsed = end_of_bulk_str + 2;
+    Ok((Value::BulkString(String::from_utf8(buffer[bytes_consumed..end_of_bulk_str].to_vec())?), total_parsed))
+}
+
+fn read_until_crlf(buffer: &[u8]) -> Option<(&[u8], usize)> {
+    for i in 1..buffer.len() {
+        if buffer[i - 1] == b'\r' && buffer[i] == b'\n' {
+            return Some((&buffer[0..(i - 1)], i + 1));
+        }
+    }
+    return None;
+}
+
+fn parse_int(buffer: &[u8]) -> Result<i64> {
+    Ok(String::from_utf8(buffer.to_vec())?.parse::<i64>()?)
+}