Key-Value Store: Group Commit (Low-Level Design)

The capacity estimate of chapter 1 arrived at 24 nodes by assuming that a node stores 20 writes with every wait for its disk. Without that assumption the same calculation gave 325 nodes. Nothing built so far does anything of the kind. The storage node of chapter 2 waits once for every write, and reaches its limit at about 180 writes per second however many clients are sending to it, since each must wait its turn for a disk that confirms 185 times a second.

This chapter adds the missing technique. Under group commit, a node stores all the writes that are waiting at a given moment with a single wait for the disk. The change is small, some 60 lines in the store and 1 thread in the server, and its effect is out of all proportion to its size: the measured rate rises from 180 writes per second to almost 5000 with 64 clients, and beyond 17000 with 256. It is among the best examples in systems programming of a large gain obtained by rearranging work, with no part of the work made faster.

The idea

A wait for the disk takes about 5.4 ms whether it stores 1 record or 100. Nearly all of that time is spent by the disk in confirming that the data is safe, and very little in transferring the data, so that the cost is close to a fixed charge for each wait and almost independent of how much is written. Writes that arrive while 1 wait is in progress can therefore share the next.

The design uses no timer, and the point deserves emphasis because the alternative is common. Some systems hold a write back for a millisecond or so in the hope that others will arrive to share its wait. This improves throughput under heavy load and penalises every write under light load, when no others are coming. Here a batch is formed only from writes that are already waiting. A write that arrives alone is stored at once, with no delay, and the batches grow of their own accord as the load grows, since the busier the node is, the more writes accumulate during each wait. The mechanism adjusts itself to the load without any parameter that an operator has to tune.

A batch in the store

Store::commit takes a list of writes and returns a result for each of them.

Listing 6.1 The imports of kv/src/store.rs. kv/src/store.rs

use std::collections::BTreeMap;
use std::io;
use std::ops::Bound;
use std::path::Path;

use thiserror::Error;

use crate::log::Log;
use crate::record::{Change, Record};

Listing 6.2 Storing a batch of writes with a single wait for the disk. kv/src/store.rs

/// One write of a batch: the change, and the condition it carries.
pub struct Write {
    pub change: Change,
    pub condition: Condition,
}

impl Store {
    /// Performs several writes with a single wait for the disk. Each write
    /// is checked as if the writes before it in the batch had already
    /// happened. The result of each is returned in the same order.
    pub fn commit(&mut self, writes: Vec<Write>) -> Vec<Result<Written, StoreError>> {
        // The versions that keys will have once the accepted writes of this
        // batch are applied. `None` is a key that a write of the batch deletes.
        let mut pending: BTreeMap<String, Option<u64>> = BTreeMap::new();
        let mut records = Vec::new();
        let mut results = Vec::new();
        let first = self.log.last_sequence() + 1;

        for write in writes {
            let key = match &write.change {
                Change::Put { key, .. } | Change::Delete { key } => key.clone(),
            };
            let current = match pending.get(&key) {
                Some(version) => *version,
                None => self.version(&key),
            };
            let deletes = matches!(write.change, Change::Delete { .. });

            let checked = if deletes && current.is_none() {
                Err(StoreError::NotFound)
            } else {
                check(write.condition, current)
            };

            if let Err(error) = checked {
                results.push(Err(error));

                continue;
            }

            let sequence = first + records.len() as u64;
            let version = if deletes { None } else { Some(sequence) };

            pending.insert(key, version);
            results.push(Ok(Written {
                version: sequence,
                created: current.is_none(),
            }));
            records.push(Record {
                sequence,
                change: write.change,
            });
        }

        // One write and one wait for the disk, for the whole batch.
        if let Err(error) = self.log.append_records(&records) {
            let failure = error.to_string();

            return results
                .into_iter()
                .map(|result| result.and(Err(StoreError::Log(io::Error::other(failure.clone())))))
                .collect();
        }

        for record in records {
            apply(&mut self.entries, record);
        }

        results
    }
}

The method works in 3 passes.

Conditional writes need care in the first pass. Suppose that a batch contains 2 writes to user-1, each carrying If-None-Match: *, which means “create only”. The first ought to succeed and the second to fail, although neither is yet in the map, and a check against the map alone would accept both. The pending map resolves the difficulty. It records the version that each key will have once the accepted writes of the batch have been applied, and a later write in the same batch is checked against it. The batch thus behaves, as far as any client can tell, exactly as if its writes had been performed in sequence, which is the property that makes batching invisible.

