GRIN/grin-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into unitdiff

Browse Source
This commit is contained in:
Ignotus Peverell
2018-10-25 14:20:41 -07:00
committed by GitHub
parent 1cfc474817 b889069722
commit a42d66efff
131 changed files with 1942 additions and 2372 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chain/src/chain.rs
+74 -93
View File
@@ -18,8 +18,9 @@
use std::collections::HashMap;
use std::fs::File;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, RwLock};
use std::sync::Arc;
use std::time::{Duration, Instant};
use util::RwLock;
use lmdb;
use lru_cache::LruCache;
@@ -37,7 +38,6 @@ use store;
use txhashset;
use types::{ChainAdapter, NoStatus, Options, Tip, TxHashSetRoots, TxHashsetWriteStatus};
use util::secp::pedersen::{Commitment, RangeProof};
use util::LOGGER;
/// Orphan pool size is limited by MAX_ORPHAN_SIZE
pub const MAX_ORPHAN_SIZE: usize = 200;
@@ -75,7 +75,7 @@ impl OrphanBlockPool {
}
fn len(&self) -> usize {
let orphans = self.orphans.read().unwrap();
let orphans = self.orphans.read();
orphans.len()
}
@@ -84,8 +84,8 @@ impl OrphanBlockPool {
}
fn add(&self, orphan: Orphan) {
let mut orphans = self.orphans.write().unwrap();
let mut height_idx = self.height_idx.write().unwrap();
let mut orphans = self.orphans.write();
let mut height_idx = self.height_idx.write();
{
let height_hashes = height_idx
.entry(orphan.block.header.height)
@@ -125,15 +125,15 @@ impl OrphanBlockPool {
/// Get an orphan from the pool indexed by the hash of its parent, removing
/// it at the same time, preventing clone
fn remove_by_height(&self, height: &u64) -> Option<Vec<Orphan>> {
let mut orphans = self.orphans.write().unwrap();
let mut height_idx = self.height_idx.write().unwrap();
let mut orphans = self.orphans.write();
let mut height_idx = self.height_idx.write();
height_idx
.remove(height)
.map(|hs| hs.iter().filter_map(|h| orphans.remove(h)).collect())
}
pub fn contains(&self, hash: &Hash) -> bool {
let orphans = self.orphans.read().unwrap();
let orphans = self.orphans.read();
orphans.contains_key(hash)
}
}
@@ -183,7 +183,6 @@ impl Chain {
let head = store.head()?;
debug!(
LOGGER,
"Chain init: {} @ {} [{}]",
head.total_difficulty.to_num(),
head.height,
@@ -221,7 +220,7 @@ impl Chain {
fn process_block_single(&self, b: Block, opts: Options) -> Result<Option<Tip>, Error> {
let maybe_new_head: Result<Option<Tip>, Error>;
{
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let batch = self.store.batch()?;
let mut ctx = self.new_ctx(opts, batch, &mut txhashset)?;
@@ -235,7 +234,7 @@ impl Chain {
let add_to_hash_cache = |hash: Hash| {
// only add to hash cache below if block is definitively accepted
// or rejected
let mut cache = self.block_hashes_cache.write().unwrap();
let mut cache = self.block_hashes_cache.write();
cache.insert(hash, true);
};
@@ -260,7 +259,6 @@ impl Chain {
&self.orphans.add(orphan);
debug!(
LOGGER,
"process_block: orphan: {:?}, # orphans {}{}",
block_hash,
self.orphans.len(),
@@ -274,7 +272,6 @@ impl Chain {
}
ErrorKind::Unfit(ref msg) => {
debug!(
LOGGER,
"Block {} at {} is unfit at this time: {}",
b.hash(),
b.header.height,
@@ -284,7 +281,6 @@ impl Chain {
}
_ => {
info!(
LOGGER,
"Rejected block {} at {}: {:?}",
b.hash(),
b.header.height,
@@ -299,7 +295,7 @@ impl Chain {
/// Process a block header received during "header first" propagation.
pub fn process_block_header(&self, bh: &BlockHeader, opts: Options) -> Result<(), Error> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let batch = self.store.batch()?;
let mut ctx = self.new_ctx(opts, batch, &mut txhashset)?;
pipe::process_block_header(bh, &mut ctx)?;
@@ -315,7 +311,7 @@ impl Chain {
headers: &Vec<BlockHeader>,
opts: Options,
) -> Result<(), Error> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let batch = self.store.batch()?;
let mut ctx = self.new_ctx(opts, batch, &mut txhashset)?;
@@ -359,7 +355,6 @@ impl Chain {
// Is there an orphan in our orphans that we can now process?
loop {
trace!(
LOGGER,
"check_orphans: at {}, # orphans {}",
height,
self.orphans.len(),
@@ -372,7 +367,6 @@ impl Chain {
let orphans_len = orphans.len();
for (i, orphan) in orphans.into_iter().enumerate() {
debug!(
LOGGER,
"check_orphans: get block {} at {}{}",
orphan.block.hash(),
height,
@@ -401,7 +395,6 @@ impl Chain {
if initial_height != height {
debug!(
LOGGER,
"check_orphans: {} blocks accepted since height {}, remaining # orphans {}",
height - initial_height,
initial_height,
@@ -417,7 +410,7 @@ impl Chain {
/// current chain state, specifically the current winning (valid, most
/// work) fork.
pub fn is_unspent(&self, output_ref: &OutputIdentifier) -> Result<Hash, Error> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let res = txhashset.is_unspent(output_ref);
match res {
Err(e) => Err(e),
@@ -427,7 +420,7 @@ impl Chain {
/// Validate the tx against the current UTXO set.
pub fn validate_tx(&self, tx: &Transaction) -> Result<(), Error> {
let txhashset = self.txhashset.read().unwrap();
let txhashset = self.txhashset.read();
txhashset::utxo_view(&txhashset, |utxo| {
utxo.validate_tx(tx)?;
Ok(())
@@ -443,7 +436,7 @@ impl Chain {
/// that has not yet sufficiently matured.
pub fn verify_coinbase_maturity(&self, tx: &Transaction) -> Result<(), Error> {
let height = self.next_block_height()?;
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
txhashset::extending_readonly(&mut txhashset, |extension| {
extension.verify_coinbase_maturity(&tx.inputs(), height)?;
Ok(())
@@ -470,7 +463,7 @@ impl Chain {
return Ok(());
}
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
// Now create an extension from the txhashset and validate against the
// latest block header. Rewind the extension to the specified header to
@@ -485,7 +478,7 @@ impl Chain {
/// Sets the txhashset roots on a brand new block by applying the block on
/// the current txhashset state.
pub fn set_txhashset_roots(&self, b: &mut Block, is_fork: bool) -> Result<(), Error> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let (prev_root, roots, sizes) =
txhashset::extending_readonly(&mut txhashset, |extension| {
if is_fork {
@@ -526,7 +519,7 @@ impl Chain {
output: &OutputIdentifier,
block_header: &BlockHeader,
) -> Result<MerkleProof, Error> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let merkle_proof = txhashset::extending_readonly(&mut txhashset, |extension| {
extension.rewind(&block_header)?;
@@ -539,13 +532,13 @@ impl Chain {
/// Return a merkle proof valid for the current output pmmr state at the
/// given pos
pub fn get_merkle_proof_for_pos(&self, commit: Commitment) -> Result<MerkleProof, String> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
txhashset.merkle_proof(commit)
}
/// Returns current txhashset roots
pub fn get_txhashset_roots(&self) -> TxHashSetRoots {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
txhashset.roots()
}
@@ -560,7 +553,7 @@ impl Chain {
// to rewind after receiving the txhashset zip.
let header = self.get_block_header(&h)?;
{
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
txhashset::extending_readonly(&mut txhashset, |extension| {
extension.rewind(&header)?;
extension.snapshot()?;
@@ -588,7 +581,6 @@ impl Chain {
txhashset: &txhashset::TxHashSet,
) -> Result<(), Error> {
debug!(
LOGGER,
"chain: validate_kernel_history: rewinding and validating kernel history (readonly)"
);
@@ -605,8 +597,8 @@ impl Chain {
})?;
debug!(
LOGGER,
"chain: validate_kernel_history: validated kernel root on {} headers", count,
"chain: validate_kernel_history: validated kernel root on {} headers",
count,
);
Ok(())
@@ -617,7 +609,7 @@ impl Chain {
/// have an MMR we can safely rewind based on the headers received from a peer.
/// TODO - think about how to optimize this.
pub fn rebuild_sync_mmr(&self, head: &Tip) -> Result<(), Error> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let mut batch = self.store.batch()?;
txhashset::sync_extending(&mut txhashset, &mut batch, |extension| {
extension.rebuild(head, &self.genesis)?;
@@ -681,10 +673,7 @@ impl Chain {
self.validate_kernel_history(&header, &txhashset)?;
// all good, prepare a new batch and update all the required records
debug!(
LOGGER,
"chain: txhashset_write: rewinding a 2nd time (writeable)"
);
debug!("chain: txhashset_write: rewinding a 2nd time (writeable)");
let mut batch = self.store.batch()?;
@@ -708,10 +697,7 @@ impl Chain {
Ok(())
})?;
debug!(
LOGGER,
"chain: txhashset_write: finished validating and rebuilding"
);
debug!("chain: txhashset_write: finished validating and rebuilding");
status.on_save();
@@ -726,21 +712,15 @@ impl Chain {
// Commit all the changes to the db.
batch.commit()?;
debug!(
LOGGER,
"chain: txhashset_write: finished committing the batch (head etc.)"
);
debug!("chain: txhashset_write: finished committing the batch (head etc.)");
// Replace the chain txhashset with the newly built one.
{
let mut txhashset_ref = self.txhashset.write().unwrap();
let mut txhashset_ref = self.txhashset.write();
*txhashset_ref = txhashset;
}
debug!(
LOGGER,
"chain: txhashset_write: replaced our txhashset with the new one"
);
debug!("chain: txhashset_write: replaced our txhashset with the new one");
// Check for any orphan blocks and process them based on the new chain state.
self.check_orphans(header.height + 1);
@@ -749,33 +729,11 @@ impl Chain {
Ok(())
}
/// Triggers chain compaction, cleaning up some unnecessary historical
/// information. We introduce a chain depth called horizon, which is
/// typically in the range of a couple days. Before that horizon, this
/// method will:
///
/// * compact the MMRs data files and flushing the corresponding remove logs
/// * delete old records from the k/v store (older blocks, indexes, etc.)
///
/// This operation can be resource intensive and takes some time to execute.
/// Meanwhile, the chain will not be able to accept new blocks. It should
/// therefore be called judiciously.
pub fn compact(&self) -> Result<(), Error> {
if self.archive_mode {
debug!(
LOGGER,
"Blockchain compaction disabled, node running in archive mode."
);
return Ok(());
}
debug!(LOGGER, "Starting blockchain compaction.");
// Compact the txhashset via the extension.
fn compact_txhashset(&self) -> Result<(), Error> {
debug!("Starting blockchain compaction.");
{
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
txhashset.compact()?;
// print out useful debug info after compaction
txhashset::extending_readonly(&mut txhashset, |extension| {
extension.dump_output_pmmr();
Ok(())
@@ -784,23 +742,33 @@ impl Chain {
// Now check we can still successfully validate the chain state after
// compacting, shouldn't be necessary once all of this is well-oiled
debug!(LOGGER, "Validating state after compaction.");
debug!("Validating state after compaction.");
self.validate(true)?;
Ok(())
}
// we need to be careful here in testing as 20 blocks is not that long
// in wall clock time
let horizon = global::cut_through_horizon() as u64;
let head = self.head()?;
if head.height <= horizon {
/// Cleanup old blocks from the db.
/// Determine the cutoff height from the horizon and the current block height.
/// *Only* runs if we are not in archive mode.
fn compact_blocks_db(&self) -> Result<(), Error> {
if self.archive_mode {
return Ok(());
}
let horizon = global::cut_through_horizon() as u64;
let head = self.head()?;
let cutoff = head.height.saturating_sub(horizon);
debug!(
LOGGER,
"Compaction remove blocks older than {}.",
head.height - horizon
"chain: compact_blocks_db: head height: {}, horizon: {}, cutoff: {}",
head.height, horizon, cutoff,
);
if cutoff == 0 {
return Ok(());
}
let mut count = 0;
let batch = self.store.batch()?;
let mut current = batch.get_header_by_height(head.height - horizon - 1)?;
@@ -830,25 +798,40 @@ impl Chain {
}
}
batch.commit()?;
debug!(LOGGER, "Compaction removed {} blocks, done.", count);
debug!("chain: compact_blocks_db: removed {} blocks.", count);
Ok(())
}
/// Triggers chain compaction.
///
/// * compacts the txhashset based on current prune_list
/// * removes historical blocks and associated data from the db (unless archive mode)
///
pub fn compact(&self) -> Result<(), Error> {
self.compact_txhashset()?;
if !self.archive_mode {
self.compact_blocks_db()?;
}
Ok(())
}
/// returns the last n nodes inserted into the output sum tree
pub fn get_last_n_output(&self, distance: u64) -> Vec<(Hash, OutputIdentifier)> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
txhashset.last_n_output(distance)
}
/// as above, for rangeproofs
pub fn get_last_n_rangeproof(&self, distance: u64) -> Vec<(Hash, RangeProof)> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
txhashset.last_n_rangeproof(distance)
}
/// as above, for kernels
pub fn get_last_n_kernel(&self, distance: u64) -> Vec<(Hash, TxKernel)> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
txhashset.last_n_kernel(distance)
}
@@ -858,7 +841,7 @@ impl Chain {
start_index: u64,
max: u64,
) -> Result<(u64, u64, Vec<Output>), Error> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let max_index = txhashset.highest_output_insertion_index();
let outputs = txhashset.outputs_by_insertion_index(start_index, max);
let rangeproofs = txhashset.rangeproofs_by_insertion_index(start_index, max);
@@ -945,7 +928,7 @@ impl Chain {
&self,
output_ref: &OutputIdentifier,
) -> Result<BlockHeader, Error> {
let mut txhashset = self.txhashset.write().unwrap();
let mut txhashset = self.txhashset.write();
let (_, pos) = txhashset.is_unspent(output_ref)?;
let mut min = 1;
let mut max = {
@@ -1051,7 +1034,6 @@ fn setup_head(
if header.height > 0 && extension.batch.get_block_sums(&header.hash()).is_err()
{
debug!(
LOGGER,
"chain: init: building (missing) block sums for {} @ {}",
header.height,
header.hash()
@@ -1072,7 +1054,6 @@ fn setup_head(
}
debug!(
LOGGER,
"chain: init: rewinding and validating before we start... {} at {}",
header.hash(),
header.height,
@@ -1109,7 +1090,7 @@ fn setup_head(
// Save the block_sums to the db for use later.
batch.save_block_sums(&genesis.hash(), &BlockSums::default())?;
info!(LOGGER, "chain: init: saved genesis: {:?}", genesis.hash());
info!("chain: init: saved genesis: {:?}", genesis.hash());
}
Err(e) => return Err(ErrorKind::StoreErr(e, "chain init load head".to_owned()))?,
};
chain/src/lib.rs
+1 -1
View File
@@ -30,7 +30,7 @@ extern crate serde;
#[macro_use]
extern crate serde_derive;
#[macro_use]
extern crate slog;
extern crate log;
extern crate chrono;
extern crate failure;
#[macro_use]
chain/src/pipe.rs
+13 -23
View File
@@ -14,7 +14,8 @@
//! Implementation of the chain block acceptance (or refusal) pipeline.
use std::sync::{Arc, RwLock};
use std::sync::Arc;
use util::RwLock;
use chrono::prelude::Utc;
use chrono::Duration;
@@ -34,7 +35,6 @@ use grin_store;
use store;
use txhashset;
use types::{Options, Tip};
use util::LOGGER;
/// Contextual information required to process a new block and either reject or
/// accept it.
@@ -70,7 +70,6 @@ pub fn process_block(b: &Block, ctx: &mut BlockContext) -> Result<Option<Tip>, E
// spend resources reading the full block when its header is invalid
debug!(
LOGGER,
"pipe: process_block {} at {} with {} inputs, {} outputs, {} kernels",
b.hash(),
b.header.height,
@@ -167,7 +166,6 @@ pub fn process_block(b: &Block, ctx: &mut BlockContext) -> Result<Option<Tip>, E
})?;
trace!(
LOGGER,
"pipe: process_block: {} at {} is valid, save and append.",
b.hash(),
b.header.height,
@@ -189,7 +187,6 @@ pub fn sync_block_headers(
) -> Result<Option<Tip>, Error> {
if let Some(header) = headers.first() {
debug!(
LOGGER,
"pipe: sync_block_headers: {} headers from {} at {}",
headers.len(),
header.hash(),
@@ -250,7 +247,6 @@ pub fn sync_block_headers(
/// it.
pub fn process_block_header(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
debug!(
LOGGER,
"pipe: process_block_header: {} at {}",
header.hash(),
header.height,
@@ -288,7 +284,7 @@ fn check_known_head(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(),
/// Keeps duplicates from the network in check.
/// Checks against the cache of recently processed block hashes.
fn check_known_cache(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), Error> {
let mut cache = ctx.block_hashes_cache.write().unwrap();
let mut cache = ctx.block_hashes_cache.write();
if cache.contains_key(&header.hash()) {
return Err(ErrorKind::Unfit("already known in cache".to_string()).into());
}
@@ -355,8 +351,8 @@ fn validate_header(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), E
// check version, enforces scheduled hard fork
if !consensus::valid_header_version(header.height, header.version) {
error!(
LOGGER,
"Invalid block header version received ({}), maybe update Grin?", header.version
"Invalid block header version received ({}), maybe update Grin?",
header.version
);
return Err(ErrorKind::InvalidBlockVersion(header.version).into());
}
@@ -377,8 +373,8 @@ fn validate_header(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), E
let edge_bits = header.pow.edge_bits();
if !(ctx.pow_verifier)(header, edge_bits).is_ok() {
error!(
LOGGER,
"pipe: error validating header with cuckoo edge_bits {}", edge_bits
"pipe: error validating header with cuckoo edge_bits {}",
edge_bits
);
return Err(ErrorKind::InvalidPow.into());
}
@@ -433,7 +429,6 @@ fn validate_header(header: &BlockHeader, ctx: &mut BlockContext) -> Result<(), E
let next_header_info = consensus::next_difficulty(header.height, diff_iter);
if target_difficulty != next_header_info.difficulty {
info!(
LOGGER,
"validate_header: header target difficulty {} != {}",
target_difficulty.to_num(),
next_header_info.difficulty.to_num()
@@ -553,8 +548,8 @@ fn update_head(b: &Block, ctx: &BlockContext) -> Result<Option<Tip>, Error> {
.map_err(|e| ErrorKind::StoreErr(e, "pipe save body".to_owned()))?;
debug!(
LOGGER,
"pipe: head updated to {} at {}", tip.last_block_h, tip.height
"pipe: head updated to {} at {}",
tip.last_block_h, tip.height
);
Ok(Some(tip))
@@ -574,7 +569,7 @@ fn update_sync_head(bh: &BlockHeader, batch: &mut store::Batch) -> Result<(), Er
batch
.save_sync_head(&tip)
.map_err(|e| ErrorKind::StoreErr(e, "pipe save sync head".to_owned()))?;
debug!(LOGGER, "sync head {} @ {}", bh.hash(), bh.height);
debug!("sync head {} @ {}", bh.hash(), bh.height);
Ok(())
}
@@ -588,8 +583,8 @@ fn update_header_head(bh: &BlockHeader, ctx: &mut BlockContext) -> Result<Option
.map_err(|e| ErrorKind::StoreErr(e, "pipe save header head".to_owned()))?;
debug!(
LOGGER,
"pipe: header_head updated to {} at {}", tip.last_block_h, tip.height
"pipe: header_head updated to {} at {}",
tip.last_block_h, tip.height
);
Ok(Some(tip))
@@ -621,7 +616,6 @@ pub fn rewind_and_apply_fork(b: &Block, ext: &mut txhashset::Extension) -> Resul
let forked_header = ext.batch.get_block_header(&current)?;
trace!(
LOGGER,
"rewind_and_apply_fork @ {} [{}], was @ {} [{}]",
forked_header.height,
forked_header.hash(),
@@ -632,11 +626,7 @@ pub fn rewind_and_apply_fork(b: &Block, ext: &mut txhashset::Extension) -> Resul
// Rewind the txhashset state back to the block where we forked from the most work chain.
ext.rewind(&forked_header)?;
trace!(
LOGGER,
"rewind_and_apply_fork: blocks on fork: {:?}",
fork_hashes,
);
trace!("rewind_and_apply_fork: blocks on fork: {:?}", fork_hashes,);
// Now re-apply all blocks on this fork.
for (_, h) in fork_hashes {
chain/src/store.rs
+14 -13
View File
@@ -14,7 +14,8 @@
//! Implements storage primitives required by the chain
use std::sync::{Arc, RwLock};
use std::sync::Arc;
use util::RwLock;
use croaring::Bitmap;
use lmdb;
@@ -96,7 +97,7 @@ impl ChainStore {
pub fn get_block_sums(&self, h: &Hash) -> Result<BlockSums, Error> {
{
let mut block_sums_cache = self.block_sums_cache.write().unwrap();
let mut block_sums_cache = self.block_sums_cache.write();
// cache hit - return the value from the cache
if let Some(block_sums) = block_sums_cache.get_mut(h) {
@@ -112,7 +113,7 @@ impl ChainStore {
// cache miss - so adding to the cache for next time
if let Ok(block_sums) = block_sums {
{
let mut block_sums_cache = self.block_sums_cache.write().unwrap();
let mut block_sums_cache = self.block_sums_cache.write();
block_sums_cache.insert(*h, block_sums.clone());
}
Ok(block_sums)
@@ -123,7 +124,7 @@ impl ChainStore {
pub fn get_block_header(&self, h: &Hash) -> Result<BlockHeader, Error> {
{
let mut header_cache = self.header_cache.write().unwrap();
let mut header_cache = self.header_cache.write();
// cache hit - return the value from the cache
if let Some(header) = header_cache.get_mut(h) {
@@ -140,7 +141,7 @@ impl ChainStore {
// cache miss - so adding to the cache for next time
if let Ok(header) = header {
{
let mut header_cache = self.header_cache.write().unwrap();
let mut header_cache = self.header_cache.write();
header_cache.insert(*h, header.clone());
}
Ok(header)
@@ -310,7 +311,7 @@ impl<'a> Batch<'a> {
let hash = header.hash();
{
let mut header_cache = self.header_cache.write().unwrap();
let mut header_cache = self.header_cache.write();
header_cache.insert(hash, header.clone());
}
@@ -350,7 +351,7 @@ impl<'a> Batch<'a> {
pub fn get_block_header(&self, h: &Hash) -> Result<BlockHeader, Error> {
{
let mut header_cache = self.header_cache.write().unwrap();
let mut header_cache = self.header_cache.write();
// cache hit - return the value from the cache
if let Some(header) = header_cache.get_mut(h) {
@@ -367,7 +368,7 @@ impl<'a> Batch<'a> {
// cache miss - so adding to the cache for next time
if let Ok(header) = header {
{
let mut header_cache = self.header_cache.write().unwrap();
let mut header_cache = self.header_cache.write();
header_cache.insert(*h, header.clone());
}
Ok(header)
@@ -390,7 +391,7 @@ impl<'a> Batch<'a> {
pub fn save_block_sums(&self, h: &Hash, sums: &BlockSums) -> Result<(), Error> {
{
let mut block_sums_cache = self.block_sums_cache.write().unwrap();
let mut block_sums_cache = self.block_sums_cache.write();
block_sums_cache.insert(*h, sums.clone());
}
@@ -400,7 +401,7 @@ impl<'a> Batch<'a> {
pub fn get_block_sums(&self, h: &Hash) -> Result<BlockSums, Error> {
{
let mut block_sums_cache = self.block_sums_cache.write().unwrap();
let mut block_sums_cache = self.block_sums_cache.write();
// cache hit - return the value from the cache
if let Some(block_sums) = block_sums_cache.get_mut(h) {
@@ -416,7 +417,7 @@ impl<'a> Batch<'a> {
// cache miss - so adding to the cache for next time
if let Ok(block_sums) = block_sums {
{
let mut block_sums_cache = self.block_sums_cache.write().unwrap();
let mut block_sums_cache = self.block_sums_cache.write();
block_sums_cache.insert(*h, block_sums.clone());
}
Ok(block_sums)
@@ -511,7 +512,7 @@ impl<'a> Batch<'a> {
self.save_block_input_bitmap(&block.hash(), &bitmap)?;
// Finally cache it locally for use later.
let mut cache = self.block_input_bitmap_cache.write().unwrap();
let mut cache = self.block_input_bitmap_cache.write();
cache.insert(block.hash(), bitmap.serialize());
Ok(bitmap)
@@ -519,7 +520,7 @@ impl<'a> Batch<'a> {
pub fn get_block_input_bitmap(&self, bh: &Hash) -> Result<Bitmap, Error> {
{
let mut cache = self.block_input_bitmap_cache.write().unwrap();
let mut cache = self.block_input_bitmap_cache.write();
// cache hit - return the value from the cache
if let Some(bytes) = cache.get_mut(bh) {
chain/src/txhashset/txhashset.rs
+39 -83
View File
@@ -40,7 +40,7 @@ use grin_store::types::prune_noop;
use store::{Batch, ChainStore};
use txhashset::{RewindableKernelView, UTXOView};
use types::{Tip, TxHashSetRoots, TxHashsetWriteStatus};
use util::{file, secp_static, zip, LOGGER};
use util::{file, secp_static, zip};
const HEADERHASHSET_SUBDIR: &'static str = "header";
const TXHASHSET_SUBDIR: &'static str = "txhashset";
@@ -328,7 +328,7 @@ where
// we explicitly rewind the extension.
let header = batch.head_header()?;
trace!(LOGGER, "Starting new txhashset (readonly) extension.");
trace!("Starting new txhashset (readonly) extension.");
let res = {
let mut extension = Extension::new(trees, &batch, header);
@@ -340,14 +340,14 @@ where
inner(&mut extension)
};
trace!(LOGGER, "Rollbacking txhashset (readonly) extension.");
trace!("Rollbacking txhashset (readonly) extension.");
trees.header_pmmr_h.backend.discard();
trees.output_pmmr_h.backend.discard();
trees.rproof_pmmr_h.backend.discard();
trees.kernel_pmmr_h.backend.discard();
trace!(LOGGER, "TxHashSet (readonly) extension done.");
trace!("TxHashSet (readonly) extension done.");
res
}
@@ -423,7 +423,7 @@ where
// index saving can be undone
let child_batch = batch.child()?;
{
trace!(LOGGER, "Starting new txhashset extension.");
trace!("Starting new txhashset extension.");
// TODO - header_mmr may be out ahead via the header_head
// TODO - do we need to handle this via an explicit rewind on the header_mmr?
@@ -436,10 +436,7 @@ where
match res {
Err(e) => {
debug!(
LOGGER,
"Error returned, discarding txhashset extension: {}", e
);
debug!("Error returned, discarding txhashset extension: {}", e);
trees.header_pmmr_h.backend.discard();
trees.output_pmmr_h.backend.discard();
trees.rproof_pmmr_h.backend.discard();
@@ -448,13 +445,13 @@ where
}
Ok(r) => {
if rollback {
trace!(LOGGER, "Rollbacking txhashset extension. sizes {:?}", sizes);
trace!("Rollbacking txhashset extension. sizes {:?}", sizes);
trees.header_pmmr_h.backend.discard();
trees.output_pmmr_h.backend.discard();
trees.rproof_pmmr_h.backend.discard();
trees.kernel_pmmr_h.backend.discard();
} else {
trace!(LOGGER, "Committing txhashset extension. sizes {:?}", sizes);
trace!("Committing txhashset extension. sizes {:?}", sizes);
child_batch.commit()?;
trees.header_pmmr_h.backend.sync()?;
trees.output_pmmr_h.backend.sync()?;
@@ -466,7 +463,7 @@ where
trees.kernel_pmmr_h.last_pos = sizes.3;
}
trace!(LOGGER, "TxHashSet extension done.");
trace!("TxHashSet extension done.");
Ok(r)
}
}
@@ -497,7 +494,7 @@ where
// index saving can be undone
let child_batch = batch.child()?;
{
trace!(LOGGER, "Starting new txhashset sync_head extension.");
trace!("Starting new txhashset sync_head extension.");
let pmmr = DBPMMR::at(&mut trees.sync_pmmr_h.backend, trees.sync_pmmr_h.last_pos);
let mut extension = HeaderExtension::new(pmmr, &child_batch, header);
@@ -510,31 +507,23 @@ where
match res {
Err(e) => {
debug!(
LOGGER,
"Error returned, discarding txhashset sync_head extension: {}", e
"Error returned, discarding txhashset sync_head extension: {}",
e
);
trees.sync_pmmr_h.backend.discard();
Err(e)
}
Ok(r) => {
if rollback {
trace!(
LOGGER,
"Rollbacking txhashset sync_head extension. size {:?}",
size
);
trace!("Rollbacking txhashset sync_head extension. size {:?}", size);
trees.sync_pmmr_h.backend.discard();
} else {
trace!(
LOGGER,
"Committing txhashset sync_head extension. size {:?}",
size
);
trace!("Committing txhashset sync_head extension. size {:?}", size);
child_batch.commit()?;
trees.sync_pmmr_h.backend.sync()?;
trees.sync_pmmr_h.last_pos = size;
}
trace!(LOGGER, "TxHashSet sync_head extension done.");
trace!("TxHashSet sync_head extension done.");
Ok(r)
}
}
@@ -564,7 +553,7 @@ where
// index saving can be undone
let child_batch = batch.child()?;
{
trace!(LOGGER, "Starting new txhashset header extension.");
trace!("Starting new txhashset header extension.");
let pmmr = DBPMMR::at(
&mut trees.header_pmmr_h.backend,
trees.header_pmmr_h.last_pos,
@@ -579,31 +568,23 @@ where
match res {
Err(e) => {
debug!(
LOGGER,
"Error returned, discarding txhashset header extension: {}", e
"Error returned, discarding txhashset header extension: {}",
e
);
trees.header_pmmr_h.backend.discard();
Err(e)
}
Ok(r) => {
if rollback {
trace!(
LOGGER,
"Rollbacking txhashset header extension. size {:?}",
size
);
trace!("Rollbacking txhashset header extension. size {:?}", size);
trees.header_pmmr_h.backend.discard();
} else {
trace!(
LOGGER,
"Committing txhashset header extension. size {:?}",
size
);
trace!("Committing txhashset header extension. size {:?}", size);
child_batch.commit()?;
trees.header_pmmr_h.backend.sync()?;
trees.header_pmmr_h.last_pos = size;
}
trace!(LOGGER, "TxHashSet header extension done.");
trace!("TxHashSet header extension done.");
Ok(r)
}
}
@@ -643,9 +624,7 @@ impl<'a> HeaderExtension<'a> {
/// This may be either the header MMR or the sync MMR depending on the
/// extension.
pub fn apply_header(&mut self, header: &BlockHeader) -> Result<(), Error> {
self.pmmr
.push(header.clone())
.map_err(&ErrorKind::TxHashSetErr)?;
self.pmmr.push(&header).map_err(&ErrorKind::TxHashSetErr)?;
self.header = header.clone();
Ok(())
}
@@ -654,7 +633,6 @@ impl<'a> HeaderExtension<'a> {
/// Note the close relationship between header height and insertion index.
pub fn rewind(&mut self, header: &BlockHeader) -> Result<(), Error> {
debug!(
LOGGER,
"Rewind header extension to {} at {}",
header.hash(),
header.height
@@ -675,7 +653,7 @@ impl<'a> HeaderExtension<'a> {
/// Used when rebuilding the header MMR by reapplying all headers
/// including the genesis block header.
pub fn truncate(&mut self) -> Result<(), Error> {
debug!(LOGGER, "Truncating header extension.");
debug!("Truncating header extension.");
self.pmmr.rewind(0).map_err(&ErrorKind::TxHashSetErr)?;
Ok(())
}
@@ -689,7 +667,6 @@ impl<'a> HeaderExtension<'a> {
/// Requires *all* header hashes to be iterated over in ascending order.
pub fn rebuild(&mut self, head: &Tip, genesis: &BlockHeader) -> Result<(), Error> {
debug!(
LOGGER,
"About to rebuild header extension from {:?} to {:?}.",
genesis.hash(),
head.last_block_h,
@@ -712,7 +689,6 @@ impl<'a> HeaderExtension<'a> {
if header_hashes.len() > 0 {
debug!(
LOGGER,
"Re-applying {} headers to extension, from {:?} to {:?}.",
header_hashes.len(),
header_hashes.first().unwrap(),
@@ -983,7 +959,7 @@ impl<'a> Extension<'a> {
fn apply_header(&mut self, header: &BlockHeader) -> Result<(), Error> {
self.header_pmmr
.push(header.clone())
.push(&header)
.map_err(&ErrorKind::TxHashSetErr)?;
Ok(())
}
@@ -995,10 +971,7 @@ impl<'a> Extension<'a> {
/// We need the hash of each sibling pos from the pos up to the peak
/// including the sibling leaf node which may have been removed.
pub fn merkle_proof(&self, output: &OutputIdentifier) -> Result<MerkleProof, Error> {
debug!(
LOGGER,
"txhashset: merkle_proof: output: {:?}", output.commit,
);
debug!("txhashset: merkle_proof: output: {:?}", output.commit,);
// then calculate the Merkle Proof based on the known pos
let pos = self.batch.get_output_pos(&output.commit)?;
let merkle_proof = self
@@ -1027,12 +1000,7 @@ impl<'a> Extension<'a> {
/// Rewinds the MMRs to the provided block, rewinding to the last output pos
/// and last kernel pos of that block.
pub fn rewind(&mut self, header: &BlockHeader) -> Result<(), Error> {
debug!(
LOGGER,
"Rewind to header {} at {}",
header.hash(),
header.height,
);
debug!("Rewind to header {} at {}", header.hash(), header.height,);
// We need to build bitmaps of added and removed output positions
// so we can correctly rewind all operations applied to the output MMR
@@ -1067,11 +1035,8 @@ impl<'a> Extension<'a> {
rewind_rm_pos: &Bitmap,
) -> Result<(), Error> {
debug!(
LOGGER,
"txhashset: rewind_to_pos: header {}, output {}, kernel {}",
header_pos,
output_pos,
kernel_pos,
header_pos, output_pos, kernel_pos,
);
self.header_pmmr
@@ -1191,7 +1156,6 @@ impl<'a> Extension<'a> {
}
debug!(
LOGGER,
"txhashset: validated the header {}, output {}, rproof {}, kernel {} mmrs, took {}s",
self.header_pmmr.unpruned_size(),
self.output_pmmr.unpruned_size(),
@@ -1270,22 +1234,22 @@ impl<'a> Extension<'a> {
/// Dumps the output MMR.
/// We use this after compacting for visual confirmation that it worked.
pub fn dump_output_pmmr(&self) {
debug!(LOGGER, "-- outputs --");
debug!("-- outputs --");
self.output_pmmr.dump_from_file(false);
debug!(LOGGER, "--");
debug!("--");
self.output_pmmr.dump_stats();
debug!(LOGGER, "-- end of outputs --");
debug!("-- end of outputs --");
}
/// Dumps the state of the 3 sum trees to stdout for debugging. Short
/// version only prints the Output tree.
pub fn dump(&self, short: bool) {
debug!(LOGGER, "-- outputs --");
debug!("-- outputs --");
self.output_pmmr.dump(short);
if !short {
debug!(LOGGER, "-- range proofs --");
debug!("-- range proofs --");
self.rproof_pmmr.dump(short);
debug!(LOGGER, "-- kernels --");
debug!("-- kernels --");
self.kernel_pmmr.dump(short);
}
}
@@ -1318,7 +1282,6 @@ impl<'a> Extension<'a> {
}
debug!(
LOGGER,
"txhashset: verified {} kernel signatures, pmmr size {}, took {}s",
kern_count,
self.kernel_pmmr.unpruned_size(),
@@ -1353,8 +1316,8 @@ impl<'a> Extension<'a> {
commits.clear();
proofs.clear();
debug!(
LOGGER,
"txhashset: verify_rangeproofs: verified {} rangeproofs", proof_count,
"txhashset: verify_rangeproofs: verified {} rangeproofs",
proof_count,
);
}
}
@@ -1370,13 +1333,12 @@ impl<'a> Extension<'a> {
commits.clear();
proofs.clear();
debug!(
LOGGER,
"txhashset: verify_rangeproofs: verified {} rangeproofs", proof_count,
"txhashset: verify_rangeproofs: verified {} rangeproofs",
proof_count,
);
}
debug!(
LOGGER,
"txhashset: verified {} rangeproofs, pmmr size {}, took {}s",
proof_count,
self.rproof_pmmr.unpruned_size(),
@@ -1452,10 +1414,7 @@ fn check_and_remove_files(txhashset_path: &PathBuf, header: &BlockHeader) -> Res
// Removing unexpected directories if needed
if !dir_difference.is_empty() {
debug!(
LOGGER,
"Unexpected folder(s) found in txhashset folder, removing."
);
debug!("Unexpected folder(s) found in txhashset folder, removing.");
for diff in dir_difference {
let diff_path = txhashset_path.join(diff);
file::delete(diff_path)?;
@@ -1492,7 +1451,6 @@ fn check_and_remove_files(txhashset_path: &PathBuf, header: &BlockHeader) -> Res
.collect();
if !difference.is_empty() {
debug!(
LOGGER,
"Unexpected file(s) found in txhashset subfolder {:?}, removing.",
&subdirectory_path
);
@@ -1520,10 +1478,8 @@ pub fn input_pos_to_rewind(
if head_header.height < block_header.height {
debug!(
LOGGER,
"input_pos_to_rewind: {} < {}, nothing to rewind",
head_header.height,
block_header.height
head_header.height, block_header.height
);
return Ok(Bitmap::create());
}