client.rs

// 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/>.

use std::collections::{HashSet, HashMap, VecDeque};
use std::ops::Deref;
use std::sync::{Arc, Weak};
use std::path::{Path};
use std::fmt;
use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering};
use std::time::{Instant};
use time::precise_time_ns;

// util
use util::{journaldb, rlp, Bytes, View, PerfTimer, Itertools, Mutex, RwLock};
use util::journaldb::JournalDB;
use util::rlp::{UntrustedRlp};
use util::numbers::*;
use util::panics::*;
use util::io::*;
use util::sha3::*;
use util::kvdb::*;

// other
use views::{BlockView, HeaderView, BodyView};
use error::{ImportError, ExecutionError, ReplayError, BlockError, ImportResult};
use header::BlockNumber;
use state::State;
use spec::Spec;
use basic_types::Seal;
use engines::Engine;
use service::ClientIoMessage;
use env_info::LastHashes;
use verification;
use verification::{PreverifiedBlock, Verifier};
use block::*;
use transaction::{LocalizedTransaction, SignedTransaction, Action};
use blockchain::extras::TransactionAddress;
use types::filter::Filter;
use log_entry::LocalizedLogEntry;
use block_queue::{BlockQueue, BlockQueueInfo};
use blockchain::{BlockChain, BlockProvider, TreeRoute, ImportRoute};
use client::{BlockID, TransactionID, UncleID, TraceId, ClientConfig, BlockChainClient, MiningBlockChainClient,
	TraceFilter, CallAnalytics, BlockImportError, Mode, ChainNotify};
use client::Error as ClientError;
use env_info::EnvInfo;
use executive::{Executive, Executed, TransactOptions, contract_address};
use receipt::LocalizedReceipt;
use trace::{TraceDB, ImportRequest as TraceImportRequest, LocalizedTrace, Database as TraceDatabase};
use trace;
use trace::FlatTransactionTraces;
use evm::Factory as EvmFactory;
use miner::{Miner, MinerService};
use util::TrieFactory;

// re-export
pub use types::blockchain_info::BlockChainInfo;
pub use types::block_status::BlockStatus;
pub use blockchain::CacheSize as BlockChainCacheSize;

const MAX_TX_QUEUE_SIZE: usize = 4096;
const MAX_QUEUE_SIZE_TO_SLEEP_ON: usize = 2;

impl fmt::Display for BlockChainInfo {
	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
		write!(f, "#{}.{}", self.best_block_number, self.best_block_hash)
	}
}

/// 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,
	/// How many transactions have been applied so far.
	pub transactions_applied: usize,
	/// How much gas has been processed so far.
	pub gas_processed: U256,
	/// Memory used by state DB
	pub state_db_mem: usize,
}

impl ClientReport {
	/// Alter internal reporting to reflect the additional `block` has been processed.
	pub fn accrue_block(&mut self, block: &PreverifiedBlock) {
		self.blocks_imported += 1;
		self.transactions_applied += block.transactions.len();
		self.gas_processed = self.gas_processed + block.header.gas_used;
	}
}

struct SleepState {
	last_activity: Option<Instant>,
	last_autosleep: Option<Instant>,
}

impl SleepState {
	fn new(awake: bool) -> Self {
		SleepState {
			last_activity: match awake { false => None, true => Some(Instant::now()) },
			last_autosleep: match awake { false => Some(Instant::now()), true => None },
		}
	}
}

/// 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 {
	mode: Mode,
	chain: Arc<BlockChain>,
	tracedb: Arc<TraceDB<BlockChain>>,
	engine: Arc<Box<Engine>>,
	db: Arc<Database>,
	state_db: Mutex<Box<JournalDB>>,
	block_queue: BlockQueue,
	report: RwLock<ClientReport>,
	import_lock: Mutex<()>,
	panic_handler: Arc<PanicHandler>,
	verifier: Box<Verifier>,
	vm_factory: Arc<EvmFactory>,
	trie_factory: TrieFactory,
	miner: Arc<Miner>,
	sleep_state: Mutex<SleepState>,
	liveness: AtomicBool,
	io_channel: IoChannel<ClientIoMessage>,
	notify: RwLock<Vec<Weak<ChainNotify>>>,
	queue_transactions: AtomicUsize,
	last_hashes: RwLock<VecDeque<H256>>,
}

const HISTORY: u64 = 1200;
// database columns
/// Column for State
pub const DB_COL_STATE: Option<u32> = Some(0);
/// Column for Block headers
pub const DB_COL_HEADERS: Option<u32> = Some(1);
/// Column for Block bodies
pub const DB_COL_BODIES: Option<u32> = Some(2);
/// Column for Extras
pub const DB_COL_EXTRA: Option<u32> = Some(3);
/// Column for Traces
pub const DB_COL_TRACE: Option<u32> = Some(4);
/// Number of columns in DB
pub const DB_NO_OF_COLUMNS: Option<u32> = Some(5);

/// Append a path element to the given path and return the string.
pub fn append_path<P>(path: P, item: &str) -> String where P: AsRef<Path> {
	let mut p = path.as_ref().to_path_buf();
	p.push(item);
	p.to_str().unwrap().to_owned()
}

impl Client {
	///  Create a new client with given spec and DB path and custom verifier.
	pub fn new(
		config: ClientConfig,
		spec: Spec,
		path: &Path,
		miner: Arc<Miner>,
		message_channel: IoChannel<ClientIoMessage>,
	) -> Result<Arc<Client>, ClientError> {
		let path = path.to_path_buf();
		let gb = spec.genesis_block();
		let mut db_config = DatabaseConfig::with_columns(DB_NO_OF_COLUMNS);
		db_config.cache_size = config.db_cache_size;
		db_config.compaction = config.db_compaction.compaction_profile();

		let db = Arc::new(Database::open(&db_config, &path.to_str().unwrap()).expect("Error opening database"));
		let chain = Arc::new(BlockChain::new(config.blockchain, &gb, db.clone()));
		let tracedb = Arc::new(try!(TraceDB::new(config.tracing, db.clone(), chain.clone())));

		let mut state_db = journaldb::new(db.clone(), config.pruning, DB_COL_STATE);
		if state_db.is_empty() && spec.ensure_db_good(state_db.as_hashdb_mut()) {
			let batch = DBTransaction::new(&db);
			state_db.commit(&batch, 0, &spec.genesis_header().hash(), None).expect("Error commiting genesis state to state DB");
			db.write(batch).expect("Error writing genesis state to state DB");
		}

		if !chain.block_header(&chain.best_block_hash()).map_or(true, |h| state_db.contains(h.state_root())) {
			warn!("State root not found for block #{} ({})", chain.best_block_number(), chain.best_block_hash().hex());
		}

		let engine = Arc::new(spec.engine);

		let block_queue = BlockQueue::new(config.queue, engine.clone(), message_channel.clone());
		let panic_handler = PanicHandler::new_in_arc();
		panic_handler.forward_from(&block_queue);