The order of the passes preserves invariant 1 of chapter 2. No result leaves commit before the wait for the disk has ended, and so no client is answered early.

The committer

A single thread, the committer, owns the store. The handlers of requests do not touch the store at all. Each hands its write to the committer through a channel and waits for the result.

Listing 6.3 The imports of kvgroup/src/main.rs. kvgroup/src/main.rs

use std::error::Error;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::mpsc::{self, Receiver, Sender};
use std::thread;

use axum::Router;
use axum::body::Bytes;
use axum::extract::{DefaultBodyLimit, Path, State};
use axum::http::header::ETAG;
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::put;
use clap::Parser;
use kv::{Change, Store, StoreError, Write, Written};
use kvd::api::{ApiError, MAX_VALUE, etag, valid_key, write_condition};
use tokio::net::TcpListener;
use tokio::sync::oneshot;

Listing 6.4 The options, and a write on its way to the committer. kvgroup/src/main.rs

#[derive(Parser)]
struct Options {
    #[arg(long, default_value = "127.0.0.1:7001")]
    listen: SocketAddr,

    #[arg(long, default_value = "data")]
    data: PathBuf,

    /// The largest number of writes stored with one wait for the disk.
    #[arg(long, default_value_t = 256)]
    batch: usize,
}

/// A write on its way to the committer, with the channel for its result.
struct Request {
    write: Write,
    reply: oneshot::Sender<Result<Written, StoreError>>,
}

Listing 6.5 The committer thread. kvgroup/src/main.rs

/// The committer: the one thread that owns the store. It takes the write
/// that has waited longest, adds every write that arrived meanwhile, and
/// stores them together.
fn run_committer(mut store: Store, requests: Receiver<Request>, limit: usize) {
    let (mut batches, mut writes) = (0u64, 0u64);

    while let Ok(first) = requests.recv() {
        // The first request, and whatever else is already waiting.
        let mut batch = vec![first];

        batch.extend(requests.try_iter().take(limit - 1));

        let (changes, replies): (Vec<Write>, Vec<_>) = batch
            .into_iter()
            .map(|request| (request.write, request.reply))
            .unzip();

        batches += 1;
        writes += changes.len() as u64;

        // A handler that has gone away no longer wants its result.
        store
            .commit(changes)
            .into_iter()
            .zip(replies)
            .for_each(|(result, reply)| drop(reply.send(result)));

        if batches.is_multiple_of(200) {
            let average = writes as f64 / batches as f64;
            println!("{writes} writes in {batches} batches, {average:.1} writes per batch");
        }
    }
}

This loop is the whole of group commit. recv blocks until a write arrives. try_iter then yields every write that is already in the channel, without waiting for more, up to a limit of 256 for the batch. While commit is waiting for the disk, new writes accumulate in the channel, and the next turn of the loop takes all of them.

Giving the store to 1 thread has a further merit. Since no other thread touches it, the store needs no lock, and the question of what 2 concurrent writers might do to the map does not arise. Concurrency has been confined to the channel, where it is handled by a well-tested library, and the logic of the store remains sequential.

Listing 6.6 Handing a write to the committer, and the HTTP handlers. kvgroup/src/main.rs

/// Hands a write to the committer and waits for its result.
async fn submit(committer: &Sender<Request>, write: Write) -> Result<Written, ApiError> {
    let (reply, result) = oneshot::channel();
    let stopped = || ApiError::Unavailable("the committer has stopped".into());

    if committer
        .send(Request { write, reply })
        .is_err()
    {
        return Err(stopped());
    }

    let Ok(outcome) = result.await else {
        return Err(stopped());
    };

    Ok(outcome?)
}

