Learning Rust Crates
libraries to investigate
rayon
- install the crate, then
use rayon::prelude::*
- instantly speed up iterators with rayon by using par_iter() instead of iter()
serde
- use
#[derive(Serialize, Deserialize)] enable structs to be enriched by serde
- use
serde_json:to_string to serialize struct into json
tokio
tokio basics
- tokio is a toolkit for async runtimes and networking applications
- use spawn thread around socket process to create multithreading responses
use tokio::net::*;
use mini_redis::{Connection, Frame};
#[tokio::main]
async fn main() {
let listener = TcpListener::bind("localhost:6379").await.unwrap();
println!("mini_redis listening on port {:?}", listener.local_addr());
loop {
let (socket, _) = listener.accept().await.unwrap();
tokio::spawn(async move {
process(socket).await;
});
}
}
async fn process(socket: TcpStream) {
let mut connection = Connection::new(socket);
if let Some(frame) = connection.read_frame().await.unwrap() {
println!("{:?}", frame);
let response = Frame::Error("unimplemented".to_string());
connection.write_frame(&response).await.unwrap();
}
}
- when passing process functions to tokio’s spawn, under the hood “tasks” are created.
- tasks are very lightweight (64bytes of mem)
- tasks will be scheduled on the same or other thread
- feel free to spawn thousands / millions of tasks
- tasks should not reference memory with a shorter lifetime than
static
- use move in the async block to make passed values get moved
tokio shared state with sync mutex
- if you want to use data amongst threads, use synchronisation such as Arc
- using a synchronous mutex from within asynchronous code is fine as long as “contention” remains low
- contention in this case means different threads and tasks fighting to access the same (locked) value
- mutex guards should die before calling new asynchronous await functions. We can achieve this by creating blocks for the mutex guard:
async fn increment_and_do_stuff(mutex: &Mutex<i32>) {
{
let mut lock: MutexGuard<i32> = mutex.lock().unwrap();
*lock += 1;
} // lock goes out of scope here
do_something_async().await;
}
- the safest way to do something similar is to create a struct and implement a non async function that mutates the value:
tokio channels (message passing)
- if we want to share sync resources amongst tokio tasks, we use a channel to manage exclusive access
- its like a mini queueing system
- variants of the channels:
- mpsc: multi-producer, single-consumer
- oneshot: single-producer, single-consumer
- broadcast: multi-producer, multi-consumer (each receiver sees every value)
- watch: multi-producer, multi-consumer (no history is kept, receivers only see most recent)
- create a channel that can hold 32 messages with
let (tx, mut rx) = mpsc::channel(32)
tokio stream piping
- io::copy can be used to copy input from one stream into the other stream
- the tcplistener implements both reader and writer, but we can’t borrow the reference to the listener twice.
- We can use socket.split to separate the reader and writer and use io::copy to copy the stream to the output
- we can also copy by hand, using
socket.read() and socket.write_all
- don’t forget to handle EOF
tokio framing
- framing is taking a byte stream and converting it into a stream of frames
- if you are writing an exact data packet, you can use a write frame function on a struct that matches the frame and writes to a stream byte by byte
tokio timeouts
- you can specify timeouts on tokio calls to avoid waiting to long for http requests for instance
log facade and env_logger logging
- use facade log to log
log::info!("opened wordlist file: {:?}", wordlist_file);
- use implementation
env_logger using env_logger::init();
- use
structured_logging crate for structured logging, e.g. adding parseable values