key-value database initial
This commit is contained in:
NikVolf
parent
2ea0310190
commit
deefe60a13
|
|
@ -6,3 +6,6 @@
|
|||
*.swo
|
||||
*.swn
|
||||
*.DS_Store
|
||||
|
||||
# idea ide
|
||||
.idea
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ version = "0.1.0"
|
|||
dependencies = [
|
||||
"bitcrypto 0.1.0",
|
||||
"clap 2.14.0 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"db 0.1.0",
|
||||
"keys 0.1.0",
|
||||
"message 0.1.0",
|
||||
"p2p 0.1.0",
|
||||
|
|
@ -88,6 +89,29 @@ name = "crossbeam"
|
|||
version = "0.2.10"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "db"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"elastic-array 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"ethcore-devtools 1.3.0",
|
||||
"parking_lot 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"rocksdb 0.4.5 (git+https://github.com/ethcore/rust-rocksdb)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "deque"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "elastic-array"
|
||||
version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "eth-secp256k1"
|
||||
version = "0.5.6"
|
||||
|
|
@ -100,6 +124,13 @@ dependencies = [
|
|||
"rustc-serialize 0.3.19 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ethcore-devtools"
|
||||
version = "1.3.0"
|
||||
dependencies = [
|
||||
"rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures"
|
||||
version = "0.1.2"
|
||||
|
|
@ -122,6 +153,9 @@ dependencies = [
|
|||
name = "gcc"
|
||||
version = "0.3.35"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"rayon 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "kernel32-sys"
|
||||
|
|
@ -236,6 +270,14 @@ dependencies = [
|
|||
"odds 0.2.22 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_cpus"
|
||||
version = "0.2.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"libc 0.2.16 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_cpus"
|
||||
version = "1.1.0"
|
||||
|
|
@ -270,6 +312,17 @@ dependencies = [
|
|||
"tokio-core 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parking_lot"
|
||||
version = "0.2.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"libc 0.2.16 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"smallvec 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parking_lot"
|
||||
version = "0.3.5"
|
||||
|
|
@ -306,6 +359,34 @@ dependencies = [
|
|||
"libc 0.2.16 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rayon"
|
||||
version = "0.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"deque 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"num_cpus 0.2.13 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rocksdb"
|
||||
version = "0.4.5"
|
||||
source = "git+https://github.com/ethcore/rust-rocksdb#ffc7c82380fe8569f85ae6743f7f620af2d4a679"
|
||||
dependencies = [
|
||||
"libc 0.2.16 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"rocksdb-sys 0.3.0 (git+https://github.com/ethcore/rust-rocksdb)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rocksdb-sys"
|
||||
version = "0.3.0"
|
||||
source = "git+https://github.com/ethcore/rust-rocksdb#ffc7c82380fe8569f85ae6743f7f620af2d4a679"
|
||||
dependencies = [
|
||||
"gcc 0.3.35 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"libc 0.2.16 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rust-crypto"
|
||||
version = "0.2.36"
|
||||
|
|
@ -461,6 +542,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
"checksum cfg-if 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "de1e760d7b6535af4241fca8bd8adf68e2e7edacc6b29f5d399050c5e48cf88c"
|
||||
"checksum clap 2.14.0 (registry+https://github.com/rust-lang/crates.io-index)" = "5fa304b03c49ccbb005784fc26e985b5d2310b1d37f2c311ce90dbcd18ea5fde"
|
||||
"checksum crossbeam 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)" = "0c5ea215664ca264da8a9d9c3be80d2eaf30923c259d03e870388eb927508f97"
|
||||
"checksum deque 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1614659040e711785ed8ea24219140654da1729f3ec8a47a9719d041112fe7bf"
|
||||
"checksum elastic-array 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)" = "4bc9250a632e7c001b741eb0ec6cee93c9a5b6d5f1879696a4b94d62b012210a"
|
||||
"checksum eth-secp256k1 0.5.6 (git+https://github.com/ethcore/rust-secp256k1)" = "<none>"
|
||||
"checksum futures 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "0bd34f72c0fffc9d2f6c570fd392bf99b9c5cd1481d79809e1cc2320befc0af0"
|
||||
"checksum futures-cpupool 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "bb982bb25cd8fa5da6a8eb3a460354c984ff1113da82bcb4f0b0862b5795db82"
|
||||
|
|
@ -475,12 +558,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
"checksum net2 0.2.26 (registry+https://github.com/rust-lang/crates.io-index)" = "5edf9cb6be97212423aed9413dd4729d62b370b5e1c571750e882cebbbc1e3e2"
|
||||
"checksum nix 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)" = "7a7bb1da2be7da3cbffda73fc681d509ffd9e665af478d2bee1907cee0bc64b2"
|
||||
"checksum nodrop 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "0dbbadd3f4c98dea0bd3d9b4be4c0cdaf1ab57035cb2e41fce3983db5add7cc5"
|
||||
"checksum num_cpus 0.2.13 (registry+https://github.com/rust-lang/crates.io-index)" = "cee7e88156f3f9e19bdd598f8d6c9db7bf4078f99f8381f43a55b09648d1a6e3"
|
||||
"checksum num_cpus 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "8890e6084723d57d0df8d2720b0d60c6ee67d6c93e7169630e4371e88765dcad"
|
||||
"checksum odds 0.2.22 (registry+https://github.com/rust-lang/crates.io-index)" = "308eb922caa165a2c4be8004abf87822fc4b444ab339f70fc172236b1905efb0"
|
||||
"checksum owning_ref 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "8d91377085359426407a287ab16884a0111ba473aa6844ff01d4ec20ce3d75e7"
|
||||
"checksum parking_lot 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)" = "968f685642555d2f7e202c48b8b11de80569e9bfea817f7f12d7c61aac62d4e6"
|
||||
"checksum parking_lot 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)" = "dbc5847584161f273e69edc63c1a86254a22f570a0b5dd87aa6f9773f6f7d125"
|
||||
"checksum parking_lot_core 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "fb1b97670a2ffadce7c397fb80a3d687c4f3060140b885621ef1653d0e5d5068"
|
||||
"checksum rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)" = "2791d88c6defac799c3f20d74f094ca33b9332612d9aef9078519c82e4fe04a5"
|
||||
"checksum rayon 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)" = "655df67c314c30fa3055a365eae276eb88aa4f3413a352a1ab32c1320eda41ea"
|
||||
"checksum rocksdb 0.4.5 (git+https://github.com/ethcore/rust-rocksdb)" = "<none>"
|
||||
"checksum rocksdb-sys 0.3.0 (git+https://github.com/ethcore/rust-rocksdb)" = "<none>"
|
||||
"checksum rust-crypto 0.2.36 (registry+https://github.com/rust-lang/crates.io-index)" = "f76d05d3993fd5f4af9434e8e436db163a12a9d40e1a58a726f27a01dfd12a2a"
|
||||
"checksum rustc-serialize 0.3.19 (registry+https://github.com/rust-lang/crates.io-index)" = "6159e4e6e559c81bd706afe9c8fd68f547d3e851ce12e76b1de7914bab61691b"
|
||||
"checksum rustc_version 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)" = "c5f5376ea5e30ce23c03eb77cbe4962b988deead10910c372b226388b594c084"
|
||||
|
|
|
|||
|
|
@ -7,12 +7,12 @@ description = "Parity bitcoin client."
|
|||
|
||||
[dependencies]
|
||||
clap = { version = "2", features = ["yaml"] }
|
||||
|
||||
bitcrypto = { path = "crypto" }
|
||||
keys = { path = "keys" }
|
||||
message = { path = "message" }
|
||||
p2p = { path = "p2p" }
|
||||
script = { path = "script" }
|
||||
db = { path = "db" }
|
||||
|
||||
[[bin]]
|
||||
path = "pbtc/main.rs"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,10 @@
|
|||
[package]
|
||||
name = "db"
|
||||
version = "0.1.0"
|
||||
authors = ["Nikolay Volf <nikvolf@gmail.com>"]
|
||||
|
||||
[dependencies]
|
||||
elastic-array = "0.5"
|
||||
rocksdb = { git = "https://github.com/ethcore/rust-rocksdb" }
|
||||
ethcore-devtools = { path = "../devtools" }
|
||||
parking_lot = "0.2"
|
||||
|
|
@ -0,0 +1,529 @@
|
|||
//! Key-Value store abstraction with `RocksDB` backend.
|
||||
|
||||
use std::io::ErrorKind;
|
||||
use elastic_array::*;
|
||||
use std::default::Default;
|
||||
use std::path::PathBuf;
|
||||
use rocksdb::{DB, Writable, WriteBatch, WriteOptions, IteratorMode, DBIterator,
|
||||
Options, DBCompactionStyle, BlockBasedOptions, Cache, Column};
|
||||
use std::{self, fs, mem};
|
||||
use parking_lot::RwLock;
|
||||
use std::collections::HashMap;
|
||||
use std::convert::From;
|
||||
|
||||
type Bytes = Vec<u8>;
|
||||
|
||||
/// Database error
|
||||
pub enum Error {
|
||||
/// Rocksdb error
|
||||
DB(String),
|
||||
/// Io error
|
||||
Io(std::io::Error),
|
||||
}
|
||||
|
||||
impl From<String> for Error {
|
||||
fn from(err: String) -> Error {
|
||||
Error::DB(err)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<std::io::Error> for Error {
|
||||
fn from(err: std::io::Error) -> Error {
|
||||
Error::Io(err)
|
||||
}
|
||||
}
|
||||
|
||||
const DB_BACKGROUND_FLUSHES: i32 = 2;
|
||||
const DB_BACKGROUND_COMPACTIONS: i32 = 2;
|
||||
|
||||
/// Write transaction. Batches a sequence of put/delete operations for efficiency.
|
||||
pub struct DBTransaction {
|
||||
ops: Vec<DBOp>,
|
||||
}
|
||||
|
||||
enum DBOp {
|
||||
Insert {
|
||||
col: Option<u32>,
|
||||
key: ElasticArray32<u8>,
|
||||
value: Bytes,
|
||||
},
|
||||
Delete {
|
||||
col: Option<u32>,
|
||||
key: ElasticArray32<u8>,
|
||||
}
|
||||
}
|
||||
|
||||
impl DBTransaction {
|
||||
/// Create new transaction.
|
||||
pub fn new(_db: &Database) -> DBTransaction {
|
||||
DBTransaction {
|
||||
ops: Vec::with_capacity(256),
|
||||
}
|
||||
}
|
||||
|
||||
/// Insert a key-value pair in the transaction. Any existing value value will be overwritten upon write.
|
||||
pub fn put(&mut self, col: Option<u32>, key: &[u8], value: &[u8]) {
|
||||
let mut ekey = ElasticArray32::new();
|
||||
ekey.append_slice(key);
|
||||
self.ops.push(DBOp::Insert {
|
||||
col: col,
|
||||
key: ekey,
|
||||
value: value.to_vec(),
|
||||
});
|
||||
}
|
||||
|
||||
/// Insert a key-value pair in the transaction. Any existing value value will be overwritten upon write.
|
||||
pub fn put_vec(&mut self, col: Option<u32>, key: &[u8], value: Bytes) {
|
||||
let mut ekey = ElasticArray32::new();
|
||||
ekey.append_slice(key);
|
||||
self.ops.push(DBOp::Insert {
|
||||
col: col,
|
||||
key: ekey,
|
||||
value: value,
|
||||
});
|
||||
}
|
||||
|
||||
/// Delete value by key.
|
||||
pub fn delete(&mut self, col: Option<u32>, key: &[u8]) {
|
||||
let mut ekey = ElasticArray32::new();
|
||||
ekey.append_slice(key);
|
||||
self.ops.push(DBOp::Delete {
|
||||
col: col,
|
||||
key: ekey,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
enum KeyState {
|
||||
Insert(Bytes),
|
||||
Delete,
|
||||
}
|
||||
|
||||
/// Compaction profile for the database settings
|
||||
#[derive(Clone, Copy)]
|
||||
pub struct CompactionProfile {
|
||||
/// L0-L1 target file size
|
||||
pub initial_file_size: u64,
|
||||
/// L2-LN target file size multiplier
|
||||
pub file_size_multiplier: i32,
|
||||
/// rate limiter for background flushes and compactions, bytes/sec, if any
|
||||
pub write_rate_limit: Option<u64>,
|
||||
}
|
||||
|
||||
impl Default for CompactionProfile {
|
||||
/// Default profile suitable for most storage
|
||||
fn default() -> CompactionProfile {
|
||||
CompactionProfile {
|
||||
initial_file_size: 32 * 1024 * 1024,
|
||||
file_size_multiplier: 2,
|
||||
write_rate_limit: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl CompactionProfile {
|
||||
/// Slow hdd compaction profile
|
||||
pub fn hdd() -> CompactionProfile {
|
||||
CompactionProfile {
|
||||
initial_file_size: 192 * 1024 * 1024,
|
||||
file_size_multiplier: 1,
|
||||
write_rate_limit: Some(8 * 1024 * 1024),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Database configuration
|
||||
#[derive(Clone)]
|
||||
pub struct DatabaseConfig {
|
||||
/// Max number of open files.
|
||||
pub max_open_files: i32,
|
||||
/// Cache sizes (in MiB) for specific columns.
|
||||
pub cache_sizes: HashMap<Option<u32>, usize>,
|
||||
/// Compaction profile
|
||||
pub compaction: CompactionProfile,
|
||||
/// Set number of columns
|
||||
pub columns: Option<u32>,
|
||||
/// Should we keep WAL enabled?
|
||||
pub wal: bool,
|
||||
}
|
||||
|
||||
impl DatabaseConfig {
|
||||
/// Create new `DatabaseConfig` with default parameters and specified set of columns.
|
||||
/// Note that cache sizes must be explicitly set.
|
||||
pub fn with_columns(columns: Option<u32>) -> Self {
|
||||
let mut config = Self::default();
|
||||
config.columns = columns;
|
||||
config
|
||||
}
|
||||
|
||||
/// Set the column cache size in MiB.
|
||||
pub fn set_cache(&mut self, col: Option<u32>, size: usize) {
|
||||
self.cache_sizes.insert(col, size);
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for DatabaseConfig {
|
||||
fn default() -> DatabaseConfig {
|
||||
DatabaseConfig {
|
||||
cache_sizes: HashMap::new(),
|
||||
max_open_files: 512,
|
||||
compaction: CompactionProfile::default(),
|
||||
columns: None,
|
||||
wal: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Database iterator for flushed data only
|
||||
pub struct DatabaseIterator {
|
||||
iter: DBIterator,
|
||||
}
|
||||
|
||||
impl<'a> Iterator for DatabaseIterator {
|
||||
type Item = (Box<[u8]>, Box<[u8]>);
|
||||
|
||||
fn next(&mut self) -> Option<Self::Item> {
|
||||
self.iter.next()
|
||||
}
|
||||
}
|
||||
|
||||
struct DBAndColumns {
|
||||
db: DB,
|
||||
cfs: Vec<Column>,
|
||||
}
|
||||
|
||||
/// Key-Value database.
|
||||
pub struct Database {
|
||||
db: RwLock<Option<DBAndColumns>>,
|
||||
config: DatabaseConfig,
|
||||
write_opts: WriteOptions,
|
||||
overlay: RwLock<Vec<HashMap<ElasticArray32<u8>, KeyState>>>,
|
||||
path: String,
|
||||
}
|
||||
|
||||
impl Database {
|
||||
/// Open database with default settings.
|
||||
pub fn open_default(path: &str) -> Result<Database, String> {
|
||||
Database::open(&DatabaseConfig::default(), path)
|
||||
}
|
||||
|
||||
/// Open database file. Creates if it does not exist.
|
||||
pub fn open(config: &DatabaseConfig, path: &str) -> Result<Database, String> {
|
||||
// default cache size for columns not specified.
|
||||
const DEFAULT_CACHE: usize = 2;
|
||||
|
||||
let mut opts = Options::new();
|
||||
if let Some(rate_limit) = config.compaction.write_rate_limit {
|
||||
try!(opts.set_parsed_options(&format!("rate_limiter_bytes_per_sec={}", rate_limit)));
|
||||
}
|
||||
try!(opts.set_parsed_options(&format!("max_total_wal_size={}", 64 * 1024 * 1024)));
|
||||
opts.set_max_open_files(config.max_open_files);
|
||||
opts.create_if_missing(true);
|
||||
opts.set_use_fsync(false);
|
||||
|
||||
opts.set_max_background_flushes(DB_BACKGROUND_FLUSHES);
|
||||
opts.set_max_background_compactions(DB_BACKGROUND_COMPACTIONS);
|
||||
|
||||
// compaction settings
|
||||
opts.set_compaction_style(DBCompactionStyle::DBUniversalCompaction);
|
||||
opts.set_target_file_size_base(config.compaction.initial_file_size);
|
||||
opts.set_target_file_size_multiplier(config.compaction.file_size_multiplier);
|
||||
|
||||
let mut cf_options = Vec::with_capacity(config.columns.unwrap_or(0) as usize);
|
||||
|
||||
for col in 0 .. config.columns.unwrap_or(0) {
|
||||
let mut opts = Options::new();
|
||||
opts.set_compaction_style(DBCompactionStyle::DBUniversalCompaction);
|
||||
opts.set_target_file_size_base(config.compaction.initial_file_size);
|
||||
opts.set_target_file_size_multiplier(config.compaction.file_size_multiplier);
|
||||
|
||||
let col_opt = config.columns.map(|_| col);
|
||||
|
||||
{
|
||||
let cache_size = config.cache_sizes.get(&col_opt).cloned().unwrap_or(DEFAULT_CACHE);
|
||||
let mut block_opts = BlockBasedOptions::new();
|
||||
// all goes to read cache.
|
||||
block_opts.set_cache(Cache::new(cache_size * 1024 * 1024));
|
||||
opts.set_block_based_table_factory(&block_opts);
|
||||
}
|
||||
|
||||
cf_options.push(opts);
|
||||
}
|
||||
|
||||
let mut write_opts = WriteOptions::new();
|
||||
if !config.wal {
|
||||
write_opts.disable_wal(true);
|
||||
}
|
||||
|
||||
let mut cfs: Vec<Column> = Vec::new();
|
||||
let db = match config.columns {
|
||||
Some(columns) => {
|
||||
let cfnames: Vec<_> = (0..columns).map(|c| format!("col{}", c)).collect();
|
||||
let cfnames: Vec<&str> = cfnames.iter().map(|n| n as &str).collect();
|
||||
match DB::open_cf(&opts, path, &cfnames, &cf_options) {
|
||||
Ok(db) => {
|
||||
cfs = cfnames.iter().map(|n| db.cf_handle(n).unwrap()).collect();
|
||||
assert!(cfs.len() == columns as usize);
|
||||
Ok(db)
|
||||
}
|
||||
Err(_) => {
|
||||
// retry and create CFs
|
||||
match DB::open_cf(&opts, path, &[], &[]) {
|
||||
Ok(mut db) => {
|
||||
cfs = cfnames.iter().enumerate().map(|(i, n)| db.create_cf(n, &cf_options[i]).unwrap()).collect();
|
||||
Ok(db)
|
||||
},
|
||||
err @ Err(_) => err,
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
None => DB::open(&opts, path)
|
||||
};
|
||||
let db = match db {
|
||||
Ok(db) => db,
|
||||
Err(ref s) if s.starts_with("Corruption:") => {
|
||||
try!(DB::repair(&opts, path));
|
||||
try!(DB::open(&opts, path))
|
||||
},
|
||||
Err(s) => { return Err(s); }
|
||||
};
|
||||
let num_cols = cfs.len();
|
||||
Ok(Database {
|
||||
db: RwLock::new(Some(DBAndColumns{ db: db, cfs: cfs })),
|
||||
config: config.clone(),
|
||||
write_opts: write_opts,
|
||||
overlay: RwLock::new((0..(num_cols + 1)).map(|_| HashMap::new()).collect()),
|
||||
path: path.to_owned(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Creates new transaction for this database.
|
||||
pub fn transaction(&self) -> DBTransaction {
|
||||
DBTransaction::new(self)
|
||||
}
|
||||
|
||||
|
||||
fn to_overlay_column(col: Option<u32>) -> usize {
|
||||
col.map_or(0, |c| (c + 1) as usize)
|
||||
}
|
||||
|
||||
/// Commit transaction to database.
|
||||
pub fn write_buffered(&self, tr: DBTransaction) {
|
||||
let mut overlay = self.overlay.write();
|
||||
let ops = tr.ops;
|
||||
for op in ops {
|
||||
match op {
|
||||
DBOp::Insert { col, key, value } => {
|
||||
let c = Self::to_overlay_column(col);
|
||||
overlay[c].insert(key, KeyState::Insert(value));
|
||||
},
|
||||
DBOp::Delete { col, key } => {
|
||||
let c = Self::to_overlay_column(col);
|
||||
overlay[c].insert(key, KeyState::Delete);
|
||||
},
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/// Commit buffered changes to database.
|
||||
pub fn flush(&self) -> Result<(), String> {
|
||||
match *self.db.read() {
|
||||
Some(DBAndColumns { ref db, ref cfs }) => {
|
||||
let batch = WriteBatch::new();
|
||||
let mut overlay = self.overlay.write();
|
||||
|
||||
for (c, column) in overlay.iter_mut().enumerate() {
|
||||
let column_data = mem::replace(column, HashMap::new());
|
||||
for (key, state) in column_data.into_iter() {
|
||||
match state {
|
||||
KeyState::Delete => {
|
||||
if c > 0 {
|
||||
try!(batch.delete_cf(cfs[c - 1], &key));
|
||||
} else {
|
||||
try!(batch.delete(&key));
|
||||
}
|
||||
},
|
||||
KeyState::Insert(value) => {
|
||||
if c > 0 {
|
||||
try!(batch.put_cf(cfs[c - 1], &key, &value));
|
||||
} else {
|
||||
try!(batch.put(&key, &value));
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
db.write_opt(batch, &self.write_opts)
|
||||
},
|
||||
None => Err("Database is closed".to_owned())
|
||||
}
|
||||
}
|
||||
|
||||
/// Commit transaction to database.
|
||||
pub fn write(&self, tr: DBTransaction) -> Result<(), String> {
|
||||
match *self.db.read() {
|
||||
Some(DBAndColumns { ref db, ref cfs }) => {
|
||||
let batch = WriteBatch::new();
|
||||
let ops = tr.ops;
|
||||
for op in ops {
|
||||
match op {
|
||||
DBOp::Insert { col, key, value } => {
|
||||
try!(col.map_or_else(|| batch.put(&key, &value), |c| batch.put_cf(cfs[c as usize], &key, &value)))
|
||||
},
|
||||
DBOp::Delete { col, key } => {
|
||||
try!(col.map_or_else(|| batch.delete(&key), |c| batch.delete_cf(cfs[c as usize], &key)))
|
||||
},
|
||||
}
|
||||
}
|
||||
db.write_opt(batch, &self.write_opts)
|
||||
},
|
||||
None => Err("Database is closed".to_owned())
|
||||
}
|
||||
}
|
||||
|
||||
/// Get value by key.
|
||||
pub fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<Bytes>, String> {
|
||||
match *self.db.read() {
|
||||
Some(DBAndColumns { ref db, ref cfs }) => {
|
||||
let overlay = &self.overlay.read()[Self::to_overlay_column(col)];
|
||||
match overlay.get(key) {
|
||||
Some(&KeyState::Insert(ref value)) => Ok(Some(value.clone())),
|
||||
Some(&KeyState::Delete) => Ok(None),
|
||||
None => {
|
||||
col.map_or_else(
|
||||
|| db.get(key).map(|r| r.map(|v| v.to_vec())),
|
||||
|c| db.get_cf(cfs[c as usize], key).map(|r| r.map(|v| v.to_vec())))
|
||||
},
|
||||
}
|
||||
},
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
/// Get database iterator for flushed data.
|
||||
pub fn iter(&self, col: Option<u32>) -> DatabaseIterator {
|
||||
//TODO: iterate over overlay
|
||||
match *self.db.read() {
|
||||
Some(DBAndColumns { ref db, ref cfs }) => {
|
||||
col.map_or_else(|| DatabaseIterator { iter: db.iterator(IteratorMode::Start) },
|
||||
|c| DatabaseIterator { iter: db.iterator_cf(cfs[c as usize], IteratorMode::Start).unwrap() })
|
||||
},
|
||||
None => panic!("Not supported yet") //TODO: return an empty iterator or change return type
|
||||
}
|
||||
}
|
||||
|
||||
/// Close the database
|
||||
fn close(&self) {
|
||||
*self.db.write() = None;
|
||||
self.overlay.write().clear();
|
||||
}
|
||||
|
||||
/// Restore the database from a copy at given path.
|
||||
pub fn restore(&self, new_db: &str) -> Result<(), Error> {
|
||||
self.close();
|
||||
|
||||
let mut backup_db = PathBuf::from(&self.path);
|
||||
backup_db.pop();
|
||||
backup_db.push("backup_db");
|
||||
|
||||
let existed = match fs::rename(&self.path, &backup_db) {
|
||||
Ok(_) => true,
|
||||
Err(e) => if let ErrorKind::NotFound = e.kind() {
|
||||
false
|
||||
} else {
|
||||
return Err(e.into());
|
||||
}
|
||||
};
|
||||
|
||||
match fs::rename(&new_db, &self.path) {
|
||||
Ok(_) => {
|
||||
// clean up the backup.
|
||||
if existed {
|
||||
try!(fs::remove_dir_all(&backup_db));
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
// restore the backup.
|
||||
if existed {
|
||||
try!(fs::rename(&backup_db, &self.path));
|
||||
}
|
||||
return Err(e.into())
|
||||
}
|
||||
}
|
||||
|
||||
// reopen the database and steal handles into self
|
||||
let db = try!(Self::open(&self.config, &self.path));
|
||||
*self.db.write() = mem::replace(&mut *db.db.write(), None);
|
||||
*self.overlay.write() = mem::replace(&mut *db.overlay.write(), Vec::new());
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use hash::*;
|
||||
use super::*;
|
||||
use devtools::*;
|
||||
use std::str::FromStr;
|
||||
|
||||
fn test_db(config: &DatabaseConfig) {
|
||||
let path = RandomTempPath::create_dir();
|
||||
let db = Database::open(config, path.as_path().to_str().unwrap()).unwrap();
|
||||
let key1 = H256::from_str("02c69be41d0b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc").unwrap();
|
||||
let key2 = H256::from_str("03c69be41d0b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc").unwrap();
|
||||
let key3 = H256::from_str("01c69be41d0b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc").unwrap();
|
||||
|
||||
let mut batch = db.transaction();
|
||||
batch.put(None, &key1, b"cat");
|
||||
batch.put(None, &key2, b"dog");
|
||||
db.write(batch).unwrap();
|
||||
|
||||
assert_eq!(&*db.get(None, &key1).unwrap().unwrap(), b"cat");
|
||||
|
||||
let contents: Vec<_> = db.iter(None).collect();
|
||||
assert_eq!(contents.len(), 2);
|
||||
assert_eq!(&*contents[0].0, &*key1);
|
||||
assert_eq!(&*contents[0].1, b"cat");
|
||||
assert_eq!(&*contents[1].0, &*key2);
|
||||
assert_eq!(&*contents[1].1, b"dog");
|
||||
|
||||
let mut batch = db.transaction();
|
||||
batch.delete(None, &key1);
|
||||
db.write(batch).unwrap();
|
||||
|
||||
assert!(db.get(None, &key1).unwrap().is_none());
|
||||
|
||||
let mut batch = db.transaction();
|
||||
batch.put(None, &key1, b"cat");
|
||||
db.write(batch).unwrap();
|
||||
|
||||
let mut transaction = db.transaction();
|
||||
transaction.put(None, &key3, b"elephant");
|
||||
transaction.delete(None, &key1);
|
||||
db.write(transaction).unwrap();
|
||||
assert!(db.get(None, &key1).unwrap().is_none());
|
||||
assert_eq!(&*db.get(None, &key3).unwrap().unwrap(), b"elephant");
|
||||
|
||||
assert_eq!(&*db.get_by_prefix(None, &key3).unwrap(), b"elephant");
|
||||
assert_eq!(&*db.get_by_prefix(None, &key2).unwrap(), b"dog");
|
||||
|
||||
let mut transaction = db.transaction();
|
||||
transaction.put(None, &key1, b"horse");
|
||||
transaction.delete(None, &key3);
|
||||
db.write_buffered(transaction);
|
||||
assert!(db.get(None, &key3).unwrap().is_none());
|
||||
assert_eq!(&*db.get(None, &key1).unwrap().unwrap(), b"horse");
|
||||
|
||||
db.flush().unwrap();
|
||||
assert!(db.get(None, &key3).unwrap().is_none());
|
||||
assert_eq!(&*db.get(None, &key1).unwrap().unwrap(), b"horse");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn kvdb() {
|
||||
let path = RandomTempPath::create_dir();
|
||||
let _ = Database::open_default(path.as_path().to_str().unwrap()).unwrap();
|
||||
test_db(&DatabaseConfig::default());
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
//! Bitcoin database
|
||||
|
||||
extern crate elastic_array;
|
||||
extern crate rocksdb;
|
||||
extern crate parking_lot;
|
||||
|
||||
pub mod kvdb;
|
||||
|
|
@ -0,0 +1,16 @@
|
|||
[package]
|
||||
description = "Ethcore development/test/build tools"
|
||||
homepage = "http://ethcore.io"
|
||||
license = "GPL-3.0"
|
||||
name = "ethcore-devtools"
|
||||
version = "1.3.0"
|
||||
authors = ["Ethcore <admin@ethcore.io>"]
|
||||
|
||||
[dependencies]
|
||||
rand = "0.3"
|
||||
|
||||
[features]
|
||||
|
||||
[lib]
|
||||
path = "src/lib.rs"
|
||||
test = true
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
// Copyright 2015, 2016 Ethcore (UK) Ltd.
|
||||
// This file is part of Parity.
|
||||
|
||||
// Parity is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Parity is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
//! dev-tools
|
||||
|
||||
|
||||
extern crate rand;
|
||||
|
||||
mod random_path;
|
||||
mod test_socket;
|
||||
mod stop_guard;
|
||||
|
||||
pub use random_path::*;
|
||||
pub use test_socket::*;
|
||||
pub use stop_guard::*;
|
||||
|
|
@ -0,0 +1,129 @@
|
|||
// Copyright 2015, 2016 Ethcore (UK) Ltd.
|
||||
// This file is part of Parity.
|
||||
|
||||
// Parity is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Parity is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
//! Random path
|
||||
|
||||
use std::path::*;
|
||||
use std::fs;
|
||||
use std::env;
|
||||
use std::ops::{Deref, DerefMut};
|
||||
use rand::random;
|
||||
|
||||
pub struct RandomTempPath {
|
||||
path: PathBuf
|
||||
}
|
||||
|
||||
pub fn random_filename() -> String {
|
||||
random_str(8)
|
||||
}
|
||||
|
||||
pub fn random_str(len: usize) -> String {
|
||||
(0..len).map(|_| ((random::<f32>() * 26.0) as u8 + 97) as char).collect()
|
||||
}
|
||||
|
||||
impl RandomTempPath {
|
||||
pub fn new() -> RandomTempPath {
|
||||
let mut dir = env::temp_dir();
|
||||
dir.push(random_filename());
|
||||
RandomTempPath {
|
||||
path: dir.clone()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn create_dir() -> RandomTempPath {
|
||||
let mut dir = env::temp_dir();
|
||||
dir.push(random_filename());
|
||||
fs::create_dir_all(dir.as_path()).unwrap();
|
||||
RandomTempPath {
|
||||
path: dir.clone()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn as_path(&self) -> &PathBuf {
|
||||
&self.path
|
||||
}
|
||||
|
||||
pub fn as_str(&self) -> &str {
|
||||
self.path.to_str().unwrap()
|
||||
}
|
||||
|
||||
pub fn new_in(&self, name: &str) -> String {
|
||||
let mut path = self.path.clone();
|
||||
path.push(name);
|
||||
path.to_str().unwrap().to_owned()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for RandomTempPath {
|
||||
fn drop(&mut self) {
|
||||
if let Err(e) = fs::remove_dir_all(self.as_path()) {
|
||||
panic!("Failed to remove temp directory. Here's what prevented this from happening: ({})", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct GuardedTempResult<T> {
|
||||
pub result: Option<T>,
|
||||
pub _temp: RandomTempPath
|
||||
}
|
||||
|
||||
impl<T> GuardedTempResult<T> {
|
||||
pub fn reference(&self) -> &T {
|
||||
self.result.as_ref().unwrap()
|
||||
}
|
||||
|
||||
pub fn reference_mut(&mut self) -> &mut T {
|
||||
self.result.as_mut().unwrap()
|
||||
}
|
||||
|
||||
pub fn take(&mut self) -> T {
|
||||
self.result.take().unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Deref for GuardedTempResult<T> {
|
||||
type Target = T;
|
||||
|
||||
fn deref(&self) -> &T { self.result.as_ref().unwrap() }
|
||||
}
|
||||
|
||||
impl<T> DerefMut for GuardedTempResult<T> {
|
||||
fn deref_mut(&mut self) -> &mut T { self.result.as_mut().unwrap() }
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn creates_dir() {
|
||||
let temp = RandomTempPath::create_dir();
|
||||
assert!(fs::metadata(temp.as_path()).unwrap().is_dir());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn destroys_dir() {
|
||||
let path_buf = {
|
||||
let temp = RandomTempPath::create_dir();
|
||||
assert!(fs::metadata(temp.as_path()).unwrap().is_dir());
|
||||
let path_buf = temp.as_path().to_path_buf();
|
||||
path_buf
|
||||
};
|
||||
|
||||
assert!(fs::metadata(&path_buf).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn provides_random() {
|
||||
let temp = RandomTempPath::create_dir();
|
||||
assert!(temp.as_path().to_str().is_some());
|
||||
}
|
||||
|
|
@ -0,0 +1,45 @@
|
|||
// Copyright 2015, 2016 Ethcore (UK) Ltd.
|
||||
// This file is part of Parity.
|
||||
|
||||
// Parity is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Parity is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
//! Stop guard mod
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::*;
|
||||
|
||||
/// Stop guard that will set a stop flag on drop
|
||||
pub struct StopGuard {
|
||||
flag: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
impl StopGuard {
|
||||
/// Create a stop guard
|
||||
pub fn new() -> StopGuard {
|
||||
StopGuard {
|
||||
flag: Arc::new(AtomicBool::new(false))
|
||||
}
|
||||
}
|
||||
|
||||
/// Share stop guard between the threads
|
||||
pub fn share(&self) -> Arc<AtomicBool> {
|
||||
self.flag.clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for StopGuard {
|
||||
fn drop(&mut self) {
|
||||
self.flag.store(true, Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,95 @@
|
|||
// Copyright 2015, 2016 Ethcore (UK) Ltd.
|
||||
// This file is part of Parity.
|
||||
|
||||
// Parity is free software: you can redistribute it and/or modify
|
||||
// it under the terms of the GNU General Public License as published by
|
||||
// the Free Software Foundation, either version 3 of the License, or
|
||||
// (at your option) any later version.
|
||||
|
||||
// Parity is distributed in the hope that it will be useful,
|
||||
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
// GNU General Public License for more details.
|
||||
|
||||
// You should have received a copy of the GNU General Public License
|
||||
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
use std::io::*;
|
||||
use std::cmp;
|
||||
|
||||
pub struct TestSocket {
|
||||
pub read_buffer: Vec<u8>,
|
||||
pub write_buffer: Vec<u8>,
|
||||
pub cursor: usize,
|
||||
pub buf_size: usize,
|
||||
}
|
||||
|
||||
impl Default for TestSocket {
|
||||
fn default() -> Self {
|
||||
TestSocket::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl TestSocket {
|
||||
pub fn new() -> Self {
|
||||
TestSocket {
|
||||
read_buffer: vec![],
|
||||
write_buffer: vec![],
|
||||
cursor: 0,
|
||||
buf_size: 0,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_buf(buf_size: usize) -> TestSocket {
|
||||
TestSocket {
|
||||
read_buffer: vec![],
|
||||
write_buffer: vec![],
|
||||
cursor: 0,
|
||||
buf_size: buf_size,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_ready(data: Vec<u8>) -> TestSocket {
|
||||
TestSocket {
|
||||
read_buffer: data,
|
||||
write_buffer: vec![],
|
||||
cursor: 0,
|
||||
buf_size: 0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Read for TestSocket {
|
||||
fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
|
||||
let end_position = cmp::min(self.read_buffer.len(), self.cursor+buf.len());
|
||||
if self.cursor > end_position { return Ok(0) }
|
||||
let len = cmp::max(end_position - self.cursor, 0);
|
||||
match len {
|
||||
0 => Ok(0),
|
||||
_ => {
|
||||
for i in self.cursor..end_position {
|
||||
buf[i-self.cursor] = self.read_buffer[i];
|
||||
}
|
||||
self.cursor = end_position;
|
||||
Ok(len)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for TestSocket {
|
||||
fn write(&mut self, buf: &[u8]) -> Result<usize> {
|
||||
if self.buf_size == 0 || buf.len() < self.buf_size {
|
||||
self.write_buffer.extend(buf.iter().cloned());
|
||||
Ok(buf.len())
|
||||
}
|
||||
else {
|
||||
self.write_buffer.extend(buf.iter().take(self.buf_size).cloned());
|
||||
Ok(self.buf_size)
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> Result<()> {
|
||||
unimplemented!();
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue