WARNING
本文内容大多翻译自原文：PNA Rust Project 3: Synchronous client-server networking。

前言#

任务：创建一个单线程、持久化的键/值存储的服务器和客户端，使用自定义协议进行同步联网。

目标：

创建一个客户端-服务器应用程序
用std库的网络API编写自定义协议
为服务端引入日志记录功能
用trait实现可插拔的后端
用sled对手写的后端进行基准测试 关键词：std::net、日志、trait、基准测试。

介绍#

在这个项目中，你将创建一个简单的键/值存储服务端和客户端，它们将用你自定义的网络协议进行通信。你将使用标准的日志crate生成日志，并正确处理网络边界上的错误。一旦有了一个可运行的客户端-服务端架构，那么你就能够抽象出trait背后的存储引擎，并将你的实现与sled引擎进行性能比较。

项目需求规格#

cargo项目kvs建立了一个名为kvs-client的命令行键值存储客户端，和一个名为kvs-server的键值存储服务端，二者又都调用了一个名为kvs的库。客户端通过一个自定义协议与服务端通信。

kvs-server可执行文件支持以下命令行参数：

kvs-server [--addr IP-PORT] [--engine ENGINE-NAME]

启动服务端并开始监听进入的连接。--addr接受一个IP地址（可以是v4或v6）以及一个端口号，格式为IP:PORT。如果没有指定--addr，则默认监听127.0.0.1:4000。

如果指定了--engine，那么ENGINE-NAME必须是”kvs”（即使用内置引擎），或者是”sled”（即使用sled引擎）。如果这是首次运行程序（没有以前保存的数据），那么默认值是”kvs”；如果以前有保存的数据，那么默认使用已经在用的引擎。如果以前保存的数据与当前选择的引擎不同，则打印一个错误并以非零的退出码退出。

如果套接字绑定失败，或ENGINE-NAME无效，或IP-PORT不能解析为一个地址，则打印一个错误并以非零的退出码退出。
kvs-server -V

打印版本信息。

kvs-client可执行文件支持以下命令行参数：

kvs-client set <KEY> <VALUE> [--addr IP-PORT]

将一个字符串键的值设置为一个字符串。

--addr接受一个IP地址（可以是v4或v6）以及一个端口号，格式为IP:PORT。如果没有指定--addr，则在默认127.0.0.1:4000上连接。

如果服务器出错，或者IP-PORT不能解析为一个地址，则打印错误并以非零的退出码退出。
kvs-client get <KEY> [--addr IP-PORT]

获取一个给定的字符串键的字符串值。

--addr接受一个IP地址（可以是v4或v6）以及一个端口号，格式为IP:PORT。如果没有指定--addr，则在默认127.0.0.1:4000上连接。

如果服务器出错，或者IP-PORT不能解析为一个地址，则打印错误并以非零的退出码退出。
kvs-client rm <KEY> [--addr IP-PORT]

删除一个给定的字符串键。

--addr接受一个IP地址（可以是v4或v6）以及一个端口号，格式为IP:PORT。如果没有指定--addr，则在默认127.0.0.1:4000上连接。

如果服务器出错，或者IP-PORT不能解析为一个地址，则打印错误并以非零的退出码退出。
kvs-client -V

打印版本信息。

所有错误信息都应打印到stderr。

kvs库包含四种类型。

KvsClient - 为kvs-client实现与kvs-server通信所需的功能。
KvsServer - 为kvs-server实现响应kvs-client请求的功能。
KvsEngine trait - 定义了KvsServer需要调用的存储接口。
KvStore - 手动实现KvsEngine trait
SledKvsEngine - 为sled存储引擎实现KvsEngine trait。

KvsClient和KvsServer的设计由你决定，并将使用你自定义的网络协议来通信。本项目的测试套件并不直接调用这两种类型，而仅是通过CLI测试它们。

KvsEngine trait支持以下方法：

KvsEngine::set(&mut self, key: String, value: String) -> Result<()>

将一个字符串键的值设置为一个字符串。

如果值没有成功写入，则返回错误。
KvsEngine::get(&mut self, key: String) -> Result<Option<String>>

获取一个字符串键的字符串值。如果键不存在，返回None。

如果没有成功读取该值，则返回错误。
KvsEngine::remove(&mut self, key: String) -> Result<()>

删除一个给定的字符串键。

如果键不存在或值未被成功读取，则返回错误。

当为一个键设置值时，KvStore会把set命令写入磁盘的顺序日志。当删除一个键时，KvStore将rm命令写入日志。在启动时，日志中的命令将被重新求值，各键的最新一条设置命令的日志指针（文件偏移量）将会被保存在内存索引中。

当用get命令检索一个键的值时，它会搜索索引，如果找到了，就从日志的相应位置中加载相应命令并求值。

当未压缩的日志条目的大小达到一个给定的阈值时，KvStore将其压缩成一个新的日志，删除多余的条目以释放磁盘空间。

项目设置#

继续你以前的项目，删除你以前的tests目录，把这本项目的tests目录复制到它的位置。这个项目应该包含一个名为kvs的库，以及两个可执行文件，kvs-server和kvs-client。

你需要在Cargo.toml中加入以下dev-dependencies依赖：

1
[dev-dependencies]
2
assert_cmd = "0.11"
3
criterion = "0.3"
4
predicates = "1.0.0"
5
rand = "0.6.5"
6
tempfile = "3.0.7"
7
walkdir = "2.2.7"

与以前的项目一样，添加足够的定义，以构建测试套件。

使测试套件编译通过只需要：

按上文要求在lib.rs中添加一个新模块mod engines；
创建新模块engines\mod.rs；
在新模块mod.rs中按要求加入KvsEngine trait的定义即可。
在lib.rs中公开引用这个trait即可：pub use engines::KvsEngine;

目录结构为：

