feat(cmd): execute commands in task pool

Cargo.toml

@@ -9,6 +9,7 @@ members = [
 	"src/server",
 	"src/conf",
 	"src/cmd",
+	"src/executor",
 	"src/client"
 ]
 
@@ -52,13 +53,22 @@ tokio = { version = "1", features = ["full"] }
 snafu = "0.8"
 tempfile = "3.8"
 crc16 = "0.4"
+async-channel = "2.5.0"
+tokio-util = "0.7.16"
+async-trait = "0.1"
 foyer = { version = "0.18", features = ["nightly"] }
 
 ## workspaces members
 storage = { path = "src/storage" }
 kstd = { path = "src/kstd" }
 common-macro = { path = "src/common/macro" }
 net = { path = "src/net" }
+resp = { path = "src/resp" }
+server = { path = "src/server" }
+conf = { path = "src/conf" }
+cmd = { path = "src/cmd" }
+executor = { path = "src/executor" }
+client = { path = "src/client" }
 
 [profile.dev]
 split-debuginfo = "unpacked"

src/client/Cargo.toml

@@ -8,3 +8,5 @@ edition = "2021"
 [dependencies]
 async-trait = "0.1"
 resp = { path = "../resp" }
+parking_lot = { workspace = true }
+tokio = { version = "1.0", features = ["sync"] }

src/client/src/lib.rs

@@ -15,8 +15,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use async_trait::async_trait;
 use resp::RespData;
+use tokio::sync::Mutex;
 
 #[async_trait]
 pub trait StreamTrait: Send + Sync {
@@ -25,73 +28,94 @@ pub trait StreamTrait: Send + Sync {
 }
 
 pub struct Client {
-    stream: Box<dyn StreamTrait>,
+    // using tokio::sync::Mutex may has risks to pass this Client object across
+    // tokio runtimes. we may require to figure out how to make a refactor to
+    // avoid passing Client across runtimes.
+    stream: Mutex<Box<dyn StreamTrait>>,
+    ctx: parking_lot::Mutex<ClientContext>,
+}
+
+struct ClientContext {
     // TODO: use &[Vec<u8>], need lifetime.
     argv: Vec<Vec<u8>>,
     // Client name.
-    name: Vec<u8>,
-    cmd_name: Vec<u8>,
+    name: Arc<Vec<u8>>,
+    cmd_name: Arc<Vec<u8>>,
     key: Vec<u8>,
     reply: RespData,
 }
 
 impl Client {
     pub fn new(stream: Box<dyn StreamTrait>) -> Self {
         Self {
-            stream,
-            argv: Vec::default(),
-            name: Vec::default(),
-            cmd_name: Vec::default(),
-            key: Vec::default(),
-            reply: RespData::default(),
+            stream: Mutex::new(stream),
+            ctx: parking_lot::Mutex::new(ClientContext {
+                argv: Vec::default(),
+                name: Arc::new(Vec::default()),
+                cmd_name: Arc::new(Vec::default()),
+                key: Vec::default(),
+                reply: RespData::default(),
+            }),
         }
     }
 
-    pub async fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
-        self.stream.read(buf).await
+    pub async fn read(&self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
+        let mut stream = self.stream.lock().await;
+        stream.read(buf).await
     }
 
-    pub async fn write(&mut self, data: &[u8]) -> Result<usize, std::io::Error> {
-        self.stream.write(data).await
+    pub async fn write(&self, data: &[u8]) -> Result<usize, std::io::Error> {
+        let mut stream = self.stream.lock().await;
+        stream.write(data).await
     }
 
-    pub fn set_argv(&mut self, argv: &[Vec<u8>]) {
-        self.argv = argv.to_vec()
+    pub fn set_argv(&self, argv: &[Vec<u8>]) {
+        let mut ctx = self.ctx.lock();
+        ctx.argv = argv.to_vec();
     }
 
-    pub fn argv(&self) -> &[Vec<u8>] {
-        &self.argv
+    pub fn argv(&self) -> Vec<Vec<u8>> {
+        let ctx = self.ctx.lock();
+        ctx.argv.clone()
     }
 
-    pub fn set_name(&mut self, name: &[u8]) {
-        self.name = name.to_vec()
+    pub fn set_name(&self, name: &[u8]) {
+        let mut ctx = self.ctx.lock();
+        ctx.name = Arc::new(name.to_vec());
     }
 
-    pub fn name(&self) -> &[u8] {
-        &self.name
+    pub fn name(&self) -> Arc<Vec<u8>> {
+        let ctx = self.ctx.lock();
+        ctx.name.clone()
     }
 
-    pub fn set_cmd_name(&mut self, name: &[u8]) {
-        self.cmd_name = name.to_vec()
+    pub fn set_cmd_name(&self, name: &[u8]) {
+        let mut ctx = self.ctx.lock();
+        ctx.cmd_name = Arc::new(name.to_vec());
     }
 
-    pub fn cmd_name(&self) -> &[u8] {
-        &self.cmd_name
+    pub fn cmd_name(&self) -> Arc<Vec<u8>> {
+        let ctx = self.ctx.lock();
+        ctx.cmd_name.clone()
     }
 
-    pub fn set_key(&mut self, key: &[u8]) {
-        self.key = key.to_vec()
+    pub fn set_key(&self, key: &[u8]) {
+        let mut ctx = self.ctx.lock();
+        ctx.key = key.to_vec();
     }
 
-    pub fn key(&self) -> &[u8] {
-        &self.key
+    pub fn key(&self) -> Vec<u8> {
+        let ctx = self.ctx.lock();
+        ctx.key.clone()
     }
 
-    pub fn reply_mut(&mut self) -> &mut RespData {
-        &mut self.reply
+    pub fn set_reply(&self, reply: RespData) {
+        let mut ctx = self.ctx.lock();
+        ctx.reply = reply;
     }
 
-    pub fn take_reply(&mut self) -> RespData {
-        std::mem::take(&mut self.reply)
+    pub fn take_reply(&self) -> RespData {
+        let mut ctx = self.ctx.lock();
+        std::mem::take(&mut ctx.reply)
     }
 }

src/cmd/src/get.rs

@@ -46,26 +46,26 @@ impl Cmd for GetCmd {
     impl_cmd_meta!();
     impl_cmd_clone_box!();
 
-    fn do_initial(&self, client: &mut Client) -> bool {
+    fn do_initial(&self, client: &Client) -> bool {
         let key = client.argv()[1].clone();
         client.set_key(&key);
         true
     }
 
-    fn do_cmd(&self, client: &mut Client, storage: Arc<Storage>) {
+    fn do_cmd(&self, client: &Client, storage: Arc<Storage>) {
         let key = client.key();
-        let result = storage.get(key);
+        let result = storage.get(&key);
 
         match result {
             Ok(value) => {
-                *client.reply_mut() = RespData::BulkString(Some(value.into()));
+                client.set_reply(RespData::BulkString(Some(value.into())));
             }
             Err(e) => match e {
                 storage::error::Error::KeyNotFound { .. } => {
-                    *client.reply_mut() = RespData::BulkString(None);
+                    client.set_reply(RespData::BulkString(None));
                 }
                 _ => {
-                    *client.reply_mut() = RespData::Error(format!("ERR {e}").into());
+                    client.set_reply(RespData::Error(format!("ERR {e}").into()));
                 }
             },
         }