[WIP] Perf: Kernel Sync Performance

chain/src/chain.rs

@@ -29,7 +29,7 @@ use core::core::hash::{Hash, Hashed, ZERO_HASH};
 use core::core::merkle_proof::MerkleProof;
 use core::core::verifier_cache::VerifierCache;
 use core::core::{
-	Block, BlockHeader, BlockSums, Output, OutputIdentifier, Transaction, TxKernelEntry,
+	Block, BlockHeader, BlockSums, Output, OutputIdentifier, Transaction, TxKernel, TxKernelEntry,
 };
 use core::global;
 use core::pow;
@@ -179,7 +179,8 @@ impl Chain {
 		let store = Arc::new(chain_store);
 
 		// open the txhashset, creating a new one if necessary
-		let mut txhashset = txhashset::TxHashSet::open(db_root.clone(), store.clone(), None)?;
+		let mut txhashset =
+			txhashset::TxHashSet::open(db_root.clone(), store.clone(), None, genesis.hash())?;
 
 		setup_head(genesis.clone(), store.clone(), &mut txhashset)?;
 
@@ -345,6 +346,22 @@ impl Chain {
 		Ok(())
 	}
 
+	/// Attempt to add new kernels to the kernel mmr.
+	/// This is only ever used during sync.
+	pub fn sync_kernels(
+		&self,
+		blocks: &Vec<(Hash, Vec<TxKernel>)>,
+		opts: Options,
+	) -> Result<(), Error> {
+		let mut txhashset = self.txhashset.write();
+		let batch = self.store.batch()?;
+		let mut ctx = self.new_ctx(opts, batch, &mut txhashset)?;
+
+		pipe::sync_kernels(blocks, &mut ctx)?;
+
+		Ok(())
+	}
+
 	fn new_ctx<'a>(
 		&self,
 		opts: Options,
@@ -789,17 +806,37 @@ impl Chain {
 		}
 
 		let header = self.get_block_header(&h)?;
-		txhashset::zip_write(self.db_root.clone(), txhashset_data, &header)?;
+		let mut kernels_already_synced = false;
+		{
+			let existing_txhashset = self.txhashset.read();
+			let num_kernels = existing_txhashset.num_kernels();
+			if num_kernels >= header.kernel_mmr_size {
+				kernels_already_synced = true;
+			}
+		}
 
-		let mut txhashset =
-			txhashset::TxHashSet::open(self.db_root.clone(), self.store.clone(), Some(&header))?;
+		txhashset::zip_write(
+			self.db_root.clone(),
+			txhashset_data,
+			&header,
+			kernels_already_synced,
+		)?;
+
+		let mut txhashset = txhashset::TxHashSet::open(
+			self.db_root.clone(),
+			self.store.clone(),
+			Some(&header),
+			header.hash(),
+		)?;
 
 		// The txhashset.zip contains the output, rangeproof and kernel MMRs.
 		// We must rebuild the header MMR ourselves based on the headers in our db.
 		self.rebuild_header_mmr(&Tip::from_header(&header), &mut txhashset)?;
 
 		// Validate the full kernel history (kernel MMR root for every block header).
-		self.validate_kernel_history(&header, &txhashset)?;
+		if !kernels_already_synced {
+			self.validate_kernel_history(&header, &txhashset)?;
+		}
 
 		// all good, prepare a new batch and update all the required records
 		debug!("txhashset_write: rewinding a 2nd time (writeable)");
@@ -973,6 +1010,24 @@ impl Chain {
 		txhashset.last_n_kernel(distance)
 	}
 
+	/// kernels by insertion index
+	pub fn get_kernels_by_insertion_index(&self, start_index: u64, max: u64) -> Vec<TxKernelEntry> {
+		let mut txhashset = self.txhashset.write();
+		txhashset.kernels_by_insertion_index(start_index, max).1
+	}
+
+	/// returns the number of leaves in the kernel mmr
+	pub fn get_num_kernels(&self) -> u64 {
+		let txhashset = self.txhashset.read();
+		txhashset.num_kernels()
+	}
+	/// returns the number of leaves in the kernel mmr
+	pub fn get_kernel_root_validated_tip(&self) -> Result<BlockHeader, Error> {
+		let txhashset = self.txhashset.read();
+		let hash = txhashset.kernel_root_validated_tip();
+		self.get_block_header(&hash)
+	}
+
 	/// outputs by insertion index
 	pub fn unspent_outputs_by_insertion_index(
 		&self,

chain/src/pipe.rs

@@ -27,7 +27,7 @@ use core::consensus;
 use core::core::hash::{Hash, Hashed};
 use core::core::verifier_cache::VerifierCache;
 use core::core::Committed;
-use core::core::{Block, BlockHeader, BlockSums};
+use core::core::{Block, BlockHeader, BlockSums, TxKernel};
 use core::global;
 use core::pow;
 use error::{Error, ErrorKind};
@@ -276,6 +276,68 @@ pub fn sync_block_headers(
 	}
 }
 
+/// Process the kernels.
+/// This is only ever used during sync.
+pub fn sync_kernels(
+	blocks: &Vec<(Hash, Vec<TxKernel>)>,
+	ctx: &mut BlockContext,
+) -> Result<(), Error> {
+	let first_block = match blocks.first() {
+		Some(block) => {
+			debug!(
+				"pipe: sync_kernels: {} blocks from {}",
+				blocks.len(),
+				block.0
+			);
+			block
+		}
+		_ => return Ok(()),
+	};
+
+	let header = ctx.batch.get_block_header(&first_block.0)?;
+
+	let next_kernel_index = match header.height {
+		0 => 0,
+		height => {
+			let header = match ctx.batch.get_header_by_height(height - 1) {
+				Ok(header) => header,
+				Err(e) => {
+					error!("sync_kernels: Could not find header: {:?}", e);
+					return Ok(());
+				}
+			};
+			header.kernel_mmr_size
+		}
+	};
+
+	let num_kernels = ctx.txhashset.num_kernels();
+	if num_kernels < next_kernel_index {
+		// TODO: A convenient way to store and process these later
+		// would allow nodes to request batches of kernels in parallel.
+		return Err(ErrorKind::TxHashSetErr("Previous kernels missing".to_string()).into());
+	}
+
+	let kernels_to_add: Vec<TxKernel> = blocks.iter().flat_map(|block| block.1.clone()).collect();
+	if num_kernels > (next_kernel_index + kernels_to_add.len() as u64) {
+		debug!(
+			"pipe: sync_kernels: kernels from index {} not needed.",
+			next_kernel_index,
+		);
+		return Ok(());
+	}
+
+	txhashset::extending(&mut ctx.txhashset, &mut ctx.batch, |extension| {
+		// Rewinding kernel mmr to correct kernel index. Probably unnecessary, but playing it safe.
+		extension.rewind_kernel_mmr(next_kernel_index)?;
+
+		for block in blocks {
+			extension.apply_kernels(&block.0, &block.1)?;
+		}
+		Ok(())
+	})?;
+	Ok(())
+}
+
 /// Process block header as part of "header first" block propagation.
 /// We validate the header but we do not store it or update header head based
 /// on this. We will update these once we get the block back after requesting