1
.
2
├── Cargo.toml
3
├── src
4
│  ├── bin
5
│  │  └── kvs.rs
6
│  ├── engines
7
│  │  └── mod.rs
8
│  ├── error.rs
9
│  ├── kv.rs
10
│  └── lib.rs
11
└── tests
12
   ├── cli.rs
13
   └── kv_store.rs

目前cli.rs中的11个测试均未通过：

1
cargo test --test cli
2
...
3
running 11 tests
4
test cli_log_configuration ... FAILED
5
test cli_access_server_kvs_engine ... FAILED
6
test cli_wrong_engine ... FAILED
7
test client_cli_invalid_get ... FAILED
8
test client_cli_invalid_rm ... FAILED
9
test client_cli_invalid_set ... FAILED
10
test cli_access_server_sled_engine ... FAILED
11
test client_cli_invalid_subcommand ... FAILED
12
test client_cli_version ... FAILED
13
test server_cli_version ... FAILED
14
test client_cli_no_args ... FAILED
15
...

kv_store.rs中的均通过测试：

1
cargo test --test kv_store
2
running 6 tests
3
test remove_non_existent_key ... ok
4
test remove_key ... ok
5
test get_stored_value ... ok
6
test get_non_existent_value ... ok
7
test overwrite_value ... ok
8
test compaction ... ok

第1部分：解析命令行#

本项目中的命令行解析同之前的项目相比没有多少新功能。kvs-client二进制文件接受与之前项目相同的命令行参数。而kvs-server有一组自己的命令行参数需要处理，就像上文描述的那样。

处理kvs-server的命令行参数。

kvs-server的参数比较简单，依然使用clap的derive特性编写，由于Rust已经为&str实现了ToSocketAddrs特性，Args中直接使用SocketAddr类型会自动触发字符串到IP地址的转换：

1
use clap::{Parser};
2
use kvs::{Result};
3
use std::net::SocketAddr;
4

5
const DEFAULT_LISTENING_ADDRESS: &str = "127.0.0.1:4000";
6
const DEFAULT_ENGINE: Engine = Engine::kvs;
7

8
#[allow(non_camel_case_types)]
9
#[derive(clap::ArgEnum, Clone, Debug)]
10
enum Engine {
11
   kvs,
12
   sled,
13
}
14

15
impl FromStr for Engine {
16
    type Err = KvsError;
17

18
    fn from_str(s: &str) -> Result<Self> {
19
        match s {
20
            "kvs" => {
21
                Ok(Engine::kvs)
22
            },
23
            "sled" => {
24
                Ok(Engine::sled)
25
            }
26
            _ => {
27
                Err(KvsError::UnexpectedEngineType)
28
            }
29
        }
30
    }
31
}
32

33
#[derive(Parser, Debug)]
34
#[clap(name = "kvs-server")]
35
#[clap(author, version, about="server of key value storage", long_about=None)]
36
struct Args {
37
    /// [IPADDR:PORT] of server.
38
    #[clap(long, default_value=DEFAULT_LISTENING_ADDRESS)]
39
    addr: SocketAddr,
40

41
    /// backend engine of key value storage.
42
    #[clap(arg_enum, long, required(false))]
43
    engine: Option<Engine>,
44
}
45

46

47
fn main() -> Result<()> {
48
    let args = Args::parse();
49
    println!("{:?} {:?}", args.addr, args.engine);
50

51
    Ok(())
52
}

kvs-client需要在子命令后面再添加可选参数（IP地址），所以可以使用clap的builder编写模式增加可读性。当然，也可以继续使用derive特性编写更简略的代码：

1
// builder模式
2

3
use clap::{Command, arg, crate_version};
4
use kvs::{Result};
5
use std::net::SocketAddr;
6

7
const DEFAULT_LISTENING_ADDRESS: &str = "127.0.0.1:4000";
8

9
fn cli() -> Command<'static> {
10
    Command::new("kvs-client")
11
        .about("client of key value storage")
12
        .version(crate_version!())
13
        .subcommand_required(true)
14
        .disable_help_subcommand(true)
15
        .args_conflicts_with_subcommands(true)
16
        .subcommand(
17
            Command::new("get")
18
                .about("Get the string value of a given string key")
19
                .args(&[
20
                    arg!(<KEY> "A string key"),
21
                    arg!(--addr <"IP:PORT"> "Set the server address as IP:PORT").required(false).default_value(DEFAULT_LISTENING_ADDRESS)
22
                ])
23
        )
24
        .subcommand(
25
            Command::new("set")
26
                .about("Set the value of a string key to a string")
27
                .args(&[
28
                    arg!(<KEY> "A string key"),
29
                    arg!(<VALUE> "The string value of the key").required(true),
30
                    arg!(--addr <"IP:PORT"> "Set the server address as IP:PORT").required(false).default_value(DEFAULT_LISTENING_ADDRESS)
31
                ])
32
        )
33
        .subcommand(
34
            Command::new("rm")
35
                .about("Remove a given string key")
36
                .args(&[
37
                    arg!(<KEY> "A string key"),
38
                    arg!(--addr <"IP:PORT"> "Set the server address as IP:PORT").required(false).default_value(DEFAULT_LISTENING_ADDRESS)
39
                ])
40
        )
41
}
42

