add code for 46

Author: tyrchen

46_kv/.gitignore

@@ -0,0 +1 @@
+fixtures

46_kv/Cargo.toml

@@ -0,0 +1,53 @@
+[package]
+name = "kv6"
+version = "0.1.0"
+edition = "2018"
+
+[[bin]]
+name = "kvs"
+path = "src/server.rs"
+doc = false
+
+[[bin]]
+name = "kvc"
+path = "src/client.rs"
+doc = false
+
+[dependencies]
+anyhow = "1" # 错误处理
+bytes = "1" # 高效处理网络 buffer 的库
+dashmap = "4" # 并发 HashMap
+flate2 = "1" # gzip 压缩
+futures = "0.3" # 提供 Stream trait
+http = "0.2" # 我们使用 HTTP status code 所以引入这个类型库
+opentelemetry-jaeger = "0.15" # opentelemetry jaeger 支持
+prost = "0.8" # 处理 protobuf 的代码
+rustls-native-certs = "0.5"
+serde = { version = "1", features = ["derive"] } # 序列化/反序列化
+sled = "0.34" # sled db
+thiserror = "1" # 错误定义和处理
+tokio = { version = "1", features = ["full" ] } # 异步网络库
+tokio-rustls = "0.22" # 处理 TLS
+tokio-stream = { version = "0.1", features = ["sync"] } # 处理 stream
+tokio-util = { version = "0.6", features = ["compat"]} # tokio 和 futures 的兼容性库
+toml = "0.5" # toml 支持
+tracing = "0.1" # 日志处理
+tracing-appender = "0.1" # 文件日志
+tracing-opentelemetry = "0.15" # opentelemetry 支持
+tracing-subscriber = { version = "0.2", features = ["json", "chrono"] } # 日志处理
+yamux = "0.9" # yamux 多路复用支持
+
+[dev-dependencies]
+async-prost = "0.2.1" # 支持把 protobuf 封装成 TCP frame
+certify = "0.3"
+criterion = { version = "0.3", features = ["async_futures", "async_tokio", "html_reports"] } # benchmark
+rand = "0.8" # 随机数处理
+tempfile = "3" # 处理临时目录和临时文件
+tokio-util = { version = "0.6", features = ["codec"]}
+
+[build-dependencies]
+prost-build = "0.8" # 编译 protobuf
+
+[[bench]]
+name = "pubsub"
+harness = false

46_kv/abi.proto

@@ -0,0 +1,119 @@
+syntax = "proto3";
+
+package abi;
+
+// 来自客户端的命令请求
+message CommandRequest {
+  oneof request_data {
+    Hget hget = 1;
+    Hgetall hgetall = 2;
+    Hmget hmget = 3;
+    Hset hset = 4;
+    Hmset hmset = 5;
+    Hdel hdel = 6;
+    Hmdel hmdel = 7;
+    Hexist hexist = 8;
+    Hmexist hmexist = 9;
+    Subscribe subscribe = 10;
+    Unsubscribe unsubscribe = 11;
+    Publish publish = 12;
+  }
+}
+
+// 服务器的响应
+message CommandResponse {
+  // 状态码；复用 HTTP 2xx/4xx/5xx 状态码
+  uint32 status = 1;
+  // 如果不是 2xx，message 里包含详细的信息
+  string message = 2;
+  // 成功返回的 values
+  repeated Value values = 3;
+  // 成功返回的 kv pairs
+  repeated Kvpair pairs = 4;
+}
+
+// 从 table 中获取一个 key，返回 value
+message Hget {
+  string table = 1;
+  string key = 2;
+}
+
+// 从 table 中获取所有的 Kvpair
+message Hgetall { string table = 1; }
+
+// 从 table 中获取一组 key，返回它们的 value
+message Hmget {
+  string table = 1;
+  repeated string keys = 2;
+}
+
+// 返回的值
+message Value {
+  oneof value {
+    string string = 1;
+    bytes binary = 2;
+    int64 integer = 3;
+    double float = 4;
+    bool bool = 5;
+  }
+}
+
+// 返回的 kvpair
+message Kvpair {
+  string key = 1;
+  Value value = 2;
+}
+
+// 往 table 里存一个 kvpair，
+// 如果 table 不存在就创建这个 table
+message Hset {
+  string table = 1;
+  Kvpair pair = 2;
+}
+
+// 往 table 中存一组 kvpair，
+// 如果 table 不存在就创建这个 table
+message Hmset {
+  string table = 1;
+  repeated Kvpair pairs = 2;
+}
+
+// 从 table 中删除一个 key，返回它之前的值
+message Hdel {
+  string table = 1;
+  string key = 2;
+}
+
+// 从 table 中删除一组 key，返回它们之前的值
+message Hmdel {
+  string table = 1;
+  repeated string keys = 2;
+}
+
+// 查看 key 是否存在
+message Hexist {
+  string table = 1;
+  string key = 2;
+}
+
+// 查看一组 key 是否存在
+message Hmexist {
+  string table = 1;
+  repeated string keys = 2;
+}
+
+// subscribe 到某个主题，任何发布到这个主题的数据都会被收到
+// 成功后，第一个返回的 CommandResponse，我们返回一个唯一的 subscription id
+message Subscribe { string topic = 1; }
+
+// 取消对某个主题的订阅
+message Unsubscribe {
+  string topic = 1;
+  uint32 id = 2;
+}
+
+// 发布数据到某个主题
+message Publish {
+  string topic = 1;
+  repeated Value data = 2;
+}

