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

Merge remote-tracking branch 'origin/master' into boxjdb

Browse Source
This commit is contained in:
Gav Wood
2016-03-11 12:56:33 +01:00
parent 756f964130 62e94e1744
commit e461916f5a
28 changed files with 1329 additions and 728 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
ethcore/src/block_queue.rs
View File

@@ -95,7 +95,7 @@ pub struct BlockQueue {
panic_handler: Arc<PanicHandler>,
engine: Arc<Box<Engine>>,
more_to_verify: Arc<Condvar>,
verification: Arc<Mutex<Verification>>,
verification: Arc<Verification>,
verifiers: Vec<JoinHandle<()>>,
deleting: Arc<AtomicBool>,
ready_signal: Arc<QueueSignal>,
@@ -132,18 +132,23 @@ impl QueueSignal {
}
}
#[derive(Default)]
struct Verification {
unverified: VecDeque<UnverifiedBlock>,
verified: VecDeque<PreverifiedBlock>,
verifying: VecDeque<VerifyingBlock>,
bad: HashSet<H256>,
// All locks must be captured in the order declared here.
unverified: Mutex<VecDeque<UnverifiedBlock>>,
verified: Mutex<VecDeque<PreverifiedBlock>>,
verifying: Mutex<VecDeque<VerifyingBlock>>,
bad: Mutex<HashSet<H256>>,
}
impl BlockQueue {
/// Creates a new queue instance.
pub fn new(config: BlockQueueConfig, engine: Arc<Box<Engine>>, message_channel: IoChannel<NetSyncMessage>) -> BlockQueue {
let verification = Arc::new(Mutex::new(Verification::default()));
let verification = Arc::new(Verification {
unverified: Mutex::new(VecDeque::new()),
verified: Mutex::new(VecDeque::new()),
verifying: Mutex::new(VecDeque::new()),
bad: Mutex::new(HashSet::new()),
});
let more_to_verify = Arc::new(Condvar::new());
let ready_signal = Arc::new(QueueSignal { signalled: AtomicBool::new(false), message_channel: message_channel });
let deleting = Arc::new(AtomicBool::new(false));
@@ -186,17 +191,17 @@ impl BlockQueue {
}
}
fn verify(verification: Arc<Mutex<Verification>>, engine: Arc<Box<Engine>>, wait: Arc<Condvar>, ready: Arc<QueueSignal>, deleting: Arc<AtomicBool>, empty: Arc<Condvar>) {
fn verify(verification: Arc<Verification>, engine: Arc<Box<Engine>>, wait: Arc<Condvar>, ready: Arc<QueueSignal>, deleting: Arc<AtomicBool>, empty: Arc<Condvar>) {
while !deleting.load(AtomicOrdering::Acquire) {
{
let mut lock = verification.lock().unwrap();
let mut unverified = verification.unverified.lock().unwrap();
if lock.unverified.is_empty() && lock.verifying.is_empty() {
if unverified.is_empty() && verification.verifying.lock().unwrap().is_empty() {
empty.notify_all();
}
while lock.unverified.is_empty() && !deleting.load(AtomicOrdering::Acquire) {
lock = wait.wait(lock).unwrap();
while unverified.is_empty() && !deleting.load(AtomicOrdering::Acquire) {
unverified = wait.wait(unverified).unwrap();
}
if deleting.load(AtomicOrdering::Acquire) {
@@ -205,39 +210,42 @@ impl BlockQueue {
}
let block = {
let mut v = verification.lock().unwrap();
if v.unverified.is_empty() {
let mut unverified = verification.unverified.lock().unwrap();
if unverified.is_empty() {
continue;
}
let block = v.unverified.pop_front().unwrap();
v.verifying.push_back(VerifyingBlock{ hash: block.header.hash(), block: None });
let mut verifying = verification.verifying.lock().unwrap();
let block = unverified.pop_front().unwrap();
verifying.push_back(VerifyingBlock{ hash: block.header.hash(), block: None });
block
};
let block_hash = block.header.hash();
match verify_block_unordered(block.header, block.bytes, engine.deref().deref()) {
Ok(verified) => {
let mut v = verification.lock().unwrap();
for e in &mut v.verifying {
let mut verifying = verification.verifying.lock().unwrap();
for e in verifying.iter_mut() {
if e.hash == block_hash {
e.block = Some(verified);
break;
}
}
if !v.verifying.is_empty() && v.verifying.front().unwrap().hash == block_hash {
if !verifying.is_empty() && verifying.front().unwrap().hash == block_hash {
// we're next!
let mut vref = v.deref_mut();
BlockQueue::drain_verifying(&mut vref.verifying, &mut vref.verified, &mut vref.bad);
let mut verified = verification.verified.lock().unwrap();
let mut bad = verification.bad.lock().unwrap();
BlockQueue::drain_verifying(&mut verifying, &mut verified, &mut bad);
ready.set();
}
},
Err(err) => {
let mut v = verification.lock().unwrap();
let mut verifying = verification.verifying.lock().unwrap();
let mut verified = verification.verified.lock().unwrap();
let mut bad = verification.bad.lock().unwrap();
warn!(target: "client", "Stage 2 block verification failed for {}\nError: {:?}", block_hash, err);
v.bad.insert(block_hash.clone());
v.verifying.retain(|e| e.hash != block_hash);
let mut vref = v.deref_mut();
BlockQueue::drain_verifying(&mut vref.verifying, &mut vref.verified, &mut vref.bad);
bad.insert(block_hash.clone());
verifying.retain(|e| e.hash != block_hash);
BlockQueue::drain_verifying(&mut verifying, &mut verified, &mut bad);
ready.set();
}
}
@@ -257,19 +265,21 @@ impl BlockQueue {
}
/// Clear the queue and stop verification activity.
pub fn clear(&mut self) {
let mut verification = self.verification.lock().unwrap();
verification.unverified.clear();
verification.verifying.clear();
verification.verified.clear();
pub fn clear(&self) {
let mut unverified = self.verification.unverified.lock().unwrap();
let mut verifying = self.verification.verifying.lock().unwrap();
let mut verified = self.verification.verified.lock().unwrap();
unverified.clear();
verifying.clear();
verified.clear();
self.processing.write().unwrap().clear();
}
/// Wait for queue to be empty
pub fn flush(&mut self) {
let mut verification = self.verification.lock().unwrap();
while !verification.unverified.is_empty() || !verification.verifying.is_empty() {
verification = self.empty.wait(verification).unwrap();
/// Wait for unverified queue to be empty
pub fn flush(&self) {
let mut unverified = self.verification.unverified.lock().unwrap();
while !unverified.is_empty() || !self.verification.verifying.lock().unwrap().is_empty() {
unverified = self.empty.wait(unverified).unwrap();
}
}
@@ -278,27 +288,28 @@ impl BlockQueue {
if self.processing.read().unwrap().contains(&hash) {
return BlockStatus::Queued;
}
if self.verification.lock().unwrap().bad.contains(&hash) {
if self.verification.bad.lock().unwrap().contains(&hash) {
return BlockStatus::Bad;
}
BlockStatus::Unknown
}
/// Add a block to the queue.
pub fn import_block(&mut self, bytes: Bytes) -> ImportResult {
pub fn import_block(&self, bytes: Bytes) -> ImportResult {
let header = BlockView::new(&bytes).header();
let h = header.hash();
if self.processing.read().unwrap().contains(&h) {
return Err(x!(ImportError::AlreadyQueued));
}
{
let mut verification = self.verification.lock().unwrap();
if verification.bad.contains(&h) {
if self.processing.read().unwrap().contains(&h) {
return Err(x!(ImportError::AlreadyQueued));
}
let mut bad = self.verification.bad.lock().unwrap();
if bad.contains(&h) {
return Err(x!(ImportError::KnownBad));
}
if verification.bad.contains(&header.parent_hash) {
verification.bad.insert(h.clone());
if bad.contains(&header.parent_hash) {
bad.insert(h.clone());
return Err(x!(ImportError::KnownBad));
}
}
@@ -306,48 +317,47 @@ impl BlockQueue {
match verify_block_basic(&header, &bytes, self.engine.deref().deref()) {
Ok(()) => {
self.processing.write().unwrap().insert(h.clone());
self.verification.lock().unwrap().unverified.push_back(UnverifiedBlock { header: header, bytes: bytes });
self.verification.unverified.lock().unwrap().push_back(UnverifiedBlock { header: header, bytes: bytes });
self.more_to_verify.notify_all();
Ok(h)
},
Err(err) => {
warn!(target: "client", "Stage 1 block verification failed for {}\nError: {:?}", BlockView::new(&bytes).header_view().sha3(), err);
self.verification.lock().unwrap().bad.insert(h.clone());
self.verification.bad.lock().unwrap().insert(h.clone());
Err(err)
}
}
}
/// Mark given block and all its children as bad. Stops verification.
pub fn mark_as_bad(&mut self, block_hashes: &[H256]) {
pub fn mark_as_bad(&self, block_hashes: &[H256]) {
if block_hashes.is_empty() {
return;
}
let mut verification_lock = self.verification.lock().unwrap();
let mut verified_lock = self.verification.verified.lock().unwrap();
let mut verified = verified_lock.deref_mut();
let mut bad = self.verification.bad.lock().unwrap();
let mut processing = self.processing.write().unwrap();
let mut verification = verification_lock.deref_mut();
verification.bad.reserve(block_hashes.len());
bad.reserve(block_hashes.len());
for hash in block_hashes {
verification.bad.insert(hash.clone());
bad.insert(hash.clone());
processing.remove(&hash);
}
let mut new_verified = VecDeque::new();
for block in verification.verified.drain(..) {
if verification.bad.contains(&block.header.parent_hash) {
verification.bad.insert(block.header.hash());
for block in verified.drain(..) {
if bad.contains(&block.header.parent_hash) {
bad.insert(block.header.hash());
processing.remove(&block.header.hash());
} else {
new_verified.push_back(block);
}
}
verification.verified = new_verified;
*verified = new_verified;
}
/// Mark given block as processed
pub fn mark_as_good(&mut self, block_hashes: &[H256]) {
pub fn mark_as_good(&self, block_hashes: &[H256]) {
if block_hashes.is_empty() {
return;
}
@@ -358,16 +368,16 @@ impl BlockQueue {
}
/// Removes up to `max` verified blocks from the queue
pub fn drain(&mut self, max: usize) -> Vec<PreverifiedBlock> {
let mut verification = self.verification.lock().unwrap();
let count = min(max, verification.verified.len());
pub fn drain(&self, max: usize) -> Vec<PreverifiedBlock> {
let mut verified = self.verification.verified.lock().unwrap();
let count = min(max, verified.len());
let mut result = Vec::with_capacity(count);
for _ in 0..count {
let block = verification.verified.pop_front().unwrap();
let block = verified.pop_front().unwrap();
result.push(block);
}
self.ready_signal.reset();
if !verification.verified.is_empty() {
if !verified.is_empty() {
self.ready_signal.set();
}
result
@@ -375,17 +385,28 @@ impl BlockQueue {
/// Get queue status.
pub fn queue_info(&self) -> BlockQueueInfo {
let verification = self.verification.lock().unwrap();
let (unverified_len, unverified_bytes) = {
let v = self.verification.unverified.lock().unwrap();
(v.len(), v.heap_size_of_children())
};
let (verifying_len, verifying_bytes) = {
let v = self.verification.verifying.lock().unwrap();
(v.len(), v.heap_size_of_children())
};
let (verified_len, verified_bytes) = {
let v = self.verification.verified.lock().unwrap();
(v.len(), v.heap_size_of_children())
};
BlockQueueInfo {
verified_queue_size: verification.verified.len(),
unverified_queue_size: verification.unverified.len(),
verifying_queue_size: verification.verifying.len(),
unverified_queue_size: unverified_len,
verifying_queue_size: verifying_len,
verified_queue_size: verified_len,
max_queue_size: self.max_queue_size,
max_mem_use: self.max_mem_use,
mem_used:
verification.unverified.heap_size_of_children()
+ verification.verifying.heap_size_of_children()
+ verification.verified.heap_size_of_children(),
unverified_bytes
+ verifying_bytes
+ verified_bytes
// TODO: https://github.com/servo/heapsize/pull/50
//+ self.processing.read().unwrap().heap_size_of_children(),
}
@@ -393,10 +414,9 @@ impl BlockQueue {
pub fn collect_garbage(&self) {
{
let mut verification = self.verification.lock().unwrap();
verification.unverified.shrink_to_fit();
verification.verifying.shrink_to_fit();
verification.verified.shrink_to_fit();
self.verification.unverified.lock().unwrap().shrink_to_fit();
self.verification.verifying.lock().unwrap().shrink_to_fit();
self.verification.verified.lock().unwrap().shrink_to_fit();
}
self.processing.write().unwrap().shrink_to_fit();
}
@@ -444,7 +464,7 @@ mod tests {
#[test]
fn can_import_blocks() {
let mut queue = get_test_queue();
let queue = get_test_queue();
if let Err(e) = queue.import_block(get_good_dummy_block()) {
panic!("error importing block that is valid by definition({:?})", e);
}
@@ -452,7 +472,7 @@ mod tests {
#[test]
fn returns_error_for_duplicates() {
let mut queue = get_test_queue();
let queue = get_test_queue();
if let Err(e) = queue.import_block(get_good_dummy_block()) {
panic!("error importing block that is valid by definition({:?})", e);
}
@@ -471,7 +491,7 @@ mod tests {
#[test]
fn returns_ok_for_drained_duplicates() {
let mut queue = get_test_queue();
let queue = get_test_queue();
let block = get_good_dummy_block();
let hash = BlockView::new(&block).header().hash().clone();
if let Err(e) = queue.import_block(block) {
@@ -488,7 +508,7 @@ mod tests {
#[test]
fn returns_empty_once_finished() {
let mut queue = get_test_queue();
let queue = get_test_queue();
queue.import_block(get_good_dummy_block()).expect("error importing block that is valid by definition");
queue.flush();
queue.drain(1);

6
ethcore/src/blockchain/block_info.rs
View File

@@ -18,6 +18,7 @@ use util::numbers::{U256,H256};
use header::BlockNumber;
/// Brief info about inserted block.
#[derive(Clone)]
pub struct BlockInfo {
/// Block hash.
pub hash: H256,
@@ -30,6 +31,7 @@ pub struct BlockInfo {
}
/// Describes location of newly inserted block.
#[derive(Clone)]
pub enum BlockLocation {
/// It's part of the canon chain.
CanonChain,
@@ -42,6 +44,8 @@ pub enum BlockLocation {
/// Hash of the newest common ancestor with old canon chain.
ancestor: H256,
/// Hashes of the blocks between ancestor and this block.
route: Vec<H256>
enacted: Vec<H256>,
/// Hashes of the blocks which were invalidated.
retracted: Vec<H256>,
}
}

163
ethcore/src/blockchain/blockchain.rs
View File

@@ -16,6 +16,7 @@
//! Blockchain database.
use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrder};
use util::*;
use header::*;
use extras::*;
@@ -28,7 +29,7 @@ use blockchain::best_block::BestBlock;
use blockchain::bloom_indexer::BloomIndexer;
use blockchain::tree_route::TreeRoute;
use blockchain::update::ExtrasUpdate;
use blockchain::CacheSize;
use blockchain::{CacheSize, ImportRoute};
const BLOOM_INDEX_SIZE: usize = 16;
const BLOOM_LEVELS: u8 = 3;
@@ -134,8 +135,9 @@ struct CacheManager {
///
/// **Does not do input data verification.**
pub struct BlockChain {
pref_cache_size: usize,
max_cache_size: usize,
// All locks must be captured in the order declared here.
pref_cache_size: AtomicUsize,
max_cache_size: AtomicUsize,
best_block: RwLock<BestBlock>,
@@ -157,6 +159,8 @@ pub struct BlockChain {
// blooms indexing
bloom_indexer: BloomIndexer,
insert_lock: Mutex<()>
}
impl FilterDataSource for BlockChain {
@@ -262,8 +266,8 @@ impl BlockChain {
(0..COLLECTION_QUEUE_SIZE).foreach(|_| cache_man.cache_usage.push_back(HashSet::new()));
let bc = BlockChain {
pref_cache_size: config.pref_cache_size,
max_cache_size: config.max_cache_size,
pref_cache_size: AtomicUsize::new(config.pref_cache_size),
max_cache_size: AtomicUsize::new(config.max_cache_size),
best_block: RwLock::new(BestBlock::default()),
blocks: RwLock::new(HashMap::new()),
block_details: RwLock::new(HashMap::new()),
@@ -275,7 +279,8 @@ impl BlockChain {
extras_db: extras_db,
blocks_db: blocks_db,
cache_man: RwLock::new(cache_man),
bloom_indexer: BloomIndexer::new(BLOOM_INDEX_SIZE, BLOOM_LEVELS)
bloom_indexer: BloomIndexer::new(BLOOM_INDEX_SIZE, BLOOM_LEVELS),
insert_lock: Mutex::new(()),
};
// load best block
@@ -318,9 +323,9 @@ impl BlockChain {
}
/// Set the cache configuration.
pub fn configure_cache(&mut self, pref_cache_size: usize, max_cache_size: usize) {
self.pref_cache_size = pref_cache_size;
self.max_cache_size = max_cache_size;
pub fn configure_cache(&self, pref_cache_size: usize, max_cache_size: usize) {
self.pref_cache_size.store(pref_cache_size, AtomicOrder::Relaxed);
self.max_cache_size.store(max_cache_size, AtomicOrder::Relaxed);
}
/// Returns a tree route between `from` and `to`, which is a tuple of:
@@ -414,16 +419,17 @@ impl BlockChain {
/// Inserts the block into backing cache database.
/// Expects the block to be valid and already verified.
/// If the block is already known, does nothing.
pub fn insert_block(&self, bytes: &[u8], receipts: Vec<Receipt>) {
pub fn insert_block(&self, bytes: &[u8], receipts: Vec<Receipt>) -> ImportRoute {
// create views onto rlp
let block = BlockView::new(bytes);
let header = block.header_view();
let hash = header.sha3();
if self.is_known(&hash) {
return;
return ImportRoute::none();
}
let _lock = self.insert_lock.lock();
// store block in db
self.blocks_db.put(&hash, &bytes).unwrap();
@@ -435,8 +441,10 @@ impl BlockChain {
block_receipts: self.prepare_block_receipts_update(receipts, &info),
transactions_addresses: self.prepare_transaction_addresses_update(bytes, &info),
blocks_blooms: self.prepare_block_blooms_update(bytes, &info),
info: info
info: info.clone(),
});
ImportRoute::from(info)
}
/// Applies extras update.
@@ -444,48 +452,58 @@ impl BlockChain {
let batch = DBTransaction::new();
batch.put(b"best", &update.info.hash).unwrap();
// update best block
let mut best_block = self.best_block.write().unwrap();
match update.info.location {
BlockLocation::Branch => (),
_ => {
*best_block = BestBlock {
hash: update.info.hash,
number: update.info.number,
total_difficulty: update.info.total_difficulty
};
{
let mut write_details = self.block_details.write().unwrap();
for (hash, details) in update.block_details.into_iter() {
batch.put_extras(&hash, &details);
self.note_used(CacheID::Extras(ExtrasIndex::BlockDetails, hash.clone()));
write_details.insert(hash, details);
}
}
let mut write_hashes = self.block_hashes.write().unwrap();
for (number, hash) in &update.block_hashes {
batch.put_extras(number, hash);
write_hashes.remove(number);
{
let mut write_receipts = self.block_receipts.write().unwrap();
for (hash, receipt) in &update.block_receipts {
batch.put_extras(hash, receipt);
write_receipts.remove(hash);
}
}
let mut write_details = self.block_details.write().unwrap();
for (hash, details) in update.block_details.into_iter() {
batch.put_extras(&hash, &details);
write_details.insert(hash.clone(), details);
self.note_used(CacheID::Extras(ExtrasIndex::BlockDetails, hash));
{
let mut write_blocks_blooms = self.blocks_blooms.write().unwrap();
for (bloom_hash, blocks_bloom) in &update.blocks_blooms {
batch.put_extras(bloom_hash, blocks_bloom);
write_blocks_blooms.remove(bloom_hash);
}
}
let mut write_receipts = self.block_receipts.write().unwrap();
for (hash, receipt) in &update.block_receipts {
batch.put_extras(hash, receipt);
write_receipts.remove(hash);
}
// These cached values must be updated last and togeterh
{
let mut best_block = self.best_block.write().unwrap();
let mut write_hashes = self.block_hashes.write().unwrap();
let mut write_txs = self.transaction_addresses.write().unwrap();
let mut write_txs = self.transaction_addresses.write().unwrap();
for (hash, tx_address) in &update.transactions_addresses {
batch.put_extras(hash, tx_address);
write_txs.remove(hash);
}
// update best block
match update.info.location {
BlockLocation::Branch => (),
_ => {
*best_block = BestBlock {
hash: update.info.hash,
number: update.info.number,
total_difficulty: update.info.total_difficulty
};
}
}
let mut write_blocks_blooms = self.blocks_blooms.write().unwrap();
for (bloom_hash, blocks_bloom) in &update.blocks_blooms {
batch.put_extras(bloom_hash, blocks_bloom);
write_blocks_blooms.remove(bloom_hash);
for (number, hash) in &update.block_hashes {
batch.put_extras(number, hash);
write_hashes.remove(number);
}
for (hash, tx_address) in &update.transactions_addresses {
batch.put_extras(hash, tx_address);
write_txs.remove(hash);
}
}
// update extras database
@@ -549,9 +567,14 @@ impl BlockChain {
match route.blocks.len() {
0 => BlockLocation::CanonChain,
_ => BlockLocation::BranchBecomingCanonChain {
ancestor: route.ancestor,
route: route.blocks.into_iter().skip(route.index).collect()
_ => {
let retracted = route.blocks.iter().take(route.index).cloned().collect::<Vec<H256>>();
BlockLocation::BranchBecomingCanonChain {
ancestor: route.ancestor,
enacted: route.blocks.into_iter().skip(route.index).collect(),
retracted: retracted.into_iter().rev().collect(),
}
}
}
} else {
@@ -572,11 +595,11 @@ impl BlockChain {
BlockLocation::CanonChain => {
block_hashes.insert(number, info.hash.clone());
},
BlockLocation::BranchBecomingCanonChain { ref ancestor, ref route } => {
BlockLocation::BranchBecomingCanonChain { ref ancestor, ref enacted, .. } => {
let ancestor_number = self.block_number(ancestor).unwrap();
let start_number = ancestor_number + 1;
for (index, hash) in route.iter().cloned().enumerate() {
for (index, hash) in enacted.iter().cloned().enumerate() {
block_hashes.insert(start_number + index as BlockNumber, hash);
}
@@ -661,11 +684,11 @@ impl BlockChain {
ChainFilter::new(self, self.bloom_indexer.index_size(), self.bloom_indexer.levels())
.add_bloom(&header.log_bloom(), header.number() as usize)
},
BlockLocation::BranchBecomingCanonChain { ref ancestor, ref route } => {
BlockLocation::BranchBecomingCanonChain { ref ancestor, ref enacted, .. } => {
let ancestor_number = self.block_number(ancestor).unwrap();
let start_number = ancestor_number + 1;
let mut blooms: Vec<H2048> = route.iter()
let mut blooms: Vec<H2048> = enacted.iter()
.map(|hash| self.block(hash).unwrap())
.map(|bytes| BlockView::new(&bytes).header_view().log_bloom())
.collect();
@@ -774,11 +797,10 @@ impl BlockChain {
/// Ticks our cache system and throws out any old data.
pub fn collect_garbage(&self) {
if self.cache_size().total() < self.pref_cache_size { return; }
if self.cache_size().total() < self.pref_cache_size.load(AtomicOrder::Relaxed) { return; }
for _ in 0..COLLECTION_QUEUE_SIZE {
{
let mut cache_man = self.cache_man.write().unwrap();
let mut blocks = self.blocks.write().unwrap();
let mut block_details = self.block_details.write().unwrap();
let mut block_hashes = self.block_hashes.write().unwrap();
@@ -786,6 +808,7 @@ impl BlockChain {
let mut block_logs = self.block_logs.write().unwrap();
let mut blocks_blooms = self.blocks_blooms.write().unwrap();
let mut block_receipts = self.block_receipts.write().unwrap();
let mut cache_man = self.cache_man.write().unwrap();
for id in cache_man.cache_usage.pop_back().unwrap().into_iter() {
cache_man.in_use.remove(&id);
@@ -812,7 +835,7 @@ impl BlockChain {
blocks_blooms.shrink_to_fit();
block_receipts.shrink_to_fit();
}
if self.cache_size().total() < self.max_cache_size { break; }
if self.cache_size().total() < self.max_cache_size.load(AtomicOrder::Relaxed) { break; }
}
// TODO: m_lastCollection = chrono::system_clock::now();
@@ -825,7 +848,7 @@ mod tests {
use rustc_serialize::hex::FromHex;
use util::hash::*;
use util::sha3::Hashable;
use blockchain::{BlockProvider, BlockChain, BlockChainConfig};
use blockchain::{BlockProvider, BlockChain, BlockChainConfig, ImportRoute};
use tests::helpers::*;
use devtools::*;
use blockchain::generator::{ChainGenerator, ChainIterator, BlockFinalizer};
@@ -943,10 +966,30 @@ mod tests {
let temp = RandomTempPath::new();
let bc = BlockChain::new(BlockChainConfig::default(), &genesis, temp.as_path());
bc.insert_block(&b1, vec![]);
bc.insert_block(&b2, vec![]);
bc.insert_block(&b3a, vec![]);
bc.insert_block(&b3b, vec![]);
let ir1 = bc.insert_block(&b1, vec![]);
let ir2 = bc.insert_block(&b2, vec![]);
let ir3b = bc.insert_block(&b3b, vec![]);
let ir3a = bc.insert_block(&b3a, vec![]);
assert_eq!(ir1, ImportRoute {
enacted: vec![b1_hash],
retracted: vec![],
});
assert_eq!(ir2, ImportRoute {
enacted: vec![b2_hash],
retracted: vec![],
});
assert_eq!(ir3b, ImportRoute {
enacted: vec![b3b_hash],
retracted: vec![],
});
assert_eq!(ir3a, ImportRoute {
enacted: vec![b3a_hash],
retracted: vec![b3b_hash],
});
assert_eq!(bc.best_block_hash(), best_block_hash);
assert_eq!(bc.block_number(&genesis_hash).unwrap(), 0);

119
ethcore/src/blockchain/import_route.rs Normal file
View File

@@ -0,0 +1,119 @@
// 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/>.
//! Import route.
use util::hash::H256;
use blockchain::block_info::{BlockInfo, BlockLocation};
/// Import route for newly inserted block.
#[derive(Debug, PartialEq)]
pub struct ImportRoute {
/// Blocks that were invalidated by new block.
pub retracted: Vec<H256>,
/// Blocks that were validated by new block.
pub enacted: Vec<H256>,
}
impl ImportRoute {
pub fn none() -> Self {
ImportRoute {
retracted: vec![],
enacted: vec![],
}
}
}
impl From<BlockInfo> for ImportRoute {
fn from(info: BlockInfo) -> ImportRoute {
match info.location {
BlockLocation::CanonChain => ImportRoute {
retracted: vec![],
enacted: vec![info.hash],
},
BlockLocation::Branch => ImportRoute::none(),
BlockLocation::BranchBecomingCanonChain { mut enacted, retracted, .. } => {
enacted.push(info.hash);
ImportRoute {
retracted: retracted,
enacted: enacted,
}
}
}
}
}
#[cfg(test)]
mod tests {
use util::hash::H256;
use util::numbers::U256;
use blockchain::block_info::{BlockInfo, BlockLocation};
use blockchain::ImportRoute;
#[test]
fn import_route_none() {
assert_eq!(ImportRoute::none(), ImportRoute {
enacted: vec![],
retracted: vec![],
});
}
#[test]
fn import_route_branch() {
let info = BlockInfo {
hash: H256::from(U256::from(1)),
number: 0,
total_difficulty: U256::from(0),
location: BlockLocation::Branch,
};
assert_eq!(ImportRoute::from(info), ImportRoute::none());
}
#[test]
fn import_route_canon_chain() {
let info = BlockInfo {
hash: H256::from(U256::from(1)),
number: 0,
total_difficulty: U256::from(0),
location: BlockLocation::CanonChain,
};
assert_eq!(ImportRoute::from(info), ImportRoute {
retracted: vec![],
enacted: vec![H256::from(U256::from(1))],
});
}
#[test]
fn import_route_branch_becoming_canon_chain() {
let info = BlockInfo {
hash: H256::from(U256::from(2)),
number: 0,
total_difficulty: U256::from(0),
location: BlockLocation::BranchBecomingCanonChain {
ancestor: H256::from(U256::from(0)),
enacted: vec![H256::from(U256::from(1))],
retracted: vec![H256::from(U256::from(3)), H256::from(U256::from(4))],
}
};
assert_eq!(ImportRoute::from(info), ImportRoute {
retracted: vec![H256::from(U256::from(3)), H256::from(U256::from(4))],
enacted: vec![H256::from(U256::from(1)), H256::from(U256::from(2))],
});
}
}

2
ethcore/src/blockchain/mod.rs
View File

@@ -23,9 +23,11 @@ mod bloom_indexer;
mod cache;
mod tree_route;
mod update;
mod import_route;
#[cfg(test)]
mod generator;
pub use self::blockchain::{BlockProvider, BlockChain, BlockChainConfig};
pub use self::cache::CacheSize;
pub use self::tree_route::TreeRoute;
pub use self::import_route::ImportRoute;

322
ethcore/src/client.rs → ethcore/src/client/client.rs
View File

@@ -20,7 +20,6 @@ use std::marker::PhantomData;
use std::sync::atomic::AtomicBool;
use util::*;
use util::panics::*;
use blockchain::{BlockChain, BlockProvider};
use views::BlockView;
use error::*;
use header::{BlockNumber};
@@ -28,7 +27,6 @@ use state::State;
use spec::Spec;
use engine::Engine;
use views::HeaderView;
use block_queue::BlockQueue;
use service::{NetSyncMessage, SyncMessage};
use env_info::LastHashes;
use verification::*;
@@ -37,33 +35,10 @@ use transaction::LocalizedTransaction;
use extras::TransactionAddress;
use filter::Filter;
use log_entry::LocalizedLogEntry;
use util::keys::store::SecretStore;
pub use block_queue::{BlockQueueConfig, BlockQueueInfo};
pub use blockchain::{TreeRoute, BlockChainConfig, CacheSize as BlockChainCacheSize};
/// Uniquely identifies block.
#[derive(Debug, PartialEq, Clone)]
pub enum BlockId {
/// Block's sha3.
/// Querying by hash is always faster.
Hash(H256),
/// Block number within canon blockchain.
Number(BlockNumber),
/// Earliest block (genesis).
Earliest,
/// Latest mined block.
Latest
}
/// Uniquely identifies transaction.
#[derive(Debug, PartialEq, Clone)]
pub enum TransactionId {
/// Transaction's sha3.
Hash(H256),
/// Block id and transaction index within this block.
/// Querying by block position is always faster.
Location(BlockId, usize)
}
use block_queue::{BlockQueue, BlockQueueInfo};
use blockchain::{BlockChain, BlockProvider, TreeRoute};
use client::{BlockId, TransactionId, ClientConfig, BlockChainClient};
pub use blockchain::CacheSize as BlockChainCacheSize;
/// General block status
#[derive(Debug, Eq, PartialEq)]
@@ -78,30 +53,6 @@ pub enum BlockStatus {
Unknown,
}
/// Client configuration. Includes configs for all sub-systems.
#[derive(Debug)]
pub struct ClientConfig {
/// Block queue configuration.
pub queue: BlockQueueConfig,
/// Blockchain configuration.
pub blockchain: BlockChainConfig,
/// Prefer journal rather than archive.
pub prefer_journal: bool,
/// The name of the client instance.
pub name: String,
}
impl Default for ClientConfig {
fn default() -> ClientConfig {
ClientConfig {
queue: Default::default(),
blockchain: Default::default(),
prefer_journal: false,
name: Default::default(),
}
}
}
/// Information about the blockchain gathered together.
#[derive(Debug)]
pub struct BlockChainInfo {
@@ -123,72 +74,8 @@ impl fmt::Display for BlockChainInfo {
}
}
/// Blockchain database client. Owns and manages a blockchain and a block queue.
pub trait BlockChainClient : Sync + Send {
/// Get raw block header data by block id.
fn block_header(&self, id: BlockId) -> Option<Bytes>;
/// Get raw block body data by block id.
/// Block body is an RLP list of two items: uncles and transactions.
fn block_body(&self, id: BlockId) -> Option<Bytes>;
/// Get raw block data by block header hash.
fn block(&self, id: BlockId) -> Option<Bytes>;
/// Get block status by block header hash.
fn block_status(&self, id: BlockId) -> BlockStatus;
/// Get block total difficulty.
fn block_total_difficulty(&self, id: BlockId) -> Option<U256>;
/// Get address nonce.
fn nonce(&self, address: &Address) -> U256;
/// Get block hash.
fn block_hash(&self, id: BlockId) -> Option<H256>;
/// Get address code.
fn code(&self, address: &Address) -> Option<Bytes>;
/// Get transaction with given hash.
fn transaction(&self, id: TransactionId) -> Option<LocalizedTransaction>;
/// Get a tree route between `from` and `to`.
/// See `BlockChain::tree_route`.
fn tree_route(&self, from: &H256, to: &H256) -> Option<TreeRoute>;
/// Get latest state node
fn state_data(&self, hash: &H256) -> Option<Bytes>;
/// Get raw block receipts data by block header hash.
fn block_receipts(&self, hash: &H256) -> Option<Bytes>;
/// Import a block into the blockchain.
fn import_block(&self, bytes: Bytes) -> ImportResult;
/// Get block queue information.
fn queue_info(&self) -> BlockQueueInfo;
/// Clear block queue and abort all import activity.
fn clear_queue(&self);
/// Get blockchain information.
fn chain_info(&self) -> BlockChainInfo;
/// Get the best block header.
fn best_block_header(&self) -> Bytes {
self.block_header(BlockId::Hash(self.chain_info().best_block_hash)).unwrap()
}
/// Returns numbers of blocks containing given bloom.
fn blocks_with_bloom(&self, bloom: &H2048, from_block: BlockId, to_block: BlockId) -> Option<Vec<BlockNumber>>;
/// Returns logs matching given filter.
fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry>;
}
#[derive(Default, Clone, Debug, Eq, PartialEq)]
/// Report on the status of a client.
#[derive(Default, Clone, Debug, Eq, PartialEq)]
pub struct ClientReport {
/// How many blocks have been imported so far.
pub blocks_imported: usize,
@@ -212,10 +99,10 @@ impl ClientReport {
/// Blockchain database client backed by a persistent database. Owns and manages a blockchain and a block queue.
/// Call `import_block()` to import a block asynchronously; `flush_queue()` flushes the queue.
pub struct Client<V = CanonVerifier> where V: Verifier {
chain: Arc<RwLock<BlockChain>>,
chain: Arc<BlockChain>,
engine: Arc<Box<Engine>>,
state_db: Mutex<Box<JournalDB + Send>>,
block_queue: RwLock<BlockQueue>,
block_queue: BlockQueue,
report: RwLock<ClientReport>,
import_lock: Mutex<()>,
panic_handler: Arc<PanicHandler>,
@@ -226,7 +113,6 @@ pub struct Client<V = CanonVerifier> where V: Verifier {
author: RwLock<Address>,
extra_data: RwLock<Bytes>,
verifier: PhantomData<V>,
secret_store: Arc<RwLock<SecretStore>>,
}
const HISTORY: u64 = 1000;
@@ -248,7 +134,7 @@ impl<V> Client<V> where V: Verifier {
dir.push(format!("v{}-sec-{}", CLIENT_DB_VER_STR, if config.prefer_journal { "pruned" } else { "archive" }));
let path = dir.as_path();
let gb = spec.genesis_block();
let chain = Arc::new(RwLock::new(BlockChain::new(config.blockchain, &gb, path)));
let chain = Arc::new(BlockChain::new(config.blockchain, &gb, path));
let mut state_path = path.to_path_buf();
state_path.push("state");
@@ -262,14 +148,11 @@ impl<V> Client<V> where V: Verifier {
let panic_handler = PanicHandler::new_in_arc();
panic_handler.forward_from(&block_queue);
let secret_store = Arc::new(RwLock::new(SecretStore::new()));
secret_store.write().unwrap().try_import_existing();
Ok(Arc::new(Client {
chain: chain,
engine: engine,
state_db: Mutex::new(state_db),
block_queue: RwLock::new(block_queue),
block_queue: block_queue,
report: RwLock::new(Default::default()),
import_lock: Mutex::new(()),
panic_handler: panic_handler,
@@ -278,22 +161,20 @@ impl<V> Client<V> where V: Verifier {
author: RwLock::new(Address::new()),
extra_data: RwLock::new(Vec::new()),
verifier: PhantomData,
secret_store: secret_store,
}))
}
/// Flush the block import queue.
pub fn flush_queue(&self) {
self.block_queue.write().unwrap().flush();
self.block_queue.flush();
}
fn build_last_hashes(&self, parent_hash: H256) -> LastHashes {
let mut last_hashes = LastHashes::new();
last_hashes.resize(256, H256::new());
last_hashes[0] = parent_hash;
let chain = self.chain.read().unwrap();
for i in 0..255 {
match chain.block_details(&last_hashes[i]) {
match self.chain.block_details(&last_hashes[i]) {
Some(details) => {
last_hashes[i + 1] = details.parent.clone();
},
@@ -303,31 +184,26 @@ impl<V> Client<V> where V: Verifier {
last_hashes
}
/// Secret store (key manager)
pub fn secret_store(&self) -> &Arc<RwLock<SecretStore>> {
&self.secret_store
}
fn check_and_close_block(&self, block: &PreverifiedBlock) -> Result<ClosedBlock, ()> {
let engine = self.engine.deref().deref();
let header = &block.header;
// Check the block isn't so old we won't be able to enact it.
let best_block_number = self.chain.read().unwrap().best_block_number();
let best_block_number = self.chain.best_block_number();
if best_block_number >= HISTORY && header.number() <= best_block_number - HISTORY {
warn!(target: "client", "Block import failed for #{} ({})\nBlock is ancient (current best block: #{}).", header.number(), header.hash(), best_block_number);
return Err(());
}
// Verify Block Family
let verify_family_result = V::verify_block_family(&header, &block.bytes, engine, self.chain.read().unwrap().deref());
let verify_family_result = V::verify_block_family(&header, &block.bytes, engine, self.chain.deref());
if let Err(e) = verify_family_result {
warn!(target: "client", "Stage 3 block verification failed for #{} ({})\nError: {:?}", header.number(), header.hash(), e);
return Err(());
};
// Check if Parent is in chain
let chain_has_parent = self.chain.read().unwrap().block_header(&header.parent_hash);
let chain_has_parent = self.chain.block_header(&header.parent_hash);
if let None = chain_has_parent {
warn!(target: "client", "Block import failed for #{} ({}): Parent not found ({}) ", header.number(), header.hash(), header.parent_hash);
return Err(());
@@ -362,7 +238,7 @@ impl<V> Client<V> where V: Verifier {
let mut bad_blocks = HashSet::new();
let _import_lock = self.import_lock.lock();
let blocks = self.block_queue.write().unwrap().drain(max_blocks_to_import);
let blocks = self.block_queue.drain(max_blocks_to_import);
let original_best = self.chain_info().best_block_hash;
@@ -383,8 +259,7 @@ impl<V> Client<V> where V: Verifier {
// Are we committing an era?
let ancient = if header.number() >= HISTORY {
let n = header.number() - HISTORY;
let chain = self.chain.read().unwrap();
Some((n, chain.block_hash(n).unwrap()))
Some((n, self.chain.block_hash(n).unwrap()))
} else {
None
};
@@ -396,9 +271,9 @@ impl<V> Client<V> where V: Verifier {
.commit(header.number(), &header.hash(), ancient)
.expect("State DB commit failed.");
// And update the chain
self.chain.write().unwrap()
.insert_block(&block.bytes, receipts);
// And update the chain after commit to prevent race conditions
// (when something is in chain but you are not able to fetch details)
self.chain.insert_block(&block.bytes, receipts);
self.report.write().unwrap().accrue_block(&block);
trace!(target: "client", "Imported #{} ({})", header.number(), header.hash());
@@ -408,18 +283,16 @@ impl<V> Client<V> where V: Verifier {
let bad_blocks = bad_blocks.into_iter().collect::<Vec<H256>>();
{
let mut block_queue = self.block_queue.write().unwrap();
if !bad_blocks.is_empty() {
block_queue.mark_as_bad(&bad_blocks);
self.block_queue.mark_as_bad(&bad_blocks);
}
if !good_blocks.is_empty() {
block_queue.mark_as_good(&good_blocks);
self.block_queue.mark_as_good(&good_blocks);
}
}
{
let block_queue = self.block_queue.read().unwrap();
if !good_blocks.is_empty() && block_queue.queue_info().is_empty() {
if !good_blocks.is_empty() && self.block_queue.queue_info().is_empty() {
io.send(NetworkIoMessage::User(SyncMessage::NewChainBlocks {
good: good_blocks,
bad: bad_blocks,
@@ -443,7 +316,7 @@ impl<V> Client<V> where V: Verifier {
/// Get info on the cache.
pub fn blockchain_cache_info(&self) -> BlockChainCacheSize {
self.chain.read().unwrap().cache_size()
self.chain.cache_size()
}
/// Get the report.
@@ -455,13 +328,13 @@ impl<V> Client<V> where V: Verifier {
/// Tick the client.
pub fn tick(&self) {
self.chain.read().unwrap().collect_garbage();
self.block_queue.read().unwrap().collect_garbage();
self.chain.collect_garbage();
self.block_queue.collect_garbage();
}
/// Set up the cache behaviour.
pub fn configure_cache(&self, pref_cache_size: usize, max_cache_size: usize) {
self.chain.write().unwrap().configure_cache(pref_cache_size, max_cache_size);
self.chain.configure_cache(pref_cache_size, max_cache_size);
}
fn block_hash(chain: &BlockChain, id: BlockId) -> Option<H256> {
@@ -476,9 +349,9 @@ impl<V> Client<V> where V: Verifier {
fn block_number(&self, id: BlockId) -> Option<BlockNumber> {
match id {
BlockId::Number(number) => Some(number),
BlockId::Hash(ref hash) => self.chain.read().unwrap().block_number(hash),
BlockId::Hash(ref hash) => self.chain.block_number(hash),
BlockId::Earliest => Some(0),
BlockId::Latest => Some(self.chain.read().unwrap().best_block_number())
BlockId::Latest => Some(self.chain.best_block_number())
}
}
@@ -504,17 +377,17 @@ impl<V> Client<V> where V: Verifier {
/// New chain head event. Restart mining operation.
pub fn prepare_sealing(&self) {
let h = self.chain.read().unwrap().best_block_hash();
let h = self.chain.best_block_hash();
let mut b = OpenBlock::new(
self.engine.deref().deref(),
self.state_db.lock().unwrap().spawn(),
match self.chain.read().unwrap().block_header(&h) { Some(ref x) => x, None => {return;} },
match self.chain.block_header(&h) { Some(ref x) => x, None => {return;} },
self.build_last_hashes(h.clone()),
self.author(),
self.extra_data()
);
self.chain.read().unwrap().find_uncle_headers(&h, self.engine.deref().deref().maximum_uncle_age()).unwrap().into_iter().take(self.engine.deref().deref().maximum_uncle_count()).foreach(|h| { b.push_uncle(h).unwrap(); });
self.chain.find_uncle_headers(&h, self.engine.deref().deref().maximum_uncle_age()).unwrap().into_iter().take(self.engine.deref().deref().maximum_uncle_count()).foreach(|h| { b.push_uncle(h).unwrap(); });
// TODO: push transactions.
@@ -522,53 +395,18 @@ impl<V> Client<V> where V: Verifier {
trace!("Sealing: number={}, hash={}, diff={}", b.hash(), b.block().header().difficulty(), b.block().header().number());
*self.sealing_block.lock().unwrap() = Some(b);
}
/// Grab the `ClosedBlock` that we want to be sealed. Comes as a mutex that you have to lock.
pub fn sealing_block(&self) -> &Mutex<Option<ClosedBlock>> {
if self.sealing_block.lock().unwrap().is_none() {
self.sealing_enabled.store(true, atomic::Ordering::Relaxed);
// TODO: Above should be on a timer that resets after two blocks have arrived without being asked for.
self.prepare_sealing();
}
&self.sealing_block
}
/// Submit `seal` as a valid solution for the header of `pow_hash`.
/// Will check the seal, but not actually insert the block into the chain.
pub fn submit_seal(&self, pow_hash: H256, seal: Vec<Bytes>) -> Result<(), Error> {
let mut maybe_b = self.sealing_block.lock().unwrap();
match *maybe_b {
Some(ref b) if b.hash() == pow_hash => {}
_ => { return Err(Error::PowHashInvalid); }
}
let b = maybe_b.take();
match b.unwrap().try_seal(self.engine.deref().deref(), seal) {
Err(old) => {
*maybe_b = Some(old);
Err(Error::PowInvalid)
}
Ok(sealed) => {
// TODO: commit DB from `sealed.drain` and make a VerifiedBlock to skip running the transactions twice.
try!(self.import_block(sealed.rlp_bytes()));
Ok(())
}
}
}
}
// TODO: need MinerService MinerIoHandler
impl<V> BlockChainClient for Client<V> where V: Verifier {
fn block_header(&self, id: BlockId) -> Option<Bytes> {
let chain = self.chain.read().unwrap();
Self::block_hash(&chain, id).and_then(|hash| chain.block(&hash).map(|bytes| BlockView::new(&bytes).rlp().at(0).as_raw().to_vec()))
Self::block_hash(&self.chain, id).and_then(|hash| self.chain.block(&hash).map(|bytes| BlockView::new(&bytes).rlp().at(0).as_raw().to_vec()))
}
fn block_body(&self, id: BlockId) -> Option<Bytes> {
let chain = self.chain.read().unwrap();
Self::block_hash(&chain, id).and_then(|hash| {
chain.block(&hash).map(|bytes| {
Self::block_hash(&self.chain, id).and_then(|hash| {
self.chain.block(&hash).map(|bytes| {
let rlp = Rlp::new(&bytes);
let mut body = RlpStream::new_list(2);
body.append_raw(rlp.at(1).as_raw(), 1);
@@ -579,24 +417,21 @@ impl<V> BlockChainClient for Client<V> where V: Verifier {
}
fn block(&self, id: BlockId) -> Option<Bytes> {
let chain = self.chain.read().unwrap();
Self::block_hash(&chain, id).and_then(|hash| {
chain.block(&hash)
Self::block_hash(&self.chain, id).and_then(|hash| {
self.chain.block(&hash)
})
}
fn block_status(&self, id: BlockId) -> BlockStatus {
let chain = self.chain.read().unwrap();
match Self::block_hash(&chain, id) {
Some(ref hash) if chain.is_known(hash) => BlockStatus::InChain,
Some(hash) => self.block_queue.read().unwrap().block_status(&hash),
match Self::block_hash(&self.chain, id) {
Some(ref hash) if self.chain.is_known(hash) => BlockStatus::InChain,
Some(hash) => self.block_queue.block_status(&hash),
None => BlockStatus::Unknown
}
}
fn block_total_difficulty(&self, id: BlockId) -> Option<U256> {
let chain = self.chain.read().unwrap();
Self::block_hash(&chain, id).and_then(|hash| chain.block_details(&hash)).map(|d| d.total_difficulty)
Self::block_hash(&self.chain, id).and_then(|hash| self.chain.block_details(&hash)).map(|d| d.total_difficulty)
}
fn nonce(&self, address: &Address) -> U256 {
@@ -604,8 +439,7 @@ impl<V> BlockChainClient for Client<V> where V: Verifier {
}
fn block_hash(&self, id: BlockId) -> Option<H256> {
let chain = self.chain.read().unwrap();
Self::block_hash(&chain, id)
Self::block_hash(&self.chain, id)
}
fn code(&self, address: &Address) -> Option<Bytes> {
@@ -613,20 +447,18 @@ impl<V> BlockChainClient for Client<V> where V: Verifier {
}
fn transaction(&self, id: TransactionId) -> Option<LocalizedTransaction> {
let chain = self.chain.read().unwrap();
match id {
TransactionId::Hash(ref hash) => chain.transaction_address(hash),
TransactionId::Location(id, index) => Self::block_hash(&chain, id).map(|hash| TransactionAddress {
TransactionId::Hash(ref hash) => self.chain.transaction_address(hash),
TransactionId::Location(id, index) => Self::block_hash(&self.chain, id).map(|hash| TransactionAddress {
block_hash: hash,
index: index
})
}.and_then(|address| chain.transaction(&address))
}.and_then(|address| self.chain.transaction(&address))
}
fn tree_route(&self, from: &H256, to: &H256) -> Option<TreeRoute> {
let chain = self.chain.read().unwrap();
match chain.is_known(from) && chain.is_known(to) {
true => Some(chain.tree_route(from.clone(), to.clone())),
match self.chain.is_known(from) && self.chain.is_known(to) {
true => Some(self.chain.tree_route(from.clone(), to.clone())),
false => None
}
}
@@ -642,43 +474,44 @@ impl<V> BlockChainClient for Client<V> where V: Verifier {
fn import_block(&self, bytes: Bytes) -> ImportResult {
{
let header = BlockView::new(&bytes).header_view();
if self.chain.read().unwrap().is_known(&header.sha3()) {
if self.chain.is_known(&header.sha3()) {
return Err(x!(ImportError::AlreadyInChain));
}
if self.block_status(BlockId::Hash(header.parent_hash())) == BlockStatus::Unknown {
return Err(x!(BlockError::UnknownParent(header.parent_hash())));
}
}
self.block_queue.write().unwrap().import_block(bytes)
self.block_queue.import_block(bytes)
}
fn queue_info(&self) -> BlockQueueInfo {
self.block_queue.read().unwrap().queue_info()
self.block_queue.queue_info()
}
fn clear_queue(&self) {
self.block_queue.write().unwrap().clear();
self.block_queue.clear();
}
fn chain_info(&self) -> BlockChainInfo {
let chain = self.chain.read().unwrap();
BlockChainInfo {
total_difficulty: chain.best_block_total_difficulty(),
pending_total_difficulty: chain.best_block_total_difficulty(),
genesis_hash: chain.genesis_hash(),
best_block_hash: chain.best_block_hash(),
best_block_number: From::from(chain.best_block_number())
total_difficulty: self.chain.best_block_total_difficulty(),
pending_total_difficulty: self.chain.best_block_total_difficulty(),
genesis_hash: self.chain.genesis_hash(),
best_block_hash: self.chain.best_block_hash(),
best_block_number: From::from(self.chain.best_block_number())
}
}
fn blocks_with_bloom(&self, bloom: &H2048, from_block: BlockId, to_block: BlockId) -> Option<Vec<BlockNumber>> {
match (self.block_number(from_block), self.block_number(to_block)) {
(Some(from), Some(to)) => Some(self.chain.read().unwrap().blocks_with_bloom(bloom, from, to)),
(Some(from), Some(to)) => Some(self.chain.blocks_with_bloom(bloom, from, to)),
_ => None
}
}
fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry> {
// TODO: lock blockchain only once
let mut blocks = filter.bloom_possibilities().iter()
.filter_map(|bloom| self.blocks_with_bloom(bloom, filter.from_block.clone(), filter.to_block.clone()))
.flat_map(|m| m)
@@ -690,9 +523,9 @@ impl<V> BlockChainClient for Client<V> where V: Verifier {
blocks.sort();
blocks.into_iter()
.filter_map(|number| self.chain.read().unwrap().block_hash(number).map(|hash| (number, hash)))
.filter_map(|(number, hash)| self.chain.read().unwrap().block_receipts(&hash).map(|r| (number, hash, r.receipts)))
.filter_map(|(number, hash, receipts)| self.chain.read().unwrap().block(&hash).map(|ref b| (number, hash, receipts, BlockView::new(b).transaction_hashes())))
.filter_map(|number| self.chain.block_hash(number).map(|hash| (number, hash)))
.filter_map(|(number, hash)| self.chain.block_receipts(&hash).map(|r| (number, hash, r.receipts)))
.filter_map(|(number, hash, receipts)| self.chain.block(&hash).map(|ref b| (number, hash, receipts, BlockView::new(b).transaction_hashes())))
.flat_map(|(number, hash, receipts, hashes)| {
let mut log_index = 0;
receipts.into_iter()
@@ -717,6 +550,39 @@ impl<V> BlockChainClient for Client<V> where V: Verifier {
})
.collect()
}
/// Grab the `ClosedBlock` that we want to be sealed. Comes as a mutex that you have to lock.
fn sealing_block(&self) -> &Mutex<Option<ClosedBlock>> {
if self.sealing_block.lock().unwrap().is_none() {
self.sealing_enabled.store(true, atomic::Ordering::Relaxed);
// TODO: Above should be on a timer that resets after two blocks have arrived without being asked for.
self.prepare_sealing();
}
&self.sealing_block
}
/// Submit `seal` as a valid solution for the header of `pow_hash`.
/// Will check the seal, but not actually insert the block into the chain.
fn submit_seal(&self, pow_hash: H256, seal: Vec<Bytes>) -> Result<(), Error> {
let mut maybe_b = self.sealing_block.lock().unwrap();
match *maybe_b {
Some(ref b) if b.hash() == pow_hash => {}
_ => { return Err(Error::PowHashInvalid); }
}
let b = maybe_b.take();
match b.unwrap().try_seal(self.engine.deref().deref(), seal) {
Err(old) => {
*maybe_b = Some(old);
Err(Error::PowInvalid)
}
Ok(sealed) => {
// TODO: commit DB from `sealed.drain` and make a VerifiedBlock to skip running the transactions twice.
try!(self.import_block(sealed.rlp_bytes()));
Ok(())
}
}
}
}
impl MayPanic for Client {

31
ethcore/src/client/config.rs Normal file
View File

@@ -0,0 +1,31 @@
// 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/>.
pub use block_queue::BlockQueueConfig;
pub use blockchain::BlockChainConfig;
/// Client configuration. Includes configs for all sub-systems.
#[derive(Debug, Default)]
pub struct ClientConfig {
/// Block queue configuration.
pub queue: BlockQueueConfig,
/// Blockchain configuration.
pub blockchain: BlockChainConfig,
/// Prefer journal rather than archive.
pub prefer_journal: bool,
/// The name of the client instance.
pub name: String,
}

44
ethcore/src/client/ids.rs Normal file
View File

@@ -0,0 +1,44 @@
// 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/>.
//! Unique identifiers.
use util::hash::H256;
use header::BlockNumber;
/// Uniquely identifies block.
#[derive(Debug, PartialEq, Clone)]
pub enum BlockId {
/// Block's sha3.
/// Querying by hash is always faster.
Hash(H256),
/// Block number within canon blockchain.
Number(BlockNumber),
/// Earliest block (genesis).
Earliest,
/// Latest mined block.
Latest
}
/// Uniquely identifies transaction.
#[derive(Debug, PartialEq, Clone)]
pub enum TransactionId {
/// Transaction's sha3.
Hash(H256),
/// Block id and transaction index within this block.
/// Querying by block position is always faster.
Location(BlockId, usize)
}

113
ethcore/src/client/mod.rs Normal file
View File

@@ -0,0 +1,113 @@
// 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/>.
//! Blockchain database client.
mod client;
mod config;
mod ids;
mod test_client;
pub use self::client::*;
pub use self::config::{ClientConfig, BlockQueueConfig, BlockChainConfig};
pub use self::ids::{BlockId, TransactionId};
pub use self::test_client::{TestBlockChainClient, EachBlockWith};
use std::sync::Mutex;
use util::bytes::Bytes;
use util::hash::{Address, H256, H2048};
use util::numbers::U256;
use blockchain::TreeRoute;
use block_queue::BlockQueueInfo;
use block::ClosedBlock;
use header::BlockNumber;
use transaction::LocalizedTransaction;
use log_entry::LocalizedLogEntry;
use filter::Filter;
use error::{ImportResult, Error};
/// Blockchain database client. Owns and manages a blockchain and a block queue.
pub trait BlockChainClient : Sync + Send {
/// Get raw block header data by block id.
fn block_header(&self, id: BlockId) -> Option<Bytes>;
/// Get raw block body data by block id.
/// Block body is an RLP list of two items: uncles and transactions.
fn block_body(&self, id: BlockId) -> Option<Bytes>;
/// Get raw block data by block header hash.
fn block(&self, id: BlockId) -> Option<Bytes>;
/// Get block status by block header hash.
fn block_status(&self, id: BlockId) -> BlockStatus;
/// Get block total difficulty.
fn block_total_difficulty(&self, id: BlockId) -> Option<U256>;
/// Get address nonce.
fn nonce(&self, address: &Address) -> U256;
/// Get block hash.
fn block_hash(&self, id: BlockId) -> Option<H256>;
/// Get address code.
fn code(&self, address: &Address) -> Option<Bytes>;
/// Get transaction with given hash.
fn transaction(&self, id: TransactionId) -> Option<LocalizedTransaction>;
/// Get a tree route between `from` and `to`.
/// See `BlockChain::tree_route`.
fn tree_route(&self, from: &H256, to: &H256) -> Option<TreeRoute>;
/// Get latest state node
fn state_data(&self, hash: &H256) -> Option<Bytes>;
/// Get raw block receipts data by block header hash.
fn block_receipts(&self, hash: &H256) -> Option<Bytes>;
/// Import a block into the blockchain.
fn import_block(&self, bytes: Bytes) -> ImportResult;
/// Get block queue information.
fn queue_info(&self) -> BlockQueueInfo;
/// Clear block queue and abort all import activity.
fn clear_queue(&self);
/// Get blockchain information.
fn chain_info(&self) -> BlockChainInfo;
/// Get the best block header.
fn best_block_header(&self) -> Bytes {
// TODO: lock blockchain only once
self.block_header(BlockId::Hash(self.chain_info().best_block_hash)).unwrap()
}
/// Returns numbers of blocks containing given bloom.
fn blocks_with_bloom(&self, bloom: &H2048, from_block: BlockId, to_block: BlockId) -> Option<Vec<BlockNumber>>;
/// Returns logs matching given filter.
fn logs(&self, filter: Filter) -> Vec<LocalizedLogEntry>;
/// Grab the `ClosedBlock` that we want to be sealed. Comes as a mutex that you have to lock.
fn sealing_block(&self) -> &Mutex<Option<ClosedBlock>>;
/// Submit `seal` as a valid solution for the header of `pow_hash`.
/// Will check the seal, but not actually insert the block into the chain.
fn submit_seal(&self, pow_hash: H256, seal: Vec<Bytes>) -> Result<(), Error>;
}

336
ethcore/src/client/test_client.rs Normal file
View File

@@ -0,0 +1,336 @@
// 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/>.
//! Test client.
use util::*;
use transaction::{Transaction, LocalizedTransaction, Action};
use blockchain::TreeRoute;
use client::{BlockChainClient, BlockChainInfo, BlockStatus, BlockId, TransactionId};
use header::{Header as BlockHeader, BlockNumber};
use filter::Filter;
use log_entry::LocalizedLogEntry;
use receipt::Receipt;
use error::{ImportResult, Error};
use block_queue::BlockQueueInfo;
use block::ClosedBlock;
/// Test client.
pub struct TestBlockChainClient {
/// Blocks.
pub blocks: RwLock<HashMap<H256, Bytes>>,
/// Mapping of numbers to hashes.
pub numbers: RwLock<HashMap<usize, H256>>,
/// Genesis block hash.
pub genesis_hash: H256,
/// Last block hash.
pub last_hash: RwLock<H256>,
/// Difficulty.
pub difficulty: RwLock<U256>,
}
#[derive(Clone)]
/// Used for generating test client blocks.
pub enum EachBlockWith {
/// Plain block.
Nothing,
/// Block with an uncle.
Uncle,
/// Block with a transaction.
Transaction,
/// Block with an uncle and transaction.
UncleAndTransaction
}
impl TestBlockChainClient {
/// Creates new test client.
pub fn new() -> TestBlockChainClient {
let mut client = TestBlockChainClient {
blocks: RwLock::new(HashMap::new()),
numbers: RwLock::new(HashMap::new()),
genesis_hash: H256::new(),
last_hash: RwLock::new(H256::new()),
difficulty: RwLock::new(From::from(0)),
};
client.add_blocks(1, EachBlockWith::Nothing); // add genesis block
client.genesis_hash = client.last_hash.read().unwrap().clone();
client
}
/// Add blocks to test client.
pub fn add_blocks(&mut self, count: usize, with: EachBlockWith) {
let len = self.numbers.read().unwrap().len();
for n in len..(len + count) {
let mut header = BlockHeader::new();
header.difficulty = From::from(n);
header.parent_hash = self.last_hash.read().unwrap().clone();
header.number = n as BlockNumber;
let uncles = match with {
EachBlockWith::Uncle | EachBlockWith::UncleAndTransaction => {
let mut uncles = RlpStream::new_list(1);
let mut uncle_header = BlockHeader::new();
uncle_header.difficulty = From::from(n);
uncle_header.parent_hash = self.last_hash.read().unwrap().clone();
uncle_header.number = n as BlockNumber;
uncles.append(&uncle_header);
header.uncles_hash = uncles.as_raw().sha3();
uncles
},
_ => RlpStream::new_list(0)
};
let txs = match with {
EachBlockWith::Transaction | EachBlockWith::UncleAndTransaction => {
let mut txs = RlpStream::new_list(1);
let keypair = KeyPair::create().unwrap();
let tx = Transaction {
action: Action::Create,
value: U256::from(100),
data: "3331600055".from_hex().unwrap(),
gas: U256::from(100_000),
gas_price: U256::one(),
nonce: U256::zero()
};
let signed_tx = tx.sign(&keypair.secret());
txs.append(&signed_tx);
txs.out()
},
_ => rlp::NULL_RLP.to_vec()
};
let mut rlp = RlpStream::new_list(3);
rlp.append(&header);
rlp.append_raw(&txs, 1);
rlp.append_raw(uncles.as_raw(), 1);
self.import_block(rlp.as_raw().to_vec()).unwrap();
}
}
/// TODO:
pub fn corrupt_block(&mut self, n: BlockNumber) {
let hash = self.block_hash(BlockId::Number(n)).unwrap();
let mut header: BlockHeader = decode(&self.block_header(BlockId::Number(n)).unwrap());
header.parent_hash = H256::new();
let mut rlp = RlpStream::new_list(3);
rlp.append(&header);
rlp.append_raw(&rlp::NULL_RLP, 1);
rlp.append_raw(&rlp::NULL_RLP, 1);
self.blocks.write().unwrap().insert(hash, rlp.out());
}
/// TODO:
pub fn block_hash_delta_minus(&mut self, delta: usize) -> H256 {
let blocks_read = self.numbers.read().unwrap();
let index = blocks_read.len() - delta;
blocks_read[&index].clone()
}
fn block_hash(&self, id: BlockId) -> Option<H256> {
match id {
BlockId::Hash(hash) => Some(hash),
BlockId::Number(n) => self.numbers.read().unwrap().get(&(n as usize)).cloned(),
BlockId::Earliest => self.numbers.read().unwrap().get(&0).cloned(),
BlockId::Latest => self.numbers.read().unwrap().get(&(self.numbers.read().unwrap().len() - 1)).cloned()
}
}
}
impl BlockChainClient for TestBlockChainClient {
fn block_total_difficulty(&self, _id: BlockId) -> Option<U256> {
Some(U256::zero())
}
fn block_hash(&self, _id: BlockId) -> Option<H256> {
unimplemented!();
}
fn nonce(&self, _address: &Address) -> U256 {
U256::zero()
}
fn code(&self, _address: &Address) -> Option<Bytes> {
unimplemented!();
}
fn transaction(&self, _id: TransactionId) -> Option<LocalizedTransaction> {
unimplemented!();
}
fn blocks_with_bloom(&self, _bloom: &H2048, _from_block: BlockId, _to_block: BlockId) -> Option<Vec<BlockNumber>> {
unimplemented!();
}
fn logs(&self, _filter: Filter) -> Vec<LocalizedLogEntry> {
unimplemented!();
}
fn sealing_block(&self) -> &Mutex<Option<ClosedBlock>> {
unimplemented!();
}
fn submit_seal(&self, _pow_hash: H256, _seal: Vec<Bytes>) -> Result<(), Error> {
unimplemented!();
}
fn block_header(&self, id: BlockId) -> Option<Bytes> {
self.block_hash(id).and_then(|hash| self.blocks.read().unwrap().get(&hash).map(|r| Rlp::new(r).at(0).as_raw().to_vec()))
}
fn block_body(&self, id: BlockId) -> Option<Bytes> {
self.block_hash(id).and_then(|hash| self.blocks.read().unwrap().get(&hash).map(|r| {
let mut stream = RlpStream::new_list(2);
stream.append_raw(Rlp::new(&r).at(1).as_raw(), 1);
stream.append_raw(Rlp::new(&r).at(2).as_raw(), 1);
stream.out()
}))
}
fn block(&self, id: BlockId) -> Option<Bytes> {
self.block_hash(id).and_then(|hash| self.blocks.read().unwrap().get(&hash).cloned())
}
fn block_status(&self, id: BlockId) -> BlockStatus {
match id {
BlockId::Number(number) if (number as usize) < self.blocks.read().unwrap().len() => BlockStatus::InChain,
BlockId::Hash(ref hash) if self.blocks.read().unwrap().get(hash).is_some() => BlockStatus::InChain,
_ => BlockStatus::Unknown
}
}
// works only if blocks are one after another 1 -> 2 -> 3
fn tree_route(&self, from: &H256, to: &H256) -> Option<TreeRoute> {
Some(TreeRoute {
ancestor: H256::new(),
index: 0,
blocks: {
let numbers_read = self.numbers.read().unwrap();
let mut adding = false;
let mut blocks = Vec::new();
for (_, hash) in numbers_read.iter().sort_by(|tuple1, tuple2| tuple1.0.cmp(tuple2.0)) {
if hash == to {
if adding {
blocks.push(hash.clone());
}
adding = false;
break;
}
if hash == from {
adding = true;
}
if adding {
blocks.push(hash.clone());
}
}
if adding { Vec::new() } else { blocks }
}
})
}
// TODO: returns just hashes instead of node state rlp(?)
fn state_data(&self, hash: &H256) -> Option<Bytes> {
// starts with 'f' ?
if *hash > H256::from("f000000000000000000000000000000000000000000000000000000000000000") {
let mut rlp = RlpStream::new();
rlp.append(&hash.clone());
return Some(rlp.out());
}
None
}
fn block_receipts(&self, hash: &H256) -> Option<Bytes> {
// starts with 'f' ?
if *hash > H256::from("f000000000000000000000000000000000000000000000000000000000000000") {
let receipt = Receipt::new(
H256::zero(),
U256::zero(),
vec![]);
let mut rlp = RlpStream::new();
rlp.append(&receipt);
return Some(rlp.out());
}
None
}
fn import_block(&self, b: Bytes) -> ImportResult {
let header = Rlp::new(&b).val_at::<BlockHeader>(0);
let h = header.hash();
let number: usize = header.number as usize;
if number > self.blocks.read().unwrap().len() {
panic!("Unexpected block number. Expected {}, got {}", self.blocks.read().unwrap().len(), number);
}
if number > 0 {
match self.blocks.read().unwrap().get(&header.parent_hash) {
Some(parent) => {
let parent = Rlp::new(parent).val_at::<BlockHeader>(0);
if parent.number != (header.number - 1) {
panic!("Unexpected block parent");
}
},
None => {
panic!("Unknown block parent {:?} for block {}", header.parent_hash, number);
}
}
}
let len = self.numbers.read().unwrap().len();
if number == len {
{
let mut difficulty = self.difficulty.write().unwrap();
*difficulty.deref_mut() = *difficulty.deref() + header.difficulty;
}
mem::replace(self.last_hash.write().unwrap().deref_mut(), h.clone());
self.blocks.write().unwrap().insert(h.clone(), b);
self.numbers.write().unwrap().insert(number, h.clone());
let mut parent_hash = header.parent_hash;
if number > 0 {
let mut n = number - 1;
while n > 0 && self.numbers.read().unwrap()[&n] != parent_hash {
*self.numbers.write().unwrap().get_mut(&n).unwrap() = parent_hash.clone();
n -= 1;
parent_hash = Rlp::new(&self.blocks.read().unwrap()[&parent_hash]).val_at::<BlockHeader>(0).parent_hash;
}
}
}
else {
self.blocks.write().unwrap().insert(h.clone(), b.to_vec());
}
Ok(h)
}
fn queue_info(&self) -> BlockQueueInfo {
BlockQueueInfo {
verified_queue_size: 0,
unverified_queue_size: 0,
verifying_queue_size: 0,
max_queue_size: 0,
max_mem_use: 0,
mem_used: 0,
}
}
fn clear_queue(&self) {
}
fn chain_info(&self) -> BlockChainInfo {
BlockChainInfo {
total_difficulty: *self.difficulty.read().unwrap(),
pending_total_difficulty: *self.difficulty.read().unwrap(),
genesis_hash: self.genesis_hash.clone(),
best_block_hash: self.last_hash.read().unwrap().clone(),
best_block_number: self.blocks.read().unwrap().len() as BlockNumber - 1,
}
}
}

2
ethcore/src/lib.rs
View File

@@ -86,6 +86,7 @@ extern crate crossbeam;
#[cfg(feature = "jit" )] extern crate evmjit;
pub mod block;
pub mod block_queue;
pub mod client;
pub mod error;
pub mod ethereum;
@@ -119,7 +120,6 @@ mod substate;
mod executive;
mod externalities;
mod verification;
mod block_queue;
mod blockchain;
#[cfg(test)]