Struct Config 

pub struct Config {
    pub path: PathBuf,
    pub shards: u8,
    pub in_memory_buffer_per_log: usize,
}

Fields

path: PathBuf

Where to open the log.

shards: u8

Number of sharded log files. Future calls to recover must always use at least this many shards, otherwise the system will not be able to recover. Defaults to 8.

in_memory_buffer_per_log: usize

The in-memory buffer size in front of each log file. Defaults to 512k.

Implementations

impl Config

pub fn recover(&self) -> Result<RecoveryIterator>

Iterate over all log shards, returning batches in the order that they were written in, even if they landed in different shards. This method does not take out an exclusive file lock on the shards in the way that Config::purge and Config::create do, so it can be used on logs that are actively being written. HOWEVER: keep in mind that while writing logs, significant data may remain in the in-memory BufWriter instances until you call ShardedLog::flush!

Examples found in repository

examples/recovery.rs (line 19)

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");
}

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

Clear the previous contents of a log directory. Requires an exclusive lock over the log directory, and may not be performed concurrently with an active ShardedLog.

Examples found in repository

examples/recovery.rs (line 9)

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 15)

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 create(&self) -> Result<ShardedLog>

Exclusively create a new log directory that may be used for writing sharded logs to.

Examples found in repository

examples/recovery.rs (line 11)

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 17)

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");
}

Trait Implementations

impl Clone for Config

fn clone(&self) -> Config

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for Config

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for Config

fn default() -> Config

Returns the “default value” for a type.