43
fn main() -> Result<()>{
44
    let matches = cli().get_matches();
45

46
    match matches.subcommand() {
47
        Some(("get", sub_matches)) => {
48
            println!("get {} {}", sub_matches.value_of("KEY").unwrap(), sub_matches.value_of("addr").unwrap());
49
            let key = sub_matches.value_of("KEY").unwrap();
50
            let addr: SocketAddr = sub_matches.value_of("addr").unwrap().parse()?;
51
        },
52
        Some(("set", sub_matches)) => {
53
            println!("set {} {} {}", sub_matches.value_of("KEY").unwrap(), sub_matches.value_of("VALUE").unwrap(), sub_matches.value_of("addr").unwrap());
54
            let key = sub_matches.value_of("KEY").unwrap();
55
            let value = sub_matches.value_of("VALUE").unwrap();
56
            let addr: SocketAddr = sub_matches.value_of("addr").unwrap().parse()?;
57
        }
58
        Some(("rm", sub_matches)) => {
59
            println!("rm {} {}", sub_matches.value_of("KEY").unwrap(), sub_matches.value_of("addr").unwrap());
60
            let key = sub_matches.value_of("KEY").unwrap();
61
            let addr: SocketAddr = sub_matches.value_of("addr").unwrap().parse()?;
62
        },
63

64
        _ => unreachable!(), // If all subcommands are defined above, anything else is unreachabe!()
65

66
    }
67

68
    Ok(())
69
}
70

71
// derive模式
72

73
use clap::{Parser, Subcommand};
74
use kvs::{KvStore, Result};
75
use std::process::exit;
76
use std::env::current_dir;
77
use std::net::SocketAddr;
78

79
const DEFAULT_LISTENING_ADDRESS: &str = "127.0.0.1:4000";
80

81
#[derive(Parser)]
82
#[clap(name = "kvs")]
83
#[clap(author, version, about="an in-memory key/value store", long_about=None, args_conflicts_with_subcommands(true))]
84
struct Cli {
85
    #[clap(subcommand)]
86
    command: Commands,
87
}
88

89
#[derive(Subcommand)]
90
enum Commands {
91
    // Set the value of a string key to a string
92
    #[clap(arg_required_else_help = true)]
93
    set {
94
        // A string key
95
        key: String,
96
        // The string value of the key
97
        value: String,
98
        #[clap(long, required(false), default_value(DEFAULT_LISTENING_ADDRESS), help="Set the server address as [IP:PORT]")]
99
        addr: SocketAddr
100
    },
101

102
    // Get the string value of a given string key
103
    #[clap(arg_required_else_help = true)]
104
    get {
105
        // A string key
106
        key: String,
107
        #[clap(long, required(false), default_value(DEFAULT_LISTENING_ADDRESS), help="Set the server address as [IP:PORT]")]
108
        addr: SocketAddr
109
    },
110

111
    // Remove a given key
112
    #[clap(arg_required_else_help = true)]
113
    rm {
114
        // A string key
115
        key: String,
116
        #[clap(long, required(false), default_value(DEFAULT_LISTENING_ADDRESS), help="Set the server address as [IP:PORT]")]
117
        addr: SocketAddr
118
    },
119
}
120

121
fn main() -> Result<()> {
122
    let cli= Cli::parse();
123
    match cli.command {
124
        Commands::get { key, addr } => {},
125
        Commands::set { key, value, addr } => {},
126
        Commands::rm { key, addr } => {},
127
    }
128

129
    Ok(())
130
}

第2部分：日志记录#

生产环境下的服务器应用程序应具有功能强大且可配置的日志记录。因此，现在我们应为kvs-server添加日志功能，并在后续功能实现的过程中记录有用的信息。在开发过程中，通常使用debug!和trace!级日志来打印调试信息。

Rust中有两个好用的日志系统：log和slog。两者均为不同级别的日志输出类似的宏，如error!、info!等。两者都是可扩展的，支持不同的后端，以输出日志到控制台、日志文件或系统日志等。

最主要的区别是，log相当简单，只记录格式化的字符串；slog功能丰富，支持“结构化日志”，其日志条目是以容易解析的格式类型化和序列化的。

log可以追溯到Rust最早的时候，它是编译器的一部分，然后是标准库的一部分，最后作为独立的crate发布。它由Rust项目组维护。slog是较新的crate，独立维护。两者均已被广泛使用。

对于这两种系统，都需要选择一个“接收器”crate，用于将日志发送到该接收器以进行显示或存储。

阅读这两个系统，选择一个对你喜欢的，将它们作为依赖项添加，然后修改kvs-server以在传递命令行参数之前初始化日志。将日志设置为输出到stderr（将日志发送到其他地方也可以，但日志必须也发送到stderr以通过本项目的测试）。

在启动时日志记录服务器端的版本信息，还要记录配置信息。目前指的是IP地址和端口，以及存储引擎的名称。

因为第一次使用日志功能，还是先用简单的log入门，在Cargo.toml里引入log和env_logger（用于配置日志）。然后按要求输出日志即可：

1
// ...
2
fn main() -> Result<()> {
3
    env_logger::builder().filter_level(LevelFilter::Debug).init();
4
    let mut args = Args::parse();
5
//...
6
    run(args)
7
}
8

9
fn run(args: Args) -> Result<()> {
10
    let engine = args.engine.unwrap_or(DEFAULT_ENGINE);
11
    info!("kvs-server {}", env!("CARGO_PKG_VERSION"));
12
    info!("Storage engine: {:?}", engine);
13
    info!("Listening on {}", args.addr);
14
// ...
15
}
16
// ...

第3部分：客户端-服务端网络设置#

接下来，我们将搭建网络模块。对于本项目，您将使用std::net中使用最基础的TCP/IP网络API：TcpListener和TcpStream。

对于本项目，服务器是同步、单线程的。这意味着您将监听一个socket，接受一个连接，然后一次执行/响应一条命令。在未来，我们将在实现异步、多线程和高性能数据库的过程中，多次重新回味这一设定。

考虑一下您的手动测试工作流程。现在有两个可执行文件要处理，您需要一种同时运行它们的方法。您可能和我们一样，同时使用两个终端，一个运行cargo run --bin kvs-server ，服务端将一直运行到你按下CTRL-D；另一个运行cargo run --bin kvs-client。

