This commit is contained in:
		@@ -13,7 +13,6 @@ harness = false
 | 
			
		||||
[dependencies]
 | 
			
		||||
chgk_ledb_lib = {path = "../lib"}
 | 
			
		||||
serde_json="1.0"
 | 
			
		||||
ledb = {git = "https://github.com/b4tman/ledb.git", rev="a646b90e", package="ledb"}
 | 
			
		||||
zip="0.6"
 | 
			
		||||
rand="0.8"
 | 
			
		||||
clap = { version = "3.2.22", features = ["derive"] }
 | 
			
		||||
@@ -22,8 +21,5 @@ clap = { version = "3.2.22", features = ["derive"] }
 | 
			
		||||
criterion = "0.4.0"
 | 
			
		||||
tempfile = "3.3"
 | 
			
		||||
bincode = "^2.0.0-rc.2"
 | 
			
		||||
ledb = {git = "https://github.com/b4tman/ledb.git", rev="a646b90e", package="ledb"}
 | 
			
		||||
ledb-derive = {git = "https://github.com/b4tman/ledb.git", rev="a646b90e", package="ledb-derive"}
 | 
			
		||||
ledb-types = {git = "https://github.com/b4tman/ledb.git", rev="a646b90e", package="ledb-types"}
 | 
			
		||||
serde="1.0"
 | 
			
		||||
serde_derive="1.0"
 | 
			
		||||
 
 | 
			
		||||
@@ -1,13 +1,10 @@
 | 
			
		||||
#[macro_use]
 | 
			
		||||
extern crate criterion;
 | 
			
		||||
extern crate bincode;
 | 
			
		||||
extern crate ledb;
 | 
			
		||||
extern crate ledb_types;
 | 
			
		||||
extern crate serde;
 | 
			
		||||
extern crate serde_derive;
 | 
			
		||||
extern crate tempfile;
 | 
			
		||||
#[macro_use]
 | 
			
		||||
extern crate serde_json;
 | 
			
		||||
extern crate tempfile;
 | 
			
		||||
 | 
			
		||||
use chgk_ledb_lib::db;
 | 
			
		||||
use std::path::PathBuf;
 | 
			
		||||
@@ -17,7 +14,6 @@ use db::{Reader, Writer, WriterOpts};
 | 
			
		||||
use criterion::{BatchSize, Criterion};
 | 
			
		||||
use tempfile::tempdir;
 | 
			
		||||
 | 
			
		||||
use ledb::{Document, Options, Storage};
 | 
			
		||||
use serde_derive::{Deserialize, Serialize};
 | 
			
		||||
 | 
			
		||||
#[derive(
 | 
			
		||||
@@ -31,10 +27,8 @@ use serde_derive::{Deserialize, Serialize};
 | 
			
		||||
    Ord,
 | 
			
		||||
    Serialize,
 | 
			
		||||
    Deserialize,
 | 
			
		||||
    Document,
 | 
			
		||||
)]
 | 
			
		||||
struct TestData {
 | 
			
		||||
    #[document(primary)]
 | 
			
		||||
    num1: u64,
 | 
			
		||||
    num2: u64,
 | 
			
		||||
    test: String,
 | 
			
		||||
@@ -48,7 +42,7 @@ fn gen_data(count: usize) -> impl Iterator<Item = TestData> {
 | 
			
		||||
        .map(|i| 143 + i as u64)
 | 
			
		||||
        .map(|i| TestData {
 | 
			
		||||
            num1: i,
 | 
			
		||||
            num2: i*100 ^ 0xDF0E441122334455,
 | 
			
		||||
            num2: i * 100 ^ 0xDF0E441122334455,
 | 
			
		||||
            test: "test ---- Test ____".repeat(123 + i as usize % 15),
 | 
			
		||||
        })
 | 
			
		||||
}
 | 
			