async fn put_key(
    State(committer): State<Sender<Request>>,
    Path(key): Path<String>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response, ApiError> {
    valid_key(&key)?;

    let write = Write {
        change: Change::Put {
            key,
            value: body.to_vec(),
        },
        condition: write_condition(&headers)?,
    };

    let written = submit(&committer, write).await?;
    let status = if written.created {
        StatusCode::CREATED
    } else {
        StatusCode::NO_CONTENT
    };

    Ok((status, [(ETAG, etag(written.version))]).into_response())
}

async fn delete_key(
    State(committer): State<Sender<Request>>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    valid_key(&key)?;

    let write = Write {
        change: Change::Delete { key },
        condition: write_condition(&headers)?,
    };

    submit(&committer, write).await?;

    Ok(StatusCode::NO_CONTENT)
}

Listing 6.7 Starting the committer and the server. kvgroup/src/main.rs

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let (store, applied, discarded) = Store::open(&options.data, true)?;

    println!("recovered {applied} records, cut {discarded} bytes from the end of the log");

    let (committer, requests) = mpsc::channel();
    thread::spawn(move || run_committer(store, requests, options.batch));

    let routes = Router::new()
        .route("/v1/keys/{key}", put(put_key).delete(delete_key))
        .layer(DefaultBodyLimit::max(MAX_VALUE))
        .with_state(committer);

    let listener = TcpListener::bind(options.listen).await?;
    println!("listening on {}", listener.local_addr()?);

    axum::serve(listener, routes).await?;

    Ok(())
}

Measurements

kvbench runs a number of clients simultaneously. Each client sends its writes in succession and waits for every answer before sending the next.

Listing 6.8 The imports of kvbench/src/main.rs. kvbench/src/main.rs

use std::error::Error;
use std::fs;
use std::path::PathBuf;
use std::thread;
use std::time::Instant;

use clap::Parser;
use ureq::Agent;

Listing 6.9 Writing from several clients at once. kvbench/src/main.rs

#[derive(Parser)]
struct Options {
    #[arg(long)]
    server: String,

    /// The number of clients that write at the same time.
    #[arg(long, default_value_t = 8)]
    clients: u32,

    /// The number of writes each client makes.
    #[arg(long, default_value_t = 200)]
    writes: u32,

    #[arg(long)]
    acked: PathBuf,
}

/// The writes of one client. It returns the keys that were acknowledged.
fn client(server: &str, number: u32, writes: u32) -> Vec<String> {
    let agent = Agent::new_with_defaults();
    let mut acknowledged = Vec::new();

    for write in 1..=writes {
        let key = format!("key-{number:03}-{write:06}");
        let address = format!("{server}/v1/keys/{key}");

        if agent
            .put(&address)
            .send(format!("value of {key}"))
            .is_ok()
        {
            acknowledged.push(key);
        }
    }

    acknowledged
}

fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let (server, writes) = (options.server.as_str(), options.writes);
    let started = Instant::now();

    let acknowledged: Vec<String> = thread::scope(|scope| {
        let clients: Vec<_> = (1..=options.clients)
            .map(|number| scope.spawn(move || client(server, number, writes)))
            .collect();

        clients
            .into_iter()
            .flat_map(|client| client.join().unwrap_or_default())
            .collect()
    });

    let seconds = started.elapsed().as_secs_f64();
    let rate = acknowledged.len() as f64 / seconds;

    fs::write(&options.acked, acknowledged.join("\n") + "\n")?;

    println!(
        "{} clients: {} writes acknowledged in {seconds:.2} s, {rate:.0} writes per second",
        options.clients,
        acknowledged.len()
    );

    Ok(())
}

$ ./target/release/kvd --data /var/tmp/kvd &
$ ./target/release/kvbench --server http://127.0.0.1:7001 --clients 64 --writes 12 --acked acked.txt
64 clients: 768 writes acknowledged in 4.27 s, 180 writes per second

$ ./target/release/kvgroup --data /var/tmp/kvgroup &
$ ./target/release/kvbench --server http://127.0.0.1:7001 --clients 64 --writes 400 --acked acked.txt
64 clients: 25600 writes acknowledged in 5.13 s, 4993 writes per second
25535 writes in 800 batches, 31.9 writes per batch

Several things can be read from the table. A single client gains nothing, since it sends its next write only when the previous one has been answered, and there is consequently never a second write with which a wait could be shared. Group commit improves throughput under concurrency and does nothing for the latency of an individual write, which remains 1 wait for the disk.

The size of a batch is about half the number of clients: 4 of 8, 32 of 64, 120 of 256. The explanation is that the clients fall into 2 groups that alternate. While 1 half is waiting in the batch that is being stored, the other half has just been answered and is sending its next writes, which form the following batch.

