Implement basic functionality of mRPC (#92)

* save

* move transport implementations to a separate directory

* save

* add mrpc register (not finished)

* save, working on socket datapath

* save, creating service and customer

* save, pass compile

* minor fix, update KOALA_PATH

* save, graph abstraction

* save, add infrastructure of building engine graph

* save, update customer and server, a lot compile error to fix

* pass compile

* minor fix

* fix minor bug in print_lat

* save

* save

* save, implementing uverbs

* save, port ulib to koala

* save

* save, implement rpc_adapter

* save, fix several issues, pass compile

* finish coding, there are many places which do not make much sense

* fix some bugs

* minor update

* minor changes, also update a todo list for the initial RPC to run

* finish rpc client side async glue code and non-blocking get_request in rpc_adapter

* implemented switch_addr_space at the koala side

* fix many bugs

* fix more bugs, now can do rpc of a round-trip

* fix many bugs, now switch_addr_space works

* now can send multiple RPC requests

* save

* fix many bugs in memory mapping, now can stably run multiple messages

* minor change

* backport bugfix from the main branch

* minor change

Co-authored-by: lsh <[email protected]>

Author: crazyboycjr

Cargo.lock

@@ -1,5 +1,14 @@
 [workspace]
-members = ["src/experimental", "src/koala", "src/ipc", "src/engine", "src/rdma", "src/utils", "src/libkoala", "src/koala_examples"]
+members = [
+  "src/experimental",
+  "src/koala",
+  "src/ipc",
+  "src/rdma",
+  "src/utils",
+  "src/libkoala",
+  "src/koala_examples",
+  "src/memmap-fixed",
+]
 
 [profile.release]
 debug=true

Cargo.toml

@@ -1,9 +1,65 @@
-modules = ["transport"]
+log_env = "KOALA_LOG"
+default_log_level = "debug"
 
-[transport]
+# modules = ["transport-rdma", "transport-tcp", "mrpc", "rpc-adapter", "overload"]
+modules = ["Mrpc", "RpcAdapter", "Overload"]
+
+[control]
+prefix = "/tmp/koala"
+path = "koala-control.sock"
+
+[transport-rdma]
 prefix = "/tmp/koala"
-module_path = "koala-transport.sock"
-engine_basename = "koala-transport-engine"
+engine_basename = "koala-transport-engine-rdma"
 datapath_wq_depth = 32
 datapath_cq_depth = 32
 command_max_interval_ms = 1000
+
+[[node]]
+id = "mrpc.0"
+type = "Mrpc"
+
+# [[node]]
+# id = "overload"
+# type = "Overload"
+
+[[node]]
+id = "rpc-adapter"
+type = "RpcAdapter"
+
+[edges]
+egress = [
+  ["mrpc.0", "rpc-adapter"],
+]
+# egress = [
+#   ["mrpc.0", "overload"],
+#   ["overload", "rpc-adapter"],
+# ]
+ingress = [
+  ["rpc-adapter", "mrpc.0"],
+]
+
+# [[egress]]
+# from = "mrpc.0"
+# to = "overload"
+
+# instances = [
+#     {
+#         "id": 1,
+#         "type":"mrpc",
+#         "tx_output": [2],
+#         "rx_input": [3]
+#     },
+#     {
+#         "id": 2,
+#         "type":"overload",
+#         "tx_input": [1],
+#         "tx_output": [3]
+#     },
+#     {
+#         "id": 3,
+#         "type":"adapter",
+#         "tx_input": [2],
+#         "rx_output": [1]
+#     }
+# ]

koala.toml

@@ -3,7 +3,7 @@
 set -euo pipefail
 
 main() {
-    cmake -DCMAKE_BUILD_TYPE=Release -Bbuild
+    cmake -DCMAKE_BUILD_TYPE=RelWithDebInfo -Bbuild
     cmake --build build
 }
 

scripts/build_release.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+
+function info {
+	echo "$(tput bold)$(tput setaf 10)"$*"$(tput sgr 0)"
+}
+
+
+info "miscellaneous code:"
+find src/experimental -type f -name "*.rs" | xargs wc -l
+info "rdma bindings:"
+find src/rdma -type f -name "*.rs" | grep -v bindings | xargs wc -l
+info "koala examples:"
+find src/koala_examples -type f -name "*.rs" | xargs wc -l
+
+info "koala total:"
+find src/ -type f -name "*.rs" | grep -v "experimental" | grep -v bindings.rs | grep -v koala_examples | xargs wc -l

scripts/count_loc.sh

@@ -24,11 +24,14 @@ serde = { version = "1.0.130", features = ["derive"] }
 bincode = "1.3.3"
 anyhow = "1.0.44"
 ipc = { path = "../ipc" }
-engine = { path = "../engine" }
 interface = { path = "../interface" }
 uuid = "0.8.2"
 thiserror = "1.0.29"
 shmem-ipc = "0.2.0"
+toml = { version = "0.5.8", features = ["preserve_order"] }
+petgraph = "0.6.0"
+linked-hash-map = "0.5.4"
+zerocopy = "0.3.0"
 
 [dev-dependencies]
 rand = "0.8.3"

src/engine/Cargo.toml

@@ -15,7 +15,7 @@ use structopt::StructOpt;
 use shmem_ipc::sharedring::{Receiver, Sender};
 
 use experimental::*;
-use ipc::{recv_fd, send_fd};
+use experimental::ipc::{recv_fd, send_fd};
 
 type Value = [u8; 64];
 

src/engine/src/lib.rs

@@ -14,7 +14,7 @@ use structopt::StructOpt;
 use shmem_ipc::sharedring::{Receiver, Sender};
 
 use experimental::*;
-use ipc::{recv_fd, send_fd};
+use experimental::ipc::{recv_fd, send_fd};
 
 type Value = [u8; 64];
 

src/experimental/Cargo.toml

@@ -0,0 +1,161 @@
+use std::fs;
+use std::path::{Path, PathBuf};
+
+use serde::{Serialize, Deserialize};
+
+use interface::engine::EngineType;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Node {
+    id: String,
+    #[serde(alias = "type")]
+    engine_type: EngineType,
+    #[serde(default)]
+    tx_input: Vec<String>,
+    #[serde(default)]
+    tx_output: Vec<String>,
+    #[serde(default)]
+    rx_input: Vec<String>,
+    #[serde(default)]
+    rx_output: Vec<String>,
+}
+
+// #[derive(Debug, Clone, Copy)]
+// struct Node {
+//     id: usize,
+//     engine_type: EngineType,
+// }
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum DatapathDir {
+    Tx,
+    Rx,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Edge {
+    dp: DatapathDir,
+}
+
+type Graph = petgraph::Graph<Node, Edge>;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct Profile {
+    log_env: String,
+    default_log_level: String,
+    modules: Vec<String>,
+    control: Control,
+    #[serde(alias = "transport-rdma")]
+    transport_rdma: Option<RdmaTransportConfig>,
+    node: Vec<Node>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Control {
+    prefix: PathBuf,
+    path: PathBuf,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RdmaTransportConfig {
+    prefix: PathBuf,
+    engine_basename: String,
+    datapath_wq_depth: usize,
+    datapath_cq_depth: usize,
+    command_max_interval_ms: u32,
+}
+
+struct Config {
+    toml_config: toml::Value,
+    profile: Profile,
+    // graph: Graph,
+}
+
+const VALID_KEYS: [&str; 4] = ["log_env", "default_log_level", "modules", "control"];
+
+impl Config {
+    fn parse_graph(config: &toml::Value, profile: &Profile) -> anyhow::Result<Graph> {
+        let modules = &profile.modules;
+        match config {
+            toml::Value::Table(config) => {
+                for key in config.keys() {
+                    let engine_type_str = if let Some((a, _)) = key.split_once("-") {
+                        a.to_owned()
+                    } else {
+                        key.clone()
+                    };
+                    if !modules.contains(&engine_type_str) { continue; }
+                    // let module: Module = &config[key].to_string().parse().unwrap();
+                    println!("engine_type_str: {:?}", engine_type_str);
+                    println!("to_str: {:?}", config[key].to_string());
+                    let module: Node = toml::from_str(&config[key].to_string()).unwrap();
+                    println!("module: {:?}", module);
+                }
+                Err(anyhow::anyhow!("Config has a wrong format"))
+            }
+            _ => Err(anyhow::anyhow!("Config has a wrong format")),
+        }
+    }
+
+    fn verify_config(config: &toml::Value, profile: &Profile) -> anyhow::Result<()> {
+        let modules = &profile.modules;
+        match config {
+            toml::Value::Table(config) => {
+                for key in config.keys() {
+                    if VALID_KEYS.contains(&key.as_str()) {
+                        continue;
+                    }
+                    if modules.contains(key) {
+                        continue;
+                    }
+                    if let Some((a, _b)) = key.split_once("-") {
+                        if modules.contains(&a.to_owned()) {
+                            continue;
+                        }
+                    }
+                    return Err(anyhow::anyhow!("Invalid key: {}", key));
+                }
+                return Ok(());
+            }
+            _ => Err(anyhow::anyhow!("Config has a wrong format")),
+        }
+    }
+
+    fn from_path<P: AsRef<Path>>(path: P) -> anyhow::Result<Self> {
+        let content = fs::read_to_string(path).unwrap();
+        let config = content.parse::<toml::Value>().unwrap();
+        let profile = toml::from_str(&content).unwrap();
+        // let graph = Self::parse_graph(&config, &profile)?;
+        // Self::verify_config(&config, &profile)?;
+        println!("profile: {:#?}", profile);
+        Ok(Self {
+            toml_config: config,
+            profile,
+            // graph,
+        })
+    }
+}
+
+fn main() {
+    let content = fs::read_to_string("koala.toml").unwrap();
+    let config = content.parse::<toml::Value>().unwrap();
+    // println!("config: {:#?}", config["log_env"]);
+    // println!("config: {:#?}", config["default_log_level"]);
+    // println!("config: {:#?}", config["modules"]);
+    // println!("config: {:#?}", config["control"]);
+    // println!("config: {:#?}", config["transport-rdma"]);
+    // println!("config: {:#?}", config["Mrpc-0"]);
+    // if let toml::Value::Table(config) = config {
+    //     for k in config.keys() {
+    //         println!("key: {:?}", k);
+    //     }
+    // }
+
+    let profile: Profile = toml::from_str(&content).unwrap();
+    println!("profile: {:#?}", profile);
+
+    let config = Config::from_path("koala.toml").unwrap();
+}