Struct ShardedLog 

pub struct ShardedLog { /* private fields */ }

Allows batches of bytes to be written with low contention, and read-back in a linearized order.

Implementations

impl ShardedLog

pub fn write_batch<B: AsRef<[u8]>>(&self, write_batch: &[B]) -> Result<u64>

Write a batch of buffers to the sharded log. Returns the logical sequence number that they will be recoverable at after your next call to flush. Writes the batch into a BufWriter in front of the sharded log file, which will be flushed on Drop, but can also be flushed using the flush method.

Examples found in repository

examples/recovery.rs (line 14)

fn main() {
    let config = Config {
        path: "path/to/logs".into(),
        ..Default::default()
    };

    config.purge().unwrap();

    let sharded_log = config.create().unwrap();

    sharded_log
        .write_batch(&[b"a", b"b", b"c", b"d"])
        .unwrap();

    sharded_log.flush().unwrap();

    for write_batch in config.recover().unwrap() {
        println!("got batch: {:?}", write_batch);
        assert_eq!(
            write_batch,
            vec![b"a", b"b", b"c", b"d"]
        );
    }

    let _ = std::fs::remove_dir_all("path");
}

examples/write_performance.rs (line 44)

fn main() {
    let number_of_threads = 16;
    let buffers_per_batch = 8;
    let buffer_size = 512;
    let batches_per_thread = 128 * 1024;

    let config = Config {
        path: "write_performance".into(),
        ..Default::default()
    };

    config.purge().unwrap();

    let log = Arc::new(config.create().unwrap());

    let barrier =
        Arc::new(Barrier::new(number_of_threads + 1));

    let mut threads = vec![];

    let before = std::time::Instant::now();

    for _ in 0..number_of_threads {
        let barrier = barrier.clone();
        let log = log.clone();

        let thread = std::thread::spawn(move || {
            barrier.wait();

            for _ in 0..batches_per_thread {
                let mut batch =
                    Vec::with_capacity(buffers_per_batch);
                for _ in 0..buffers_per_batch {
                    let mut buffer =
                        Vec::with_capacity(buffer_size);
                    unsafe {
                        buffer.set_len(buffer_size);
                    }
                    batch.push(buffer);
                }
                log.write_batch(&batch).unwrap();
            }
        });

        threads.push(thread);
    }

    barrier.wait();

    for thread in threads.into_iter() {
        thread.join().unwrap();
    }

    log.flush().unwrap();

    let bytes_written = number_of_threads
        * batches_per_thread
        * buffers_per_batch
        * buffer_size;
    let throughput = bytes_written as u128
        / before.elapsed().as_micros();

    println!("wrote {} megabytes per second, {} threads, {} batches per thread, {} buffers per batch, {} bytes per buffer",
        throughput,
        number_of_threads,
        batches_per_thread,
        buffers_per_batch,
        buffer_size
    );

    let _ = std::fs::remove_dir_all("write_performance");
}

pub fn reservation(&self) -> Reservation<'_>

Reserve a log slot at a particular index, which can then be completed using Reservation::write_batch or aborted using Reservation::abort. This is particularly useful for logging fallible lock-free operations to disk in an order-preserving manner.

pub fn flush(&self) -> Result<()>

Flushes and fsyncs in-memory shard data that have been written to directly from this instance of ShardedLog, but skipping shards that have been written to by others. To flush all shards, use flush_all.

Examples found in repository

examples/recovery.rs (line 17)

fn main() {
    let config = Config {
        path: "path/to/logs".into(),
        ..Default::default()
    };

    config.purge().unwrap();

    let sharded_log = config.create().unwrap();

    sharded_log
        .write_batch(&[b"a", b"b", b"c", b"d"])
        .unwrap();

    sharded_log.flush().unwrap();

    for write_batch in config.recover().unwrap() {
        println!("got batch: {:?}", write_batch);
        assert_eq!(
            write_batch,
            vec![b"a", b"b", b"c", b"d"]
        );
    }

    let _ = std::fs::remove_dir_all("path");
}

examples/write_performance.rs (line 57)

fn main() {
    let number_of_threads = 16;
    let buffers_per_batch = 8;
    let buffer_size = 512;
    let batches_per_thread = 128 * 1024;

    let config = Config {
        path: "write_performance".into(),
        ..Default::default()
    };

    config.purge().unwrap();

    let log = Arc::new(config.create().unwrap());

    let barrier =
        Arc::new(Barrier::new(number_of_threads + 1));

    let mut threads = vec![];

    let before = std::time::Instant::now();

    for _ in 0..number_of_threads {
        let barrier = barrier.clone();
        let log = log.clone();

        let thread = std::thread::spawn(move || {
            barrier.wait();

            for _ in 0..batches_per_thread {
                let mut batch =
                    Vec::with_capacity(buffers_per_batch);
                for _ in 0..buffers_per_batch {
                    let mut buffer =
                        Vec::with_capacity(buffer_size);
                    unsafe {
                        buffer.set_len(buffer_size);
                    }
                    batch.push(buffer);
                }
                log.write_batch(&batch).unwrap();
            }
        });

        threads.push(thread);
    }

    barrier.wait();

    for thread in threads.into_iter() {
        thread.join().unwrap();
    }

    log.flush().unwrap();

    let bytes_written = number_of_threads
        * batches_per_thread
        * buffers_per_batch
        * buffer_size;
    let throughput = bytes_written as u128
        / before.elapsed().as_micros();

    println!("wrote {} megabytes per second, {} threads, {} batches per thread, {} buffers per batch, {} bytes per buffer",
        throughput,
        number_of_threads,
        batches_per_thread,
        buffers_per_batch,
        buffer_size
    );

    let _ = std::fs::remove_dir_all("write_performance");
}

pub fn purge(&self) -> Result<()>

Delete all logs in the system

Trait Implementations

impl Clone for ShardedLog

fn clone(&self) -> ShardedLog

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.