And the assumption of chapter 1 is confirmed with something to spare. The estimate supposed 20 writes for each wait and 3700 writes per second. With 64 clients the node achieved 31.9 writes for each wait and 4993 writes per second.

A crash in the middle of a batch

The server is killed while 64 clients are writing. The program kvd of chapter 2 then opens the same data directory, which incidentally demonstrates that the 2 programs use the same format for the log.

$ ./target/release/kvgroup --data /var/tmp/kvgroup &
$ ./target/release/kvbench --server http://127.0.0.1:7001 --clients 64 --writes 2000 --acked acked.txt &
$ sleep 2; kill -9 <pid of kvgroup>
64 clients: 9341 writes acknowledged in 2.74 s, 3412 writes per second
$ ./target/release/kvd --data /var/tmp/kvgroup &
recovered 9366 keys from 9366 records, cut 0 bytes from the end of the log
$ ./target/release/kvctl verify --server http://127.0.0.1:7001 --acked acked.txt
9341 acknowledged, 9341 found, 0 missing

Every one of the 9341 acknowledged writes is present. The log contains 25 more, which are the batch that had reached the disk when the process was killed and whose answers had not yet been sent. These are the durable and unacknowledged writes of chapter 2, the third case of its figure of crash points, now occurring 25 together. Group commit has not altered the guarantee in any respect. It has enlarged the number of writes that may be in that uncertain state at the moment of a crash from 1 to the size of a batch, and each of the 25 clients concerned resolves the uncertainty in the usual way, by repeating its request.

What is still missing

The complete programs

Store::commit is part of kv/src/store.rs, which is listed in full in chapter 2.

Listing 6.10 The complete kvgroup/src/main.rs. kvgroup/src/main.rs

//! Command kvgroup is the storage node of chapter 2 with group commit: one
//! thread writes the log, and it stores every write that is waiting with a
//! single wait for the disk.
//!
//! Run it from code/ with:
//!
//!     cargo run --release -p kvgroup -- --listen 127.0.0.1:7001 --data /var/tmp/kvgroup

use std::error::Error;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::mpsc::{self, Receiver, Sender};
use std::thread;

use axum::Router;
use axum::body::Bytes;
use axum::extract::{DefaultBodyLimit, Path, State};
use axum::http::header::ETAG;
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::put;
use clap::Parser;
use kv::{Change, Store, StoreError, Write, Written};
use kvd::api::{ApiError, MAX_VALUE, etag, valid_key, write_condition};
use tokio::net::TcpListener;
use tokio::sync::oneshot;

#[derive(Parser)]
struct Options {
    #[arg(long, default_value = "127.0.0.1:7001")]
    listen: SocketAddr,

    #[arg(long, default_value = "data")]
    data: PathBuf,

    /// The largest number of writes stored with one wait for the disk.
    #[arg(long, default_value_t = 256)]
    batch: usize,
}

/// A write on its way to the committer, with the channel for its result.
struct Request {
    write: Write,
    reply: oneshot::Sender<Result<Written, StoreError>>,
}

/// The committer: the one thread that owns the store. It takes the write
/// that has waited longest, adds every write that arrived meanwhile, and
/// stores them together.
fn run_committer(mut store: Store, requests: Receiver<Request>, limit: usize) {
    let (mut batches, mut writes) = (0u64, 0u64);

    while let Ok(first) = requests.recv() {
        // The first request, and whatever else is already waiting.
        let mut batch = vec![first];

        batch.extend(requests.try_iter().take(limit - 1));

        let (changes, replies): (Vec<Write>, Vec<_>) = batch
            .into_iter()
            .map(|request| (request.write, request.reply))
            .unzip();

        batches += 1;
        writes += changes.len() as u64;

        // A handler that has gone away no longer wants its result.
        store
            .commit(changes)
            .into_iter()
            .zip(replies)
            .for_each(|(result, reply)| drop(reply.send(result)));

        if batches.is_multiple_of(200) {
            let average = writes as f64 / batches as f64;
            println!("{writes} writes in {batches} batches, {average:.1} writes per batch");
        }
    }
}

/// Hands a write to the committer and waits for its result.
async fn submit(committer: &Sender<Request>, write: Write) -> Result<Written, ApiError> {
    let (reply, result) = oneshot::channel();
    let stopped = || ApiError::Unavailable("the committer has stopped".into());

    if committer
        .send(Request { write, reply })
        .is_err()
    {
        return Err(stopped());
    }

    let Ok(outcome) = result.await else {
        return Err(stopped());
    };

    Ok(outcome?)
}