这是使用日志进行调试的好机会。继续记录有关每个已接受连接的信息。

在考虑协议之前，先修改kvs-server来监听和接受连接，在修改kvs-client来发起连接。

是时候把KvStore变成真正的KV引擎了，把为KvStore实现的get、set、remove拿出来，用于为KvStore实现KvsEngine特性：

1
//...
2
impl KvsEngine for KvStore {
3
//...
4
    fn set(&mut self, key: String, value: String) -> Result<()> {
5
        let cmd = Command::set(key, value);
6
        let pos = self.writer.pos;
7
        serde_json::to_writer(&mut self.writer, &cmd)?;
8
        self.writer.flush()?;
9
        if let Command::Set { key, .. } = cmd {
10
            if let Some(old_cmd) = self
11
                .index
12
                .insert(key, (self.current_gen, pos..self.writer.pos).into())
13
            {
14
                self.uncompacted += old_cmd.len;
15
            }
16
        }
17

18
        if self.uncompacted > COMPACTION_THRESHOLD {
19
            self.compact()?;
20
        }
21
        Ok(())
22
    }
23
//...
24
    fn get(&mut self, key: String) -> Result<Option<String>> {
25
        if let Some(cmd_pos) = self.index.get(&key) {
26
            let reader = self
27
                .readers
28
                .get_mut(&cmd_pos.gen)
29
                .expect("Cannot find log reader");
30
            reader.seek(SeekFrom::Start(cmd_pos.pos))?;
31
            let cmd_reader = reader.take(cmd_pos.len);
32
            if let Command::Set { value, .. } = serde_json::from_reader(cmd_reader)? {
33
                Ok(Some(value))
34
            } else {
35
                Err(KvsError::UnexpectedCommandType)
36
            }
37
        } else {
38
            Ok(None)
39
        }
40
    }
41
//...
42
    fn remove(&mut self, key: String) -> Result<()> {
43
        if self.index.contains_key(&key) {
44
            let cmd = Command::remove(key);
45
            serde_json::to_writer(&mut self.writer, &cmd)?;
46
            self.writer.flush()?;
47
            if let Command::Remove { key } = cmd {
48
                let old_cmd = self.index.remove(&key).expect("key not found");
49
                self.uncompacted += old_cmd.len;
50
            }
51
            Ok(())
52
        } else {
53
            Err(KvsError::KeyNotFound)
54
        }
55
    }
56

57
}

在实现客户端-服务端之前，可以先定义用于序列化的枚举类型，作为客户端与服务端之间的通信协议（在后面编写服务端和客户端时，可以方便的将这些支持序列化的类型放在TcpStream上进行读写）：

1
use serde::{Serialize, Deserialize};
2

3
#[derive(Debug, Serialize, Deserialize)]
4
pub enum Request {
5
    Get { key: String },
6
    Set { key: String, value: String },
7
    Remove { key: String },
8
}
9

10
#[derive(Debug, Serialize, Deserialize)]
11
pub enum GetResponse {
12
    Ok(Option<String>),
13
    Err(String),
14
}
15

16
#[derive(Debug, Serialize, Deserialize)]
17
pub enum SetResponse {
18
    Ok(()),
19
    Err(String),
20
}
21

22
#[derive(Debug, Serialize, Deserialize)]
23
pub enum RemoveResponse {
24
    Ok(()),
25
    Err(String),
26
}

然后在server.rs中实现KvsServer：

1
use std::{net::{ToSocketAddrs, TcpListener, TcpStream}, io::{BufReader, BufWriter, Write}};
2
use log::{error, debug};
3
use serde_json::Deserializer;
4
use crate::{KvsEngine, Result, common::{Request, GetResponse, SetResponse, RemoveResponse}};
5

6
/// The server of a key value store.
7
pub struct KvsServer<E: KvsEngine> {
8
    engine: E,
9
}
10

11
impl<E: KvsEngine> KvsServer<E>  {
12
    /// Create a `KvsServer` with a given storage engine.
13
    pub fn new(engine: E) -> Self {
14
        KvsServer { engine }
15
    }
16

17
    /// Run the server listening on the given address
18
    pub fn run(&mut self, addr: impl ToSocketAddrs) -> Result<()> {
19
        let listener = TcpListener::bind(addr)?;
20
        for stream in listener.incoming() {
21
            match stream {
22
                Ok(stream) => {
23
                    if let Err(e) = self.serve(stream) {
24
                        error!("Error on serving client: {}", e);
25
                    }
26
                }
27
                Err(e) => error!("Connection failed: {}", e),
28
            }
29
        }
30
        Ok(())
31
    }
32

33
    fn serve(&mut self, tcp: TcpStream) -> Result<()> {
34
        let peer_addr = tcp.peer_addr()?;
35
        let reader = BufReader::new(&tcp);
36
        let mut writer = BufWriter::new(&tcp);
37
        let req_reader = Deserializer::from_reader(reader).into_iter::<Request>();
38

39
        macro_rules! send_resp {
40
            ($resp:expr) => {{
41
                let resp = $resp;
42
                serde_json::to_writer(&mut writer, &resp)?;
43
                writer.flush()?;
44
                debug!("Response sent to {}: {:?}", peer_addr, resp);
45
            };};
46
        }
47

48
        for req in req_reader {
49
            let req = req?;
50
            debug!("Receive request from {} {:?}", peer_addr, req);
51
            match req {
52
                Request::Get { key } => send_resp!(
53
                    match self.engine.get(key) {
54
                        Ok(value) => { GetResponse::Ok(value) },
55
                        Err(e) => { GetResponse::Err(format!("{}", e)) },
56
                    }
57
                ),
58
                Request::Set { key, value } => send_resp!(
59
                    match self.engine.set(key, value) {
60
                        Ok(_) => SetResponse::Ok(()),
61
                        Err(e) => SetResponse::Err(format!("{}", e)),
62
                    }
63
                ),
64
                Request::Remove { key } => send_resp!(
65
                    match self.engine.remove(key) {
66
                        Ok(_) => RemoveResponse::Ok(()),
67
                        Err(e) => RemoveResponse::Err(format!("{}", e)),
68
                    }
69
                ),
70
            }
71
        }
72

73
        Ok(())
74
    }
75
}

