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

Backports to beta (#1919)

Browse Source 
* RPC errors & logs (#1845)

* Refactoring errors in RPC

* Updating jsonrpc-core

* Fixing code_at

* Avoid mentioning obvious segments in proof

[ci:skip]

* fixed cache_manager lock order

* Purging .derefs, fixing clippy warnings. (#1890)

* Fixing clippy warnings

* Purging derefs

* Simplifying engine derefs

* Simplifying more engine derefs

* Adding more details to miner log

* fixed #1889, .DS_Store is no longer treated as key file (#1892)

* fixed #1889, .DS_Store is no longer treated as key file

* ethstore filters directories, hidden files and common system files

* fixed compiling

* fix regression with geth dir

* fix regression with geth dir

* Fix ipc compilation and add ipc feature to test targets (#1902)

* fix compilation and add it to the ci run

* no separator?

* use quotes and spaces

* RocksDB version bump

* Don't return deleted nodes that are not yet flushed (#1908)

* polling & connection timeouts (#1910)

* Peers RPC + UI displaying active/connected/max peers (#1915)

* Peers API

* Bumping Parity-UI

* Fixing tests

* Save nodes removed from backing_overlay until commit (#1917)
This commit is contained in:
Arkadiy Paronyan
2016-08-12 12:16:20 +02:00
committed by GitHub
parent 52ac5a00f5
commit dfb2ddfdc2
77 changed files with 1139 additions and 858 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
util/src/hash.rs
View File

@@ -577,9 +577,9 @@ impl Hasher for PlainHasher {
}
}
/// Specialized version of HashMap with H256 keys and fast hashing function.
/// Specialized version of `HashMap` with H256 keys and fast hashing function.
pub type H256FastMap<T> = HashMap<H256, T, BuildHasherDefault<PlainHasher>>;
/// Specialized version of HashSet with H256 keys and fast hashing function.
/// Specialized version of `HashSet` with H256 keys and fast hashing function.
pub type H256FastSet = HashSet<H256, BuildHasherDefault<PlainHasher>>;
#[cfg(test)]

2
util/src/journaldb/archivedb.rs
View File

@@ -74,7 +74,7 @@ impl HashDB for ArchiveDB {
fn keys(&self) -> HashMap<H256, i32> {
let mut ret: HashMap<H256, i32> = HashMap::new();
for (key, _) in self.backing.iter(self.column) {
let h = H256::from_slice(key.deref());
let h = H256::from_slice(&*key);
ret.insert(h, 1);
}

2
util/src/journaldb/earlymergedb.rs
View File

@@ -263,7 +263,7 @@ impl HashDB for EarlyMergeDB {
fn keys(&self) -> HashMap<H256, i32> {
let mut ret: HashMap<H256, i32> = HashMap::new();
for (key, _) in self.backing.iter(self.column) {
let h = H256::from_slice(key.deref());
let h = H256::from_slice(&*key);
ret.insert(h, 1);
}

41
util/src/journaldb/overlayrecentdb.rs
View File

@@ -66,7 +66,8 @@ pub struct OverlayRecentDB {
#[derive(PartialEq)]
struct JournalOverlay {
backing_overlay: MemoryDB,
backing_overlay: MemoryDB, // Nodes added in the history period
pending_overlay: H256FastMap<Bytes>, // Nodes being transfered from backing_overlay to backing db
journal: HashMap<u64, Vec<JournalEntry>>,
latest_era: Option<u64>,
}
@@ -173,7 +174,11 @@ impl OverlayRecentDB {
}
}
trace!("Recovered {} overlay entries, {} journal entries", count, journal.len());
JournalOverlay { backing_overlay: overlay, journal: journal, latest_era: latest_era }
JournalOverlay {
backing_overlay: overlay,
pending_overlay: HashMap::default(),
journal: journal,
latest_era: latest_era }
}
}
@@ -194,6 +199,7 @@ impl JournalDB for OverlayRecentDB {
let mut mem = self.transaction_overlay.mem_used();
let overlay = self.journal_overlay.read();
mem += overlay.backing_overlay.mem_used();
mem += overlay.pending_overlay.heap_size_of_children();
mem += overlay.journal.heap_size_of_children();
mem
}
@@ -209,14 +215,19 @@ impl JournalDB for OverlayRecentDB {
fn latest_era(&self) -> Option<u64> { self.journal_overlay.read().latest_era }
fn state(&self, key: &H256) -> Option<Bytes> {
let v = self.journal_overlay.read().backing_overlay.get(&to_short_key(key)).map(|v| v.to_vec());
v.or_else(|| self.backing.get_by_prefix(self.column, &key[0..DB_PREFIX_LEN]).map(|b| b.to_vec()))
let journal_overlay = self.journal_overlay.read();
let key = to_short_key(key);
journal_overlay.backing_overlay.get(&key).map(|v| v.to_vec())
.or_else(|| journal_overlay.pending_overlay.get(&key).map(|v| v.clone()))
.or_else(|| self.backing.get_by_prefix(self.column, &key[0..DB_PREFIX_LEN]).map(|b| b.to_vec()))
}
fn commit(&mut self, batch: &DBTransaction, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> {
// record new commit's details.
trace!("commit: #{} ({}), end era: {:?}", now, id, end);
let mut journal_overlay = self.journal_overlay.write();
// flush previous changes
journal_overlay.pending_overlay.clear();
{
let mut r = RlpStream::new_list(3);
let mut tx = self.transaction_overlay.drain();
@@ -280,7 +291,8 @@ impl JournalDB for OverlayRecentDB {
}
// apply canon inserts first
for (k, v) in canon_insertions {
try!(batch.put_vec(self.column, &k, v));
try!(batch.put(self.column, &k, &v));
journal_overlay.pending_overlay.insert(to_short_key(&k), v);
}
// update the overlay
for k in overlay_deletions {
@@ -298,6 +310,10 @@ impl JournalDB for OverlayRecentDB {
Ok(0)
}
fn flush(&self) {
self.journal_overlay.write().pending_overlay.clear();
}
fn inject(&mut self, batch: &DBTransaction) -> Result<u32, UtilError> {
let mut ops = 0;
for (key, (value, rc)) in self.transaction_overlay.drain() {
@@ -329,7 +345,7 @@ impl HashDB for OverlayRecentDB {
fn keys(&self) -> HashMap<H256, i32> {
let mut ret: HashMap<H256, i32> = HashMap::new();
for (key, _) in self.backing.iter(self.column) {
let h = H256::from_slice(key.deref());
let h = H256::from_slice(&*key);
ret.insert(h, 1);
}
@@ -345,14 +361,19 @@ impl HashDB for OverlayRecentDB {
match k {
Some((d, rc)) if rc > 0 => Some(d),
_ => {
let v = self.journal_overlay.read().backing_overlay.get(&to_short_key(key)).map(|v| v.to_vec());
let v = {
let journal_overlay = self.journal_overlay.read();
let key = to_short_key(key);
journal_overlay.backing_overlay.get(&key).map(|v| v.to_vec())
.or_else(|| journal_overlay.pending_overlay.get(&key).map(|v| v.clone()))
};
match v {
Some(x) => {
Some(&self.transaction_overlay.denote(key, x).0)
Some(self.transaction_overlay.denote(key, x).0)
}
_ => {
if let Some(x) = self.payload(key) {
Some(&self.transaction_overlay.denote(key, x).0)
Some(self.transaction_overlay.denote(key, x).0)
}
else {
None
@@ -920,4 +941,4 @@ mod tests {
assert!(jdb.get(&key).is_none());
}
}
}

2
util/src/journaldb/refcounteddb.rs
View File

@@ -190,7 +190,7 @@ impl JournalDB for RefCountedDB {
for remove in self.removes.drain(..) {
self.forward.remove(&remove);
}
self.forward.commit_to_batch(&batch)
self.forward.commit_to_batch(batch)
}
}

8
util/src/journaldb/traits.rs
View File

@@ -57,12 +57,18 @@ pub trait JournalDB: HashDB {
/// Get backing database.
fn backing(&self) -> &Arc<Database>;
/// Clear internal strucutres. This should called after changes have been written
/// to the backing strage
fn flush(&self) {}
/// Commit all changes in a single batch
#[cfg(test)]
fn commit_batch(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> {
let batch = self.backing().transaction();
let res = try!(self.commit(&batch, now, id, end));
self.backing().write(batch).map(|_| res).map_err(Into::into)
let result = self.backing().write(batch).map(|_| res).map_err(Into::into);
self.flush();
result
}
/// Inject all changes in a single batch.

132
util/src/kvdb.rs
View File

@@ -105,10 +105,10 @@ impl DBTransaction {
}
}
struct DBColumnOverlay {
insertions: HashMap<ElasticArray32<u8>, Bytes>,
compressed_insertions: HashMap<ElasticArray32<u8>, Bytes>,
deletions: HashSet<ElasticArray32<u8>>,
enum KeyState {
Insert(Bytes),
InsertCompressed(Bytes),
Delete,
}
/// Compaction profile for the database settings
@@ -198,7 +198,7 @@ pub struct Database {
db: DB,
write_opts: WriteOptions,
cfs: Vec<Column>,
overlay: RwLock<Vec<DBColumnOverlay>>,
overlay: RwLock<Vec<HashMap<ElasticArray32<u8>, KeyState>>>,
}
impl Database {
@@ -275,11 +275,7 @@ impl Database {
Ok(Database {
db: db,
write_opts: write_opts,
overlay: RwLock::new((0..(cfs.len() + 1)).map(|_| DBColumnOverlay {
insertions: HashMap::new(),
compressed_insertions: HashMap::new(),
deletions: HashSet::new(),
}).collect()),
overlay: RwLock::new((0..(cfs.len() + 1)).map(|_| HashMap::new()).collect()),
cfs: cfs,
})
}
@@ -302,21 +298,15 @@ impl Database {
match op {
DBOp::Insert { col, key, value } => {
let c = Self::to_overlay_column(col);
overlay[c].deletions.remove(&key);
overlay[c].compressed_insertions.remove(&key);
overlay[c].insertions.insert(key, value);
overlay[c].insert(key, KeyState::Insert(value));
},
DBOp::InsertCompressed { col, key, value } => {
let c = Self::to_overlay_column(col);
overlay[c].deletions.remove(&key);
overlay[c].insertions.remove(&key);
overlay[c].compressed_insertions.insert(key, value);
overlay[c].insert(key, KeyState::InsertCompressed(value));
},
DBOp::Delete { col, key } => {
let c = Self::to_overlay_column(col);
overlay[c].insertions.remove(&key);
overlay[c].compressed_insertions.remove(&key);
overlay[c].deletions.insert(key);
overlay[c].insert(key, KeyState::Delete);
},
}
};
@@ -328,34 +318,34 @@ impl Database {
let batch = WriteBatch::new();
let mut overlay = self.overlay.write();
let mut c = 0;
for column in overlay.iter_mut() {
let insertions = mem::replace(&mut column.insertions, HashMap::new());
let compressed_insertions = mem::replace(&mut column.compressed_insertions, HashMap::new());
let deletions = mem::replace(&mut column.deletions, HashSet::new());
for d in deletions.into_iter() {
if c > 0 {
try!(batch.delete_cf(self.cfs[c - 1], &d));
} else {
try!(batch.delete(&d));
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(self.cfs[c - 1], &key));
} else {
try!(batch.delete(&key));
}
},
KeyState::Insert(value) => {
if c > 0 {
try!(batch.put_cf(self.cfs[c - 1], &key, &value));
} else {
try!(batch.put(&key, &value));
}
},
KeyState::InsertCompressed(value) => {
let compressed = UntrustedRlp::new(&value).compress(RlpType::Blocks);
if c > 0 {
try!(batch.put_cf(self.cfs[c - 1], &key, &compressed));
} else {
try!(batch.put(&key, &value));
}
}
}
}
for (key, value) in insertions.into_iter() {
if c > 0 {
try!(batch.put_cf(self.cfs[c - 1], &key, &value));
} else {
try!(batch.put(&key, &value));
}
}
for (key, value) in compressed_insertions.into_iter() {
let compressed = UntrustedRlp::new(&value).compress(RlpType::Blocks);
if c > 0 {
try!(batch.put_cf(self.cfs[c - 1], &key, &compressed));
} else {
try!(batch.put(&key, &compressed));
}
}
c += 1;
}
self.db.write_opt(batch, &self.write_opts)
}
@@ -385,14 +375,19 @@ impl Database {
/// Get value by key.
pub fn get(&self, col: Option<u32>, key: &[u8]) -> Result<Option<Bytes>, String> {
let overlay = &self.overlay.read()[Self::to_overlay_column(col)];
overlay.insertions.get(key).or_else(|| overlay.compressed_insertions.get(key)).map_or_else(||
col.map_or_else(
|| self.db.get(key).map(|r| r.map(|v| v.to_vec())),
|c| self.db.get_cf(self.cfs[c as usize], key).map(|r| r.map(|v| v.to_vec()))),
|value| Ok(Some(value.clone())))
match overlay.get(key) {
Some(&KeyState::Insert(ref value)) | Some(&KeyState::InsertCompressed(ref value)) => Ok(Some(value.clone())),
Some(&KeyState::Delete) => Ok(None),
None => {
col.map_or_else(
|| self.db.get(key).map(|r| r.map(|v| v.to_vec())),
|c| self.db.get_cf(self.cfs[c as usize], key).map(|r| r.map(|v| v.to_vec())))
},
}
}
/// Get value by partial key. Prefix size should match configured prefix size.
/// Get value by partial key. Prefix size should match configured prefix size. Only searches flushed values.
// TODO: support prefix seek for unflushed ata
pub fn get_by_prefix(&self, col: Option<u32>, prefix: &[u8]) -> Option<Box<[u8]>> {
let mut iter = col.map_or_else(|| self.db.iterator(IteratorMode::From(prefix, Direction::Forward)),
|c| self.db.iterator_cf(self.cfs[c as usize], IteratorMode::From(prefix, Direction::Forward)).unwrap());
@@ -403,13 +398,9 @@ impl Database {
}
}
/// Check if there is anything in the database.
pub fn is_empty(&self, col: Option<u32>) -> bool {
self.iter(col).next().is_none()
}
/// Get database iterator.
/// Get database iterator for flushed data.
pub fn iter(&self, col: Option<u32>) -> DatabaseIterator {
//TODO: iterate over overlay
col.map_or_else(|| DatabaseIterator { iter: self.db.iterator(IteratorMode::Start) },
|c| DatabaseIterator { iter: self.db.iterator_cf(self.cfs[c as usize], IteratorMode::Start).unwrap() })
}
@@ -421,7 +412,6 @@ mod tests {
use super::*;
use devtools::*;
use std::str::FromStr;
use std::ops::Deref;
fn test_db(config: &DatabaseConfig) {
let path = RandomTempPath::create_dir();
@@ -435,13 +425,13 @@ mod tests {
batch.put(None, &key2, b"dog").unwrap();
db.write(batch).unwrap();
assert_eq!(db.get(None, &key1).unwrap().unwrap().deref(), b"cat");
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.deref());
assert_eq!(&*contents[0].0, &*key1);
assert_eq!(&*contents[0].1, b"cat");
assert_eq!(&*contents[1].0, key2.deref());
assert_eq!(&*contents[1].0, &*key2);
assert_eq!(&*contents[1].1, b"dog");
let batch = db.transaction();
@@ -459,17 +449,27 @@ mod tests {
transaction.delete(None, &key1).unwrap();
db.write(transaction).unwrap();
assert!(db.get(None, &key1).unwrap().is_none());
assert_eq!(db.get(None, &key3).unwrap().unwrap().deref(), b"elephant");
assert_eq!(&*db.get(None, &key3).unwrap().unwrap(), b"elephant");
assert_eq!(db.get_by_prefix(None, &key3).unwrap().deref(), b"elephant");
assert_eq!(db.get_by_prefix(None, &key2).unwrap().deref(), b"dog");
assert_eq!(&*db.get_by_prefix(None, &key3).unwrap(), b"elephant");
assert_eq!(&*db.get_by_prefix(None, &key2).unwrap(), b"dog");
let transaction = db.transaction();
transaction.put(None, &key1, b"horse").unwrap();
transaction.delete(None, &key3).unwrap();
db.write_buffered(transaction).unwrap();
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 smoke = Database::open_default(path.as_path().to_str().unwrap()).unwrap();
assert!(smoke.is_empty(None));
let _ = Database::open_default(path.as_path().to_str().unwrap()).unwrap();
test_db(&DatabaseConfig::default());
}
}

7
util/src/overlaydb.rs
View File

@@ -22,7 +22,6 @@ use bytes::*;
use rlp::*;
use hashdb::*;
use memorydb::*;
use std::ops::*;
use std::sync::*;
use std::collections::HashMap;
use kvdb::{Database, DBTransaction};
@@ -130,7 +129,7 @@ impl HashDB for OverlayDB {
fn keys(&self) -> HashMap<H256, i32> {
let mut ret: HashMap<H256, i32> = HashMap::new();
for (key, _) in self.backing.iter(self.column) {
let h = H256::from_slice(key.deref());
let h = H256::from_slice(&*key);
let r = self.payload(&h).unwrap().1;
ret.insert(h, r as i32);
}
@@ -305,7 +304,7 @@ fn playpen() {
batch.put(None, b"test", b"test2").unwrap();
db.write(batch).unwrap();
match db.get(None, b"test") {
Ok(Some(value)) => println!("Got value {:?}", value.deref()),
Ok(Some(value)) => println!("Got value {:?}", &*value),
Ok(None) => println!("No value for that key"),
Err(..) => println!("Gah"),
}
@@ -314,4 +313,4 @@ fn playpen() {
db.write(batch).unwrap();
}
fs::remove_dir_all("/tmp/test").unwrap();
}
}

6
util/src/rlp/rlpstream.rs
View File

@@ -14,8 +14,6 @@
// 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::ops::Deref;
use std::default::Default;
use elastic_array::*;
use rlp::bytes::{ToBytes, VecLike};
use rlp::{Stream, Encoder, Encodable};
@@ -293,7 +291,7 @@ impl<'a> Encodable for &'a[u8] {
impl Encodable for Vec<u8> {
fn rlp_append(&self, s: &mut RlpStream) {
s.append_value(&U8Slice(self.deref()))
s.append_value(&U8Slice(self))
}
}
@@ -334,7 +332,7 @@ impl<'a, T> Encodable for &'a[T] where T: Encodable {
impl<T> Encodable for Vec<T> where T: Encodable {
fn rlp_append(&self, s: &mut RlpStream) {
Encodable::rlp_append(&self.deref(), s);
Encodable::rlp_append(&self.as_slice(), s);
}
}

5
util/src/rlp/rlptraits.rs
View File

@@ -15,7 +15,6 @@
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
//! Common RLP traits
use std::ops::Deref;
use rlp::bytes::VecLike;
use rlp::{DecoderError, UntrustedRlp};
use rlp::rlpstream::RlpStream;
@@ -244,7 +243,7 @@ pub trait ByteEncodable {
fn bytes_len(&self) -> usize;
}
/// Structure encodable to RLP. Implement this trait for
/// Structure encodable to RLP. Implement this trait for
pub trait Encodable {
/// Append a value to the stream
fn rlp_append(&self, s: &mut RlpStream);
@@ -257,7 +256,7 @@ pub trait Encodable {
}
/// Get the hash or RLP encoded representation
fn rlp_sha3(&self) -> H256 { self.rlp_bytes().deref().sha3() }
fn rlp_sha3(&self) -> H256 { (&*self.rlp_bytes()).sha3() }
}
/// Encodable wrapper trait required to handle special case of encoding a &[u8] as string and not as list

2
util/src/trie/triedb.rs
View File

@@ -192,7 +192,7 @@ impl<'db> TrieDB<'db> {
where 'db: 'key
{
let root_rlp = try!(self.root_data());
self.get_from_node(&root_rlp, key)
self.get_from_node(root_rlp, key)
}
/// Recursible function to retrieve the value given a `node` and a partial `key`. `None` if no