synchronization_chain.rs

use std::fmt;
use std::sync::Arc;
use std::collections::{VecDeque, HashSet};
use linked_hash_map::LinkedHashMap;
use parking_lot::RwLock;
use chain::{BlockHeader, Transaction};
use db::{self, IndexedBlock};
use best_headers_chain::{BestHeadersChain, Information as BestHeadersInformation};
use primitives::bytes::Bytes;
use primitives::hash::H256;
use hash_queue::{HashQueueChain, HashPosition};
use miner::{MemoryPool, MemoryPoolOrderingStrategy, MemoryPoolInformation};

/// Thread-safe reference to `Chain`
pub type ChainRef = Arc<RwLock<Chain>>;

/// Index of 'verifying' queue
const VERIFYING_QUEUE: usize = 0;
/// Index of 'requested' queue
const REQUESTED_QUEUE: usize = 1;
/// Index of 'scheduled' queue
const SCHEDULED_QUEUE: usize = 2;
/// Number of hash queues
const NUMBER_OF_QUEUES: usize = 3;

/// Block insertion result
#[derive(Debug, Default, PartialEq)]
pub struct BlockInsertionResult {
	/// Hashes of blocks, which were canonized during this insertion procedure. Order matters
	pub canonized_blocks_hashes: Vec<H256>,
	/// Transaction to 'reverify'. Order matters
	pub transactions_to_reverify: Vec<(H256, Transaction)>,
}

impl BlockInsertionResult {
	#[cfg(test)]
	pub fn with_canonized_blocks(canonized_blocks_hashes: Vec<H256>) -> Self {
		BlockInsertionResult {
			canonized_blocks_hashes: canonized_blocks_hashes,
			transactions_to_reverify: Vec::new(),
		}
	}
}

/// Block synchronization state
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
pub enum BlockState {
	/// Block is unknown
	Unknown,
	/// Scheduled for requesting
	Scheduled,
	/// Requested from peers
	Requested,
	/// Currently verifying
	Verifying,
	/// In storage
	Stored,
	/// This block has been marked as dead-end block
	DeadEnd,
}

/// Transactions synchronization state
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
pub enum TransactionState {
	/// Transaction is unknown
	Unknown,
	/// Currently verifying
	Verifying,
	/// In memory pool
	InMemory,
	/// In storage
	Stored,
}

/// Synchronization chain information
pub struct Information {
	/// Number of blocks hashes currently scheduled for requesting
	pub scheduled: u32,
	/// Number of blocks hashes currently requested from peers
	pub requested: u32,
	/// Number of blocks currently verifying
	pub verifying: u32,
	/// Number of blocks in the storage
	pub stored: u32,
	/// Information on memory pool
	pub transactions: MemoryPoolInformation,
	/// Information on headers chain
	pub headers: BestHeadersInformation,
}

/// Result of intersecting chain && inventory
#[derive(Debug, PartialEq)]
pub enum HeadersIntersection {
	/// 3.2: No intersection with in-memory queue && no intersection with db
	NoKnownBlocks(usize),
	/// 2.3: Inventory has no new blocks && some of blocks in inventory are in in-memory queue
	InMemoryNoNewBlocks,
	/// 2.4.2: Inventory has new blocks && these blocks are right after chain' best block
	InMemoryMainNewBlocks(usize),
	/// 2.4.3: Inventory has new blocks && these blocks are forked from our chain' best block
	InMemoryForkNewBlocks(usize),
	/// 3.3: No intersection with in-memory queue && has intersection with db && all blocks are already stored in db
	DbAllBlocksKnown,
	/// 3.4: No intersection with in-memory queue && has intersection with db && some blocks are not yet stored in db
	DbForkNewBlocks(usize),
	/// Dead-end blocks are starting from given index
	DeadEnd(usize),
}

/// Blockchain from synchroniation point of view, consisting of:
/// 1) all blocks from the `storage` [oldest blocks]
/// 2) all blocks currently verifying by `verification_queue`
/// 3) all blocks currently requested from peers
/// 4) all blocks currently scheduled for requesting [newest blocks]
pub struct Chain {
	/// Genesis block hash (stored for optimizations)
	genesis_block_hash: H256,
	/// Best storage block (stored for optimizations)
	best_storage_block: db::BestBlock,
	/// Local blocks storage
	storage: db::SharedStore,
	/// In-memory queue of blocks hashes
	hash_chain: HashQueueChain,
	/// In-memory queue of blocks headers
	headers_chain: BestHeadersChain,
	/// Currently verifying transactions
	verifying_transactions: LinkedHashMap<H256, Transaction>,
	/// Transactions memory pool
	memory_pool: MemoryPool,
	/// Blocks that have been marked as dead-ends
	dead_end_blocks: HashSet<H256>,
}

impl BlockState {
	pub fn from_queue_index(queue_index: usize) -> BlockState {
		match queue_index {
			SCHEDULED_QUEUE => BlockState::Scheduled,
			REQUESTED_QUEUE => BlockState::Requested,
			VERIFYING_QUEUE => BlockState::Verifying,
			_ => panic!("Unsupported queue_index: {}", queue_index),
		}
	}

	pub fn to_queue_index(&self) -> usize {
		match *self {
			BlockState::Scheduled => SCHEDULED_QUEUE,
			BlockState::Requested => REQUESTED_QUEUE,
			BlockState::Verifying => VERIFYING_QUEUE,
			_ => panic!("Unsupported queue: {:?}", self),
		}
	}
}

impl Chain {
	/// Create new `Chain` with given storage
	pub fn new(storage: db::SharedStore) -> Self {
		// we only work with storages with genesis block
		let genesis_block_hash = storage.block_hash(0)
			.expect("storage with genesis block is required");
		let best_storage_block = storage.best_block()
			.expect("non-empty storage is required");
		let best_storage_block_hash = best_storage_block.hash.clone();

		Chain {
			genesis_block_hash: genesis_block_hash,
			best_storage_block: best_storage_block,
			storage: storage,
			hash_chain: HashQueueChain::with_number_of_queues(NUMBER_OF_QUEUES),
			headers_chain: BestHeadersChain::new(best_storage_block_hash),
			verifying_transactions: LinkedHashMap::new(),
			memory_pool: MemoryPool::new(),
			dead_end_blocks: HashSet::new(),
		}
	}

	/// Get information on current blockchain state
	pub fn information(&self) -> Information {
		Information {
			scheduled: self.hash_chain.len_of(SCHEDULED_QUEUE),
			requested: self.hash_chain.len_of(REQUESTED_QUEUE),
			verifying: self.hash_chain.len_of(VERIFYING_QUEUE),
			stored: self.best_storage_block.number + 1,
			transactions: self.memory_pool.information(),
			headers: self.headers_chain.information(),
		}
	}

	/// Get storage
	pub fn storage(&self) -> db::SharedStore {
		self.storage.clone()
	}

	/// Get memory pool
	pub fn memory_pool(&self) -> &MemoryPool {
		&self.memory_pool
	}

	/// Get number of blocks in given state
	pub fn length_of_blocks_state(&self, state: BlockState) -> u32 {
		match state {