		||||
@@ -108,83 +102,9 @@ fn db_write(c: &mut Criterion) {
 | 
			
		||||
    });
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn ledb_write(c: &mut Criterion) {
 | 
			
		||||
    let dir = tempdir().expect("tempdir");
 | 
			
		||||
    let tmp_dir = dir.as_ref();
 | 
			
		||||
    c.bench_function("ledb_write", |b| {
 | 
			
		||||
        b.iter_batched(
 | 
			
		||||
            || {
 | 
			
		||||
                let src = gen_data(N).collect::<Vec<TestData>>().into_iter();
 | 
			
		||||
                let options: Options = serde_json::from_value(json!({
 | 
			
		||||
                    "map_size": 100 * 1024 * 1024, // 100mb
 | 
			
		||||
                    "write_map": true,
 | 
			
		||||
                    "map_async": true,
 | 
			
		||||
                    "no_lock": true,
 | 
			
		||||
                    "no_meta_sync": true,
 | 
			
		||||
                    "no_sync": true,
 | 
			
		||||
                }))
 | 
			
		||||
                .unwrap();
 | 
			
		||||
 | 
			
		||||
                let storage = Storage::new(tmp_dir, options).unwrap();
 | 
			
		||||
                let collection = storage.collection("test").unwrap();
 | 
			
		||||
                (src, collection)
 | 
			
		||||
            },
 | 
			
		||||
            |(src, collection)| collection.load(src).expect("load"),
 | 
			
		||||
            BatchSize::SmallInput,
 | 
			
		||||
        )
 | 
			
		||||
    });
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn ledb_read(c: &mut Criterion) {
 | 
			
		||||
    let dir = tempdir().expect("tempdir");
 | 
			
		||||
    let tmp_dir = dir.as_ref();
 | 
			
		||||
 | 
			
		||||
    let write_options: Options = serde_json::from_value(json!({
 | 
			
		||||
        "map_size": 100 * 1024 * 1024, // 100mb
 | 
			
		||||
        "write_map": true,
 | 
			
		||||
        "map_async": true,
 | 
			
		||||
        "no_lock": true,
 | 
			
		||||
        "no_meta_sync": true,
 | 
			
		||||
        "no_sync": true,
 | 
			
		||||
    }))
 | 
			
		||||
    .unwrap();
 | 
			
		||||
 | 
			
		||||
    let storage = Storage::new(&tmp_dir, write_options).unwrap();
 | 
			
		||||
    let collection = storage.collection("test").unwrap();
 | 
			
		||||
    let items_iter = gen_data(N).collect::<Vec<TestData>>().into_iter();
 | 
			
		||||
    collection.load(items_iter).expect("load");
 | 
			
		||||
    drop(collection);
 | 
			
		||||
    drop(storage);
 | 
			
		||||
 | 
			
		||||
    c.bench_function("ledb_read", |b| {
 | 
			
		||||
        b.iter_batched(
 | 
			
		||||
            || {
 | 
			
		||||
                let options: Options = serde_json::from_value(json!({
 | 
			
		||||
                    "read_only": true,
 | 
			
		||||
                    "map_async": true,
 | 
			
		||||
                    "no_lock": true,
 | 
			
		||||
                }))
 | 
			
		||||
                .unwrap();
 | 
			
		||||
 | 
			
		||||
                let storage = Storage::new(tmp_dir, options).unwrap();
 | 
			
		||||
                let collection = storage.collection("test").unwrap();
 | 
			
		||||
                collection
 | 
			
		||||
            },
 | 
			
		||||
            |collection| {
 | 
			
		||||
                let mut collection_iter = collection.dump::<TestData>().expect("dump");
 | 
			
		||||
                while let Some(item) = collection_iter.next() {
 | 
			
		||||
                    drop(item);
 | 
			
		||||
                }
 | 
			
		||||
            },
 | 
			
		||||
            BatchSize::SmallInput,
 | 
			
		||||
        )
 | 
			
		||||
    });
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn config() -> Criterion {
 | 
			
		||||
    Criterion::default().sample_size(40)
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
criterion_group! {name=ledb; config = config(); targets = ledb_read, ledb_write}
 | 
			
		||||
criterion_group! {name=benches; config = config(); targets = db_read, db_write}
 | 
			
		||||
criterion_main!(benches, ledb);
 | 
			
		||||
criterion_main!(benches);
 | 
			
		||||
 
 | 
			
		||||
							
								
								
									
										116
									
								
								app/src/main.rs
									
									
									
									
									
								
							
							
						
						
									
										116
									
								
								app/src/main.rs
									
									
									
									
									
								
							@@ -1,15 +1,11 @@
 | 
			
		||||
#[macro_use]
 | 
			
		||||
extern crate serde_json;
 | 
			
		||||
use clap::{Parser, Subcommand};
 | 
			
		||||
use rand::seq::IteratorRandom;
 | 
			
		||||
 | 
			
		||||
use std::io;
 | 
			
		||||
use std::path::PathBuf;
 | 
			
		||||
use std::time::Instant;
 | 
			
		||||
use std::{fs, sync::mpsc, thread};
 | 
			
		||||
 | 
			
		||||
use ledb::{Options, Storage};
 | 
			
		||||
 | 
			
		||||
use chgk_ledb_lib::db;
 | 
			
		||||
use chgk_ledb_lib::questions;
 | 
			
		||||
use chgk_ledb_lib::source;
 | 
			
		||||
@@ -19,12 +15,10 @@ use crate::source::ReadSourceQuestionsBatches;
 | 
			
		||||
 | 
			
		||||
const ZIP_FILENAME: &str = "json.zip";
 | 
			
		||||
const NEW_DB_FILENAME: &str = "db.dat";
 | 
			
		||||
const DB_DIR: &str = "db";
 | 
			
		||||
 | 
			
		||||
#[derive(Subcommand, Debug)]
 | 
			
		||||
enum Command {
 | 
			
		||||
    Write,
 | 
			
		||||
    Compact,
 | 
			
		||||
    Print {
 | 
			
		||||
        #[clap(value_parser, default_value = "0")]
 | 
			
		||||
        id: u32,
 | 
			
		||||
@@ -35,11 +29,6 @@ enum Command {
 | 
			
		||||
        #[clap(value_parser, default_value = "0")]
 | 
			
		||||
        num: usize,
 | 
			
		||||
    },
 | 
			
		||||
    Write2,
 | 
			
		||||
    Print2 {
 | 
			
		||||
        #[clap(value_parser, default_value = "0")]
 | 
			
		||||
        id: u32,
 | 
			
		||||
    },
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
#[derive(Parser, Debug)]
 | 
			
		||||
@@ -73,60 +62,12 @@ fn zip_reader_task(tx: mpsc::Sender<Question>) {
 | 
			
		||||
    }
 | 
			
		||||
    println!("read done");
 | 
			
		||||
}
 | 
			
		||||
fn db_writer_task(rx: mpsc::Receiver<Question>) {
 | 
			
		||||
    let out_file: PathBuf = [DB_DIR, "data.mdb"].into_iter().collect();
 | 
			
		||||
    match fs::metadata(&out_file) {
 | 
			
		||||
        Ok(x) if x.is_file() => {
 | 
			
		||||
            fs::remove_file(&out_file).unwrap();
 | 
			
		||||
            println!(r#""{}" removed"#, out_file.to_str().unwrap());
 | 
			
		||||
        }
 | 
			
		||||
        _ => {}
 | 
			
		||||
    };
 | 
			
		||||
 | 
			
		||||
    let options: Options = serde_json::from_value(json!({
 | 
			
		||||
        "map_size": 900 * 1024 * 1024, // 900mb
 | 
			
		||||
        "write_map": true,
 | 
			
		||||
        "map_async": true,
 | 
			
		||||
        "no_lock": true,
 | 
			
		||||
        "no_meta_sync": true,
 | 
			
		||||
        "no_sync": true,
 | 
			
		||||
    }))
 | 
			
		||||
    .unwrap();
 | 
			
		||||
 | 
			
		||||
    let storage = Storage::new(DB_DIR, options).unwrap();
 | 
			
		||||
    let collection = storage.collection("questions").unwrap();
 | 
			
		||||
 | 
			
		||||
    let count = collection.load(rx).expect("load");
 | 
			
		||||
 | 
			
		||||
    println!("loaded {count}");
 | 
			
		||||
 | 
			
		||||
    println!("syncing to disk...");
 | 
			
		||||
    storage.sync(true).unwrap();
 | 
			
		||||
 | 
			
		||||
    print!("stats: ");
 | 
			
		||||
    let stats = storage.stat().unwrap();
 | 
			
		||||
    println!("{:?}", stats);
 | 
			
		||||
 | 
			
		||||
    drop(storage);
 | 
			
		||||
    println!("write done");
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn write_db() {
 | 
			
		||||
    let (tx, rx) = mpsc::channel::<Question>();
 | 
			
		||||
    [
 | 
			
		||||
        thread::spawn(move || zip_reader_task(tx)),
 | 
			
		||||
        thread::spawn(move || db_writer_task(rx)),
 | 
			
		||||
    ]
 | 
			
		||||
    .into_iter()
 | 
			
		||||
    .for_each(|handle| handle.join().expect("thread panic"));
 | 
			
		||||
    println!("all done");
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn print_question_from<F>(get_q: F)
 | 
			
		||||
where
 | 
			
		||||
    F: FnOnce() -> Option<Question>,
 | 
			
		||||
{
 | 
			
		||||
    let q = get_q().unwrap();
 | 
			
		||||
    let q = get_q().expect("question not found");
 | 
			
		||||
    println!("{:#?}", q)
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
@@ -151,45 +92,6 @@ fn read_from_zip(file_num: usize, mut num: usize) -> Option<Question> {
 | 
			
		||||
    Some(questions[num - 1].clone())
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn compact_db() {
 | 
			
		||||
    let options: Options = serde_json::from_value(json!({
 | 
			
		||||
        "write_map": true,
 | 
			
		||||
        "map_async": true,
 | 
			
		||||
        "no_lock": true,
 | 
			
		||||
        "no_meta_sync": true,
 | 
			
		||||
        "no_sync": true,
 | 
			
		||||
        "compact": true,
 | 
			
		||||
    }))
 | 
			
		||||
    .unwrap();
 | 
			
		||||
 | 
			
		||||
    let storage = Storage::new(DB_DIR, options).unwrap();
 | 
			
		||||
 | 
			
		||||
    storage.sync(true).unwrap();
 | 
			
		||||
    let stats = storage.stat().unwrap();
 | 
			
		||||
    println!("{:?}", stats);
 | 
			
		||||
    drop(storage);
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn read_from_db(mut id: u32) -> Option<Question> {
 | 
			
		||||
    let options: Options = serde_json::from_value(json!({
 | 
			
		||||
        "read_only": true,
 | 
			
		||||
        "map_async": true,
 | 
			
		||||
        "no_lock": true,
 | 
			
		||||
    }))
 | 
			
		||||
    .unwrap();
 | 
			
		||||
 | 
			
		||||
    let storage = Storage::new(DB_DIR, options).unwrap();
 | 
			
		||||
    let collection = storage.collection("questions").unwrap();
 | 
			
		||||
    let mut rng = rand::thread_rng();
 | 
			
		||||
 | 
			
		||||
    if id == 0 {
 | 
			
		||||
        let last_id = collection.last_id().unwrap();
 | 
			
		||||
        id = (1..=last_id).choose(&mut rng).unwrap();
 | 
			
		||||
    }
 | 
			
		||||
 | 
			
		||||
    collection.get::<Question>(id).unwrap()
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// measure and return time elapsed in `func` in seconds
 | 
			
		||||
pub fn measure<F: FnOnce()>(func: F) -> f64 {
 | 
			
		||||
    let start = Instant::now();
 | 
			
		||||
@@ -208,7 +110,6 @@ fn main() {
 | 
			
		||||
 | 
			
		||||
    let mut action: Box<dyn FnOnce()> = match &args.command {
 | 
			
		||||
        Command::Write => Box::new(write_db),
 | 
			
		||||
        Command::Compact => Box::new(compact_db),
 | 
			
		||||
        Command::Print { id } => {
 | 
			
		||||
            let get_question = Box::new(|| read_from_db(*id));
 | 
			
		||||
            Box::new(|| print_question_from(get_question))
 | 
			
		||||
@@ -217,11 +118,6 @@ fn main() {
 | 
			
		||||
            let get_question = Box::new(|| read_from_zip(*file_num, *num));
 | 
			
		||||
            Box::new(|| print_question_from(get_question))
 | 
			
		||||
        }
 | 
			
		||||
        Command::Write2 => Box::new(write_db2),
 | 
			
		||||
        Command::Print2 { id } => {
 | 
			
		||||
            let get_question = Box::new(|| read_from_db2(*id));
 | 
			
		||||
            Box::new(|| print_question_from(get_question))
 | 
			
		||||
        }
 | 
			
		||||
    };
 | 
			
		||||
 | 
			
		||||
    if args.measure {
 | 
			
		||||
@@ -231,11 +127,11 @@ fn main() {
 | 
			
		||||
    action();
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn read_from_db2(id: u32) -> Option<Question> {
 | 
			
		||||
fn read_from_db(id: u32) -> Option<Question> {
 | 
			
		||||
    let reader: db::Reader<Question> =
 | 
			
		||||
        db::Reader::new(NEW_DB_FILENAME, 2048).expect("new db reader");
 | 
			
		||||
 | 
			
		||||
    let mut questions = reader.iter();
 | 
			
		||||
    let mut questions = reader.into_iter();
 | 
			
		||||
 | 
			
		||||
    match id {
 | 
			
		||||
        0 => {
 | 
			
		||||
@@ -245,17 +141,17 @@ fn read_from_db2(id: u32) -> Option<Question> {
 | 
			
		||||
        _ => questions.nth((id - 1) as usize),
 | 
			
		||||
    }
 | 
			
		||||
}
 | 
			
		||||
fn write_db2() {
 | 
			
		||||
fn write_db() {
 | 
			
		||||
    let (tx, rx) = mpsc::channel::<Question>();
 | 
			
		||||
    [
 | 
			
		||||
        thread::spawn(move || zip_reader_task(tx)),
 | 
			
		||||
        thread::spawn(move || db_writer2_task(rx)),
 | 
			
		||||
        thread::spawn(move || db_writer_task(rx)),
 | 
			
		||||
    ]
 | 
			
		||||
    .into_iter()
 | 
			
		||||
    .for_each(|handle| handle.join().expect("thread panic"));
 | 
			
		||||
    println!("all done");
 | 
			
		||||
}
 | 
			
		||||
fn db_writer2_task(rx: mpsc::Receiver<Question>) {
 | 
			
		||||
fn db_writer_task(rx: mpsc::Receiver<Question>) {
 | 
			
		||||
    let writer_opts = db::WriterOpts::default();
 | 
			
		||||
    let mut writer: db::Writer<Question> =
 | 
			
		||||
        db::Writer::new(NEW_DB_FILENAME, writer_opts).expect("new db writer");
 | 
			
		||||
 
 | 
			
		||||
		Reference in New Issue
	
	Block a user