其中send_resp!宏的作用与下面的代码相同，编写这个宏可以减少重复：

1
// 使用代码
2
Request::Get { key } => {
3
    let resp = match self.engine.get(key) {
4
        Ok(value) => { GetResponse::Ok(value) },
5
        Err(e) => { GetResponse::Err(format!("{}", e)) },
6
    };
7
    serde_json::to_writer(writer, &resp);
8
    writer.flush()?;
9
    debug!("Response sent to {}: {:?}", peer_addr, resp);
10
},
11
// 使用宏
12
macro_rules! send_resp {
13
    ($resp:expr) => {{
14
        let resp = $resp;
15
        serde_json::to_writer(&mut writer, &resp)?;
16
        writer.flush()?;
17
        debug!("Response sent to {}: {:?}", peer_addr, resp);
18
    };};
19
}
20

21
Request::Get { key } => send_resp!(
22
    match self.engine.get(key) {
23
        Ok(value) => { GetResponse::Ok(value) },
24
        Err(e) => { GetResponse::Err(format!("{}", e)) },
25
    }
26
),

在kvs-server.rs中使用KvsServer，此时可以将日志等级调整为Debug以方便调试：

1
use clap::Parser;
2
use kvs::{Result, KvsError, KvStore, KvsEngine, KvsServer};
3
use std::env::current_dir;
4
use std::fs;
5
use std::net::SocketAddr;
6
use std::process::exit;
7
use log::LevelFilter;
8
use log::{error, info, warn};
9
use std::str::FromStr;
10

11

12
const DEFAULT_LISTENING_ADDRESS: &str = "127.0.0.1:4000";
13
const DEFAULT_ENGINE: Engine = Engine::kvs;
14

15
#[allow(non_camel_case_types)]
16
#[derive(clap::ArgEnum, Copy, Clone, Debug, PartialEq, Eq)]
17
enum Engine {
18
   kvs,
19
   sled,
20
}
21

22
impl FromStr for Engine {
23
    type Err = KvsError;
24

25
    fn from_str(s: &str) -> Result<Self> {
26
        match s {
27
            "kvs" => {
28
                Ok(Engine::kvs)
29
            },
30
            "sled" => {
31
                Ok(Engine::sled)
32
            }
33
            _ => {
34
                Err(KvsError::UnexpectedEngineType)
35
            }
36
        }
37
    }
38
}
39

40
#[derive(Parser, Debug)]
41
#[clap(name = "kvs-server")]
42
#[clap(author, version, about="server of key value storage", long_about=None)]
43
struct Args {
44
    /// [IPADDR:PORT] of server.
45
    #[clap(long, default_value=DEFAULT_LISTENING_ADDRESS)]
46
    addr: SocketAddr,
47

48
    /// backend engine of key value storage.
49
    #[clap(arg_enum, long, required(false))]
50
    engine: Option<Engine>,
51
}
52

53

54
fn main() -> Result<()>{
55
    env_logger::builder().filter_level(LevelFilter::Debug).init();
56
    let mut args = Args::parse();
57

58
    let res = current_engine().and_then(move |curr_engine| {
59
        if args.engine.is_none() {
60
            args.engine = curr_engine;
61
        }
62
        if curr_engine.is_some() && args.engine != curr_engine {
63
            error!("Wrong engine!");
64
            exit(1);
65
        }
66
        run(args)
67
    });
68
    if let Err(e) = res {
69
        error!("{}", e);
70
        exit(1);
71
    }
72
    Ok(())
73
}
74

75
fn run(args: Args) -> Result<()> {
76
    let engine = args.engine.unwrap_or(DEFAULT_ENGINE);
77
    info!("kvs-server {}", env!("CARGO_PKG_VERSION"));
78
    info!("Storage engine: {:?}", engine);
79
    info!("Listening on {}", args.addr);
80

81
    // write engine to engine file
82
    fs::write(current_dir()?.join("engine"), format!("{:?}", engine))?;
83

84
    match engine {
85
        Engine::kvs => run_with_engine(KvStore::open(current_dir()?)?, args.addr),
86
        Engine::sled => Ok(()), //TODO
87
    }
88
}
89

90
fn run_with_engine(engine: impl KvsEngine, addr: SocketAddr) -> Result<()> {
91
    let mut server = KvsServer::new(engine);
92
    server.run(addr)
93
}
94

95
fn current_engine() -> Result<Option<Engine>> {
96
    let engine = current_dir()?.join("engine");
97
    if !engine.exists() {
98
        return Ok(None);
99
    }
100

101
    match fs::read_to_string(engine)?.parse() {
102
        Ok(engine) => Ok(Some(engine)),
103
        Err(e) => {
104
            warn!("The content of engine file is invalid: {}", e);
105
            Ok(None)
106
        }
107
    }
108
}

接下来在client.rs中实现KvsClient：

1
use std::{io::{BufReader, BufWriter, Write}, net::{TcpStream, ToSocketAddrs}};
2
use crate::{Result, common::{Request, GetResponse, SetResponse, RemoveResponse}, KvsError};
3
use serde::Deserialize;
4
use serde_json::{Deserializer, de::IoRead};
5

6
/// Key value store client
7
pub struct KvsClient {
8
    reader: Deserializer<IoRead<BufReader<TcpStream>>>,
9
    writer: BufWriter<TcpStream>,
10
}
11

