В этой статье блога мы создадим сервер Redis на языке программирования Rust. Если вы новичок в Rust и хотите изучить язык, эта статья станет хорошей отправной точкой. В этой заметке не предполагается никаких предварительных знаний языка. Однако, если вы читали книгу по Rust или какой-нибудь начальный учебник, это будет замечательно. Если вы уже изучили некоторые основы rust, то эта статья идеально подходит для более глубокого погружения, чтобы создать полномасштабную программу.
Вторая часть этой серии блогов.
Окончательная реализация будет иметь полностью функционирующий сервер Redis с командами get и set. Код для этого блога вы можете получить здесь.
Это будет серия из двух постов.
- В первой статье мы создадим базовое однопоточное взаимодействие сервера с клиентом и сохранение и получение данных на сервере.
- Во втором посте мы реализуем многопоточное взаимодействие клиент-сервер и реализуем уведомление клиента о выключении с помощью каналов.
Необходимые условия
На вашем компьютере должен быть установлен rust, если нет, следуйте инструкциям в официальном руководстве по установке.
Настройка проекта
Здесь мы создадим шаблон приложения, установим все необходимые зависимости и настроим проект.
Чтобы начать, создайте новый проект rust с помощью команды.
$ cargo new my_redis
$ cd my_redis
Откройте проект my_redis в visual studio code или любом другом редакторе по вашему выбору, чтобы начать кодирование.
Вы получите 2 основных файла в коде. Первый — Cargo.toml и второй — /src/main.rs. Файл Cargo.toml содержит все метаданные проекта, такие как название проекта и, самое главное, зависимости, используемые в проекте. Позже мы добавим библиотеки в раздел dependecies этого файла.
Используемые зависимости
- Clap: Используется для чтения аргументов командной строки
- Tokio: Используется для асинхронной реализации.
main.rs имеет функцию main, которая является точкой входа нашей rust-программы всякий раз, когда мы выполняем команду cargo run из терминала.
Redis состоит из двух частей, первая — сервер, вторая — клиент. Клиент делает запрос к серверу, чтобы сохранить или получить данные. Сервер отвечает за сохранение данных на сервере и их получение. Таким образом, в командной строке/терминале у нас будет 2 окна, в каждом из которых запущены сервер и клиент.
По умолчанию исполняемым файлом является main.rs. Однако, поскольку нам нужно 2 исполняемых файла для сервера и клиента, мы поместим исполняемые файлы сервера и клиента в папку src/bin, которую мы скоро создадим. Эта папка содержит столько исполняемых файлов, сколько мы захотим. Это идеально подходит для нашего случая, так как нам нужно 2 отдельных исполняемых файла для сервера и клиента для запуска в терминале.
Итак, давайте создадим новую папку внутри src под названием bin. Затем создадим server.rs и client.rs внутри папки bin. Структура папок на этом этапе должна быть такой:
src/
│ ├── main.rs
│ └── bin/
│ └── server.rs
│ └── client.rs
Добавьте функцию main() в оба файла, чтобы избавиться от ошибки, которая жалуется на отсутствие основных функций. Поскольку это исполняемые файлы, они должны содержать основные функции, которые являются точкой входа для программы rust.
Чтобы запустить эти исполняемые файлы, просто выполните команду следующим образом:
$ cargo run --bin server
$ cargo run --bin client // in new terminal window
Аргумент --bin указывает Cargo, какой исполняемый файл запускать в случае, если исполняемых файлов несколько.
После этого шага нам не нужен файл main.rs, поскольку у нас есть отдельные исполняемые файлы, готовые к запуску. Поэтому продолжайте и удалите файл main.rs.
Взаимодействие сервера и клиента будет происходить с помощью сокетного соединения. Сервер будет слушать входящие запросы на сокетное соединение. Клиент будет пытаться соединиться с сервером.
Знакомство с Tokio
Tokio предоставляет асинхронные возможности и утилиты для работы в rust. В этой статье блога основное использование Tokio будет представлено в виде реализаций TcpListener и TcpStream из набора tokio::net. Tokio раскрыл функциональность .await для этих модулей, сделав их действительно асинхронными. Короче говоря, всякий раз, когда мы используем ключевое слово await в вызовах функций async, поток управления ожидает завершения выполнения функции.
Чтобы использовать tokio, добавьте зависимость tokio в файл Cargo.toml.
Файл Cargo.toml
[dependencies]
tokio = {version="1", features = ["full"]}
Реализация сервера
Теперь добавьте макрос #[tokio::main] прямо перед строкой async fn main(). Это преобразует функцию async main в синхронную функцию main и обернет код в функции main в блок async. Причина помещения кода в блок async заключается в том, что использование .await на функциях tokio потребует, чтобы главная функция была типа async. Но поскольку среда выполнения rust ожидает, что главная функция точки входа будет синхронной по своей природе, следовательно, #[tokio::main] делает это преобразование за нас.
#[tokio::main]
async fn main() {
println!("in main");
}
//gets transformed to
fn main() {
let rt = tokio::runtime::Runtime::new().unwrap()
rt.block_on(async { // block_on takes a block of code which is async in nature, so any `.await` inside the block will be fine.
println!("in main");
})
}
Всегда помните, если вы хотите использовать .await внутри функции или блока кода, то функция или блок кода должны быть типа async.
Далее мы хотим прослушивать входящие сокетные соединения в файле server.rs. Импортируйте TcpListener из tokio crate, используя use tokio::net::TcpListener.
Теперь в функции main используем метод TcpListen::bind() и передаем ему адрес с портом. Поскольку метод bind является асинхронным методом, мы будем использовать await, чтобы задержать выполнение, пока bind не вернет значение. Оператор ? — это специальный оператор, который возвращает значение, обернутое внутри Result или распространяет ошибку из функции на один уровень вверх. Подробнее о нем вы можете прочитать здесь.
Файл bin/server.rs.
pub async fn main() -> Result<(), std::io::Error> {
let listener = TcpListener::bind("127.0.0.1:8081").await?;
loop {
let (mut socket, _) = listener.accept().await?;
println!("connection accepted {:?}", socket);
}
Ok(())
}
В этот момент, пожалуйста, обратите внимание, что мы изменили тип возврата главной функции на Result это происходит из core::Result<T, E>. Ошибка распространяется от оператора ?, как объяснялось выше.
Если мы запустим сервер командой $ cargo run --bin server Он запустит программу и будет ждать в listener.accept() новых соединений. Таким образом, наш сервер прослушивает сокетные соединения в одном потоке.
Теперь перейдем к клиенту для установления сокетного соединения с сервером.
Поскольку клиент инициирует сокет-соединение, мы будем использовать TcpStream из tokio::net crate. Мы сделаем главную функцию async и аннотируем макросом #[tokio::main], то же самое мы делали при настройке сервера.
Для создания нового соединения используйте тот же адрес хоста и порт, что и для сервера.
pub async fn main() -> Result<(), std::io::Error> {
let mut stream = TcpStream::connect("127.0.0.1:8081").await?;
Ok(())
}
На этом этапе обратите внимание, что мы изменили тип возврата функции main на Result, что происходит из core::Result<T, E>. Ошибка распространяется от оператора ?, как объяснялось выше.
Давайте проверим это минимальное взаимодействие сервера и клиента. Мы ожидаем, что сервер остановится после принятия одного клиентского соединения. Он также выведет сообщение connection accepted.
Запустите сервер и клиент в отдельных терминальных окнах с командами, приведенными выше, и вы заметите, что сервер выведет что-то вроде этого.
connection accepted PollEvented { io: Some(TcpStream { addr: 127.0.0.1:8081, peer: 127.0.0.1:61643, fd: 10 }) }
Это означает, что наше соединение установлено и сокет готов начать отправку и прием данных.
Запись в сокет от клиента
Первое, что мы хотим сделать, это отправить команду от клиента к серверу. Для простоты мы будем отправлять данные в виде строки (на самом деле байта) на сокет.
Используем функцию write_all() на сокете, которая ожидает байты для записи данных на сокет.
stream.write_all(b"set foo bar").await?;
В rust любой строковый литерал с префиксом b перед ним делает его последовательностью байтов. Если вы хотите понять разницу между строковым литералом, строкой и последовательностью байтов, посмотрите этот замечательный пост.
Это все, что нам нужно для записи данных в сокетное соединение.
Чтение из сокета на сервере
Для чтения данных из сокета на стороне сервера мы будем использовать метод read_buf. Этот метод принимает в качестве аргумента буфер. Он копирует данные из сокета в указанный буфер.
Мы создадим новый буфер из BytesMut. Он берется из крейта bytes. Сначала добавьте зависимость bytes в [dependencies] в файл Cargo.toml.
[dependencies]
bytes = { version = "1" }
Затем импортируйте BytesMut из bytes crate в файле server.rs. используйте bytes::BytesMut.
Создадим мутабельный BytesMut с емкостью. Этот объект buf является мутабельным, так как в него можно будет записывать байты. Далее читаем данные из сокета в созданный буфер, используя метод read_buf.
Bytes::with_capacity() принимает количество байт, которое необходимо выделить для буфера. Этот буфер будет автоматически увеличиваться в размере по мере необходимости, но эффективнее начинать с большим размером буфера.
Например, «foo» займет 3 байта, а «ƒoo» — 4 байта. Давайте начнем с 1024 байт в качестве стандартного размера буфера.
let mut buf = BytesMut::with_capacity(1024);
socket.read_buf(&mut buf).await?;
println!("buffer {:?}", buf); // printing the data in buffer
Это все, что нам нужно для чтения данных из сокета в сервер. Давайте запустим сервер и клиент, чтобы проверить, работает ли наша программа так, как ожидалось.
Вы получите buffer b "set foo bar" в качестве вывода на терминале сервера.
Если вы дошли до этого момента, поздравляем вас, так как вы успешно передали данные с клиента на сервер, да еще и в rust.
Сопоставление команд
Теперь, когда мы прочитали данные от клиента, которые представляют собой команду типа "set foo bar". Первый элемент — это имя команды, а следующие два — ключ и значение.
В зависимости от того, что является командой, мы выполним либо get, либо set.
Итак, давайте начнем с преобразования буфера в вектор (своего рода массив) строк. Затем мы можем взять первую строку из вектора и получить команду из Command Enum. Мы создадим Command enum позже в этом посте.
Давайте создадим метод для этого преобразования под названием buffer_to_array. Создайте новый файл src/helper.rs и поместите эту функцию в этот файл. Поскольку этот файл находится в папке src, нам нужно создать еще один файл lib.rs и импортировать модуль Helper, который будет создан в результате создания нового файла cmd.rs. Добавьте pub mod helper; в файл src/lib.rs. Это раскрывает модуль helper для кода библиотеки. В rust, main.rs — это двоичный исполняемый файл, а lib.rs — разделяемый код. Если вы создаете библиотеку lib.rs — это место, куда вы будете импортировать все ваши файлы.
Файл src/helper.rs.
use bytes::Buf; // get_u8
fn buffer_to_array(buf: &mut BytesMut) -> Vec<String> {
let mut vec = vec![];
let length = buf.len();
let mut word = "".to_string();
for i in 0..length {
match buf.get_u8() {
b' ' => { // match for space
vec.push(word);
word = "".to_string();
}
other => {
// increase the word
word.push(other as char);
let new = word.clone();
if i == length - 1 {
vec.push(new);
}
}
}
}
vec
}
Метод get_u8 происходит из трейта use bytes::Buf. Поскольку BytesMut реализует Buf trait, мы можем get_u8 на BytesMut.
Метод buffer_at_array по сути разрывает строку на каждом встречающемся пробеле (b' ') и возвращает вектор строк vec<String>. Вызовем этот метод в файле server.rs в цикле для получения списка строк, например, ["set foo bar"].
use blog_redis::helper::buffer_to_array;
main() {
loop {
...// previous code
let attrs = buffer_to_array(&mut buf);
}
}
Следующим шагом будет определение команды. Для этого возьмите первую строку из вектора и сравните ее со списком известных команд. В нашем случае, Get или Set.
Создадим еще один файл src/cmd.rs. В этом файле мы создадим перечисление для команд. Вот набор команд, которые мы будем реализовывать.
pub enum Command {
Get,
Set,
Invalid,
}
Теперь мы попытаемся найти соответствие первой строке и, в зависимости от соответствия, вернуть правильную команду. Создадим функцию внутри src/cmd.rs как реализацию Command для возврата команды (enum) на основе совпадения первой строки.
impl Command {
pub fn get_command(str: &String) -> Command {
match str.as_bytes() {
b"set" => Command::Set,
b"get" => Command::Get,
_ => Command::Invalid,
}
}
}
Поскольку этот файл находится в папке src, давайте импортируем код в lib.rs.
Файл lib.rs
pub mod cmd;
pub use cmd::Command;
Давайте воспользуемся этой функцией для получения команды.
loop {
.. // previous code
let attrs = buffer_to_array(&mut buf);
let command = Command::get_command(&attrs[0]);
}
Сохранение ключа и значения
Для нашего случая нам необходимо хранить пары ключ-значение в базе данных. Структура данных, которая хорошо подходит для хранения пар ключ-значение, это HashMap. Любой новый запрос на соединение будет принимать ссылку на созданную базу данных и либо обновлять (устанавливать/записывать) данные в нее, либо получать (читать) данные из нее. При первом запуске сервера мы создадим экземпляр db и при любых новых подключениях будем использовать экземпляр db для чтения или записи данных в него.
Давайте создадим структуру для Database, которая будет хранить наши пары ключ-значение внутри Hashmap. Этот struct должен быть добавлен в собственный файл src/db.rs.
Давайте импортируем этот модуль в наш src/lib.rs, как мы это делали ранее для cmd.rs и helper.rs.
Файл lib.rs
pub mod db;
pub use db::Db;
Файл src/db.rs
use bytes::Bytes;
use std::collections::HashMap;
pub struct Db {
entries: HashMap<String, Bytes>
}
Далее создадим новый экземпляр этой структуры Db в главной функции server.rs. Это будет сделано до того, как сервер начнет принимать запросы на сокетное соединение в loop {}. Поскольку мы хотим сохранить объект базы данных во всей области действия главной функции файла server.rs. Если мы создадим db в цикле, то в конце первого цикла объект будет удален из области видимости. Это не то, чего мы хотим.
Файл server.rs
let db = Db::new();
loop {
let (socket, _) = listener.accept().await?;
}
Давайте создадим новую функцию process_query() в файле server.rs. Она будет принимать параметры command, attrs, socket, db и записывать атрибуты в объект db.
pub async fn main() -> Result<(), std::io::Error> {
// previous main fn code...
process_query(command, attrs, &mut socket, &mut db).await?;
}
async fn process_query(
command: Command,
attrs: Vec<String>,
socket: &mut TcpStream,
db: &mut Db,
) -> std::io::Result<()> {
match command {
Command::Get => {
Ok(())
}
Command::Set => {
let resp = db.write(&attrs);
match resp {
Ok(result) => {
println!("set result: {}", result);
socket.write_all(&result.as_bytes()).await?;
}
Err(_err) => {
socket.write_all(b"").await?;
}
}
Ok(())
}
Command::Invalid => Ok(()),
}
В методе process_query() мы будем сопоставлять команду для Command::Get, Command::Set и Command::Invalid.
В Command::Set мы будем вызывать db.write(&attrs). На данный момент этот метод write на db не реализован, поэтому давайте сделаем это.
Файл src/db.rs.
impl Db {
pub fn write(&mut self, arr: &[String]) -> Result<&str, &'static str> {
let key = &arr[1];
let value = &arr[2];
let val = value.clone();
let res: &Option<Bytes> = &self.entries.insert(String::from(key), Bytes::from(val));
match res {
Some(_res) => Ok("r Ok"),
None => Ok("Ok"),
}
}
Нам нужно клонировать ссылаемое значение, так как функция Bytes::from() ожидает переменную с временем жизни 'static, но value имеет неизвестное время жизни в контексте этой функции, так как определяется вне области видимости этой функции.
Функция insert() возвращает None, если ключ не присутствует в хэшмапе, и возвращает Some(old_value), если ключ уже присутствовал в хэшмапе. Мы возвращаем «r Ok» в случае, если ключ уже присутствует, и «Ok», если ключ еще не присутствовал в хэше. Это помогает на стороне клиента выдать соответствующее сообщение клиенту/потребителю.
Some и None являются типами на типе Option. Option используется, когда мы ожидаем необязательные значения, то есть либо будет значение с Some(value), либо None.
Теперь, возвращаясь к bin/server.rs, обратите внимание, что мы получаем результат db.write(&attrs) в объект result. Мы используем match для перехвата ответа от db.write(...) и записываем значение непосредственно в socket с помощью socket.write_all(&result).await?;. Это отправит байты по сокету для дальнейшего чтения клиентским сокетным соединением.
результат на стороне клиента
Файл bin/client.rs.
// previous main function code...
let mut buf = BytesMut::with_capacity(1024);
let _length = stream.read_buf(&mut buf).await?;
match std::str::from_utf8(&mut buf) {
Ok(resp) => {
if resp == "r Ok" {
println!("key updated");
} else if resp == "Ok" {
println!("key set");
}
}
Err(err: Utf8Error) => {
// failed to convert bytes into string slice
println!("error: {}", err);
}
}
Мы создадим buf типа BytesMut и прочитаем данные из сокета в буфер. Почему мы используем BytesMut, объяснено выше.
Далее мы сопоставим строковые данные из сокета и вернем ответ в зависимости от того, свежая ли это запись ключа или обновленный ключ. Мы также фиксируем ошибку от std::str::from_utf8(&mut buf), так как она вернет Utf8Error, если не сможет преобразовать байтовый фрагмент в строковый.
На данный момент наша реализация команды set в некоторой степени завершена. Мы не можем проверить это из командной строки.
Давайте проверим это, запустив клиента и надеясь увидеть сообщение об успехе. Это будет либо «ключ установлен», либо «ключ обновлен».
$ cargo run --bin client
// response should be "key set" in terminal.
Итак, мы установили пару ключ-значение на нашем сервере. Давайте попробуем реализовать команду get и получить соответствующее значение. Давайте реализуем это в следующем разделе.
Считывание данных с сервера
Чтобы получить значение, связанное с ключом, мы создадим еще одно сокетное соединение в main() из client.rs. На этот сокет мы запишем данные get foo.
Файл bin/client.rs.
let mut stream = TcpStream::connect("127.0.0.1:8081").await?;
stream.write_all(b"get foo").await?;
Далее мы изменим bin/server.rs, чтобы принять Command::Get в функции match в методе process_query.
Файл bin/server.rs.
match command {
Command::Get => {
let result = db.read(&attrs);
match result {
Ok(result) => {
socket.write_all(&result).await?;
}
Err(_err) => {
println!("no key found {:?}", _err);
socket.write_all(b"").await?;
}
}
Ok(())
}
}
Command::Set => {
// already implemented above
}
Сервер вернул ответ, записав в сокет ответ из базы данных. Далее мы обновим файл src/db.rs для реализации метода read().
pub fn read(&mut self, arr: &[String]) -> Result<&Bytes, &'static str> {
let key = &arr[1];
let query_result = self.entries.get(key);
if let Some(value) = query_result {
return Ok(value);
} else {
return Err("no such key found");
}
}
Возвращаемая ошибка представляет собой фрагмент строки str. Время жизни &'static обычно означает, что значение будет сохраняться до конца программы. Причина добавления &'static времени жизни к str заключается в том, что строка создается в этой области видимости функции и будет удалена, как только вызов функции завершится. Следовательно, добавление &'static помогает передать компилятору, что эта строка живет вечно. Время жизни в Rust — непростая концепция, для лучшего понимания посмотрите этот замечательный учебник.
Последним шагом будет чтение ответа в файле client.rs и вывод значения в командную строку.
let mut buf = BytesMut::with_capacity(1024);
let _length = stream.read_buf(&mut buf).await?;
println!("buffer: {:?}", &buf);
match std::str::from_utf8(&mut buf) {
Ok(resp) => {
if resp == "" {
println!("no such key found");
} else {
println!("value: {}", resp);
}
}
Err(_err) => {
println!("in errr");
}
}
Это все, что нам нужно для реализации базового однопоточного сервера redis с командами set и get.
Если вы запустите сервер, а затем клиента, то получите в терминале ответ value: bar. Ура!!!
Реализация CLAP
До сих пор мы жестко кодировали наши команды Redis в файле client.rs, но это не то, как клиент должен отправлять данные. Итак, давайте попробуем принять команду из командной строки/терминала.
Clap — это крейт парсера командной строки в rust. С его помощью принимать аргументы командной строки очень просто.
Прежде всего, добавьте зависимость в Cargo.toml.
clap = { version = "3.1.18", features = ["derive"] }
Теперь давайте импортируем это в файл client.rs и начнем добавлять необходимый код. Мы меняем способ передачи команды из файла клиента в файл сервера. Ранее мы создавали сокетные соединения из файла клиента. Теперь мы будем создавать только одно сокет-соединение на каждый запуск клиента в командной строке с аргументами командной строки.
Файл bin/client.rs.
use clap::{Parser, Subcommand};
#[derive(Parser, Debug)]
struct Cli {
#[clap(subcommand)]
command: Command,
}
#[derive(Subcommand, Debug)]
enum Command {
Get {
key: String,
},
Set {
key: String,
value: String,
},
}
main() {
let args = Cli::parse();
let mut stream = TcpStream::connect("127.0.0.1:8081").await.unwrap();
match args.command {
Command::Set { key, value } => {
stream.write_all(b"set").await?;
stream.write_all(b" ").await?;
stream.write_all(&key.as_bytes()).await?;
stream.write_all(b" ").await?;
stream.write_all(&value.as_bytes()).await?;
let mut buf = BytesMut::with_capacity(1024);
let _length = stream.read_buf(&mut buf).await?;
match std::str::from_utf8(&mut buf) {
Ok(resp) => {
if resp == "r Ok" {
println!("updated key");
} else if resp == "Ok" {
println!("key set");
}
}
Err(err) => {
// failed to convert bytes into string slice
println!("error: {}", err);
}
}
}
Command::Get { key } => {
stream.write_all(b"get").await?;
stream.write_all(b" ").await?;
stream.write_all(&key.as_bytes()).await?;
let mut buf = BytesMut::with_capacity(1024);
let _length = stream.read_buf(&mut buf).await?;
match std::str::from_utf8(&mut buf) {
Ok(resp) => {
if resp == "" {
println!("no such key found");
} else {
println!("key: {} => value: {}", key, resp);
}
}
Err(_err) => {
println!("in errr");
}
}
return Ok(());
}
}
}
В приведенном выше коде мы определяем clap с подкомандой, для этого мы создаем Enum команд с дополнительными параметрами и типами.
Мы выполняем простое сопоставление команды, введенной из командной строки, и записываем соответствующую команду в сокетное соединение для чтения и выполнения сервером.
Что дальше?
Текущая реализация представляет собой однопоточную программу. В действительности это не то, что нам нужно. Представьте, что на наш сервер Redis поступает 5000 запросов, текущая программа будет ждать завершения первого запроса, чтобы обработать следующий. Однако мы хотим, чтобы все запросы обрабатывались без ожидания. Этого можно достичь с помощью многопоточности. В следующем посте мы изменим эту однопоточную реализацию на многопоточную. Именно здесь Tokio сияет больше всего благодаря своим продвинутым функциям tokio::spawn и tokio::select.
Мы также реализуем механизм отключения с помощью каналов tokio, таких как broadcast и mpsc стратегии. Представьте, что есть 5000 активных соединений, что произойдет, если сервер будет выключен. Грациозная реализация выключения для всех потоков будет рассмотрена в следующем посте.
Счастливого программирования!