DistriBeautie

Have you ever heard of a distributed client-server app called DistriBeautie? Me neither. Let’s write it together from scratch…

We will write it in Rust with no additional crates and libraries support, just using standard features of the language. We are going to cover the following aspects: working with files, network, threads and scheduling the tasks.

Agenda

Create an app, which runs in master and replica modes
Accept read + write requests
Use HTTP 1.1 protocol as a transport
Store the value from HTTP request line
Eventually propagate the values written to master — to replicas

Configs

It’s essential to externalize configs from the app, and our configs have the following key-value structure:

server.host=127.0.0.1
server.port=7878
server.ismasterhost=true
sync.interval.ms=15000
replicas.addr=127.0.0.1:7871

We read the config:

fn read_config() -> Vec<String> {
    fs::read_to_string("server.conf")
        .unwrap()
        .lines()
        .map(String::from)
        .collect()
}

And its properties the following way:

fn read_config_property(config: Vec<String>, property_name: &str) -> String {
    let config_property_value = config
        .iter()
        .find(|&x| x.contains(property_name))
        .unwrap()
        .split("=")
        .collect::<Vec<_>>()[1];
    String::from(config_property_value)
}

Starting the Server

Using standard library’s socket listener, we listen to a port specified in our configs, and serve a connection in a new thread:

let listener = TcpListener::bind(server_addr).unwrap();

for stream in listener.incoming() {
    let stream = stream.unwrap();
    thread::spawn(|| networking::handle_connection(stream));
}

To handle a connection, we perform the following steps:

Read HTTP request from TCP stream
Find out if it’s a read (GET) or write (POST) access mode
In case of write path — store received string variable in array
In case of read path — return T/F if a string variable exists
Write a response back to TCP stream

A healthy HTTP request taken from healthy network usually consists of 4 parts:

Request line
Headers
Empty line
Body

Now we are mostly interested only in the first request line, which consists of:

Request method
A whitespace
Requested URL
Yet one more whitespace
The protocol version

Since we don’t need the whole HTTP request, we can read just a request line the following way:

let buf_reader = BufReader::new(&mut stream);
let request_line_optional = buf_reader.lines().next();
let request_line = if request_line_optional.is_some() {
    request_line_optional.unwrap().unwrap()
} else {
    return;
};

Eventually these are just characters, that we need to read from TCP stream. At this point inside request line we have:

POST /foo HTTP/1.1

Simple string manipulation helps us with extracting request method and a resource. Then we store the variable in the array — upon write request, and read it — upon read request:

static STORAGE: Mutex<Vec<String>> = Mutex::new(Vec::new());

pub fn add_item(item: String) {
    let mut vec = STORAGE.lock().unwrap();
    vec.push(item);
}

pub fn get_item(item: String) -> bool {
    let vec = STORAGE.lock().unwrap();
    vec.contains(&item)
}

When it comes to responding to a client — we write directly to TCP stream:

let status_line = "HTTP/1.1 201 DISTRIBEAUTIE_OBJECT_CREATED";
let contents = format!("The following object has been created: {object_to_create}");
let length = contents.len();
let response = format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
stream.write_all(response.as_bytes()).unwrap();

Since the implementation details are in our hands, we can even respond with some deviations, like custom HTTP statuses. Note that 201 DISTRIBEAUTIE_OBJECT_CREATED doesn’t really exist, however is well understood by popular 3rd party tools like Postman:

Postman’s view of a custom HTTP status

image::postman-status.png[Postman View]postman-status.png

Scheduler

To keep other replicas in sync, let’s make use of a scheduler.

Do we need 3rd party crates like clockwerk or tokio-cron-scheduler? Why, a sleep inside a loop will do the trick:

loop {
    thread::sleep(Duration::from_millis(config::get_sync_interval_ms()));
    sync_replicas();
}

Syncing Replicas

To keep replicas in sync, we perform write request against all replicas mentioned in our config:

let replicas = config::get_replicas_addr();
let all_items = storage::get_all_items();

for replica_addr in replicas {
    match TcpStream::connect(replica_addr.clone()) {
        Ok(mut stream) => {
            for item in &all_items {
                // ...
            }
        }
        Err(e) => {
            println!("Failed to connect: {} {}", replica_addr, e);
        }
    }
}