Merge pull request #3 from hindenbug/project-4-threadpool

Project 4 - Threadpool

Author: hindenbug

kvs-2/src/kv.rs

@@ -41,8 +41,8 @@ impl KvStore {
         let log = Self::new_log_file(&path)?;
 
         let mut store = KvStore {
-            path: path,
-            log: log,
+            path,
+            log,
             map: BTreeMap::new(),
         };
 

kvs-client-server/Cargo.lock

@@ -7,23 +7,28 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-clap = "2.33.0"
+clap = "2.33.1"
 failure = "0.1.5"
-serde = { version = "1.0.89", features = ["derive"] }
-serde_json = "1.0.39"
+serde = { version = "1.0.111", features = ["derive", "rc"] }
+serde_json = "1.0.53"
 bincode = "1.2.1"
 structopt = "0.3"
 log = "0.4"
 env_logger = "0.7"
 sled = "0.31"
+num_cpus = "1.13.0"
+crossbeam = "0.7.3"
+
 
 [dev-dependencies]
-assert_cmd = "0.11.0"
+assert_cmd = "1.0.1"
 predicates = "1.0.0"
-tempfile = "3.0.7"
+tempfile = "3.1.0"
 walkdir = "2.2.7"
 rand = "0.6.5"
 criterion = "0.3.0"
+crossbeam-utils = "0.7.2"
+panic-control = "0.1.4"
 
 [lib]
 test = false

kvs-client-server/Cargo.toml

@@ -79,7 +79,7 @@ fn main() -> Result<()> {
 }
 
 fn start_server_with<E: KvsEngine>(addr: &str, engine: E) -> Result<()> {
-    let mut server = Server::new(addr, engine);
+    let server = Server::new(addr, engine);
     server.serve()?;
     Ok(())
 }

kvs-client-server/src/bin/kvs-server.rs

@@ -5,12 +5,18 @@ use std::collections::BTreeMap;
 use std::fs::{self, File};
 use std::io::{BufReader, BufWriter, Seek, SeekFrom, Write};
 use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
 
 const COMPACTION_THRESHOLD: u64 = 1024;
 const LOG_FILE_NAME: &str = "current.db";
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct KvStore {
+    inner: Arc<Mutex<InnerKvStore>>,
+}
+
+#[derive(Debug)]
+struct InnerKvStore {
     path: PathBuf,
     log: File,
     map: BTreeMap<String, LogPointer>,
@@ -36,44 +42,53 @@ struct LogPointer {
 }
 
 impl KvsEngine for KvStore {
-    fn set(&mut self, key: String, value: String) -> Result<()> {
-        let offset = self.log.seek(SeekFrom::End(0))?;
+    fn set(&self, key: String, value: String) -> Result<()> {
+        let mut inner = self.inner.lock().unwrap();
+
+        let offset = inner.log.seek(SeekFrom::End(0))?;
         let command = Command {
             cmd: CommandType::Set,
             key: key.clone(),
             value,
         };
         // encoding before writing to log
-        serde_json::to_writer(&mut self.log, &command)?;
-        self.log.flush()?;
-        let current_offset = self.log.seek(SeekFrom::End(0))?;
-        self.map.insert(
+        serde_json::to_writer(&mut inner.log, &command)?;
+        inner.log.flush()?;
+        let current_offset = inner.log.seek(SeekFrom::End(0))?;
+        inner.map.insert(
             key,
             LogPointer {
                 offset,
                 len: current_offset - offset,
             },
         );
         if current_offset > COMPACTION_THRESHOLD {
-            self.compact()?;
+            inner.compact()?;
         }
         Ok(())
     }
 
-    fn get(&mut self, key: String) -> Result<Option<String>> {
-        if let Some(pointer) = self.map.get(&key) {
-            &self.log.seek(SeekFrom::Start(pointer.offset));
-            let mut de = serde_json::Deserializer::from_reader(&self.log);
-            let cmd: Command = serde::de::Deserialize::deserialize(&mut de)?;
-            Ok(Some(cmd.value))
-        } else {
-            Ok(None)
+    fn get(&self, key: String) -> Result<Option<String>> {
+        let mut inner = self.inner.lock().unwrap();
+
+        if !inner.map.contains_key(&key) {
+            return Ok(None);
         }
+
+        let pointer = inner.map.get(&key).unwrap();
+        let pos = SeekFrom::Start(pointer.offset);
+
+        inner.log.seek(pos)?;
+        let mut de = serde_json::Deserializer::from_reader(&inner.log);
+        let cmd: Command = serde::de::Deserialize::deserialize(&mut de)?;
+        Ok(Some(cmd.value))
     }
 
-    fn remove(&mut self, key: String) -> Result<()> {
-        self.log.seek(SeekFrom::End(0))?;
-        if self.map.get(&key).is_none() {
+    fn remove(&self, key: String) -> Result<()> {
+        let mut inner = self.inner.lock().unwrap();
+
+        inner.log.seek(SeekFrom::End(0))?;
+        if inner.map.get(&key).is_none() {
             return Err(KvsError::KeyNotFound);
         }
         let command = Command {
@@ -82,30 +97,35 @@ impl KvsEngine for KvStore {
             value: String::new(),
         };
         // encoding before writing to log
-        serde_json::to_writer(&mut self.log, &command)?;
-        self.log.flush()?;
-        self.map.remove(&key);
+        serde_json::to_writer(&mut inner.log, &command)?;
+        inner.log.flush()?;
+        inner.map.remove(&key);
         Ok(())
     }
 }
 
 impl KvStore {
     pub fn open(path: impl Into<PathBuf>) -> Result<KvStore> {
         let path = path.into();
-        fs::create_dir_all(&path)?;
-        let log = Self::new_log_file(&path)?;
+        fs::create_dir_all(&*path)?;
+        let log = InnerKvStore::new_log_file(&path)?;
 
-        let mut store = KvStore {
-            path: path,
-            log: log,
+        let mut inner = InnerKvStore {
+            path,
+            log,
             map: BTreeMap::new(),
         };
 
         // Load from log files
-        store.load_from_log()?;
-        Ok(store)
+        inner.load_from_log()?;
+
+        Ok(KvStore {
+            inner: Arc::new(Mutex::new(inner)),
+        })
     }
+}
 
+impl InnerKvStore {
     fn load_from_log(&mut self) -> Result<()> {
         let mut reader = BufReader::new(File::open(&self.path.join(LOG_FILE_NAME))?);
         let mut offset = reader.seek(SeekFrom::Start(0))?;
@@ -141,7 +161,7 @@ impl KvStore {
         Ok(())
     }
 
-    fn compact(&mut self) -> Result<()> {
+    fn compact(&self) -> Result<()> {
         let tmp_path: PathBuf = self.path.join("tmp.db");
         let file_path: PathBuf = self.path.join(LOG_FILE_NAME);
         let mut work_file = self.log.try_clone()?;

kvs-client-server/src/engines/kvs.rs

@@ -1,23 +1,23 @@
 use crate::Result;
 
 /// Trait for a key value storage engine.
-pub trait KvsEngine {
+pub trait KvsEngine: Clone + Send + 'static {
     /// Sets the value of a string key to a string.
     ///
     /// If the key already exists, the previous value will be overwritten.
-    fn set(&mut self, key: String, value: String) -> Result<()>;
+    fn set(&self, key: String, value: String) -> Result<()>;
 
     /// Gets the string value of a given string key.
     ///
     /// Returns `None` if the given key does not exist.
-    fn get(&mut self, key: String) -> Result<Option<String>>;
+    fn get(&self, key: String) -> Result<Option<String>>;
 
     /// Removes a given key.
     ///
     /// # Errors
     ///
     /// It returns `KvsError::KeyNotFound` if the given key is not found.
-    fn remove(&mut self, key: String) -> Result<()>;
+    fn remove(&self, key: String) -> Result<()>;
 }
 
 mod kvs;

kvs-client-server/src/engines/mod.rs

@@ -1,6 +1,8 @@
 use crate::{KvsEngine, KvsError, Result};
 use sled::Db;
 use std::path::PathBuf;
+
+#[derive(Clone)]
 pub struct SledKvsEngine {
     tree: Db,
 }
@@ -13,13 +15,13 @@ impl SledKvsEngine {
 }
 
 impl KvsEngine for SledKvsEngine {
-    fn set(&mut self, key: String, value: String) -> Result<()> {
+    fn set(&self, key: String, value: String) -> Result<()> {
         self.tree.insert(key, value.as_bytes())?;
         self.tree.flush()?;
         Ok(())
     }
 
-    fn get(&mut self, key: String) -> Result<Option<String>> {
+    fn get(&self, key: String) -> Result<Option<String>> {
         Ok(self
             .tree
             .get(key)?
@@ -28,7 +30,7 @@ impl KvsEngine for SledKvsEngine {
             .transpose()?)
     }
 
-    fn remove(&mut self, key: String) -> Result<()> {
+    fn remove(&self, key: String) -> Result<()> {
         self.tree.remove(key)?.ok_or(KvsError::KeyNotFound)?;
         self.tree.flush()?;
         Ok(())

kvs-client-server/src/engines/sled.rs

@@ -3,9 +3,11 @@ pub use engines::{KvStore, KvsEngine, SledKvsEngine};
 pub use error::{KvsError, Result};
 pub use network::Request;
 pub use server::Server;
+pub use thread_pool::{NaiveThreadPool, SharedQueueThreadPool, ThreadPool};
 
 mod client;
 mod engines;
 mod error;
 mod network;
 mod server;
+mod thread_pool;

kvs-client-server/src/lib.rs

@@ -1,5 +1,5 @@
 use crate::network::{GetResponse, RemoveResponse, Request, SetResponse};
-use crate::{KvsEngine, Result};
+use crate::{KvsEngine, Result, SharedQueueThreadPool, ThreadPool};
 use log::{debug, error, info};
 use serde_json::Deserializer;
 use std::io::{BufReader, BufWriter, Write};
@@ -22,23 +22,25 @@ impl<E: KvsEngine> Server<E> {
         Server { listener, engine }
     }
 
-    pub fn serve(&mut self) -> Result<()> {
+    pub fn serve(&self) -> Result<()> {
         debug!("Waiting for connections...");
+        let thread_pool = SharedQueueThreadPool::new(num_cpus::get() as u32)?;
         let listnr = self.listener.try_clone().unwrap();
         for stream in listnr.incoming() {
-            match stream {
+            let engine = self.engine.clone();
+            thread_pool.spawn(move || match stream {
                 Ok(stream) => {
-                    if let Err(e) = self.handle_client(stream) {
+                    if let Err(e) = Self::handle_client(engine, stream) {
                         error!("Error on serving client: {}", e);
                     }
                 }
                 Err(e) => error!("Connection failed, reason: {:?}", e),
-            }
+            })
         }
         Ok(())
     }
 
-    fn handle_client(&mut self, stream: TcpStream) -> Result<()> {
+    fn handle_client(engine: E, stream: TcpStream) -> Result<()> {
         debug!(
             "Connection established from {}, waiting for data...",
             stream.peer_addr()?
@@ -64,21 +66,21 @@ impl<E: KvsEngine> Server<E> {
             debug!("Received request from {}: {:?}", peer_addr, req);
             match req {
                 Request::Get { key } => {
-                    let engine_response = match self.engine.get(key) {
+                    let engine_response = match engine.get(key) {
                         Ok(value) => GetResponse::Ok(value),
                         Err(err) => GetResponse::Err(format!("{}", err)),
                     };
                     send_response!(engine_response);
                 }
                 Request::Set { key, value } => {
-                    let engine_response = match self.engine.set(key, value) {
+                    let engine_response = match engine.set(key, value) {
                         Ok(_) => SetResponse::Ok(()),
                         Err(err) => SetResponse::Err(format!("{}", err)),
                     };
                     send_response!(engine_response);
                 }
                 Request::Remove { key } => {
-                    let engine_response = match self.engine.remove(key) {
+                    let engine_response = match engine.remove(key) {
                         Ok(_) => RemoveResponse::Ok(()),
                         Err(err) => RemoveResponse::Err(format!("{}", err)),
                     };

kvs-client-server/src/server.rs

@@ -0,0 +1,20 @@
+use crate::Result;
+
+mod naive;
+mod shared_queue;
+
+pub use self::naive::NaiveThreadPool;
+pub use self::shared_queue::SharedQueueThreadPool;
+
+pub trait ThreadPool {
+    /// Creates a thread pool
+    ///
+    /// return error if failed to create any thread
+    fn new(threads: u32) -> Result<Self>
+    where
+        Self: Sized;
+
+    fn spawn<F>(&self, job: F)
+    where
+        F: FnOnce() + Send + 'static;
+}

kvs-client-server/src/thread_pool/mod.rs

@@ -0,0 +1,18 @@
+use crate::{Result, ThreadPool};
+
+use std::thread;
+
+pub struct NaiveThreadPool;
+
+impl ThreadPool for NaiveThreadPool {
+    fn new(_threads: u32) -> Result<Self> {
+        Ok(NaiveThreadPool)
+    }
+
+    fn spawn<F>(&self, job: F)
+    where
+        F: FnOnce() + Send + 'static,
+    {
+        thread::spawn(job);
+    }
+}