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

Apply all removes after updating all counters

Browse Source
This commit is contained in:
arkpar 2016-02-07 21:18:51 +01:00
parent 28dd73e340
commit 6c36a7e1a6
1 changed files with 30 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
util/src/journaldb.rs
View File

@ -20,7 +20,7 @@ use common::*;
use rlp::*; use rlp::*;
use hashdb::*; use hashdb::*;
use memorydb::*; use memorydb::*;
use rocksdb::{DB, Writable, WriteBatch, IteratorMode, DBVector}; use rocksdb::{DB, Writable, WriteBatch, IteratorMode};
#[cfg(test)] #[cfg(test)]
use std::env; use std::env;
@ -105,12 +105,16 @@ impl JournalDB {
// we remove all of its removes assuming it is canonical and all // we remove all of its removes assuming it is canonical and all
// of its inserts otherwise. // of its inserts otherwise.
// //
// we also keep track of the counters for each key inserted in the journal to handle // We also keep reference counters for each key inserted in the journal to handle
// the following cases where key K must not be deleted from the DB: // the following cases where key K must not be deleted from the DB when processing removals :
// Given H is the journal size in eras, 0 <= C <= H. // Given H is the journal size in eras, 0 <= C <= H.
// Key K is removed in era A(N) and re-inserted in canonical era B(N + C). // Key K is removed in era A(N) and re-inserted in canonical era B(N + C).
// Key K is removed in era A(N) and re-inserted in non-canonical era B`(N + C). // Key K is removed in era A(N) and re-inserted in non-canonical era B`(N + C).
// Key K is added in non-canonical era A'(N) canonical B(N + C). // Key K is added in non-canonical era A'(N) canonical B(N + C).
//
// The counter is encreased each time a key is inserted in the journal in the commit. The list of insertions
// is saved with the era record. When the era becomes end_era and goes out of journal the counter is decreased
// and the key is safe to delete.
// record new commit's details. // record new commit's details.
let batch = WriteBatch::new(); let batch = WriteBatch::new();
@ -131,7 +135,7 @@ impl JournalDB {
let mut r = RlpStream::new_list(3); let mut r = RlpStream::new_list(3);
let inserts: Vec<H256> = self.overlay.keys().iter().filter(|&(_, &c)| c > 0).map(|(key, _)| key.clone()).collect(); let inserts: Vec<H256> = self.overlay.keys().iter().filter(|&(_, &c)| c > 0).map(|(key, _)| key.clone()).collect();
// Increase counter for each insrted key no matter if the block is canonical or not. // Increase counter for each inserted key no matter if the block is canonical or not.
for i in &inserts { for i in &inserts {
*counters.entry(i.clone()).or_insert(0) += 1; *counters.entry(i.clone()).or_insert(0) += 1;
} }
@ -146,7 +150,8 @@ impl JournalDB {
if let Some((end_era, canon_id)) = end { if let Some((end_era, canon_id)) = end {
let mut index = 0usize; let mut index = 0usize;
let mut last; let mut last;
let mut canon_data: Option<DBVector> = None; let mut to_remove: Vec<H256> = Vec::new();
let mut canon_inserts: Vec<H256> = Vec::new();
while let Some(rlp_data) = try!(self.backing.get({ while let Some(rlp_data) = try!(self.backing.get({
let mut r = RlpStream::new_list(2); let mut r = RlpStream::new_list(2);
r.append(&end_era); r.append(&end_era);
@ -154,30 +159,30 @@ impl JournalDB {
last = r.drain(); last = r.drain();
&last &last
})) { })) {
let canon = {
let rlp = Rlp::new(&rlp_data); let rlp = Rlp::new(&rlp_data);
if canon_id != rlp.val_at(0) { let inserts: Vec<H256> = rlp.val_at(1);
let to_add: Vec<H256> = rlp.val_at(1); JournalDB::decrease_counters(&inserts, &mut counters);
JournalDB::apply_removes(&to_add, &to_add, &mut counters, &batch); // Collect keys to be removed. These are removed keys for canonical block, inserted for non-canonical
false if canon_id == rlp.val_at(0) {
} else { true } to_remove.extend(rlp.at(2).iter().map(|r| r.as_val::<H256>()));
}; canon_inserts = inserts;
if canon { }
canon_data = Some(rlp_data) else {
to_remove.extend(inserts);
} }
try!(batch.delete(&last)); try!(batch.delete(&last));
index += 1; index += 1;
} }
// Canon must be commited last to handle a case when counter reaches 0 in a sibling block
if let Some(ref c) = canon_data {
let rlp = Rlp::new(&c);
let deleted = JournalDB::apply_removes(&rlp.val_at::<Vec<H256>>(1), &rlp.val_at::<Vec<H256>>(2), &mut counters, &batch);
trace!("JournalDB: delete journal for time #{}.{}, (canon was {}): {} entries", end_era, index, canon_id, deleted);
}
let canon_inserts = canon_inserts.drain(..).collect::<HashSet<_>>();
// Purge removed keys if they are not referenced and not re-inserted in the canon commit
for h in to_remove.iter().filter(|h| !counters.contains_key(h) && !canon_inserts.contains(h)) {
try!(batch.delete(&h));
}
try!(batch.put(&LAST_ERA_KEY, &encode(&end_era))); try!(batch.put(&LAST_ERA_KEY, &encode(&end_era)));
} }
// Commit overlay insertions
let mut ret = 0u32; let mut ret = 0u32;
let mut deletes = 0usize; let mut deletes = 0usize;
for i in self.overlay.drain().into_iter() { for i in self.overlay.drain().into_iter() {
@ -199,10 +204,10 @@ impl JournalDB {
Ok(ret) Ok(ret)
} }
fn apply_removes(added: &[H256], removed: &[H256], counters: &mut HashMap<H256, i32>, batch: &WriteBatch) -> usize {
let mut deleted = 0usize; // Decrease counters for given keys. Deletes obsolete counters
// Decrease the counters first fn decrease_counters(keys: &[H256], counters: &mut HashMap<H256, i32>) {
for i in added.iter() { for i in keys.iter() {
let delete_counter = { let delete_counter = {
if let Some(mut cnt) = counters.get_mut(i) { if let Some(mut cnt) = counters.get_mut(i) {
*cnt -= 1; *cnt -= 1;
@ -214,12 +219,6 @@ impl JournalDB {
counters.remove(i); counters.remove(i);
} }
} }
// Remove only if counter reached zero
for i in removed.iter().filter(|i| !counters.contains_key(i)) {
batch.delete(&i).expect("Low-level database error. Some issue with your hard disk?");
deleted += 1;
}
deleted
} }
fn payload(&self, key: &H256) -> Option<Bytes> { fn payload(&self, key: &H256) -> Option<Bytes> {