async fn put_key(
    State(committer): State<Sender<Request>>,
    Path(key): Path<String>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Response, ApiError> {
    valid_key(&key)?;

    let write = Write {
        change: Change::Put {
            key,
            value: body.to_vec(),
        },
        condition: write_condition(&headers)?,
    };

    let written = submit(&committer, write).await?;
    let status = if written.created {
        StatusCode::CREATED
    } else {
        StatusCode::NO_CONTENT
    };

    Ok((status, [(ETAG, etag(written.version))]).into_response())
}

async fn delete_key(
    State(committer): State<Sender<Request>>,
    Path(key): Path<String>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    valid_key(&key)?;

    let write = Write {
        change: Change::Delete { key },
        condition: write_condition(&headers)?,
    };

    submit(&committer, write).await?;

    Ok(StatusCode::NO_CONTENT)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let (store, applied, discarded) = Store::open(&options.data, true)?;

    println!("recovered {applied} records, cut {discarded} bytes from the end of the log");

    let (committer, requests) = mpsc::channel();
    thread::spawn(move || run_committer(store, requests, options.batch));

    let routes = Router::new()
        .route("/v1/keys/{key}", put(put_key).delete(delete_key))
        .layer(DefaultBodyLimit::max(MAX_VALUE))
        .with_state(committer);

    let listener = TcpListener::bind(options.listen).await?;
    println!("listening on {}", listener.local_addr()?);

    axum::serve(listener, routes).await?;

    Ok(())
}

Listing 6.11 The complete kvbench/src/main.rs. kvbench/src/main.rs

//! Command kvbench writes keys to a server from several clients at once and
//! prints the number of writes per second. Every acknowledged key is added
//! to a file, in the format that `kvctl verify` reads.
//!
//! Run it from code/ with:
//!
//!     cargo run --release -p kvbench -- --server http://127.0.0.1:7001 --clients 64 --writes 200 --acked acked.txt

use std::error::Error;
use std::fs;
use std::path::PathBuf;
use std::thread;
use std::time::Instant;

use clap::Parser;
use ureq::Agent;

#[derive(Parser)]
struct Options {
    #[arg(long)]
    server: String,

    /// The number of clients that write at the same time.
    #[arg(long, default_value_t = 8)]
    clients: u32,

    /// The number of writes each client makes.
    #[arg(long, default_value_t = 200)]
    writes: u32,

    #[arg(long)]
    acked: PathBuf,
}

/// The writes of one client. It returns the keys that were acknowledged.
fn client(server: &str, number: u32, writes: u32) -> Vec<String> {
    let agent = Agent::new_with_defaults();
    let mut acknowledged = Vec::new();

    for write in 1..=writes {
        let key = format!("key-{number:03}-{write:06}");
        let address = format!("{server}/v1/keys/{key}");

        if agent
            .put(&address)
            .send(format!("value of {key}"))
            .is_ok()
        {
            acknowledged.push(key);
        }
    }

    acknowledged
}

fn main() -> Result<(), Box<dyn Error>> {
    let options = Options::parse();
    let (server, writes) = (options.server.as_str(), options.writes);
    let started = Instant::now();

    let acknowledged: Vec<String> = thread::scope(|scope| {
        let clients: Vec<_> = (1..=options.clients)
            .map(|number| scope.spawn(move || client(server, number, writes)))
            .collect();

        clients
            .into_iter()
            .flat_map(|client| client.join().unwrap_or_default())
            .collect()
    });

    let seconds = started.elapsed().as_secs_f64();
    let rate = acknowledged.len() as f64 / seconds;

    fs::write(&options.acked, acknowledged.join("\n") + "\n")?;

    println!(
        "{} clients: {} writes acknowledged in {seconds:.2} s, {rate:.0} writes per second",
        options.clients,
        acknowledged.len()
    );

    Ok(())
}

Further reading

• D. DeWitt and others, “Implementation Techniques for Main Memory Database Systems”, SIGMOD 1984, which introduced group commit.
• J. Gray and A. Reuter, Transaction Processing: Concepts and Techniques, Morgan Kaufmann, 1992.