Single DB (#1741)

* Consolidation migration

* Started db amalgamation

* Using client constants for columns

* Adding with_columns constructor

* Migrating to single db

* Fixing tests.

* test.sh without verbose

* Fixing warnings

* add migration tests that catch the bug

* make multiple migrations more robust

* add moved v9

* Merge branch 'noop-migrations' into single-db

* spurious line

* clean up migrations ordering

* update comment [ci skip]

* Bumping default number of max_open_files & re-ordering columns.

* fix merge

* fix ignored analysis tests

* Caching best block content

* Faster best_block_header

* Adding progress to v8 migration

* clean up warnings

* Separate hashes and bodies in the DB

* Separate hashes and bodies in the DB

* Fixed tests

ethcore/src/client/client.rs

@@ -24,9 +24,9 @@ use std::time::{Instant};
 use time::precise_time_ns;
 
 // util
-use util::{journaldb, rlp, Bytes, Stream, View, PerfTimer, Itertools, Mutex, RwLock};
+use util::{journaldb, rlp, Bytes, View, PerfTimer, Itertools, Mutex, RwLock};
 use util::journaldb::JournalDB;
-use util::rlp::{RlpStream, Rlp, UntrustedRlp};
+use util::rlp::{UntrustedRlp};
 use util::numbers::*;
 use util::panics::*;
 use util::io::*;
@@ -34,14 +34,13 @@ use util::sha3::*;
 use util::kvdb::*;
 
 // other
-use views::BlockView;
+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 views::HeaderView;
 use service::ClientIoMessage;
 use env_info::LastHashes;
 use verification;
@@ -123,6 +122,7 @@ pub struct Client {
 	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>,
@@ -141,6 +141,19 @@ pub struct Client {
 }
 
 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> {
@@ -160,36 +173,25 @@ impl Client {
 	) -> Result<Arc<Client>, ClientError> {
 		let path = path.to_path_buf();
 		let gb = spec.genesis_block();
-		let chain = Arc::new(BlockChain::new(config.blockchain, &gb, &path));
-		let tracedb = Arc::new(try!(TraceDB::new(config.tracing, &path, chain.clone())));
+		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 mut state_db_config = match config.db_cache_size {
-			None => DatabaseConfig::default(),
-			Some(cache_size) => DatabaseConfig::with_cache(cache_size),
-		};
-
-		state_db_config = state_db_config.compaction(config.db_compaction.compaction_profile());
-
-		let mut state_db = journaldb::new(
-			&append_path(&path, "state"),
-			config.pruning,
-			state_db_config
-		);
+		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()) {
-			state_db.commit(0, &spec.genesis_header().hash(), None).expect("Error commiting genesis state to state DB");
+			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());
 		}
 
-		/* TODO: enable this once the best block issue is resolved
-		while !chain.block_header(&chain.best_block_hash()).map_or(true, |h| state_db.contains(h.state_root())) {
-			warn!("State root not found for block #{} ({}), recovering...", chain.best_block_number(), chain.best_block_hash().hex());
-			chain.rewind();
-		}*/
-
 		let engine = Arc::new(spec.engine);
 
 		let block_queue = BlockQueue::new(config.queue, engine.clone(), message_channel.clone());
@@ -204,6 +206,7 @@ impl Client {
 			chain: chain,
 			tracedb: tracedb,
 			engine: engine,
+			db: db,
 			state_db: Mutex::new(state_db),
 			block_queue: block_queue,
 			report: RwLock::new(Default::default()),
@@ -432,21 +435,23 @@ impl Client {
 
 		//let traces = From::from(block.traces().clone().unwrap_or_else(Vec::new));
 
+		let batch = DBTransaction::new(&self.db);
 		// CHECK! I *think* this is fine, even if the state_root is equal to another
 		// already-imported block of the same number.
 		// TODO: Prove it with a test.
-		block.drain().commit(number, hash, ancient).expect("State DB commit failed.");
+		block.drain().commit(&batch, number, hash, ancient).expect("State DB commit failed.");
 
-		// And update the chain after commit to prevent race conditions
-		// (when something is in chain but you are not able to fetch details)
-		let route = self.chain.insert_block(block_data, receipts);
-		self.tracedb.import(TraceImportRequest {
+		let route = self.chain.insert_block(&batch, block_data, receipts);
+		self.tracedb.import(&batch, TraceImportRequest {
 			traces: traces.into(),
 			block_hash: hash.clone(),
 			block_number: number,
 			enacted: route.enacted.clone(),
 			retracted: route.retracted.len()
 		});
+		// Final commit to the DB
+		self.db.write(batch).expect("State DB write failed.");
+
 		self.update_last_hashes(&parent, hash);
 		route
 	}
@@ -674,17 +679,17 @@ impl BlockChainClient for Client {
 
 	fn replay(&self, id: TransactionID, analytics: CallAnalytics) -> Result<Executed, ReplayError> {
 		let address = try!(self.transaction_address(id).ok_or(ReplayError::TransactionNotFound));
-		let block_data = try!(self.block(BlockID::Hash(address.block_hash)).ok_or(ReplayError::StatePruned));
+		let header_data = try!(self.block_header(BlockID::Hash(address.block_hash)).ok_or(ReplayError::StatePruned));
+		let body_data = try!(self.block_body(BlockID::Hash(address.block_hash)).ok_or(ReplayError::StatePruned));
 		let mut state = try!(self.state_at_beginning(BlockID::Hash(address.block_hash)).ok_or(ReplayError::StatePruned));
-		let block = BlockView::new(&block_data);
-		let txs = block.transactions();
+		let txs = BodyView::new(&body_data).transactions();
 
 		if address.index >= txs.len() {
 			return Err(ReplayError::TransactionNotFound);
 		}
 
 		let options = TransactOptions { tracing: analytics.transaction_tracing, vm_tracing: analytics.vm_tracing, check_nonce: false };
-		let view = block.header_view();
+		let view = HeaderView::new(&header_data);
 		let last_hashes = self.build_last_hashes(view.hash());
 		let mut env_info = EnvInfo {
 			number: view.number(),
@@ -719,20 +724,16 @@ impl BlockChainClient for Client {
 		}
 	}
 
+	fn best_block_header(&self) -> Bytes {
+		self.chain.best_block_header()
+	}
+
 	fn block_header(&self, id: BlockID) -> Option<Bytes> {
-		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()))
+		Self::block_hash(&self.chain, id).and_then(|hash| self.chain.block_header_data(&hash))
 	}
 
 	fn block_body(&self, id: BlockID) -> Option<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);
-				body.append_raw(rlp.at(2).as_raw(), 1);
-				body.out()
-			})
-		})
+		Self::block_hash(&self.chain, id).and_then(|hash| self.chain.block_body(&hash))
 	}
 
 	fn block(&self, id: BlockID) -> Option<Bytes> {
@@ -789,13 +790,13 @@ impl BlockChainClient for Client {
 
 	fn uncle(&self, id: UncleID) -> Option<Bytes> {
 		let index = id.position;
-		self.block(id.block).and_then(|block| BlockView::new(&block).uncle_rlp_at(index))
+		self.block_body(id.block).and_then(|body| BodyView::new(&body).uncle_rlp_at(index))
 	}
 
 	fn transaction_receipt(&self, id: TransactionID) -> Option<LocalizedReceipt> {
-		self.transaction_address(id).and_then(|address| {
-			let t = self.chain.block(&address.block_hash)
-				.and_then(|block| BlockView::new(&block).localized_transaction_at(address.index));
+		self.transaction_address(id).and_then(|address| self.chain.block_number(&address.block_hash).and_then(|block_number| {
+			let t = self.chain.block_body(&address.block_hash)
+				.and_then(|block| BodyView::new(&block).localized_transaction_at(&address.block_hash, block_number, address.index));
 
 			match (t, self.chain.transaction_receipt(&address)) {
 				(Some(tx), Some(receipt)) => {
@@ -834,7 +835,7 @@ impl BlockChainClient for Client {
 				},
 				_ => None
 			}
-		})
+		}))
 	}
 
 	fn tree_route(&self, from: &H256, to: &H256) -> Option<TreeRoute> {
@@ -910,7 +911,7 @@ impl BlockChainClient for Client {
 		blocks.into_iter()
 			.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())))
+			.filter_map(|(number, hash, receipts)| self.chain.block_body(&hash).map(|ref b| (number, hash, receipts, BodyView::new(b).transaction_hashes())))
 			.flat_map(|(number, hash, receipts, hashes)| {
 				let mut log_index = 0;
 				receipts.into_iter()

ethcore/src/client/test_client.rs

@@ -248,7 +248,8 @@ impl TestBlockChainClient {
 
 pub fn get_temp_journal_db() -> GuardedTempResult<Box<JournalDB>> {
 	let temp = RandomTempPath::new();
-	let journal_db = journaldb::new(temp.as_str(), journaldb::Algorithm::EarlyMerge, DatabaseConfig::default());
+	let db = Database::open_default(temp.as_str()).unwrap();
+	let journal_db = journaldb::new(Arc::new(db), journaldb::Algorithm::EarlyMerge, None);
 	GuardedTempResult {
 		_temp: temp,
 		result: Some(journal_db)
@@ -363,6 +364,10 @@ impl BlockChainClient for TestBlockChainClient {
 		unimplemented!();
 	}
 
+	fn best_block_header(&self) -> Bytes {
+		self.block_header(BlockID::Hash(self.chain_info().best_block_hash)).expect("Best block always have header.")
+	}
+
 	fn block_header(&self, id: BlockID) -> Option<Bytes> {
 		self.block_hash(id).and_then(|hash| self.blocks.read().get(&hash).map(|r| Rlp::new(r).at(0).as_raw().to_vec()))
 	}

ethcore/src/client/traits.rs

@@ -145,10 +145,7 @@ pub trait BlockChainClient : Sync + Send {
 	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()
-	}
+	fn best_block_header(&self) -> Bytes;
 
 	/// Returns numbers of blocks containing given bloom.
 	fn blocks_with_bloom(&self, bloom: &H2048, from_block: BlockID, to_block: BlockID) -> Option<Vec<BlockNumber>>;