46_kv/benches/pubsub.rs

@@ -0,0 +1,126 @@
+use anyhow::Result;
+use criterion::{criterion_group, criterion_main, Criterion};
+use futures::StreamExt;
+use kv6::{
+    start_client_with_config, start_server_with_config, ClientConfig, CommandRequest, ServerConfig,
+    StorageConfig, YamuxCtrl,
+};
+use rand::prelude::SliceRandom;
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio::runtime::Builder;
+use tokio::time;
+use tokio_rustls::client::TlsStream;
+use tracing::{info, span};
+use tracing_subscriber::{layer::SubscriberExt, prelude::*, EnvFilter};
+
+async fn start_server() -> Result<()> {
+    let addr = "127.0.0.1:9999";
+    let mut config: ServerConfig = toml::from_str(include_str!("../fixtures/server.conf"))?;
+    config.general.addr = addr.into();
+    config.storage = StorageConfig::MemTable;
+
+    tokio::spawn(async move {
+        start_server_with_config(&config).await.unwrap();
+    });
+
+    Ok(())
+}
+
+async fn connect() -> Result<YamuxCtrl<TlsStream<TcpStream>>> {
+    let addr = "127.0.0.1:9999";
+    let mut config: ClientConfig = toml::from_str(include_str!("../fixtures/client.conf"))?;
+    config.general.addr = addr.into();
+
+    Ok(start_client_with_config(&config).await?)
+}
+
+async fn start_subscribers(topic: &'static str) -> Result<()> {
+    let mut ctrl = connect().await?;
+    let stream = ctrl.open_stream().await?;
+    info!("C(subscriber): stream opened");
+    let cmd = CommandRequest::new_subscribe(topic.to_string());
+    tokio::spawn(async move {
+        let mut stream = stream.execute_streaming(&cmd).await.unwrap();
+        while let Some(Ok(data)) = stream.next().await {
+            drop(data);
+        }
+    });
+
+    Ok(())
+}
+
+async fn start_publishers(topic: &'static str, values: &'static [&'static str]) -> Result<()> {
+    let mut rng = rand::thread_rng();
+    let v = values.choose(&mut rng).unwrap();
+
+    let mut ctrl = connect().await.unwrap();
+    let mut stream = ctrl.open_stream().await.unwrap();
+    info!("C(publisher): stream opened");
+
+    let cmd = CommandRequest::new_publish(topic.to_string(), vec![(*v).into()]);
+    stream.execute_unary(&cmd).await.unwrap();
+
+    Ok(())
+}
+
+fn pubsub(c: &mut Criterion) {
+    let tracer = opentelemetry_jaeger::new_pipeline()
+        .with_service_name("kv-bench")
+        .install_simple()
+        .unwrap();
+    let opentelemetry = tracing_opentelemetry::layer().with_tracer(tracer);
+
+    tracing_subscriber::registry()
+        .with(EnvFilter::from_default_env())
+        .with(opentelemetry)
+        .init();
+
+    let root = span!(tracing::Level::INFO, "app_start", work_units = 2);
+    let _enter = root.enter();
+    // 创建 Tokio runtime
+    let runtime = Builder::new_multi_thread()
+        .worker_threads(4)
+        .thread_name("pubsub")
+        .enable_all()
+        .build()
+        .unwrap();
+
+    let base_str = include_str!("../fixtures/server.conf"); // 891 bytes
+
+    let values: &'static [&'static str] = Box::leak(
+        vec![
+            &base_str[..64],
+            &base_str[..128],
+            &base_str[..256],
+            &base_str[..512],
+        ]
+        .into_boxed_slice(),
+    );
+    let topic = "lobby";
+
+    // 运行服务器和 100 个 subscriber，为测试准备
+    runtime.block_on(async {
+        eprint!("preparing server and subscribers");
+        start_server().await.unwrap();
+        time::sleep(Duration::from_millis(50)).await;
+        for _ in 0..1000 {
+            start_subscribers(topic).await.unwrap();
+            eprint!(".");
+        }
+        eprintln!("Done!");
+    });
+
+    // 进行 benchmark
+    c.bench_function("publishing", move |b| {
+        b.to_async(&runtime)
+            .iter(|| async { start_publishers(topic, values).await })
+    });
+}
+
+criterion_group! {
+    name = benches;
+    config = Criterion::default().sample_size(10);
+    targets = pubsub
+}
+criterion_main!(benches);

46_kv/build.rs

@@ -0,0 +1,11 @@
+fn main() {
+    let mut config = prost_build::Config::new();
+    config.bytes(&["."]);
+    config.type_attribute(".", "#[derive(PartialOrd)]");
+    config
+        .out_dir("src/pb")
+        .compile_protos(&["abi.proto"], &["."])
+        .unwrap();
+    println!("cargo:rerun-if-changed=build.rs");
+    println!("cargo:rerun-if-changed=abi.proto");
+}