12
impl KvsClient {
13
    /// Connect to `addr` to access `KvsServer`.
14
    pub fn connect(addr: impl ToSocketAddrs) -> Result<Self> {
15
        let tcp_reader = TcpStream::connect(addr)?;
16
        let tcp_writer = tcp_reader.try_clone()?;
17
        Ok(KvsClient{
18
            reader: Deserializer::from_reader(BufReader::new(tcp_reader)),
19
            writer: BufWriter::new(tcp_writer),
20
        })
21
    }
22

23
    /// Get the value of a given key from the server.
24
    pub fn get(&mut self, key: String) -> Result<Option<String>> {
25
        serde_json::to_writer(&mut self.writer, &Request::Get { key })?;
26
        self.writer.flush()?;
27
        let resp = GetResponse::deserialize(&mut self.reader)?;
28
        match resp {
29
            GetResponse::Ok(value) => Ok(value),
30
            GetResponse::Err(msg) => Err(KvsError::StringError(msg)),
31
        }
32
    }
33

34
    /// Set the value of a string key in the server.
35
    pub fn set(&mut self, key: String, value: String) -> Result<()> {
36
        serde_json::to_writer(&mut self.writer, &Request::Set { key, value })?;
37
        self.writer.flush()?;
38
        let resp = SetResponse::deserialize(&mut self.reader)?;
39
        match resp {
40
            SetResponse::Ok(_) => Ok(()),
41
            SetResponse::Err(msg) => Err(KvsError::StringError(msg)),
42
        }
43
    }
44

45
    /// Remove a string key in the server.
46
    pub fn remove(&mut self, key: String) -> Result<()> {
47
        serde_json::to_writer(&mut self.writer, &Request::Remove { key })?;
48
        self.writer.flush()?;
49
        let resp = RemoveResponse::deserialize(&mut self.reader)?;
50
        match resp {
51
            RemoveResponse::Ok(_) => Ok(()),
52
            RemoveResponse::Err(msg) => Err(KvsError::StringError(msg)),
53
        }
54
    }
55

56
}

并在kvs-client.rs中使用：

1
use clap::{Parser, Subcommand};
2
use kvs::{Result, KvsClient};
3
use std::net::SocketAddr;
4

5
const DEFAULT_LISTENING_ADDRESS: &str = "127.0.0.1:4000";
6

7
#[derive(Parser)]
8
#[clap(name = "kvs")]
9
#[clap(author, version, about="an in-memory key/value store", long_about=None, args_conflicts_with_subcommands(true))]
10
struct Cli {
11
    #[clap(subcommand)]
12
    command: Commands,
13
}
14

15
#[derive(Subcommand)]
16
#[allow(non_camel_case_types)]
17
enum Commands {
18
    // Set the value of a string key to a string
19
    #[clap(arg_required_else_help=true)]
20
    set {
21
        // A string key
22
        key: String,
23
        // The string value of the key
24
        value: String,
25

26
        #[clap(long, required(false), default_value(DEFAULT_LISTENING_ADDRESS), help="Set the server address as [IP:PORT]")]
27
        addr: SocketAddr
28
    },
29

30
    // Get the string value of a given string key
31
    #[clap(arg_required_else_help=true)]
32
    get {
33
        // A string key
34
        key: String,
35

36
        #[clap(long, required(false), default_value(DEFAULT_LISTENING_ADDRESS), help="Set the server address as [IP:PORT]")]
37
        addr: SocketAddr
38
    },
39

40
    // Remove a given key
41
    #[clap(arg_required_else_help=true)]
42
    rm {
43
        // A string key
44
        key: String,
45

46
        #[clap(long, required(false), default_value(DEFAULT_LISTENING_ADDRESS), help="Set the server address as [IP:PORT]")]
47
        addr: SocketAddr
48
    },
49
}
50

51
fn main() -> Result<()> {
52
    let cli= Cli::parse();
53
    match cli.command {
54
        Commands::get { key, addr } => {
55
            let mut client = KvsClient::connect(addr)?;
56
            if let Some(value) = client.get(key.to_string())? {
57
                println!("{}", value);
58
            } else {
59
                println!("Key not found");
60
            }
61
        },
62
        Commands::set { key, value, addr } => {
63
            let mut client = KvsClient::connect(addr)?;
64
            client.set(key.to_string(), value.to_string())?;
65
        }
66
        Commands::rm { key, addr } => {
67
            let mut client = KvsClient::connect(addr)?;
68
            client.remove(key)?;
69
        }
70
    }
71

72
    Ok(())
73
}

此时一个终端打开cargo run --bin kvs-server，另一个终端运行cargo run --bin kvs-client -- get key，就可以从网络上将“训练课程2”中存入的键: 值取出来了：

1
# server
2
(base) ➜  kvs git:(master) ✗ cargo run --bin kvs-server
3
   Compiling kvs v0.1.0 (/Users/zealot/RustProjects/pingcap-projects/kvs)
4
    Finished dev [unoptimized + debuginfo] target(s) in 1.14s
5
     Running `target/debug/kvs-server`
6
[2022-06-03T10:38:45Z INFO  kvs_server] kvs-server 0.1.0
7
[2022-06-03T10:38:45Z INFO  kvs_server] Storage engine: kvs
8
[2022-06-03T10:38:45Z INFO  kvs_server] Listening on 127.0.0.1:4000
9
[2022-06-03T10:38:49Z DEBUG kvs::server] Receive request from 127.0.0.1:55017 Get { key: "键" }
10
[2022-06-03T10:38:49Z DEBUG kvs::server] Response sent to 127.0.0.1:55017: Ok(Some("值"))
11

12
# client
13
(base) ➜  kvs git:(master) ✗ cargo run --bin kvs-client -- get 键
14
    Finished dev [unoptimized + debuginfo] target(s) in 0.08s
