The length of values [default: 100].
- --get-prop= The relative proportion of get requests [default: 94].
- --set-prop=
The relative proportion of set requests [default: 2].
- --del-prop=
The relative proportion of del requests [default: 1].
- --cas-prop=
The relative proportion of cas requests [default: 1].
- --scan-prop=
The relative proportion of scan requests [default: 1].
- --merge-prop=
The relative proportion of merge requests [default: 1].
- --entries= The total keyspace [default: 100000].
- --sequential Run the test in sequential mode instead of random.
- --total-ops= Stop test after executing a total number of operations.
- --flush-every= Flush and sync the database every ms [default: 200].
- --cache-mb= Size of the page cache in megabytes [default: 1024].
-";
-
-#[derive(Debug, Clone, Copy)]
-struct Args {
- threads: usize,
- burn_in: bool,
- duration: u64,
- key_len: usize,
- val_len: usize,
- get_prop: usize,
- set_prop: usize,
- del_prop: usize,
- cas_prop: usize,
- scan_prop: usize,
- merge_prop: usize,
- entries: usize,
- sequential: bool,
- total_ops: Option,
- flush_every: u64,
- cache_mb: usize,
-}
-
-impl Default for Args {
- fn default() -> Args {
- Args {
- threads: 4,
- burn_in: false,
- duration: 10,
- key_len: 10,
- val_len: 100,
- get_prop: 94,
- set_prop: 2,
- del_prop: 1,
- cas_prop: 1,
- scan_prop: 1,
- merge_prop: 1,
- entries: 100000,
- sequential: false,
- total_ops: None,
- flush_every: 200,
- cache_mb: 1024,
- }
- }
-}
-
-fn parse<'a, I, T>(mut iter: I) -> T
-where
- I: Iterator- ,
- T: std::str::FromStr,
- ::Err: std::fmt::Debug,
-{
- iter.next().expect(USAGE).parse().expect(USAGE)
-}
-
-impl Args {
- fn parse() -> Args {
- let mut args = Args::default();
- for raw_arg in std::env::args().skip(1) {
- let mut splits = raw_arg[2..].split('=');
- match splits.next().unwrap() {
- "threads" => args.threads = parse(&mut splits),
- "burn-in" => args.burn_in = true,
- "duration" => args.duration = parse(&mut splits),
- "key-len" => args.key_len = parse(&mut splits),
- "val-len" => args.val_len = parse(&mut splits),
- "get-prop" => args.get_prop = parse(&mut splits),
- "set-prop" => args.set_prop = parse(&mut splits),
- "del-prop" => args.del_prop = parse(&mut splits),
- "cas-prop" => args.cas_prop = parse(&mut splits),
- "scan-prop" => args.scan_prop = parse(&mut splits),
- "merge-prop" => args.merge_prop = parse(&mut splits),
- "entries" => args.entries = parse(&mut splits),
- "sequential" => args.sequential = true,
- "total-ops" => args.total_ops = Some(parse(&mut splits)),
- "flush-every" => args.flush_every = parse(&mut splits),
- "cache-mb" => args.cache_mb = parse(&mut splits),
- other => panic!("unknown option: {}, {}", other, USAGE),
- }
- }
- args
- }
-}
-
-fn report(shutdown: Arc) {
- let mut last = 0;
- while !shutdown.load(Ordering::Relaxed) {
- thread::sleep(std::time::Duration::from_secs(1));
- let total = TOTAL.load(Ordering::Acquire);
-
- println!(
- "did {} ops, {}mb RSS",
- (total - last).to_formatted_string(&Locale::en),
- rss() / (1024 * 1024)
- );
-
- last = total;
- }
-}
-
-fn concatenate_merge(
- _key: &[u8], // the key being merged
- old_value: Option<&[u8]>, // the previous value, if one existed
- merged_bytes: &[u8], // the new bytes being merged in
-) -> Option> {
- // set the new value, return None to delete
- let mut ret = old_value.map(|ov| ov.to_vec()).unwrap_or_else(Vec::new);
-
- ret.extend_from_slice(merged_bytes);
-
- Some(ret)
-}
-
-fn run(args: Args, tree: Arc, shutdown: Arc) {
- let get_max = args.get_prop;
- let set_max = get_max + args.set_prop;
- let del_max = set_max + args.del_prop;
- let cas_max = del_max + args.cas_prop;
- let merge_max = cas_max + args.merge_prop;
- let scan_max = merge_max + args.scan_prop;
-
- let keygen = |len| -> sled::IVec {
- static SEQ: AtomicUsize = AtomicUsize::new(0);
- let i = if args.sequential {
- SEQ.fetch_add(1, Ordering::Relaxed)
- } else {
- thread_rng().gen::()
- } % args.entries;
-
- let start = if len < 8 { 8 - len } else { 0 };
-
- let i_keygen = &i.to_be_bytes()[start..];
-
- i_keygen.iter().cycle().take(len).copied().collect()
- };
-
- let valgen = |len| -> sled::IVec {
- if len == 0 {
- return vec![].into();
- }
-
- let i: usize = thread_rng().gen::() % (len * 8);
-
- let i_keygen = i.to_be_bytes();
-
- i_keygen
- .iter()
- .skip_while(|v| **v == 0)
- .cycle()
- .take(len)
- .copied()
- .collect()
- };
-
- let mut rng = thread_rng();
-
- while !shutdown.load(Ordering::Relaxed) {
- let op = TOTAL.fetch_add(1, Ordering::Release);
- let key = keygen(args.key_len);
- let choice = rng.gen_range(0, scan_max + 1);
-
- match choice {
- v if v <= get_max => {
- tree.get_zero_copy(&key, |_| {}).unwrap();
- }
- v if v > get_max && v <= set_max => {
- let value = valgen(args.val_len);
- tree.insert(&key, value).unwrap();
- }
- v if v > set_max && v <= del_max => {
- tree.remove(&key).unwrap();
- }
- v if v > del_max && v <= cas_max => {
- let old = if rng.gen::() {
- let value = valgen(args.val_len);
- Some(value)
- } else {
- None
- };
-
- let new = if rng.gen::() {
- let value = valgen(args.val_len);
- Some(value)
- } else {
- None
- };
-
- if let Err(e) = tree.compare_and_swap(&key, old, new) {
- panic!("operational error: {:?}", e);
- }
- }
- v if v > cas_max && v <= merge_max => {
- let value = valgen(args.val_len);
- tree.merge(&key, value).unwrap();
- }
- _ => {
- let iter = tree.range(key..).map(|res| res.unwrap());
-
- if op % 2 == 0 {
- let _ = iter.take(rng.gen_range(0, 15)).collect::>();
- } else {
- let _ = iter
- .rev()
- .take(rng.gen_range(0, 15))
- .collect::>();
- }
- }
- }
- }
-}
-
-fn rss() -> usize {
- #[cfg(target_os = "linux")]
- {
- use std::io::prelude::*;
- use std::io::BufReader;
-
- let mut buf = String::new();
- let mut f =
- BufReader::new(std::fs::File::open("/proc/self/statm").unwrap());
- f.read_line(&mut buf).unwrap();
- let mut parts = buf.split_whitespace();
- let rss_pages = parts.nth(1).unwrap().parse::().unwrap();
- rss_pages * 4096
- }
- #[cfg(not(target_os = "linux"))]
- {
- 0
- }
-}
-
-fn main() {
- #[cfg(feature = "logging")]
- setup_logger();
-
- #[cfg(feature = "dh")]
- let _dh = Dhat::start_heap_profiling();
-
- let args = Args::parse();
-
- let shutdown = Arc::new(AtomicBool::new(false));
-
- dbg!(args);
-
- let config = sled::Config::new()
- .cache_capacity(args.cache_mb * 1024 * 1024)
- .flush_every_ms(if args.flush_every == 0 {
- None
- } else {
- Some(args.flush_every)
- });
-
- let tree = Arc::new(config.open().unwrap());
- tree.set_merge_operator(concatenate_merge);
-
- let mut threads = vec![];
-
- let now = std::time::Instant::now();
-
- let n_threads = args.threads;
-
- for i in 0..=n_threads {
- let tree = tree.clone();
- let shutdown = shutdown.clone();
-
- let t = if i == 0 {
- thread::Builder::new()
- .name("reporter".into())
- .spawn(move || report(shutdown))
- .unwrap()
- } else {
- thread::spawn(move || run(args, tree, shutdown))
- };
-
- threads.push(t);
- }
-
- if let Some(ops) = args.total_ops {
- assert!(!args.burn_in, "don't set both --burn-in and --total-ops");
- while TOTAL.load(Ordering::Relaxed) < ops {
- thread::sleep(std::time::Duration::from_millis(50));
- }
- shutdown.store(true, Ordering::SeqCst);
- } else if !args.burn_in {
- thread::sleep(std::time::Duration::from_secs(args.duration));
- shutdown.store(true, Ordering::SeqCst);
- }
-
- for t in threads.into_iter() {
- t.join().unwrap();
- }
- let ops = TOTAL.load(Ordering::SeqCst);
- let time = now.elapsed().as_secs() as usize;
-
- println!(
- "did {} total ops in {} seconds. {} ops/s",
- ops.to_formatted_string(&Locale::en),
- time,
- ((ops * 1_000) / (time * 1_000)).to_formatted_string(&Locale::en)
- );
-
- #[cfg(feature = "measure_allocs")]
- println!(
- "allocated {} bytes in {} allocations",
- alloc::ALLOCATED_BYTES
- .load(Ordering::Acquire)
- .to_formatted_string(&Locale::en),
- alloc::ALLOCATIONS
- .load(Ordering::Acquire)
- .to_formatted_string(&Locale::en),
- );
-
- #[cfg(feature = "metrics")]
- sled::print_profile();
-}
-
-#[cfg(feature = "logging")]
-pub fn setup_logger() {
- use std::io::Write;
-
- color_backtrace::install();
-
- fn tn() -> String {
- std::thread::current().name().unwrap_or("unknown").to_owned()
- }
-
- let mut builder = env_logger::Builder::new();
- builder
- .format(|buf, record| {
- writeln!(
- buf,
- "{:05} {:25} {:10} {}",
- record.level(),
- tn(),
- record.module_path().unwrap().split("::").last().unwrap(),
- record.args()
- )
- })
- .filter(None, log::LevelFilter::Info);
-
- if let Ok(env) = std::env::var("RUST_LOG") {
- builder.parse_filters(&env);
- }
-
- let _r = builder.try_init();
-}
diff --git a/benchmarks/stress2/tsan.sh b/benchmarks/stress2/tsan.sh
deleted file mode 100755
index f71d53345..000000000
--- a/benchmarks/stress2/tsan.sh
+++ /dev/null
@@ -1,10 +0,0 @@
-#!/usr/bin/env bash
-
-set -euxo pipefail
-
-echo "tsan"
-export RUSTFLAGS="-Z sanitizer=thread"
-export TSAN_OPTIONS="suppressions=/home/t/src/sled/tsan_suppressions.txt"
-sudo rm -rf default.sled
-cargo +nightly run --features=lock_free_delays,no_jemalloc --target x86_64-unknown-linux-gnu -- --duration=6
-cargo +nightly run --features=lock_free_delays,no_jemalloc --target x86_64-unknown-linux-gnu -- --duration=6
diff --git a/bindings/sled-native/Cargo.toml b/bindings/sled-native/Cargo.toml
deleted file mode 100644
index d8a6d800a..000000000
--- a/bindings/sled-native/Cargo.toml
+++ /dev/null
@@ -1,19 +0,0 @@
-[package]
-name = "sled-native"
-version = "0.34.6"
-authors = ["Tyler Neely "]
-description = "a C-compatible API for sled"
-license = "Apache-2.0"
-homepage = "https://github.com/spacejam/sled"
-repository = "https://github.com/spacejam/sled/sled-native"
-keywords = ["database", "embedded", "concurrent", "persistent", "c"]
-documentation = "https://docs.rs/sled-native/"
-edition = "2018"
-
-[lib]
-name = "sled"
-crate-type = ["cdylib", "staticlib"]
-
-[dependencies]
-libc = "0.2.62"
-sled = {version = "0.34.6", path = "../.."}
diff --git a/bindings/sled-native/README.md b/bindings/sled-native/README.md
deleted file mode 100644
index 980f43a75..000000000
--- a/bindings/sled-native/README.md
+++ /dev/null
@@ -1,11 +0,0 @@
-# Native C-API for sled
-
-## Building
-
-```
-$ cargo install cargo-c
-$ cargo cinstall --prefix=/usr --destdir=/tmp/staging
-$ sudo cp -a /tmp/staging/* /
-```
-
-
diff --git a/bindings/sled-native/cbindgen.toml b/bindings/sled-native/cbindgen.toml
deleted file mode 100644
index d8657dd8a..000000000
--- a/bindings/sled-native/cbindgen.toml
+++ /dev/null
@@ -1,20 +0,0 @@
-header = "// SPDX-License-Identifier: Apache-2.0"
-sys_includes = ["stddef.h", "stdint.h", "stdlib.h"]
-no_includes = true
-include_guard = "SLED_H"
-tab_width = 4
-style = "Type"
-# language = "C"
-cpp_compat = true
-
-[parse]
-parse_deps = true
-include = ['sled']
-
-[export]
-prefix = "Sled"
-item_types = ["enums", "structs", "unions", "typedefs", "opaque", "functions"]
-
-[enum]
-rename_variants = "ScreamingSnakeCase"
-prefix_with_name = true
diff --git a/bindings/sled-native/src/lib.rs b/bindings/sled-native/src/lib.rs
deleted file mode 100644
index c5dc6dcc6..000000000
--- a/bindings/sled-native/src/lib.rs
+++ /dev/null
@@ -1,236 +0,0 @@
-use sled;
-
-use std::ffi::CString;
-use std::mem;
-use std::ptr;
-use std::slice;
-
-use libc::*;
-
-use sled::{Config, Db, IVec, Iter};
-
-fn leak_buf(v: Vec, vallen: *mut size_t) -> *mut c_char {
- unsafe {
- *vallen = v.len();
- }
- let mut bsv = v.into_boxed_slice();
- let val = bsv.as_mut_ptr() as *mut _;
- mem::forget(bsv);
- val
-}
-
-/// Create a new configuration.
-#[no_mangle]
-pub unsafe extern "C" fn sled_create_config() -> *mut Config {
- Box::into_raw(Box::new(Config::new()))
-}
-
-/// Destroy a configuration.
-#[no_mangle]
-pub unsafe extern "C" fn sled_free_config(config: *mut Config) {
- drop(Box::from_raw(config));
-}
-
-/// Set the configured file path. The caller is responsible for freeing the path
-/// string after calling this (it is copied in this function).
-#[no_mangle]
-pub unsafe extern "C" fn sled_config_set_path(
- config: *mut Config,
- path: *const c_char,
-) -> *mut Config {
- let c_str = CString::from_raw(path as *mut _);
- let value = c_str.into_string().unwrap();
-
- let config = Box::from_raw(config);
- Box::into_raw(Box::from(config.path(value)))
-}
-
-/// Set the configured cache capacity in bytes.
-#[no_mangle]
-pub unsafe extern "C" fn sled_config_set_cache_capacity(
- config: *mut Config,
- capacity: size_t,
-) -> *mut Config {
- let config = Box::from_raw(config);
- Box::into_raw(Box::from(config.cache_capacity(capacity as u64)))
-}
-
-/// Configure the use of the zstd compression library.
-#[no_mangle]
-pub unsafe extern "C" fn sled_config_use_compression(
- config: *mut Config,
- use_compression: c_uchar,
-) -> *mut Config {
- let config = Box::from_raw(config);
- Box::into_raw(Box::from(config.use_compression(use_compression == 1)))
-}
-
-/// Set the configured IO buffer flush interval in milliseconds.
-#[no_mangle]
-pub unsafe extern "C" fn sled_config_flush_every_ms(
- config: *mut Config,
- flush_every: c_int,
-) -> *mut Config {
- let val = if flush_every < 0 { None } else { Some(flush_every as u64) };
- let config = Box::from_raw(config);
- Box::into_raw(Box::from(config.flush_every_ms(val)))
-}
-
-/// Open a sled lock-free log-structured tree. Consumes the passed-in config.
-#[no_mangle]
-pub unsafe extern "C" fn sled_open_db(config: *mut Config) -> *mut Db {
- let config = Box::from_raw(config);
- Box::into_raw(Box::new(config.open().unwrap()))
-}
-
-/// Close a sled lock-free log-structured tree.
-#[no_mangle]
-pub unsafe extern "C" fn sled_close(db: *mut Db) {
- drop(Box::from_raw(db));
-}
-
-/// Free a buffer originally allocated by sled.
-#[no_mangle]
-pub unsafe extern "C" fn sled_free_buf(buf: *mut c_char, sz: size_t) {
- drop(Vec::from_raw_parts(buf, sz, sz));
-}
-
-/// Free an iterator.
-#[no_mangle]
-pub unsafe extern "C" fn sled_free_iter(iter: *mut Iter) {
- drop(Box::from_raw(iter));
-}
-
-/// Set a key to a value.
-#[no_mangle]
-pub unsafe extern "C" fn sled_set(
- db: *mut Db,
- key: *const c_uchar,
- keylen: size_t,
- val: *const c_uchar,
- vallen: size_t,
-) {
- let k = IVec::from(slice::from_raw_parts(key, keylen));
- let v = IVec::from(slice::from_raw_parts(val, vallen));
- (*db).insert(k, v).unwrap();
-}
-
-/// Get the value of a key.
-/// Caller is responsible for freeing the returned value with `sled_free_buf` if
-/// it's non-null.
-#[no_mangle]
-pub unsafe extern "C" fn sled_get(
- db: *mut Db,
- key: *const c_char,
- keylen: size_t,
- vallen: *mut size_t,
-) -> *mut c_char {
- let k = slice::from_raw_parts(key as *const u8, keylen);
- let res = (*db).get(k);
- match res {
- Ok(Some(v)) => leak_buf(v.to_vec(), vallen),
- Ok(None) => ptr::null_mut(),
- // TODO proper error propagation
- Err(e) => panic!("{:?}", e),
- }
-}
-
-/// Delete the value of a key.
-#[no_mangle]
-pub unsafe extern "C" fn sled_del(
- db: *mut Db,
- key: *const c_char,
- keylen: size_t,
-) {
- let k = slice::from_raw_parts(key as *const u8, keylen);
- (*db).remove(k).unwrap();
-}
-
-/// Compare and swap.
-/// Returns 1 if successful, 0 if unsuccessful.
-/// Otherwise sets `actual_val` and `actual_vallen` to the current value,
-/// which must be freed using `sled_free_buf` by the caller if non-null.
-/// `actual_val` will be null and `actual_vallen` 0 if the current value is not
-/// set.
-#[no_mangle]
-pub unsafe extern "C" fn sled_compare_and_swap(
- db: *mut Db,
- key: *const c_char,
- keylen: size_t,
- old_val: *const c_uchar,
- old_vallen: size_t,
- new_val: *const c_uchar,
- new_vallen: size_t,
- actual_val: *mut *const c_uchar,
- actual_vallen: *mut size_t,
-) -> c_uchar {
- let k = IVec::from(slice::from_raw_parts(key as *const u8, keylen));
-
- let old = if old_vallen == 0 {
- None
- } else {
- let copy =
- IVec::from(slice::from_raw_parts(old_val as *const u8, old_vallen));
- Some(copy)
- };
-
- let new = if new_vallen == 0 {
- None
- } else {
- let copy =
- IVec::from(slice::from_raw_parts(new_val as *const u8, new_vallen));
- Some(copy)
- };
-
- let res = (*db).compare_and_swap(k, old, new);
-
- match res {
- Ok(Ok(())) => 1,
- Ok(Err(sled::CompareAndSwapError { current: None, .. })) => {
- *actual_vallen = 0;
- 0
- }
- Ok(Err(sled::CompareAndSwapError { current: Some(v), .. })) => {
- *actual_val = leak_buf(v.to_vec(), actual_vallen) as *const u8;
- 0
- }
- // TODO proper error propagation
- Err(e) => panic!("{:?}", e),
- }
-}
-
-/// Iterate over tuples which have specified key prefix.
-/// Caller is responsible for freeing the returned iterator with
-/// `sled_free_iter`.
-#[no_mangle]
-pub unsafe extern "C" fn sled_scan_prefix(
- db: *mut Db,
- key: *const c_char,
- keylen: size_t,
-) -> *mut Iter {
- let k = slice::from_raw_parts(key as *const u8, keylen);
- Box::into_raw(Box::new((*db).scan_prefix(k)))
-}
-
-/// Get they next kv pair from an iterator.
-/// Caller is responsible for freeing the key and value with `sled_free_buf`.
-/// Returns 0 when exhausted.
-#[no_mangle]
-pub unsafe extern "C" fn sled_iter_next(
- iter: *mut Iter,
- key: *mut *const c_char,
- keylen: *mut size_t,
- val: *mut *const c_char,
- vallen: *mut size_t,
-) -> c_uchar {
- match (*iter).next() {
- Some(Ok((k, v))) => {
- *key = leak_buf(k.to_vec(), keylen);
- *val = leak_buf(v.to_vec(), vallen);
- 1
- }
- // TODO proper error propagation
- Some(Err(e)) => panic!("{:?}", e),
- None => 0,
- }
-}
diff --git a/examples/bench.rs b/examples/bench.rs
new file mode 100644
index 000000000..c524ab909
--- /dev/null
+++ b/examples/bench.rs
@@ -0,0 +1,610 @@
+use std::path::Path;
+use std::sync::Barrier;
+use std::thread::scope;
+use std::time::{Duration, Instant};
+use std::{fs, io};
+
+use num_format::{Locale, ToFormattedString};
+
+use sled::{Config, Db as SledDb};
+
+type Db = SledDb<1024>;
+
+const N_WRITES_PER_THREAD: u32 = 4 * 1024 * 1024;
+const MAX_CONCURRENCY: u32 = 4;
+const CONCURRENCY: &[usize] = &[/*1, 2, 4,*/ MAX_CONCURRENCY as _];
+const BYTES_PER_ITEM: u32 = 8;
+
+trait Databench: Clone + Send {
+ type READ: AsRef<[u8]>;
+ const NAME: &'static str;
+ const PATH: &'static str;
+ fn open() -> Self;
+ fn remove_generic(&self, key: &[u8]);
+ fn insert_generic(&self, key: &[u8], value: &[u8]);
+ fn get_generic(&self, key: &[u8]) -> Option;
+ fn flush_generic(&self);
+ fn print_stats(&self);
+}
+
+impl Databench for Db {
+ type READ = sled::InlineArray;
+
+ const NAME: &'static str = "sled 1.0.0-alpha";
+ const PATH: &'static str = "timing_test.sled-new";
+
+ fn open() -> Self {
+ sled::Config {
+ path: Self::PATH.into(),
+ zstd_compression_level: 3,
+ cache_capacity_bytes: 1024 * 1024 * 1024,
+ entry_cache_percent: 20,
+ flush_every_ms: Some(200),
+ ..Config::default()
+ }
+ .open()
+ .unwrap()
+ }
+
+ fn insert_generic(&self, key: &[u8], value: &[u8]) {
+ self.insert(key, value).unwrap();
+ }
+ fn remove_generic(&self, key: &[u8]) {
+ self.remove(key).unwrap();
+ }
+ fn get_generic(&self, key: &[u8]) -> Option {
+ self.get(key).unwrap()
+ }
+ fn flush_generic(&self) {
+ self.flush().unwrap();
+ }
+ fn print_stats(&self) {
+ dbg!(self.stats());
+ }
+}
+
+/*
+impl Databench for old_sled::Db {
+ type READ = old_sled::IVec;
+
+ const NAME: &'static str = "sled 0.34.7";
+ const PATH: &'static str = "timing_test.sled-old";
+
+ fn open() -> Self {
+ old_sled::open(Self::PATH).unwrap()
+ }
+ fn insert_generic(&self, key: &[u8], value: &[u8]) {
+ self.insert(key, value).unwrap();
+ }
+ fn get_generic(&self, key: &[u8]) -> Option {
+ self.get(key).unwrap()
+ }
+ fn flush_generic(&self) {
+ self.flush().unwrap();
+ }
+}
+*/
+
+/*
+impl Databench for Arc {
+ type READ = Vec;
+
+ const NAME: &'static str = "rocksdb 0.21.0";
+ const PATH: &'static str = "timing_test.rocksdb";
+
+ fn open() -> Self {
+ Arc::new(rocksdb::DB::open_default(Self::PATH).unwrap())
+ }
+ fn insert_generic(&self, key: &[u8], value: &[u8]) {
+ self.put(key, value).unwrap();
+ }
+ fn get_generic(&self, key: &[u8]) -> Option {
+ self.get(key).unwrap()
+ }
+ fn flush_generic(&self) {
+ self.flush().unwrap();
+ }
+}
+*/
+
+/*
+struct Lmdb {
+ env: heed::Env,
+ db: heed::Database<
+ heed::types::UnalignedSlice,
+ heed::types::UnalignedSlice,
+ >,
+}
+
+impl Clone for Lmdb {
+ fn clone(&self) -> Lmdb {
+ Lmdb { env: self.env.clone(), db: self.db.clone() }
+ }
+}
+
+impl Databench for Lmdb {
+ type READ = Vec;
+
+ const NAME: &'static str = "lmdb";
+ const PATH: &'static str = "timing_test.lmdb";
+
+ fn open() -> Self {
+ let _ = std::fs::create_dir_all(Self::PATH);
+ let env = heed::EnvOpenOptions::new()
+ .map_size(1024 * 1024 * 1024)
+ .open(Self::PATH)
+ .unwrap();
+ let db = env.create_database(None).unwrap();
+ Lmdb { env, db }
+ }
+ fn insert_generic(&self, key: &[u8], value: &[u8]) {
+ let mut wtxn = self.env.write_txn().unwrap();
+ self.db.put(&mut wtxn, key, value).unwrap();
+ wtxn.commit().unwrap();
+ }
+ fn get_generic(&self, key: &[u8]) -> Option {
+ let rtxn = self.env.read_txn().unwrap();
+ let ret = self.db.get(&rtxn, key).unwrap().map(Vec::from);
+ rtxn.commit().unwrap();
+ ret
+ }
+ fn flush_generic(&self) {
+ // NOOP
+ }
+}
+*/
+
+/*
+struct Sqlite {
+ connection: rusqlite::Connection,
+}
+
+impl Clone for Sqlite {
+ fn clone(&self) -> Sqlite {
+ Sqlite { connection: rusqlite::Connection::open(Self::PATH).unwrap() }
+ }
+}
+
+impl Databench for Sqlite {
+ type READ = Vec;
+
+ const NAME: &'static str = "sqlite";
+ const PATH: &'static str = "timing_test.sqlite";
+
+ fn open() -> Self {
+ let connection = rusqlite::Connection::open(Self::PATH).unwrap();
+ connection
+ .execute(
+ "create table if not exists bench (
+ key integer primary key,
+ val integer not null
+ )",
+ [],
+ )
+ .unwrap();
+ Sqlite { connection }
+ }
+ fn insert_generic(&self, key: &[u8], value: &[u8]) {
+ loop {
+ let res = self.connection.execute(
+ "insert or ignore into bench (key, val) values (?1, ?2)",
+ [
+ format!("{}", u32::from_be_bytes(key.try_into().unwrap())),
+ format!(
+ "{}",
+ u32::from_be_bytes(value.try_into().unwrap())
+ ),
+ ],
+ );
+ if res.is_ok() {
+ break;
+ }
+ }
+ }
+ fn get_generic(&self, key: &[u8]) -> Option {
+ let mut stmt = self
+ .connection
+ .prepare("SELECT b.val from bench b WHERE key = ?1")
+ .unwrap();
+ let mut rows =
+ stmt.query([u32::from_be_bytes(key.try_into().unwrap())]).unwrap();
+
+ let value = rows.next().unwrap()?;
+ value.get(0).ok()
+ }
+ fn flush_generic(&self) {
+ // NOOP
+ }
+}
+*/
+
+fn allocated() -> usize {
+ #[cfg(feature = "testing-count-allocator")]
+ {
+ return sled::alloc::allocated();
+ }
+ 0
+}
+
+fn freed() -> usize {
+ #[cfg(feature = "testing-count-allocator")]
+ {
+ return sled::alloc::freed();
+ }
+ 0
+}
+
+fn resident() -> usize {
+ #[cfg(feature = "testing-count-allocator")]
+ {
+ return sled::alloc::resident();
+ }
+ 0
+}
+
+fn inserts(store: &D) -> Vec {
+ println!("{} inserts", D::NAME);
+ let mut i = 0_u32;
+
+ let factory = move || {
+ i += 1;
+ (store.clone(), i - 1)
+ };
+
+ let f = |state: (D, u32)| {
+ let (store, offset) = state;
+ let start = N_WRITES_PER_THREAD * offset;
+ let end = N_WRITES_PER_THREAD * (offset + 1);
+ for i in start..end {
+ let k: &[u8] = &i.to_be_bytes();
+ store.insert_generic(k, k);
+ }
+ };
+
+ let mut ret = vec![];
+
+ for concurrency in CONCURRENCY {
+ let insert_elapsed =
+ execute_lockstep_concurrent(factory, f, *concurrency);
+
+ let flush_timer = Instant::now();
+ store.flush_generic();
+
+ let wps = (N_WRITES_PER_THREAD * *concurrency as u32) as u64
+ * 1_000_000_u64
+ / u64::try_from(insert_elapsed.as_micros().max(1))
+ .unwrap_or(u64::MAX);
+
+ ret.push(InsertStats {
+ thread_count: *concurrency,
+ inserts_per_second: wps,
+ });
+
+ println!(
+ "{} inserts/s with {concurrency} threads over {:?}, then {:?} to flush {}",
+ wps.to_formatted_string(&Locale::en),
+ insert_elapsed,
+ flush_timer.elapsed(),
+ D::NAME,
+ );
+ }
+
+ ret
+}
+
+fn removes(store: &D) -> Vec {
+ println!("{} removals", D::NAME);
+ let mut i = 0_u32;
+
+ let factory = move || {
+ i += 1;
+ (store.clone(), i - 1)
+ };
+
+ let f = |state: (D, u32)| {
+ let (store, offset) = state;
+ let start = N_WRITES_PER_THREAD * offset;
+ let end = N_WRITES_PER_THREAD * (offset + 1);
+ for i in start..end {
+ let k: &[u8] = &i.to_be_bytes();
+ store.remove_generic(k);
+ }
+ };
+
+ let mut ret = vec![];
+
+ for concurrency in CONCURRENCY {
+ let remove_elapsed =
+ execute_lockstep_concurrent(factory, f, *concurrency);
+
+ let flush_timer = Instant::now();
+ store.flush_generic();
+
+ let wps = (N_WRITES_PER_THREAD * *concurrency as u32) as u64
+ * 1_000_000_u64
+ / u64::try_from(remove_elapsed.as_micros().max(1))
+ .unwrap_or(u64::MAX);
+
+ ret.push(RemoveStats {
+ thread_count: *concurrency,
+ removes_per_second: wps,
+ });
+
+ println!(
+ "{} removes/s with {concurrency} threads over {:?}, then {:?} to flush {}",
+ wps.to_formatted_string(&Locale::en),
+ remove_elapsed,
+ flush_timer.elapsed(),
+ D::NAME,
+ );
+ }
+
+ ret
+}
+
+fn gets(store: &D) -> Vec {
+ println!("{} reads", D::NAME);
+
+ let factory = || store.clone();
+
+ let f = |store: D| {
+ let start = 0;
+ let end = N_WRITES_PER_THREAD * MAX_CONCURRENCY;
+ for i in start..end {
+ let k: &[u8] = &i.to_be_bytes();
+ store.get_generic(k);
+ }
+ };
+
+ let mut ret = vec![];
+
+ for concurrency in CONCURRENCY {
+ let get_stone_elapsed =
+ execute_lockstep_concurrent(factory, f, *concurrency);
+
+ let rps = (N_WRITES_PER_THREAD * MAX_CONCURRENCY * *concurrency as u32)
+ as u64
+ * 1_000_000_u64
+ / u64::try_from(get_stone_elapsed.as_micros().max(1))
+ .unwrap_or(u64::MAX);
+
+ ret.push(GetStats { thread_count: *concurrency, gets_per_second: rps });
+
+ println!(
+ "{} gets/s with concurrency of {concurrency}, {:?} total reads {}",
+ rps.to_formatted_string(&Locale::en),
+ get_stone_elapsed,
+ D::NAME
+ );
+ }
+ ret
+}
+
+fn execute_lockstep_concurrent<
+ State: Send,
+ Factory: FnMut() -> State,
+ F: Sync + Fn(State),
+>(
+ mut factory: Factory,
+ f: F,
+ concurrency: usize,
+) -> Duration {
+ let barrier = &Barrier::new(concurrency + 1);
+ let f = &f;
+
+ scope(|s| {
+ let mut threads = vec![];
+
+ for _ in 0..concurrency {
+ let state = factory();
+
+ let thread = s.spawn(move || {
+ barrier.wait();
+ f(state);
+ });
+
+ threads.push(thread);
+ }
+
+ barrier.wait();
+ let get_stone = Instant::now();
+
+ for thread in threads.into_iter() {
+ thread.join().unwrap();
+ }
+
+ get_stone.elapsed()
+ })
+}
+
+#[derive(Debug, Clone, Copy)]
+struct InsertStats {
+ thread_count: usize,
+ inserts_per_second: u64,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct GetStats {
+ thread_count: usize,
+ gets_per_second: u64,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct RemoveStats {
+ thread_count: usize,
+ removes_per_second: u64,
+}
+
+#[allow(unused)]
+#[derive(Debug, Clone)]
+struct Stats {
+ post_insert_disk_space: u64,
+ post_remove_disk_space: u64,
+ allocated_memory: usize,
+ freed_memory: usize,
+ resident_memory: usize,
+ insert_stats: Vec,
+ get_stats: Vec,
+ remove_stats: Vec,
+}
+
+impl Stats {
+ fn print_report(&self) {
+ println!(
+ "bytes on disk after inserts: {}",
+ self.post_insert_disk_space.to_formatted_string(&Locale::en)
+ );
+ println!(
+ "bytes on disk after removes: {}",
+ self.post_remove_disk_space.to_formatted_string(&Locale::en)
+ );
+ println!(
+ "bytes in memory: {}",
+ self.resident_memory.to_formatted_string(&Locale::en)
+ );
+ for stats in &self.insert_stats {
+ println!(
+ "{} threads {} inserts per second",
+ stats.thread_count,
+ stats.inserts_per_second.to_formatted_string(&Locale::en)
+ );
+ }
+ for stats in &self.get_stats {
+ println!(
+ "{} threads {} gets per second",
+ stats.thread_count,
+ stats.gets_per_second.to_formatted_string(&Locale::en)
+ );
+ }
+ for stats in &self.remove_stats {
+ println!(
+ "{} threads {} removes per second",
+ stats.thread_count,
+ stats.removes_per_second.to_formatted_string(&Locale::en)
+ );
+ }
+ }
+}
+
+fn bench() -> Stats {
+ let store = D::open();
+
+ let insert_stats = inserts(&store);
+
+ let before_flush = Instant::now();
+ store.flush_generic();
+ println!("final flush took {:?} for {}", before_flush.elapsed(), D::NAME);
+
+ let post_insert_disk_space = du(D::PATH.as_ref()).unwrap();
+
+ let get_stats = gets(&store);
+
+ let remove_stats = removes(&store);
+
+ store.print_stats();
+
+ Stats {
+ post_insert_disk_space,
+ post_remove_disk_space: du(D::PATH.as_ref()).unwrap(),
+ allocated_memory: allocated(),
+ freed_memory: freed(),
+ resident_memory: resident(),
+ insert_stats,
+ get_stats,
+ remove_stats,
+ }
+}
+
+fn du(path: &Path) -> io::Result {
+ fn recurse(mut dir: fs::ReadDir) -> io::Result {
+ dir.try_fold(0, |acc, file| {
+ let file = file?;
+ let size = match file.metadata()? {
+ data if data.is_dir() => recurse(fs::read_dir(file.path())?)?,
+ data => data.len(),
+ };
+ Ok(acc + size)
+ })
+ }
+
+ recurse(fs::read_dir(path)?)
+}
+
+fn main() {
+ let _ = env_logger::try_init();
+
+ let new_stats = bench::();
+
+ println!(
+ "raw data size: {}",
+ (MAX_CONCURRENCY * N_WRITES_PER_THREAD * BYTES_PER_ITEM)
+ .to_formatted_string(&Locale::en)
+ );
+ println!("sled 1.0 space stats:");
+ new_stats.print_report();
+
+ /*
+ let old_stats = bench::();
+ dbg!(old_stats);
+
+ let new_sled_vs_old_sled_storage_ratio =
+ new_stats.disk_space as f64 / old_stats.disk_space as f64;
+ let new_sled_vs_old_sled_allocated_memory_ratio =
+ new_stats.allocated_memory as f64 / old_stats.allocated_memory as f64;
+ let new_sled_vs_old_sled_freed_memory_ratio =
+ new_stats.freed_memory as f64 / old_stats.freed_memory as f64;
+ let new_sled_vs_old_sled_resident_memory_ratio =
+ new_stats.resident_memory as f64 / old_stats.resident_memory as f64;
+
+ dbg!(new_sled_vs_old_sled_storage_ratio);
+ dbg!(new_sled_vs_old_sled_allocated_memory_ratio);
+ dbg!(new_sled_vs_old_sled_freed_memory_ratio);
+ dbg!(new_sled_vs_old_sled_resident_memory_ratio);
+
+ let rocksdb_stats = bench::>();
+
+ bench::();
+
+ bench::();
+ */
+
+ /*
+ let new_sled_vs_rocksdb_storage_ratio =
+ new_stats.disk_space as f64 / rocksdb_stats.disk_space as f64;
+ let new_sled_vs_rocksdb_allocated_memory_ratio =
+ new_stats.allocated_memory as f64 / rocksdb_stats.allocated_memory as f64;
+ let new_sled_vs_rocksdb_freed_memory_ratio =
+ new_stats.freed_memory as f64 / rocksdb_stats.freed_memory as f64;
+ let new_sled_vs_rocksdb_resident_memory_ratio =
+ new_stats.resident_memory as f64 / rocksdb_stats.resident_memory as f64;
+
+ dbg!(new_sled_vs_rocksdb_storage_ratio);
+ dbg!(new_sled_vs_rocksdb_allocated_memory_ratio);
+ dbg!(new_sled_vs_rocksdb_freed_memory_ratio);
+ dbg!(new_sled_vs_rocksdb_resident_memory_ratio);
+ */
+
+ /*
+ let scan = Instant::now();
+ let count = stone.iter().count();
+ assert_eq!(count as u64, N_WRITES_PER_THREAD);
+ let scan_elapsed = scan.elapsed();
+ println!(
+ "{} scanned items/s, total {:?}",
+ (N_WRITES_PER_THREAD * 1_000_000) / u64::try_from(scan_elapsed.as_micros().max(1)).unwrap_or(u64::MAX),
+ scan_elapsed
+ );
+ */
+
+ /*
+ let scan_rev = Instant::now();
+ let count = stone.range(..).rev().count();
+ assert_eq!(count as u64, N_WRITES_PER_THREAD);
+ let scan_rev_elapsed = scan_rev.elapsed();
+ println!(
+ "{} reverse-scanned items/s, total {:?}",
+ (N_WRITES_PER_THREAD * 1_000_000) / u64::try_from(scan_rev_elapsed.as_micros().max(1)).unwrap_or(u64::MAX),
+ scan_rev_elapsed
+ );
+ */
+}
diff --git a/examples/playground.rs b/examples/playground.rs
deleted file mode 100644
index 78afa41f7..000000000
--- a/examples/playground.rs
+++ /dev/null
@@ -1,81 +0,0 @@
-extern crate sled;
-
-use sled::{Config, Result};
-
-fn basic() -> Result<()> {
- let config = Config::new().temporary(true);
-
- let db = config.open()?;
-
- let k = b"k".to_vec();
- let v1 = b"v1".to_vec();
- let v2 = b"v2".to_vec();
-
- // set and get
- db.insert(k.clone(), v1.clone())?;
- assert_eq!(db.get(&k).unwrap().unwrap(), (v1));
-
- // compare and swap
- match db.compare_and_swap(k.clone(), Some(&v1), Some(v2.clone()))? {
- Ok(()) => println!("it worked!"),
- Err(sled::CompareAndSwapError { current: cur, proposed: _ }) => {
- println!("the actual current value is {:?}", cur)
- }
- }
-
- // scan forward
- let mut iter = db.range(k.as_slice()..);
- let (k1, v1) = iter.next().unwrap().unwrap();
- assert_eq!(v1, v2);
- assert_eq!(k1, k);
- assert_eq!(iter.next(), None);
-
- // deletion
- db.remove(&k)?;
-
- Ok(())
-}
-
-fn merge_operator() -> Result<()> {
- fn concatenate_merge(
- _key: &[u8], // the key being merged
- old_value: Option<&[u8]>, // the previous value, if one existed
- merged_bytes: &[u8], // the new bytes being merged in
- ) -> Option> {
- // set the new value, return None to delete
- let mut ret = old_value.map_or_else(Vec::new, |ov| ov.to_vec());
-
- ret.extend_from_slice(merged_bytes);
-
- Some(ret)
- }
-
- let config = Config::new().temporary(true);
-
- let db = config.open()?;
- db.set_merge_operator(concatenate_merge);
-
- let k = b"k".to_vec();
-
- db.insert(k.clone(), vec![0])?;
- db.merge(k.clone(), vec![1])?;
- db.merge(k.clone(), vec![2])?;
- assert_eq!(db.get(&*k).unwrap().unwrap(), (vec![0, 1, 2]));
-
- // sets replace previously merged data,
- // bypassing the merge function.
- db.insert(k.clone(), vec![3])?;
- assert_eq!(db.get(&*k).unwrap().unwrap(), (vec![3]));
-
- // merges on non-present values will add them
- db.remove(&*k)?;
- db.merge(k.clone(), vec![4])?;
- assert_eq!(db.get(&*k).unwrap().unwrap(), (vec![4]));
-
- Ok(())
-}
-
-fn main() -> Result<()> {
- basic()?;
- merge_operator()
-}
diff --git a/examples/structured.rs b/examples/structured.rs
deleted file mode 100644
index 7b900915f..000000000
--- a/examples/structured.rs
+++ /dev/null
@@ -1,238 +0,0 @@
-//! This example demonstrates how to work with structured
-//! keys and values without paying expensive (de)serialization
-//! costs.
-//!
-//! The `upsert` function shows how to use structured keys and values.
-//!
-//! The `variable_lengths` function shows how to put a variable length
-//! component in either the beginning or the end of your value.
-//!
-//! The `hash_join` function shows how to do some SQL-like joins.
-//!
-//! Running this example several times via `cargo run --example structured`
-//! will initialize the count field to 0, and on subsequent runs it will
-//! increment it.
-use {
- byteorder::{BigEndian, LittleEndian},
- zerocopy::{
- byteorder::U64, AsBytes, FromBytes, LayoutVerified, Unaligned, U16, U32,
- },
-};
-
-fn upsert(db: &sled::Db) -> sled::Result<()> {
- // We use `BigEndian` for key types because
- // they preserve lexicographic ordering,
- // which is nice if we ever want to iterate
- // over our items in order. We use the
- // `U64` type from zerocopy because it
- // does not have alignment requirements.
- // sled does not guarantee any particular
- // value alignment as of now.
- #[derive(FromBytes, AsBytes, Unaligned)]
- #[repr(C)]
- struct Key {
- a: U64,
- b: U64,
- }
-
- // We use `LittleEndian` for values because
- // it's possibly cheaper, but the difference
- // isn't likely to be measurable, so honestly
- // use whatever you want for values.
- #[derive(FromBytes, AsBytes, Unaligned)]
- #[repr(C)]
- struct Value {
- count: U64,
- whatever: [u8; 16],
- }
-
- let key = Key { a: U64::new(21), b: U64::new(890) };
-
- // "UPSERT" functionality
- db.update_and_fetch(key.as_bytes(), |value_opt| {
- if let Some(existing) = value_opt {
- // We need to make a copy that will be written back
- // into the database. This allows other threads that
- // may have witnessed the old version to keep working
- // without taking out any locks. IVec will be
- // stack-allocated until it reaches 22 bytes
- let mut backing_bytes = sled::IVec::from(existing);
-
- // this verifies that our value is the correct length
- // and alignment (in this case we don't need it to be
- // aligned, because we use the `U64` type from zerocopy)
- let layout: LayoutVerified<&mut [u8], Value> =
- LayoutVerified::new_unaligned(&mut *backing_bytes)
- .expect("bytes do not fit schema");
-
- // this lets us work with the underlying bytes as
- // a mutable structured value.
- let value: &mut Value = layout.into_mut();
-
- let new_count = value.count.get() + 1;
-
- println!("incrementing count to {}", new_count);
-
- value.count.set(new_count);
-
- Some(backing_bytes)
- } else {
- println!("setting count to 0");
-
- Some(sled::IVec::from(
- Value { count: U64::new(0), whatever: [0; 16] }.as_bytes(),
- ))
- }
- })?;
-
- Ok(())
-}
-
-// Cat values will be:
-// favorite_number + battles_won +
-#[derive(FromBytes, AsBytes, Unaligned)]
-#[repr(C)]
-struct CatValue {
- favorite_number: U64,
- battles_won: U64,
-}
-
-// Dog values will be:
-// + woof_count + postal_code
-#[derive(FromBytes, AsBytes, Unaligned)]
-#[repr(C)]
-struct DogValue {
- woof_count: U32,
- postal_code: U16,
-}
-
-fn variable_lengths(db: &sled::Db) -> sled::Result<()> {
- // here we will show how we can use zerocopy for inserting
- // fixed-size components, mixed with variable length
- // records on the end or beginning.
-
- // the hash_join example below shows how to read items
- // out in a way that accounts for the variable portion,
- // using `zerocopy::LayoutVerified::{new_from_prefix, new_from_suffix}`
-
- let dogs = db.open_tree(b"dogs")?;
-
- let mut dog2000_value = vec![];
- dog2000_value.extend_from_slice(b"science zone");
- dog2000_value.extend_from_slice(
- DogValue { woof_count: U32::new(666), postal_code: U16::new(42) }
- .as_bytes(),
- );
- dogs.insert("dog2000", dog2000_value)?;
-
- let mut zed_pup_value = vec![];
- zed_pup_value.extend_from_slice(b"bowling alley");
- zed_pup_value.extend_from_slice(
- DogValue { woof_count: U32::new(32113231), postal_code: U16::new(0) }
- .as_bytes(),
- );
- dogs.insert("zed pup", zed_pup_value)?;
-
- // IMPORTANT NOTE: German dogs eat food called "barf"
- let mut klaus_value = vec![];
- klaus_value.extend_from_slice(b"barf shop");
- klaus_value.extend_from_slice(
- DogValue { woof_count: U32::new(0), postal_code: U16::new(12045) }
- .as_bytes(),
- );
- dogs.insert("klaus", klaus_value)?;
-
- let cats = db.open_tree(b"cats")?;
-
- let mut laser_cat_value = vec![];
- laser_cat_value.extend_from_slice(
- CatValue {
- favorite_number: U64::new(11),
- battles_won: U64::new(321231321),
- }
- .as_bytes(),
- );
- laser_cat_value.extend_from_slice(b"science zone");
- cats.insert("laser cat", laser_cat_value)?;
-
- let mut pulsar_cat_value = vec![];
- pulsar_cat_value.extend_from_slice(
- CatValue {
- favorite_number: U64::new(11),
- battles_won: U64::new(321231321),
- }
- .as_bytes(),
- );
- pulsar_cat_value.extend_from_slice(b"science zone");
- cats.insert("pulsar cat", pulsar_cat_value)?;
-
- let mut fluffy_value = vec![];
- fluffy_value.extend_from_slice(
- CatValue {
- favorite_number: U64::new(11),
- battles_won: U64::new(321231321),
- }
- .as_bytes(),
- );
- fluffy_value.extend_from_slice(b"bowling alley");
- cats.insert("fluffy", fluffy_value)?;
-
- Ok(())
-}
-
-fn hash_join(db: &sled::Db) -> sled::Result<()> {
- // here we will try to find cats and dogs who
- // live in the same home.
-
- let cats = db.open_tree(b"cats")?;
- let dogs = db.open_tree(b"dogs")?;
-
- let mut join = std::collections::HashMap::new();
-
- for name_value_res in &cats {
- // cats are stored as name -> favorite_number + battles_won + home name
- // variable bytes
- let (name, value_bytes) = name_value_res?;
- let (_, home_name): (LayoutVerified<&[u8], CatValue>, &[u8]) =
- LayoutVerified::new_from_prefix(&*value_bytes).unwrap();
- let (ref mut cat_names, _dog_names) =
- join.entry(home_name.to_vec()).or_insert((vec![], vec![]));
- cat_names.push(std::str::from_utf8(&*name).unwrap().to_string());
- }
-
- for name_value_res in &dogs {
- // dogs are stored as name -> home name variable bytes + woof count +
- // postal code
- let (name, value_bytes) = name_value_res?;
-
- // note that this is reversed from the cat example above, where
- // the variable bytes are at the other end of the value, and are
- // extracted using new_from_prefix instead of new_from_suffix.
- let (home_name, _dog_value): (_, LayoutVerified<&[u8], DogValue>) =
- LayoutVerified::new_from_suffix(&*value_bytes).unwrap();
-
- if let Some((_cat_names, ref mut dog_names)) = join.get_mut(home_name) {
- dog_names.push(std::str::from_utf8(&*name).unwrap().to_string());
- }
- }
-
- for (home, (cats, dogs)) in join {
- println!(
- "the cats {:?} and the dogs {:?} live in the same home of {}",
- cats,
- dogs,
- std::str::from_utf8(&home).unwrap()
- );
- }
-
- Ok(())
-}
-
-fn main() -> sled::Result<()> {
- let db = sled::open("my_database")?;
- upsert(&db)?;
- variable_lengths(&db)?;
- hash_join(&db)?;
-
- Ok(())
-}
diff --git a/experiments/epoch/Cargo.toml b/experiments/epoch/Cargo.toml
deleted file mode 100644
index 9dae2b3f1..000000000
--- a/experiments/epoch/Cargo.toml
+++ /dev/null
@@ -1,8 +0,0 @@
-[package]
-name = "epoch"
-version = "0.1.0"
-authors = ["Tyler Neely "]
-edition = "2018"
-
-[profile.release]
-debug = true
diff --git a/experiments/epoch/sanitizers.sh b/experiments/epoch/sanitizers.sh
deleted file mode 100755
index 2e6a9e293..000000000
--- a/experiments/epoch/sanitizers.sh
+++ /dev/null
@@ -1,22 +0,0 @@
-#!/bin/bash
-set -eo pipefail
-
-echo "asan"
-cargo clean
-export RUSTFLAGS="-Z sanitizer=address"
-# export ASAN_OPTIONS="detect_odr_violation=0"
-cargo +nightly run --target x86_64-unknown-linux-gnu
-unset ASAN_OPTIONS
-
-echo "lsan"
-cargo clean
-export RUSTFLAGS="-Z sanitizer=leak"
-cargo +nightly run --target x86_64-unknown-linux-gnu
-
-echo "tsan"
-cargo clean
-export RUSTFLAGS="-Z sanitizer=thread"
-export TSAN_OPTIONS=suppressions=../../tsan_suppressions.txt
-cargo +nightly run --target x86_64-unknown-linux-gnu
-unset RUSTFLAGS
-unset TSAN_OPTIONS
diff --git a/experiments/epoch/src/main.rs b/experiments/epoch/src/main.rs
deleted file mode 100644
index e08b21818..000000000
--- a/experiments/epoch/src/main.rs
+++ /dev/null
@@ -1,213 +0,0 @@
-/// A simple implementation of epoch-based reclamation.
-///
-/// Using the `pin` method, a thread checks into an epoch
-/// before operating on a shared resource. If that thread
-/// makes a shared resource inaccessible, it can defer its
-/// destruction until all threads that may have already
-/// checked in have moved on.
-use std::{
- cell::RefCell,
- sync::{
- atomic::{AtomicPtr, AtomicUsize, Ordering::SeqCst},
- Arc,
- },
-};
-
-const EPOCH_SZ: usize = 16;
-
-#[derive(Default)]
-struct Epoch {
- garbage: [AtomicPtr>; EPOCH_SZ],
- offset: AtomicUsize,
- next: AtomicPtr,
- id: u64,
-}
-
-impl Drop for Epoch {
- fn drop(&mut self) {
- let count = std::cmp::min(EPOCH_SZ, self.offset.load(SeqCst));
- for offset in 0..count {
- let mut garbage_ptr: *mut Box =
- self.garbage[offset].load(SeqCst);
- while garbage_ptr.is_null() {
- // maybe this is impossible, but this is to
- // be defensive against race conditions.
- garbage_ptr = self.garbage[offset].load(SeqCst);
- }
-
- let garbage: Box> =
- unsafe { Box::from_raw(garbage_ptr) };
-
- drop(garbage);
- }
-
- let next = self.next.swap(std::ptr::null_mut(), SeqCst);
- if !next.is_null() {
- let arc = unsafe { Arc::from_raw(next) };
- drop(arc);
- }
- }
-}
-
-struct Collector {
- head: AtomicPtr,
-}
-
-unsafe impl Send for Collector {}
-unsafe impl Sync for Collector {}
-
-impl Default for Collector {
- fn default() -> Collector {
- let ptr = Arc::into_raw(Arc::new(Epoch::default())) as *mut Epoch;
- Collector { head: AtomicPtr::new(ptr) }
- }
-}
-
-impl Collector {
- fn pin(&self) -> Guard {
- let head_ptr = self.head.load(SeqCst);
- assert!(!head_ptr.is_null());
- let mut head = unsafe { Arc::from_raw(head_ptr) };
- let mut next = head.next.load(SeqCst);
- let mut last_head = head_ptr;
-
- // forward head to current tip
- while !next.is_null() {
- std::mem::forget(head);
-
- let res = self.head.compare_and_swap(last_head, next, SeqCst);
- if res == last_head {
- head = unsafe { Arc::from_raw(next) };
- last_head = next;
- } else {
- head = unsafe { Arc::from_raw(res) };
- last_head = res;
- }
-
- next = head.next.load(SeqCst);
- }
-
- let (a1, a2) = (head.clone(), head.clone());
- std::mem::forget(head);
-
- Guard {
- _entry_epoch: a1,
- current_epoch: a2,
- trash_sack: RefCell::new(vec![]),
- }
- }
-}
-
-impl Drop for Collector {
- fn drop(&mut self) {
- let head_ptr = self.head.load(SeqCst);
- assert!(!head_ptr.is_null());
- unsafe {
- let head = Arc::from_raw(head_ptr);
- drop(head);
- }
- }
-}
-
-pub(crate) struct Guard {
- _entry_epoch: Arc,
- current_epoch: Arc,
- trash_sack: RefCell>>,
-}
-
-impl Guard {
- pub fn defer(&self, f: F)
- where
- F: FnOnce() + Send + 'static,
- {
- let garbage_ptr =
- Box::into_raw(Box::new(Box::new(f) as Box));
- let mut trash_sack = self.trash_sack.borrow_mut();
- trash_sack.push(garbage_ptr);
- }
-}
-
-impl Drop for Guard {
- fn drop(&mut self) {
- let trash_sack = self.trash_sack.replace(vec![]);
-
- for garbage_ptr in trash_sack.into_iter() {
- // try to reserve
- let mut offset = self.current_epoch.offset.fetch_add(1, SeqCst);
- while offset >= EPOCH_SZ {
- let next = self.current_epoch.next.load(SeqCst);
- if !next.is_null() {
- unsafe {
- let raced_arc = Arc::from_raw(next);
- self.current_epoch = raced_arc.clone();
- std::mem::forget(raced_arc);
- }
- offset = self.current_epoch.offset.fetch_add(1, SeqCst);
- continue;
- }
-
- // push epoch forward if we're full
- let mut next_epoch = Epoch::default();
- next_epoch.id = self.current_epoch.id + 1;
-
- let next_epoch_arc = Arc::new(next_epoch);
- let next_ptr =
- Arc::into_raw(next_epoch_arc.clone()) as *mut Epoch;
- let old = self.current_epoch.next.compare_and_swap(
- std::ptr::null_mut(),
- next_ptr,
- SeqCst,
- );
- if old != std::ptr::null_mut() {
- // somebody else already installed a new segment
- unsafe {
- let unneeded = Arc::from_raw(next_ptr);
- drop(unneeded);
-
- let raced_arc = Arc::from_raw(old);
- self.current_epoch = raced_arc.clone();
- std::mem::forget(raced_arc);
- }
- offset = self.current_epoch.offset.fetch_add(1, SeqCst);
- continue;
- }
-
- self.current_epoch = next_epoch_arc;
- offset = self.current_epoch.offset.fetch_add(1, SeqCst);
- }
-
- let old =
- self.current_epoch.garbage[offset].swap(garbage_ptr, SeqCst);
- assert!(old.is_null());
- }
- }
-}
-
-#[derive(Debug)]
-struct S(usize);
-
-fn main() {
- let collector = Arc::new(Collector::default());
-
- let mut threads = vec![];
-
- for t in 0..100 {
- use std::thread::spawn;
-
- let collector = collector.clone();
- let thread = spawn(move || {
- for _ in 0..1000000 {
- let guard = collector.pin();
- guard.defer(move || {
- S(t as usize);
- });
- }
- });
-
- threads.push(thread);
- }
-
- for thread in threads.into_iter() {
- thread.join().unwrap();
- }
-}
diff --git a/experiments/new_segment_ownership/Cargo.lock b/experiments/new_segment_ownership/Cargo.lock
deleted file mode 100644
index 839910bd3..000000000
--- a/experiments/new_segment_ownership/Cargo.lock
+++ /dev/null
@@ -1,6 +0,0 @@
-# This file is automatically @generated by Cargo.
-# It is not intended for manual editing.
-[[package]]
-name = "new_segment_ownership"
-version = "0.1.0"
-
diff --git a/experiments/new_segment_ownership/Cargo.toml b/experiments/new_segment_ownership/Cargo.toml
deleted file mode 100644
index 1d59d65da..000000000
--- a/experiments/new_segment_ownership/Cargo.toml
+++ /dev/null
@@ -1,9 +0,0 @@
-[package]
-name = "new_segment_ownership"
-version = "0.1.0"
-authors = ["Tyler Neely "]
-edition = "2018"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
diff --git a/experiments/new_segment_ownership/src/main.rs b/experiments/new_segment_ownership/src/main.rs
deleted file mode 100644
index 21fb2ebe0..000000000
--- a/experiments/new_segment_ownership/src/main.rs
+++ /dev/null
@@ -1,109 +0,0 @@
-use std::sync::{
- atomic::{AtomicUsize, Ordering},
- Arc,
-};
-
-const SZ: usize = 128;
-
-#[derive(Default, Debug)]
-struct Log {
- segment_accountant: Arc,
- io_buf: Arc,
-}
-
-impl Log {
- fn new() -> Log {
- let io_buf = Arc::new(IoBuf::default());
- let segment_accountant = io_buf.segment.segment_accountant.clone();
- Log { io_buf, segment_accountant }
- }
-
- fn reserve(&mut self, size: usize) -> Reservation {
- assert!(size <= SZ);
- if self.io_buf.buf.load(Ordering::SeqCst) + size > SZ {
- let segment = self.segment_accountant.clone().next_segment();
- let buf = AtomicUsize::new(0);
- self.io_buf = Arc::new(IoBuf { segment, buf });
- }
- let io_buf = self.io_buf.clone();
- io_buf.buf.fetch_add(size, Ordering::SeqCst);
- Reservation { io_buf }
- }
-}
-
-#[derive(Default, Debug)]
-struct Reservation {
- io_buf: Arc,
-}
-
-#[derive(Default, Debug)]
-struct IoBuf {
- segment: Arc,
- buf: AtomicUsize,
-}
-
-#[derive(Default, Debug)]
-struct Segment {
- offset: usize,
- segment_accountant: Arc,
-}
-
-#[derive(Default, Debug)]
-struct SegmentAccountant {
- tip: AtomicUsize,
- free: Vec,
-}
-
-impl SegmentAccountant {
- fn next_segment(self: Arc) -> Arc {
- let offset = SZ + self.tip.fetch_add(SZ, Ordering::SeqCst);
- println!("setting new segment {}", offset);
- Arc::new(Segment { segment_accountant: self, offset })
- }
-}
-
-fn main() {
- let mut log = Log::new();
- {
- let _ = log.reserve(64);
- let _ = log.reserve(64);
- }
- println!("src/main.rs:70");
- {
- let _ = log.reserve(128);
- }
- println!("src/main.rs:74");
- {
- let _ = log.reserve(128);
- }
- println!("src/main.rs:78");
- {
- let _ = log.reserve(128);
- }
- println!("src/main.rs:77");
-}
-
-mod dropz {
- use super::*;
-
- impl Drop for IoBuf {
- fn drop(&mut self) {
- println!("IoBuf::drop");
- }
- }
- impl Drop for Segment {
- fn drop(&mut self) {
- println!("dropping Segment {:?}", self.offset);
- }
- }
- impl Drop for SegmentAccountant {
- fn drop(&mut self) {
- println!("SegmentAccountant::drop");
- }
- }
- impl Drop for Reservation {
- fn drop(&mut self) {
- println!("Reservation::drop");
- }
- }
-}
diff --git a/fuzz/.gitignore b/fuzz/.gitignore
new file mode 100644
index 000000000..a0925114d
--- /dev/null
+++ b/fuzz/.gitignore
@@ -0,0 +1,3 @@
+target
+corpus
+artifacts
diff --git a/fuzz/Cargo.toml b/fuzz/Cargo.toml
new file mode 100644
index 000000000..481ec21fc
--- /dev/null
+++ b/fuzz/Cargo.toml
@@ -0,0 +1,31 @@
+[package]
+name = "bloodstone-fuzz"
+version = "0.0.0"
+authors = ["Automatically generated"]
+publish = false
+edition = "2018"
+
+[package.metadata]
+cargo-fuzz = true
+
+[dependencies.libfuzzer-sys]
+version = "0.4.0"
+features = ["arbitrary-derive"]
+
+[dependencies]
+arbitrary = { version = "1.0.3", features = ["derive"] }
+tempfile = "3.5.0"
+
+[dependencies.sled]
+path = ".."
+features = []
+
+# Prevent this from interfering with workspaces
+[workspace]
+members = ["."]
+
+[[bin]]
+name = "fuzz_model"
+path = "fuzz_targets/fuzz_model.rs"
+test = false
+doc = false
diff --git a/fuzz/fuzz_targets/fuzz_model.rs b/fuzz/fuzz_targets/fuzz_model.rs
new file mode 100644
index 000000000..4af511255
--- /dev/null
+++ b/fuzz/fuzz_targets/fuzz_model.rs
@@ -0,0 +1,146 @@
+#![no_main]
+#[macro_use]
+extern crate libfuzzer_sys;
+extern crate arbitrary;
+extern crate sled;
+
+use arbitrary::Arbitrary;
+
+use sled::{Config, Db as SledDb, InlineArray};
+
+type Db = SledDb<3>;
+
+const KEYSPACE: u64 = 128;
+
+#[derive(Debug)]
+enum Op {
+ Get { key: InlineArray },
+ Insert { key: InlineArray, value: InlineArray },
+ Reboot,
+ Remove { key: InlineArray },
+ Cas { key: InlineArray, old: Option, new: Option },
+ Range { start: InlineArray, end: InlineArray },
+}
+
+fn keygen(
+ u: &mut arbitrary::Unstructured<'_>,
+) -> arbitrary::Result {
+ let key_i: u64 = u.int_in_range(0..=KEYSPACE)?;
+ Ok(key_i.to_be_bytes().as_ref().into())
+}
+
+impl<'a> Arbitrary<'a> for Op {
+ fn arbitrary(
+ u: &mut arbitrary::Unstructured<'a>,
+ ) -> arbitrary::Result {
+ Ok(if u.ratio(1, 2)? {
+ Op::Insert { key: keygen(u)?, value: keygen(u)? }
+ } else if u.ratio(1, 2)? {
+ Op::Get { key: keygen(u)? }
+ } else if u.ratio(1, 2)? {
+ Op::Reboot
+ } else if u.ratio(1, 2)? {
+ Op::Remove { key: keygen(u)? }
+ } else if u.ratio(1, 2)? {
+ Op::Cas {
+ key: keygen(u)?,
+ old: if u.ratio(1, 2)? { Some(keygen(u)?) } else { None },
+ new: if u.ratio(1, 2)? { Some(keygen(u)?) } else { None },
+ }
+ } else {
+ let start = u.int_in_range(0..=KEYSPACE)?;
+ let end = (start + 1).max(u.int_in_range(0..=KEYSPACE)?);
+
+ Op::Range {
+ start: start.to_be_bytes().as_ref().into(),
+ end: end.to_be_bytes().as_ref().into(),
+ }
+ })
+ }
+}
+
+fuzz_target!(|ops: Vec| {
+ let tmp_dir = tempfile::TempDir::new().unwrap();
+ let tmp_path = tmp_dir.path().to_owned();
+ let config = Config::new().path(tmp_path);
+
+ let mut tree: Db = config.open().unwrap();
+ let mut model = std::collections::BTreeMap::new();
+
+ for (_i, op) in ops.into_iter().enumerate() {
+ match op {
+ Op::Insert { key, value } => {
+ assert_eq!(
+ tree.insert(key.clone(), value.clone()).unwrap(),
+ model.insert(key, value)
+ );
+ }
+ Op::Get { key } => {
+ assert_eq!(tree.get(&key).unwrap(), model.get(&key).cloned());
+ }
+ Op::Reboot => {
+ drop(tree);
+ tree = config.open().unwrap();
+ }
+ Op::Remove { key } => {
+ assert_eq!(tree.remove(&key).unwrap(), model.remove(&key));
+ }
+ Op::Range { start, end } => {
+ let mut model_iter =
+ model.range::(&start..&end);
+ let mut tree_iter = tree.range(start..end);
+
+ for (k1, v1) in &mut model_iter {
+ let (k2, v2) = tree_iter
+ .next()
+ .expect("None returned from iter when Some expected")
+ .expect("IO issue encountered");
+ assert_eq!((k1, v1), (&k2, &v2));
+ }
+
+ assert!(tree_iter.next().is_none());
+ }
+ Op::Cas { key, old, new } => {
+ let succ = if old == model.get(&key).cloned() {
+ if let Some(n) = &new {
+ model.insert(key.clone(), n.clone());
+ } else {
+ model.remove(&key);
+ }
+ true
+ } else {
+ false
+ };
+
+ let res = tree
+ .compare_and_swap(key, old.as_ref(), new)
+ .expect("hit IO error");
+
+ if succ {
+ assert!(res.is_ok());
+ } else {
+ assert!(res.is_err());
+ }
+ }
+ };
+
+ for (key, value) in &model {
+ assert_eq!(tree.get(key).unwrap().unwrap(), value);
+ }
+
+ for kv_res in &tree {
+ let (key, value) = kv_res.unwrap();
+ assert_eq!(model.get(&key), Some(&value));
+ }
+ }
+
+ let mut model_iter = model.iter();
+ let mut tree_iter = tree.iter();
+
+ for (k1, v1) in &mut model_iter {
+ let (k2, v2) = tree_iter.next().unwrap().unwrap();
+ assert_eq!((k1, v1), (&k2, &v2));
+ }
+
+ assert!(tree_iter.next().is_none());
+});
diff --git a/src/alloc.rs b/src/alloc.rs
new file mode 100644
index 000000000..474ca6bf7
--- /dev/null
+++ b/src/alloc.rs
@@ -0,0 +1,81 @@
+#[cfg(any(
+ feature = "testing-shred-allocator",
+ feature = "testing-count-allocator"
+))]
+pub use alloc::*;
+
+// the memshred feature causes all allocated and deallocated
+// memory to be set to a specific non-zero value of 0xa1 for
+// uninitialized allocations and 0xde for deallocated memory,
+// in the hope that it will cause memory errors to surface
+// more quickly.
+
+#[cfg(feature = "testing-shred-allocator")]
+mod alloc {
+ use std::alloc::{Layout, System};
+
+ #[global_allocator]
+ static ALLOCATOR: ShredAllocator = ShredAllocator;
+
+ #[derive(Default, Debug, Clone, Copy)]
+ struct ShredAllocator;
+
+ unsafe impl std::alloc::GlobalAlloc for ShredAllocator {
+ unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+ let ret = System.alloc(layout);
+ assert_ne!(ret, std::ptr::null_mut());
+ std::ptr::write_bytes(ret, 0xa1, layout.size());
+ ret
+ }
+
+ unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+ std::ptr::write_bytes(ptr, 0xde, layout.size());
+ System.dealloc(ptr, layout)
+ }
+ }
+}
+
+#[cfg(feature = "testing-count-allocator")]
+mod alloc {
+ use std::alloc::{Layout, System};
+
+ #[global_allocator]
+ static ALLOCATOR: CountingAllocator = CountingAllocator;
+
+ static ALLOCATED: AtomicUsize = AtomicUsize::new(0);
+ static FREED: AtomicUsize = AtomicUsize::new(0);
+ static RESIDENT: AtomicUsize = AtomicUsize::new(0);
+
+ fn allocated() -> usize {
+ ALLOCATED.swap(0, Ordering::Relaxed)
+ }
+
+ fn freed() -> usize {
+ FREED.swap(0, Ordering::Relaxed)
+ }
+
+ fn resident() -> usize {
+ RESIDENT.load(Ordering::Relaxed)
+ }
+
+ #[derive(Default, Debug, Clone, Copy)]
+ struct CountingAllocator;
+
+ unsafe impl std::alloc::GlobalAlloc for CountingAllocator {
+ unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+ let ret = System.alloc(layout);
+ assert_ne!(ret, std::ptr::null_mut());
+ ALLOCATED.fetch_add(layout.size(), Ordering::Relaxed);
+ RESIDENT.fetch_add(layout.size(), Ordering::Relaxed);
+ std::ptr::write_bytes(ret, 0xa1, layout.size());
+ ret
+ }
+
+ unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+ std::ptr::write_bytes(ptr, 0xde, layout.size());
+ FREED.fetch_add(layout.size(), Ordering::Relaxed);
+ RESIDENT.fetch_sub(layout.size(), Ordering::Relaxed);
+ System.dealloc(ptr, layout)
+ }
+ }
+}
diff --git a/src/atomic_shim.rs b/src/atomic_shim.rs
deleted file mode 100644
index f134008b7..000000000
--- a/src/atomic_shim.rs
+++ /dev/null
@@ -1,262 +0,0 @@
-///! Inline of `https://github.com/bltavares/atomic-shim`
-
-#[cfg(not(any(
- target_arch = "mips",
- target_arch = "powerpc",
- feature = "mutex"
-)))]
-pub use std::sync::atomic::{AtomicI64, AtomicU64};
-#[cfg(any(target_arch = "mips", target_arch = "powerpc", feature = "mutex"))]
-mod shim {
- use parking_lot::{const_rwlock, RwLock};
- use std::sync::atomic::Ordering;
-
- #[derive(Debug, Default)]
- pub struct AtomicU64 {
- value: RwLock,
- }
-
- impl AtomicU64 {
- pub const fn new(v: u64) -> Self {
- Self { value: const_rwlock(v) }
- }
-
- #[allow(dead_code)]
- pub fn load(&self, _: Ordering) -> u64 {
- *self.value.read()
- }
-
- #[allow(dead_code)]
- pub fn store(&self, value: u64, _: Ordering) {
- let mut lock = self.value.write();
- *lock = value;
- }
-
- #[allow(dead_code)]
- pub fn swap(&self, value: u64, _: Ordering) -> u64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = value;
- prev
- }
-
- #[allow(dead_code)]
- pub fn compare_exchange(
- &self,
- current: u64,
- new: u64,
- _: Ordering,
- _: Ordering,
- ) -> Result {
- let mut lock = self.value.write();
- let prev = *lock;
- if prev == current {
- *lock = new;
- Ok(current)
- } else {
- Err(prev)
- }
- }
-
- #[allow(dead_code)]
- pub fn compare_exchange_weak(
- &self,
- current: u64,
- new: u64,
- success: Ordering,
- failure: Ordering,
- ) -> Result {
- self.compare_exchange(current, new, success, failure)
- }
-
- #[allow(dead_code)]
- pub fn fetch_add(&self, val: u64, _: Ordering) -> u64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev.wrapping_add(val);
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_sub(&self, val: u64, _: Ordering) -> u64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev.wrapping_sub(val);
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_and(&self, val: u64, _: Ordering) -> u64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev & val;
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_nand(&self, val: u64, _: Ordering) -> u64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = !(prev & val);
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_or(&self, val: u64, _: Ordering) -> u64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev | val;
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_xor(&self, val: u64, _: Ordering) -> u64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev ^ val;
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_max(&self, val: u64, _: Ordering) -> u64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev.max(val);
- prev
- }
- }
-
- impl From for AtomicU64 {
- fn from(value: u64) -> Self {
- AtomicU64::new(value)
- }
- }
-
- #[derive(Debug, Default)]
- pub struct AtomicI64 {
- value: RwLock,
- }
-
- impl AtomicI64 {
- pub const fn new(v: i64) -> Self {
- Self { value: const_rwlock(v) }
- }
-
- #[allow(dead_code)]
- pub fn load(&self, _: Ordering) -> i64 {
- *self.value.read()
- }
-
- #[allow(dead_code)]
- pub fn store(&self, value: i64, _: Ordering) {
- let mut lock = self.value.write();
- *lock = value;
- }
-
- #[allow(dead_code)]
- pub fn swap(&self, value: i64, _: Ordering) -> i64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = value;
- prev
- }
-
- #[allow(dead_code)]
- pub fn compare_exchange(
- &self,
- current: i64,
- new: i64,
- _: Ordering,
- _: Ordering,
- ) -> Result {
- let mut lock = self.value.write();
- let prev = *lock;
- if prev == current {
- *lock = new;
- Ok(current)
- } else {
- Err(prev)
- }
- }
-
- #[allow(dead_code)]
- pub fn compare_exchange_weak(
- &self,
- current: i64,
- new: i64,
- success: Ordering,
- failure: Ordering,
- ) -> Result {
- self.compare_exchange(current, new, success, failure)
- }
-
- #[allow(dead_code)]
- pub fn fetch_add(&self, val: i64, _: Ordering) -> i64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev.wrapping_add(val);
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_sub(&self, val: i64, _: Ordering) -> i64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev.wrapping_sub(val);
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_and(&self, val: i64, _: Ordering) -> i64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev & val;
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_nand(&self, val: i64, _: Ordering) -> i64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = !(prev & val);
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_or(&self, val: i64, _: Ordering) -> i64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev | val;
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_xor(&self, val: i64, _: Ordering) -> i64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev ^ val;
- prev
- }
-
- #[allow(dead_code)]
- pub fn fetch_max(&self, val: i64, _: Ordering) -> i64 {
- let mut lock = self.value.write();
- let prev = *lock;
- *lock = prev.max(val);
- prev
- }
- }
-
- impl From for AtomicI64 {
- fn from(value: i64) -> Self {
- AtomicI64::new(value)
- }
- }
-}
-
-#[cfg(any(
- target_arch = "mips",
- target_arch = "powerpc",
- feature = "mutex"
-))]
-pub use shim::{AtomicI64, AtomicU64};
diff --git a/src/backoff.rs b/src/backoff.rs
deleted file mode 100644
index 46a65d7f4..000000000
--- a/src/backoff.rs
+++ /dev/null
@@ -1,43 +0,0 @@
-/// Vendored and simplified from crossbeam-utils
-use core::cell::Cell;
-use core::sync::atomic;
-
-const SPIN_LIMIT: u32 = 6;
-
-/// Performs exponential backoff in spin loops.
-///
-/// Backing off in spin loops reduces contention and improves overall performance.
-///
-/// This primitive can execute *YIELD* and *PAUSE* instructions, yield the current thread to the OS
-/// scheduler, and tell when is a good time to block the thread using a different synchronization
-/// mechanism. Each step of the back off procedure takes roughly twice as long as the previous
-/// step.
-pub struct Backoff {
- step: Cell,
-}
-
-impl Backoff {
- /// Creates a new `Backoff`.
- pub const fn new() -> Self {
- Backoff { step: Cell::new(0) }
- }
-
- /// Backs off in a lock-free loop.
- ///
- /// This method should be used when we need to retry an operation because another thread made
- /// progress.
- ///
- /// The processor may yield using the *YIELD* or *PAUSE* instruction.
- #[inline]
- pub fn spin(&self) {
- for _ in 0..1 << self.step.get().min(SPIN_LIMIT) {
- // `hint::spin_loop` requires Rust 1.49.
- #[allow(deprecated)]
- atomic::spin_loop_hint();
- }
-
- if self.step.get() <= SPIN_LIMIT {
- self.step.set(self.step.get() + 1);
- }
- }
-}
diff --git a/src/batch.rs b/src/batch.rs
deleted file mode 100644
index e960fd8b5..000000000
--- a/src/batch.rs
+++ /dev/null
@@ -1,60 +0,0 @@
-#![allow(unused_results)]
-
-use super::*;
-
-/// A batch of updates that will
-/// be applied atomically to the
-/// Tree.
-///
-/// # Examples
-///
-/// ```
-/// # fn main() -> Result<(), Box> {
-/// use sled::{Batch, open};
-///
-/// # let _ = std::fs::remove_dir_all("batch_db_2");
-/// let db = open("batch_db_2")?;
-/// db.insert("key_0", "val_0")?;
-///
-/// let mut batch = Batch::default();
-/// batch.insert("key_a", "val_a");
-/// batch.insert("key_b", "val_b");
-/// batch.insert("key_c", "val_c");
-/// batch.remove("key_0");
-///
-/// db.apply_batch(batch)?;
-/// // key_0 no longer exists, and key_a, key_b, and key_c
-/// // now do exist.
-/// # let _ = std::fs::remove_dir_all("batch_db_2");
-/// # Ok(()) }
-/// ```
-#[derive(Debug, Default, Clone, PartialEq, Eq)]
-pub struct Batch {
- pub(crate) writes: Map>,
-}
-
-impl Batch {
- /// Set a key to a new value
- pub fn insert(&mut self, key: K, value: V)
- where
- K: Into,
- V: Into,
- {
- self.writes.insert(key.into(), Some(value.into()));
- }
-
- /// Remove a key
- pub fn remove(&mut self, key: K)
- where
- K: Into,
- {
- self.writes.insert(key.into(), None);
- }
-
- /// Get a value if it is present in the `Batch`.
- /// `Some(None)` means it's present as a deletion.
- pub fn get>(&self, k: K) -> Option> {
- let inner = self.writes.get(k.as_ref())?;
- Some(inner.as_ref())
- }
-}
diff --git a/src/cache_padded.rs b/src/cache_padded.rs
deleted file mode 100644
index 4495ff206..000000000
--- a/src/cache_padded.rs
+++ /dev/null
@@ -1,66 +0,0 @@
-/// Vendored and simplified from crossbeam-utils.
-use core::fmt;
-use core::ops::{Deref, DerefMut};
-
-// Starting from Intel's Sandy Bridge, spatial prefetcher is now pulling pairs of 64-byte cache
-// lines at a time, so we have to align to 128 bytes rather than 64.
-//
-// Sources:
-// - https://www.intel.com/content/dam/www/public/us/en/documents/manuals/64-ia-32-architectures-optimization-manual.pdf
-// - https://github.com/facebook/folly/blob/1b5288e6eea6df074758f877c849b6e73bbb9fbb/folly/lang/Align.h#L107
-//
-// ARM's big.LITTLE architecture has asymmetric cores and "big" cores have 128 byte cache line size
-// Sources:
-// - https://www.mono-project.com/news/2016/09/12/arm64-icache/
-//
-#[cfg_attr(
- any(target_arch = "x86_64", target_arch = "aarch64"),
- repr(align(128))
-)]
-#[cfg_attr(
- not(any(target_arch = "x86_64", target_arch = "aarch64")),
- repr(align(64))
-)]
-#[derive(Default, PartialEq, Eq)]
-pub struct CachePadded {
- value: T,
-}
-
-#[allow(unsafe_code)]
-unsafe impl Send for CachePadded {}
-
-#[allow(unsafe_code)]
-unsafe impl Sync for CachePadded {}
-
-impl CachePadded {
- /// Pads and aligns a value to the length of a cache line.
- pub const fn new(t: T) -> CachePadded {
- CachePadded:: { value: t }
- }
-}
-
-impl Deref for CachePadded {
- type Target = T;
-
- fn deref(&self) -> &T {
- &self.value
- }
-}
-
-impl DerefMut for CachePadded {
- fn deref_mut(&mut self) -> &mut T {
- &mut self.value
- }
-}
-
-impl fmt::Debug for CachePadded {
- fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
- f.debug_struct("CachePadded").field("value", &self.value).finish()
- }
-}
-
-impl From for CachePadded {
- fn from(t: T) -> Self {
- CachePadded::new(t)
- }
-}
diff --git a/src/concurrency_control.rs b/src/concurrency_control.rs
deleted file mode 100644
index 81f8d997b..000000000
--- a/src/concurrency_control.rs
+++ /dev/null
@@ -1,108 +0,0 @@
-#[cfg(feature = "testing")]
-use std::cell::RefCell;
-use std::sync::atomic::AtomicBool;
-
-use parking_lot::{RwLockReadGuard, RwLockWriteGuard};
-
-use super::*;
-
-#[cfg(feature = "testing")]
-thread_local! {
- pub static COUNT: RefCell = RefCell::new(0);
-}
-
-const RW_REQUIRED_BIT: usize = 1 << 31;
-
-#[derive(Default)]
-pub(crate) struct ConcurrencyControl {
- active: AtomicUsize,
- upgrade_complete: AtomicBool,
- rw: RwLock<()>,
-}
-
-static CONCURRENCY_CONTROL: Lazy<
- ConcurrencyControl,
- fn() -> ConcurrencyControl,
-> = Lazy::new(init_cc);
-
-fn init_cc() -> ConcurrencyControl {
- ConcurrencyControl::default()
-}
-
-#[derive(Debug)]
-#[must_use]
-pub(crate) enum Protector<'a> {
- Write(RwLockWriteGuard<'a, ()>),
- Read(RwLockReadGuard<'a, ()>),
- None(&'a AtomicUsize),
-}
-
-impl<'a> Drop for Protector<'a> {
- fn drop(&mut self) {
- if let Protector::None(active) = self {
- active.fetch_sub(1, Release);
- }
- #[cfg(feature = "testing")]
- COUNT.with(|c| {
- let mut c = c.borrow_mut();
- *c -= 1;
- assert_eq!(*c, 0);
- });
- }
-}
-
-pub(crate) fn read<'a>() -> Protector<'a> {
- CONCURRENCY_CONTROL.read()
-}
-
-pub(crate) fn write<'a>() -> Protector<'a> {
- CONCURRENCY_CONTROL.write()
-}
-
-impl ConcurrencyControl {
- fn enable(&self) {
- if self.active.fetch_or(RW_REQUIRED_BIT, SeqCst) < RW_REQUIRED_BIT {
- // we are the first to set this bit
- while self.active.load(Acquire) != RW_REQUIRED_BIT {
- // `hint::spin_loop` requires Rust 1.49.
- #[allow(deprecated)]
- std::sync::atomic::spin_loop_hint()
- }
- self.upgrade_complete.store(true, Release);
- }
- }
-
- fn read(&self) -> Protector<'_> {
- #[cfg(feature = "testing")]
- COUNT.with(|c| {
- let mut c = c.borrow_mut();
- *c += 1;
- assert_eq!(*c, 1);
- });
-
- let active = self.active.fetch_add(1, Release);
-
- if active >= RW_REQUIRED_BIT {
- self.active.fetch_sub(1, Release);
- Protector::Read(self.rw.read())
- } else {
- Protector::None(&self.active)
- }
- }
-
- fn write(&self) -> Protector<'_> {
- #[cfg(feature = "testing")]
- COUNT.with(|c| {
- let mut c = c.borrow_mut();
- *c += 1;
- assert_eq!(*c, 1);
- });
- self.enable();
- while !self.upgrade_complete.load(Acquire) {
- // `hint::spin_loop` requires Rust 1.49.
- #[allow(deprecated)]
- std::sync::atomic::spin_loop_hint()
- }
- Protector::Write(self.rw.write())
- }
-}
diff --git a/src/config.rs b/src/config.rs
index edfda4748..7eb6dab4b 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -1,824 +1,102 @@
-use std::{
- fs,
- fs::File,
- io,
- io::{BufRead, BufReader, ErrorKind, Read, Seek, Write},
- ops::Deref,
- path::{Path, PathBuf},
- sync::atomic::AtomicUsize,
-};
+use std::io;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
-use crate::pagecache::{arr_to_u32, u32_to_arr, Heap};
-use crate::*;
+use fault_injection::{annotate, fallible};
+use tempdir::TempDir;
-const DEFAULT_PATH: &str = "default.sled";
-
-/// The high-level database mode, according to
-/// the trade-offs of the RUM conjecture.
-#[derive(Debug, Clone, Copy)]
-pub enum Mode {
- /// In this mode, the database will make
- /// decisions that favor using less space
- /// instead of supporting the highest possible
- /// write throughput. This mode will also
- /// rewrite data more frequently as it
- /// strives to reduce fragmentation.
- LowSpace,
- /// In this mode, the database will try
- /// to maximize write throughput while
- /// potentially using more disk space.
- HighThroughput,
-}
-
-/// A persisted configuration about high-level
-/// storage file information
-#[derive(Debug, Eq, PartialEq, Clone, Copy)]
-struct StorageParameters {
- pub segment_size: usize,
- pub use_compression: bool,
- pub version: (usize, usize),
-}
-
-impl StorageParameters {
- pub fn serialize(&self) -> Vec {
- let mut out = vec![];
-
- writeln!(&mut out, "segment_size: {}", self.segment_size).unwrap();
- writeln!(&mut out, "use_compression: {}", self.use_compression)
- .unwrap();
- writeln!(&mut out, "version: {}.{}", self.version.0, self.version.1)
- .unwrap();
-
- out
- }
-
- pub fn deserialize(bytes: &[u8]) -> Result {
- let reader = BufReader::new(bytes);
-
- let mut lines = Map::new();
-
- for line in reader.lines() {
- let line = if let Ok(l) = line {
- l
- } else {
- error!(
- "failed to parse persisted config as UTF-8. \
- This changed in sled version 0.29"
- );
- return Err(Error::Unsupported(
- "failed to open database that may \
- have been created using a sled version \
- earlier than 0.29",
- ));
- };
- let mut split = line.split(": ").map(String::from);
- let k = if let Some(k) = split.next() {
- k
- } else {
- error!("failed to parse persisted config line: {}", line);
- return Err(Error::corruption(None));
- };
- let v = if let Some(v) = split.next() {
- v
- } else {
- error!("failed to parse persisted config line: {}", line);
- return Err(Error::corruption(None));
- };
- lines.insert(k, v);
- }
-
- let segment_size: usize = if let Some(raw) = lines.get("segment_size") {
- if let Ok(parsed) = raw.parse() {
- parsed
- } else {
- error!("failed to parse segment_size value: {}", raw);
- return Err(Error::corruption(None));
- }
- } else {
- error!(
- "failed to retrieve required configuration parameter: segment_size"
- );
- return Err(Error::corruption(None));
- };
-
- let use_compression: bool = if let Some(raw) =
- lines.get("use_compression")
- {
- if let Ok(parsed) = raw.parse() {
- parsed
- } else {
- error!("failed to parse use_compression value: {}", raw);
- return Err(Error::corruption(None));
- }
- } else {
- error!(
- "failed to retrieve required configuration parameter: use_compression"
- );
- return Err(Error::corruption(None));
- };
-
- let version: (usize, usize) = if let Some(raw) = lines.get("version") {
- let mut split = raw.split('.');
- let major = if let Some(raw_major) = split.next() {
- if let Ok(parsed_major) = raw_major.parse() {
- parsed_major
- } else {
- error!(
- "failed to parse major version value from line: {}",
- raw
- );
- return Err(Error::corruption(None));
- }
- } else {
- error!("failed to parse major version value: {}", raw);
- return Err(Error::corruption(None));
- };
-
- let minor = if let Some(raw_minor) = split.next() {
- if let Ok(parsed_minor) = raw_minor.parse() {
- parsed_minor
- } else {
- error!(
- "failed to parse minor version value from line: {}",
- raw
- );
- return Err(Error::corruption(None));
- }
- } else {
- error!("failed to parse minor version value: {}", raw);
- return Err(Error::corruption(None));
- };
-
- (major, minor)
- } else {
- error!(
- "failed to retrieve required configuration parameter: version"
- );
- return Err(Error::corruption(None));
- };
-
- Ok(StorageParameters { segment_size, use_compression, version })
- }
-}
-
-/// Top-level configuration for the system.
-///
-/// # Examples
-///
-/// ```
-/// let _config = sled::Config::default()
-/// .path("/path/to/data".to_owned())
-/// .cache_capacity(10_000_000_000)
-/// .flush_every_ms(Some(1000));
-/// ```
-#[derive(Default, Debug, Clone)]
-pub struct Config(Arc);
-
-impl Deref for Config {
- type Target = Inner;
-
- fn deref(&self) -> &Inner {
- &self.0
- }
-}
-
-#[doc(hidden)]
-#[derive(Debug, Clone)]
-pub struct Inner {
- #[doc(hidden)]
- pub cache_capacity: usize,
- #[doc(hidden)]
- pub flush_every_ms: Option,
- #[doc(hidden)]
- pub segment_size: usize,
- #[doc(hidden)]
- pub path: PathBuf,
- #[doc(hidden)]
- pub create_new: bool,
- #[doc(hidden)]
- pub mode: Mode,
- #[doc(hidden)]
- pub temporary: bool,
- #[doc(hidden)]
- pub use_compression: bool,
- #[doc(hidden)]
- pub compression_factor: i32,
- #[doc(hidden)]
- pub idgen_persist_interval: u64,
- #[doc(hidden)]
- pub snapshot_after_ops: u64,
- #[doc(hidden)]
- pub version: (usize, usize),
- tmp_path: PathBuf,
- pub(crate) global_error: Arc>,
- #[cfg(feature = "event_log")]
- /// an event log for concurrent debugging
- pub event_log: Arc,
-}
-
-impl Default for Inner {
- fn default() -> Self {
- Self {
- // generally useful
- path: PathBuf::from(DEFAULT_PATH),
- tmp_path: Config::gen_temp_path(),
- create_new: false,
- cache_capacity: 1024 * 1024 * 1024, // 1gb
- mode: Mode::LowSpace,
- use_compression: false,
- compression_factor: 5,
- temporary: false,
- version: crate_version(),
-
- // useful in testing
- segment_size: 512 * 1024, // 512kb in bytes
- flush_every_ms: Some(500),
- idgen_persist_interval: 1_000_000,
- snapshot_after_ops: if cfg!(feature = "testing") {
- 10
- } else {
- 1_000_000
- },
- global_error: Arc::new(Atomic::default()),
- #[cfg(feature = "event_log")]
- event_log: Arc::new(crate::event_log::EventLog::default()),
- }
- }
-}
-
-impl Inner {
- // Get the path of the database
- #[doc(hidden)]
- pub fn get_path(&self) -> PathBuf {
- if self.temporary && self.path == PathBuf::from(DEFAULT_PATH) {
- self.tmp_path.clone()
- } else {
- self.path.clone()
- }
- }
-
- fn db_path(&self) -> PathBuf {
- self.get_path().join("db")
- }
-
- fn config_path(&self) -> PathBuf {
- self.get_path().join("conf")
- }
-
- pub(crate) fn normalize(&self, value: T) -> T
- where
- T: Copy
- + TryFrom
- + std::ops::Div
+