grassrootseconomics/openethereum
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #6792 from paritytech/kvdb_error

Browse Source 
consistent KeyValueDB errors
This commit is contained in:
Marek Kotewicz
2017-10-16 18:19:18 +02:00
committed by GitHub
parent 5b29b6d064 7661e93a56
commit bb1be15dc4
25 changed files with 154 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
util/Cargo.toml
View File

@@ -26,7 +26,6 @@ parking_lot = "0.4"
tiny-keccak= "1.0"
ethcore-logger = { path = "../logger" }
triehash = { path = "triehash" }
error-chain = "0.11.0-rc.2"
hashdb = { path = "hashdb" }
patricia_trie = { path = "patricia_trie" }
ethcore-bytes = { path = "bytes" }

3
util/error/Cargo.toml
View File

@@ -5,6 +5,7 @@ authors = ["Parity Technologies <admin@parity.io>"]
[dependencies]
rlp = { path = "../rlp" }
kvdb = { path = "../kvdb" }
ethcore-bigint = { path = "../bigint" }
error-chain = "0.11.0-rc.2"
error-chain = "0.11.0"
rustc-hex = "1.0"

5
util/error/src/lib.rs
View File

@@ -25,6 +25,7 @@ extern crate error_chain;
extern crate ethcore_bigint as bigint;
extern crate rlp;
extern crate rustc_hex;
extern crate kvdb;
use std::fmt;
use rustc_hex::FromHexError;
@@ -62,6 +63,10 @@ error_chain! {
UtilError, ErrorKind, ResultExt, Result;
}
links {
Db(kvdb::Error, kvdb::ErrorKind);
}
foreign_links {
Io(::std::io::Error);
FromHex(FromHexError);

13
util/kvdb-memorydb/src/lib.rs
View File

@@ -20,7 +20,7 @@ extern crate rlp;
use std::collections::{BTreeMap, HashMap};
use parking_lot::RwLock;
use kvdb::{DBValue, Error, DBTransaction, KeyValueDB, DBOp};
use kvdb::{DBValue, DBTransaction, KeyValueDB, DBOp, Result};
use rlp::{RlpType, UntrustedRlp, Compressible};
/// A key-value database fulfilling the `KeyValueDB` trait, living in memory.
@@ -46,10 +46,10 @@ pub fn create(num_cols: u32) -> InMemory {
}
impl KeyValueDB for InMemory {
fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<DBValue>, String> {
fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<DBValue>> {
let columns = self.columns.read();
match columns.get(&col) {
None => Err(format!("No such column family: {:?}", col)),
None => Err(format!("No such column family: {:?}", col).into()),
Some(map) => Ok(map.get(key).cloned()),
}
}
@@ -92,7 +92,10 @@ impl KeyValueDB for InMemory {
}
}
fn flush(&self) -> Result<(), String> { Ok(()) }
fn flush(&self) -> Result<()> {
Ok(())
}
fn iter<'a>(&'a self, col: Option<u32>) -> Box<Iterator<Item=(Box<[u8]>, Box<[u8]>)> + 'a> {
match self.columns.read().get(&col) {
Some(map) => Box::new( // TODO: worth optimizing at all?
@@ -118,7 +121,7 @@ impl KeyValueDB for InMemory {
}
}
fn restore(&self, _new_db: &str) -> Result<(), Error> {
fn restore(&self, _new_db: &str) -> Result<()> {
Err("Attempted to restore in-memory database".into())
}
}

41
util/kvdb-rocksdb/src/lib.rs
View File

@@ -40,7 +40,7 @@ use rocksdb::{
use elastic_array::ElasticArray32;
use rlp::{UntrustedRlp, RlpType, Compressible};
use kvdb::{KeyValueDB, DBTransaction, DBValue, Error, DBOp};
use kvdb::{KeyValueDB, DBTransaction, DBValue, DBOp, Result};
#[cfg(target_os = "linux")]
use regex::Regex;
@@ -257,12 +257,12 @@ pub struct Database {
impl Database {
/// Open database with default settings.
pub fn open_default(path: &str) -> Result<Database, String> {
pub fn open_default(path: &str) -> Result<Database> {
Database::open(&DatabaseConfig::default(), path)
}
/// Open database file. Creates if it does not exist.
pub fn open(config: &DatabaseConfig, path: &str) -> Result<Database, String> {
pub fn open(config: &DatabaseConfig, path: &str) -> Result<Database> {
let mut opts = Options::new();
if let Some(rate_limit) = config.compaction.write_rate_limit {
opts.set_parsed_options(&format!("rate_limiter_bytes_per_sec={}", rate_limit))?;
@@ -312,7 +312,7 @@ impl Database {
// 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])).collect::<Result<_, _>>()?;
cfs = cfnames.iter().enumerate().map(|(i, n)| db.create_cf(n, &cf_options[i])).collect::<::std::result::Result<_, _>>()?;
Ok(db)
},
err @ Err(_) => err,
@@ -335,7 +335,7 @@ impl Database {
false => DB::open_cf(&opts, path, &cfnames, &cf_options)?
}
},
Err(s) => { return Err(s); }
Err(s) => { return Err(s.into()); }
};
let num_cols = cfs.len();
Ok(Database {
@@ -383,7 +383,7 @@ impl Database {
}
/// Commit buffered changes to database. Must be called under `flush_lock`
fn write_flushing_with_lock(&self, _lock: &mut MutexGuard<bool>) -> Result<(), String> {
fn write_flushing_with_lock(&self, _lock: &mut MutexGuard<bool>) -> Result<()> {
match *self.db.read() {
Some(DBAndColumns { ref db, ref cfs }) => {
let batch = WriteBatch::new();
@@ -425,18 +425,18 @@ impl Database {
}
Ok(())
},
None => Err("Database is closed".to_owned())
None => Err("Database is closed".into())
}
}
/// Commit buffered changes to database.
pub fn flush(&self) -> Result<(), String> {
pub fn flush(&self) -> Result<()> {
let mut lock = self.flushing_lock.lock();
// If RocksDB batch allocation fails the thread gets terminated and the lock is released.
// The value inside the lock is used to detect that.
if *lock {
// This can only happen if another flushing thread is terminated unexpectedly.
return Err("Database write failure. Running low on memory perhaps?".to_owned());
return Err("Database write failure. Running low on memory perhaps?".into());
}
*lock = true;
let result = self.write_flushing_with_lock(&mut lock);
@@ -445,7 +445,7 @@ impl Database {
}
/// Commit transaction to database.
pub fn write(&self, tr: DBTransaction) -> Result<(), String> {
pub fn write(&self, tr: DBTransaction) -> Result<()> {
match *self.db.read() {
Some(DBAndColumns { ref db, ref cfs }) => {
let batch = WriteBatch::new();
@@ -464,14 +464,14 @@ impl Database {
},
}
}
db.write_opt(batch, &self.write_opts)
db.write_opt(batch, &self.write_opts).map_err(Into::into)
},
None => Err("Database is closed".to_owned())
None => Err("Database is closed".into())
}
}
/// Get value by key.
pub fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<DBValue>, String> {
pub fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<DBValue>> {
match *self.db.read() {
Some(DBAndColumns { ref db, ref cfs }) => {
let overlay = &self.overlay.read()[Self::to_overlay_column(col)];
@@ -487,6 +487,7 @@ impl Database {
col.map_or_else(
|| db.get_opt(key, &self.read_opts).map(|r| r.map(|v| DBValue::from_slice(&v))),
|c| db.get_cf_opt(cfs[c as usize], key, &self.read_opts).map(|r| r.map(|v| DBValue::from_slice(&v))))
.map_err(Into::into)
},
}
},
@@ -552,7 +553,7 @@ impl Database {
}
/// Restore the database from a copy at given path.
pub fn restore(&self, new_db: &str) -> Result<(), Error> {
pub fn restore(&self, new_db: &str) -> Result<()> {
self.close();
let mut backup_db = PathBuf::from(&self.path);
@@ -601,7 +602,7 @@ impl Database {
}
/// Drop a column family.
pub fn drop_column(&self) -> Result<(), String> {
pub fn drop_column(&self) -> Result<()> {
match *self.db.write() {
Some(DBAndColumns { ref mut db, ref mut cfs }) => {
if let Some(col) = cfs.pop() {
@@ -616,7 +617,7 @@ impl Database {
}
/// Add a column family.
pub fn add_column(&self) -> Result<(), String> {
pub fn add_column(&self) -> Result<()> {
match *self.db.write() {
Some(DBAndColumns { ref mut db, ref mut cfs }) => {
let col = cfs.len() as u32;
@@ -632,7 +633,7 @@ impl Database {
// duplicate declaration of methods here to avoid trait import in certain existing cases
// at time of addition.
impl KeyValueDB for Database {
fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<DBValue>, String> {
fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<DBValue>> {
Database::get(self, col, key)
}
@@ -644,11 +645,11 @@ impl KeyValueDB for Database {
Database::write_buffered(self, transaction)
}
fn write(&self, transaction: DBTransaction) -> Result<(), String> {
fn write(&self, transaction: DBTransaction) -> Result<()> {
Database::write(self, transaction)
}
fn flush(&self) -> Result<(), String> {
fn flush(&self) -> Result<()> {
Database::flush(self)
}
@@ -664,7 +665,7 @@ impl KeyValueDB for Database {
Box::new(unboxed.into_iter().flat_map(|inner| inner))
}
fn restore(&self, new_db: &str) -> Result<(), Error> {
fn restore(&self, new_db: &str) -> Result<()> {
Database::restore(self, new_db)
}
}

2
util/kvdb/Cargo.toml
View File

@@ -5,5 +5,5 @@ authors = ["Parity Technologies <admin@parity.io>"]
[dependencies]
elastic-array = "0.9"
error-chain = "0.11.0-rc.2"
error-chain = "0.11.0"
ethcore-bytes = { path = "../bytes" }

10
util/kvdb/src/lib.rs
View File

@@ -33,7 +33,7 @@ pub type DBValue = ElasticArray128<u8>;
error_chain! {
types {
Error, ErrorKind, ResultExt;
Error, ErrorKind, ResultExt, Result;
}
foreign_links {
@@ -148,7 +148,7 @@ pub trait KeyValueDB: Sync + Send {
fn transaction(&self) -> DBTransaction { DBTransaction::new() }
/// Get a value by key.
fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<DBValue>, String>;
fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<DBValue>>;
/// Get a value by partial key. Only works for flushed data.
fn get_by_prefix(&self, col: Option<u32>, prefix: &[u8]) -> Option<Box<[u8]>>;
@@ -157,13 +157,13 @@ pub trait KeyValueDB: Sync + Send {
fn write_buffered(&self, transaction: DBTransaction);
/// Write a transaction of changes to the backing store.
fn write(&self, transaction: DBTransaction) -> Result<(), String> {
fn write(&self, transaction: DBTransaction) -> Result<()> {
self.write_buffered(transaction);
self.flush()
}
/// Flush all buffered data.
fn flush(&self) -> Result<(), String>;
fn flush(&self) -> Result<()>;
/// Iterate over flushed data for a given column.
fn iter<'a>(&'a self, col: Option<u32>) -> Box<Iterator<Item=(Box<[u8]>, Box<[u8]>)> + 'a>;
@@ -173,5 +173,5 @@ pub trait KeyValueDB: Sync + Send {
-> Box<Iterator<Item=(Box<[u8]>, Box<[u8]>)> + 'a>;
/// Attempt to replace this database with a new one located at the given path.
fn restore(&self, new_db: &str) -> Result<(), Error>;
fn restore(&self, new_db: &str) -> Result<()>;
}

1
util/migration/Cargo.toml
View File

@@ -9,3 +9,4 @@ macros = { path = "../macros" }
kvdb = { path = "../kvdb" }
kvdb-rocksdb = { path = "../kvdb-rocksdb" }
ethcore-devtools = { path = "../../devtools" }
error-chain = "0.11.0"

100
util/migration/src/lib.rs
View File

@@ -22,6 +22,8 @@ mod tests;
extern crate log;
#[macro_use]
extern crate macros;
#[macro_use]
extern crate error_chain;
extern crate ethcore_devtools as devtools;
extern crate kvdb;
@@ -30,11 +32,37 @@ extern crate kvdb_rocksdb;
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::{fs, fmt};
use std::{fs, io};
use kvdb::DBTransaction;
use kvdb_rocksdb::{CompactionProfile, Database, DatabaseConfig};
error_chain! {
types {
Error, ErrorKind, ResultExt, Result;
}
links {
Db(kvdb::Error, kvdb::ErrorKind);
}
foreign_links {
Io(io::Error);
}
errors {
CannotAddMigration {
description("Cannot add migration"),
display("Cannot add migration"),
}
MigrationImpossible {
description("Migration impossible"),
display("Migration impossible"),
}
}
}
/// Migration config.
#[derive(Clone)]
pub struct Config {
@@ -71,7 +99,7 @@ impl Batch {
}
/// Insert a value into the batch, committing if necessary.
pub fn insert(&mut self, key: Vec<u8>, value: Vec<u8>, dest: &mut Database) -> Result<(), Error> {
pub fn insert(&mut self, key: Vec<u8>, value: Vec<u8>, dest: &mut Database) -> Result<()> {
self.inner.insert(key, value);
if self.inner.len() == self.batch_size {
self.commit(dest)?;
@@ -80,7 +108,7 @@ impl Batch {
}
/// Commit all the items in the batch to the given database.
pub fn commit(&mut self, dest: &mut Database) -> Result<(), Error> {
pub fn commit(&mut self, dest: &mut Database) -> Result<()> {
if self.inner.is_empty() { return Ok(()) }
let mut transaction = DBTransaction::new();
@@ -90,43 +118,7 @@ impl Batch {
}
self.inner.clear();
dest.write(transaction).map_err(Error::Custom)
}
}
/// Migration error.
#[derive(Debug)]
pub enum Error {
/// Error returned when it is impossible to add new migration rules.
CannotAddMigration,
/// Error returned when migration from specific version can not be performed.
MigrationImpossible,
/// Io Error.
Io(::std::io::Error),
/// Custom error.
Custom(String),
}
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
match *self {
Error::CannotAddMigration => write!(f, "Cannot add migration"),
Error::MigrationImpossible => write!(f, "Migration impossible"),
Error::Io(ref err) => write!(f, "{}", err),
Error::Custom(ref err) => write!(f, "{}", err),
}
}
}
impl From<::std::io::Error> for Error {
fn from(e: ::std::io::Error) -> Self {
Error::Io(e)
}
}
impl From<String> for Error {
fn from(e: String) -> Self {
Error::Custom(e)
dest.write(transaction).map_err(Into::into)
}
}
@@ -142,7 +134,7 @@ pub trait Migration: 'static {
/// Version of the database after the migration.
fn version(&self) -> u32;
/// Migrate a source to a destination.
fn migrate(&mut self, source: Arc<Database>, config: &Config, destination: &mut Database, col: Option<u32>) -> Result<(), Error>;
fn migrate(&mut self, source: Arc<Database>, config: &Config, destination: &mut Database, col: Option<u32>) -> Result<()>;
}
/// A simple migration over key-value pairs.
@@ -163,7 +155,7 @@ impl<T: SimpleMigration> Migration for T {
fn alters_existing(&self) -> bool { true }
fn migrate(&mut self, source: Arc<Database>, config: &Config, dest: &mut Database, col: Option<u32>) -> Result<(), Error> {
fn migrate(&mut self, source: Arc<Database>, config: &Config, dest: &mut Database, col: Option<u32>) -> Result<()> {
let mut batch = Batch::new(config, col);
let iter = match source.iter(col) {
@@ -196,7 +188,7 @@ impl Migration for ChangeColumns {
fn columns(&self) -> Option<u32> { self.post_columns }
fn version(&self) -> u32 { self.version }
fn alters_existing(&self) -> bool { false }
fn migrate(&mut self, _: Arc<Database>, _: &Config, _: &mut Database, _: Option<u32>) -> Result<(), Error> {
fn migrate(&mut self, _: Arc<Database>, _: &Config, _: &mut Database, _: Option<u32>) -> Result<()> {
Ok(())
}
}
@@ -250,7 +242,7 @@ impl Manager {
}
/// Adds new migration rules.
pub fn add_migration<T>(&mut self, migration: T) -> Result<(), Error> where T: Migration {
pub fn add_migration<T>(&mut self, migration: T) -> Result<()> where T: Migration {
let is_new = match self.migrations.last() {
Some(last) => migration.version() > last.version(),
None => true,
@@ -258,17 +250,19 @@ impl Manager {
match is_new {
true => Ok(self.migrations.push(Box::new(migration))),
false => Err(Error::CannotAddMigration),
false => Err(ErrorKind::CannotAddMigration.into()),
}
}
/// Performs migration in order, starting with a source path, migrating between two temporary databases,
/// and producing a path where the final migration lives.
pub fn execute(&mut self, old_path: &Path, version: u32) -> Result<PathBuf, Error> {
pub fn execute(&mut self, old_path: &Path, version: u32) -> Result<PathBuf> {
let config = self.config.clone();
let migrations = self.migrations_from(version);
trace!(target: "migration", "Total migrations to execute for version {}: {}", version, migrations.len());
if migrations.is_empty() { return Err(Error::MigrationImpossible) };
if migrations.is_empty() {
return Err(ErrorKind::MigrationImpossible.into())
};
let columns = migrations.get(0).and_then(|m| m.pre_columns());
@@ -286,8 +280,8 @@ impl Manager {
let mut temp_path = old_path.to_path_buf();
// start with the old db.
let old_path_str = old_path.to_str().ok_or(Error::MigrationImpossible)?;
let mut cur_db = Arc::new(Database::open(&db_config, old_path_str).map_err(Error::Custom)?);
let old_path_str = old_path.to_str().ok_or(ErrorKind::MigrationImpossible)?;
let mut cur_db = Arc::new(Database::open(&db_config, old_path_str)?);
for migration in migrations {
trace!(target: "migration", "starting migration to version {}", migration.version());
@@ -300,8 +294,8 @@ impl Manager {
temp_path = temp_idx.path(&db_root);
// open the target temporary database.
let temp_path_str = temp_path.to_str().ok_or(Error::MigrationImpossible)?;
let mut new_db = Database::open(&db_config, temp_path_str).map_err(Error::Custom)?;
let temp_path_str = temp_path.to_str().ok_or(ErrorKind::MigrationImpossible)?;
let mut new_db = Database::open(&db_config, temp_path_str)?;
match current_columns {
// migrate only default column
@@ -324,11 +318,11 @@ impl Manager {
// we can do this in-place.
let goal_columns = migration.columns().unwrap_or(0);
while cur_db.num_columns() < goal_columns {
cur_db.add_column().map_err(Error::Custom)?;
cur_db.add_column().map_err(kvdb::Error::from)?;
}
while cur_db.num_columns() > goal_columns {
cur_db.drop_column().map_err(Error::Custom)?;
cur_db.drop_column().map_err(kvdb::Error::from)?;
}
}
}

11
util/patricia_trie/src/lib.rs
View File

@@ -28,7 +28,7 @@ extern crate ethcore_logger;
#[macro_use]
extern crate log;
use std::fmt;
use std::{fmt, error};
use bigint::hash::H256;
use keccak::KECCAK_NULL_RLP;
use hashdb::{HashDB, DBValue};
@@ -86,6 +86,15 @@ impl fmt::Display for TrieError {
}
}
impl error::Error for TrieError {
fn description(&self) -> &str {
match *self {
TrieError::InvalidStateRoot(_) => "Invalid state root",
TrieError::IncompleteDatabase(_) => "Incomplete database",
}
}
}
/// Trie result type. Boxed to avoid copying around extra space for `H256`s on successful queries.
pub type Result<T> = ::std::result::Result<T, Box<TrieError>>;