15
     Running `target/debug/kvs-client get '键'`
16
值

第4部分：通过网络处理命令#

在上一个项目中，您定义了数据库接受的命令，并学习了如何使用serde对日志进行序列化和反序列化。

现在是时候通过网络实现键/值存储，以通过前面实现的同步单进程模型远程执行命令。与您在上一个项目中创建日志的文件I/O一样，您将使用Read和Write特征序列化和流式传输命令。

现在您需要设计一个网络协议。有多种方法可以在TCP流上数读写数据，与上一个项目一样，也需要做出许多选择。该协议是基于文本还是二进制？数据如何从内存中的类型格式转换为字节流格式？每个连接传输一个还是多个请求？

请记住，该协议应支持执行成功并取回结果，也支持执行时遇到的错误，现在有两种错误：由您的存储引擎产生的错误，以及由网络通信产生的错误。

协议的所有细节都由您决定。测试套件不关心数据如何从一端到达另一端，只关心结果是否正确。

实现您的网络协议。

前面我们已经一并实现了该协议，需要注意的是，只要进行适当的定义，序列化的所有类型转换都是自动完成的。

第5部分：可插拔存储引擎#

您的数据库目前有一个由您实现的存储引擎KvStore。现在将添加第二个存储引擎。

这样做有多种原因：

不同的业务负载需要不同的性能特征。对于特定的业务负载，某些存储引擎可能比其他存储引擎工作得更好；
搭建了一个熟悉的框架来比较不同的后端；
给了我们一个创建和使用trait的理由；
给了我们一个编写比较基准的环境。

所以你要从KvStore接口中抽象出一个新的特征：KvsEngine。这是一个典型的重构，将现有代码逐步修改为新的形式。一般在重构时，我们希望将工作最小化分解，以便进行持续构建和修改。

这是您最终需要的API：

KvsEnginetrait具有与KvStore具有相同签名的get、set和remove方法。
KvStore实现了KvsEngine，不再有自己的get、set和remove方法。
KvsEngine有一个新的实现——SledKvsEngine。稍后您需要使用sled库填充其get和set方法。

如果您的测试套件通过编译，则可能已经了编写了这些定义的签名，现在是时候实现定义的内容了。将重构分解为渐进式修改，并确保项目能够持续构建，并通过先前已经通过的测试，然后再继续下一步重构。

作为最后一步，您需要考虑当kvs-server使用一个引擎启动，然后进程被终结，接下来再使用不同的引擎重新启动时会发生什么。这种情况应当产生错误，你需要想清楚如何检测这种情况才能报告错误。测试cli_wrong_engine反映了这种情况。

本项目一开始，为了通过测试，我们已经实现了KvsEngine的签名，又在第3部分提前将KvStore的几个方法移动至KvsEngine的实现中。而在本项目需求说明部分，为了实现引擎切换报错，我们已经实现了启动时检测engine文件内容的功能：

若未检测到engine文件，则将本次使用的引擎写入engine文件（不指定时默认使用kvs）。
若检测到engine文件，当文件记录引擎与当前指定引擎不一致时，报错；一致时则继续执行。

在engine/sled.rs中为sled实现KvsEngine：

1
use sled::{Db, Tree};
2
use crate::{KvsEngine, Result, KvsError};
3

4
/// Wrapper of `sled::Db`
5
pub struct SledKvStore {
6
    db: Db,
7
}
8

9
impl SledKvStore {
10
    /// Creates a `SledKvsEngine` from `sled::Db`.
11
    pub fn new(db: Db) -> Self {
12
        Self { db }
13
    }
14
}
15

16
impl KvsEngine for SledKvStore {
17
    fn get(&mut self, key: String) -> Result<Option<String>> {
18
        let tree: &Tree = &self.db;
19
        Ok(
20
            tree.get(key)?
21
                .map(|iv| AsRef::<[u8]>::as_ref(&iv).to_vec())
22
                .map(String::from_utf8)
23
                .transpose()?
24
        )
25
    }
26

27
    fn set(&mut self, key: String, value: String) -> Result<()> {
28
        let tree: &Tree = &self.db;
29
        tree.insert(key, value.into_bytes()).map(|_| ())?;
30
        tree.flush()?;
31
        Ok(())
32
    }
33

34
    fn remove(&mut self, key: String) -> Result<()> {
35
        let tree: &Tree = &self.db;
36
        tree.remove(key)?.ok_or(KvsError::KeyNotFound)?;
37
        tree.flush()?;
38
        Ok(())
39
    }
40
}

然后在kvs-server.rs中调用SledKvStore，以补全最后一个//TODO：

1
// ...
2
match engine {
3
    Engine::kvs => run_with_engine(KvStore::open(current_dir()?)?, args.addr),
4
    Engine::sled => run_with_engine(SledKvStore::new(sled::open(current_dir()?)?), args.addr),
5
}
6
//...

第6部分：基准测试#

随着课程的进展，我们将越来越关注数据库的性能，探索不同架构对性能产生的影响。我们希望您不要局限于本文所展示的内容，多尝试自己的优化。

性能比对需要基准测试，目前有很多方法可以对数据库进行基准测试：例如使用ycsb和sysbench等标准测试套件。Rust内置了一些基准测试工具，我们将从此处入手。

Cargo支持使用cargo bench进行基准测试。基准测试可以使用Rust内置的基准测试工具编写，也可以使用外部工具编写。

为函数添加#[bench]属性即可启用内置基准测试套件。不过这一功能并不在Rust stable channel中提供，仅在the unstable book和test crate docs中有简略描述。尽管它在整个Rust生态中已被广泛使用—这些crates即使使用stable编译，也会用nightly进行基准测试。

尽管该系统实际上已被弃用—没有更新，且似乎永远不会被提升到stable channel。

不过，Rust也有更好的基准测试工具。本项目将使用criterion来对比kvs引擎与sled引擎的性能。

这这些基准测试工具的工作原理是定义一个基准测试函数，然后在该函数内循环迭代要执行的基准测试操作。基准测试工具将根据需要进行多次迭代，以获取具有统计意义的操作持续时间。

这是criterion指南中的这个基础示例：

1
fn criterion_benchmark(c: &mut Criterion) {
2
    c.bench_function("fib 20", |b| {
3
        b.iter(|| {
4
            fibonacci(20)
5
        });
6
    });
7
}

调用bench_function定义基准，调用iter定义为基准需要测试的代码。调用iter之前和之后的代码不计入基准测试时间。

通过创建一个名为benches/engine_benches.rs的文件来准备编写基准测试。同tests/tests.rs一样，cargo会自动找到这个文件并将其编译为基准测试。

首先编写以下基准测试：

kvs_write—使用kvs引擎，使用长度为1-100000字节的随机键和长度为1-100000字节的随机值写入100次；
sled_write—使用sled引擎，使用长度为1-100000字节的随机键和长度为1-100000字节的随机值写入100次；
kvs_read—使用kvs引擎，从先前写入的键中读取1000个值，键和值的长度是随机的；
sled_read—使用sled引擎，从先前写入的键中读取1000个值，键和值的长度是随机的；

（除了像上面描述的这样编写4个基准测试，您也可以用这2个引擎名作为参数编写参数化的基准测试，参见criterion手册）。

对于实现这些基准测试的技术细节，我们至少需要考虑三个因素：

哪些代码应该计时（写在基准测试循环内），哪些代码不应该计时（写在基准测试循环之外）？
尽管使用“随机”数字，如何使每次迭代执行相同操作。
在”read”的基准测试中，如何使用之前写入的同一组“随机”键读取值。

这些都是相互关联的：用作基准测试环境设置的代码应为非计时代码，此外还需要适当地重用随机数种子以复现测试环境。

在各种情况下，可能返回错误的操作都应该断言（使用assert!）它们没有返回错误；在”read”测试中，“get”操作应该断言找到了该键的值。

随机数可以使用randcrate生成。

编写完基准测试函数，就使用cargo bench运行它们。

完成上面的基准测试，对比kvs和sled的性能差距。

注意：请在没什么负载的主机上运行基准测试。基准测试对其运行环境非常敏感，虽然criterion库会尽力修正这些“噪音”，但最好在没有其他活动进程的干净机器上进行基准测试。如果您有一台仅用于开发的备用机器，请使用它。如果没有，AWS或其他云实例可能会产生比本地桌面更一致的结果。

在Cargo.toml中声明基准测试：

1
[[bench]]
2
name = "engine_benches"
3
harness = false

在engine_benches.rs中编写基准测试：

1
use criterion::{Criterion, BatchSize, criterion_group, criterion_main};
2
use kvs::{KvStore, KvsEngine, SledKvStore};
3
use rand::{rngs::SmallRng, SeedableRng, Rng};
4
use sled;
5
use tempfile::TempDir;
6

7
fn set_benches(c: &mut Criterion) {
8
    let mut group = c.benchmark_group("set_benches");
9
    group.bench_function("kvs", |b| {
10
        b.iter_batched(
11
            || {
12
            let temp_dir = TempDir::new().unwrap();
13
            (KvStore::open(temp_dir.path()).unwrap(), temp_dir)
14
            },
15
            |(mut store, _temp_dir)| {
16
                for i in 1..(1 << 12) {
17
                    store.set(format!("key-{}", i), format!("value-{}", i)).unwrap();
18
                }
19
            },
20
            BatchSize::SmallInput
21
        );
22
    });
23
    group.bench_function("sled", |b| {
24
        b.iter_batched(
25
            || {
26
            let temp_dir = TempDir::new().unwrap();
27
            (SledKvStore::new(sled::open(&temp_dir).unwrap()), temp_dir)
28
            },
29
            |(mut store, _temp_dir)| {
30
                for i in 1..(1 << 12) {
31
                    store.set(format!("key-{}", i), format!("value-{}", i)).unwrap();
32
                }
33
            },
34
            BatchSize::SmallInput
35
        );
36
    });
37
}
38

39
fn get_benches(c: &mut Criterion) {
40
    let mut group = c.benchmark_group("get_benches");
41
    for i in &vec![8, 12, 16, 20] {
42
        group.bench_with_input(format!("kvs_{}", i), i, |b, i| {
43
            let temp_dir = TempDir::new().unwrap();
44
            let mut store = KvStore::open(temp_dir.path()).unwrap();
45
            for key_i in 1..(1 << i) {
46
                store.set(format!("key-{}", key_i), format!("value-{}", key_i)).unwrap();
47
            }
48
            let mut rng = SmallRng::from_seed([0; 16]);
49
            b.iter(|| {
50
                store.get(format!("key-{}", rng.gen_range(1, 1 << i))).unwrap();
51
            });
52
        });
53
    }
54
    for i in &vec![8, 12, 16, 20] {
55
        group.bench_with_input(format!("sled_{}", i), i, |b, i| {
56
            let temp_dir = TempDir::new().unwrap();
57
            let mut store = SledKvStore::new(sled::open(&temp_dir).unwrap());
58
            for key_i in 1..(1 << i) {
59
                store.set(format!("key-{}", key_i), format!("value-{}", key_i)).unwrap();
60
            }
61
            let mut rng = SmallRng::from_seed([0; 16]);
62
            b.iter(|| {
63
                store.get(format!("key-{}", rng.gen_range(1, 1 << i))).unwrap();
64
            });
65
        });
66
    }
67

68
}
69

70
criterion_group!(benches, set_benches, get_benches);
71
criterion_main!(benches);

干得漂亮，朋友。休息一下吧。