pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-12 08:51:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

DB-based light client backend (#250)

Browse Source 
* use db in light clients

* fixed comment

* fixed grumbles
This commit is contained in:
Svyatoslav Nikolsky
2018-06-29 17:10:10 +03:00
committed by Gav Wood
parent 159ff27a5d
commit 1aa7cb680d
23 changed files with 1233 additions and 624 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/polkadot/service/src/components.rs
+9 -5
View File
@@ -116,15 +116,19 @@ impl Components for FullComponents {
pub struct LightComponents;
impl Components for LightComponents {
type Backend = client::light::Backend<Block>;
type Backend = client::light::backend::Backend<client_db::light::LightStorage<Block>, network::OnDemand<Block, network::Service<Block>>>;
type Api = polkadot_api::light::RemotePolkadotApiWrapper<Self::Backend, Self::Executor>;
type Executor = client::RemoteCallExecutor<client::light::Backend<Block>, network::OnDemand<Block, network::Service<Block>>>;
type Executor = client::light::call_executor::RemoteCallExecutor<
client::light::blockchain::Blockchain<client_db::light::LightStorage<Block>, network::OnDemand<Block, network::Service<Block>>>,
network::OnDemand<Block, network::Service<Block>>>;
fn build_client(&self, _settings: client_db::DatabaseSettings, executor: CodeExecutor, genesis_storage: MakeStorage)
fn build_client(&self, db_settings: client_db::DatabaseSettings, executor: CodeExecutor, genesis_storage: MakeStorage)
-> Result<(Arc<client::Client<Self::Backend, Self::Executor, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error> {
let client_backend = client::light::new_light_backend();
let fetch_checker = Arc::new(client::light::new_fetch_checker(client_backend.clone(), executor));
let db_storage = client_db::light::LightStorage::new(db_settings)?;
let light_blockchain = client::light::new_light_blockchain(db_storage);
let fetch_checker = Arc::new(client::light::new_fetch_checker(light_blockchain.clone(), executor));
let fetcher = Arc::new(network::OnDemand::new(fetch_checker));
let client_backend = client::light::new_light_backend(light_blockchain, fetcher.clone());
let client = client::light::new_light(client_backend, fetcher.clone(), genesis_storage)?;
Ok((Arc::new(client), Some(fetcher)))
}
substrate/polkadot/service/src/lib.rs
+1 -1
View File
@@ -136,7 +136,7 @@ impl<Components> Service<Components>
},
network_config: config.network,
chain: client.clone(),
on_demand: on_demand.clone().map(|d| d as Arc<network::OnDemandService>),
on_demand: on_demand.clone().map(|d| d as Arc<network::OnDemandService<Block>>),
transaction_pool: transaction_pool_adapter,
};
let network = network::Service::new(network_params)?;
substrate/substrate/client/db/src/lib.rs
+42 -119
View File
@@ -35,11 +35,14 @@ extern crate log;
#[cfg(test)]
extern crate kvdb_memorydb;
pub mod light;
mod utils;
use std::sync::Arc;
use std::path::PathBuf;
use codec::Slicable;
use kvdb_rocksdb::{Database, DatabaseConfig};
use kvdb::{KeyValueDB, DBTransaction};
use memorydb::MemoryDB;
use parking_lot::RwLock;
@@ -50,6 +53,7 @@ use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, As, Hashing
use runtime_primitives::BuildStorage;
use state_machine::backend::Backend as StateBackend;
use state_machine::{CodeExecutor, TrieH256, DBValue};
use utils::{Meta, db_err, meta_keys, number_to_db_key, open_database, read_db, read_id, read_meta};
use state_db::StateDb;
pub use state_db::PruningMode;
@@ -94,11 +98,6 @@ mod columns {
pub const HEADER: Option<u32> = Some(4);
pub const BODY: Option<u32> = Some(5);
pub const JUSTIFICATION: Option<u32> = Some(6);
pub const NUM_COLUMNS: u32 = 7;
}
mod meta {
pub const BEST_BLOCK: &[u8; 4] = b"best";
}
struct PendingBlock<Block: BlockT> {
@@ -108,37 +107,6 @@ struct PendingBlock<Block: BlockT> {
is_best: bool,
}
#[derive(Clone)]
struct Meta<N, H> {
best_hash: H,
best_number: N,
genesis_hash: H,
}
type BlockKey = [u8; 4];
// Little endian
fn number_to_db_key<N>(n: N) -> BlockKey where N: As<u32> {
let n: u32 = n.as_();
[
(n >> 24) as u8,
((n >> 16) & 0xff) as u8,
((n >> 8) & 0xff) as u8,
(n & 0xff) as u8
]
}
// Maps database error to client error
fn db_err(err: kvdb::Error) -> client::error::Error {
use std::error::Error;
match err.kind() {
&kvdb::ErrorKind::Io(ref err) => client::error::ErrorKind::Backend(err.description().into()).into(),
&kvdb::ErrorKind::Msg(ref m) => client::error::ErrorKind::Backend(m.clone()).into(),
_ => client::error::ErrorKind::Backend("Unknown backend error".into()).into(),
}
}
// wrapper that implements trait required for state_db
struct StateMetaDb<'a>(&'a KeyValueDB);
@@ -157,59 +125,14 @@ pub struct BlockchainDb<Block: BlockT> {
}
impl<Block: BlockT> BlockchainDb<Block> where <Block::Header as HeaderT>::Number: As<u32> {
fn id(&self, id: BlockId<Block>) -> Result<Option<BlockKey>, client::error::Error> {
match id {
BlockId::Hash(h) => {
{
let meta = self.meta.read();
if meta.best_hash == h {
return Ok(Some(number_to_db_key(meta.best_number)));
}
}
self.db.get(columns::BLOCK_INDEX, h.as_ref()).map(|v| v.map(|v| {
let mut key: [u8; 4] = [0; 4];
key.copy_from_slice(&v);
key
})).map_err(db_err)
},
BlockId::Number(n) => Ok(Some(number_to_db_key(n))),
}
}
fn new(db: Arc<KeyValueDB>) -> Result<Self, client::error::Error> {
let (best_hash, best_number) = if let Some(Some(header)) = db.get(columns::META, meta::BEST_BLOCK).and_then(|id|
match id {
Some(id) => db.get(columns::HEADER, &id).map(|h| h.map(|b| Block::Header::decode(&mut &b[..]))),
None => Ok(None),
}).map_err(db_err)?
{
let hash = header.hash();
debug!("DB Opened blockchain db, best {:?} ({})", hash, header.number());
(hash, header.number().clone())
} else {
(Default::default(), Zero::zero())
};
let genesis_hash = db.get(columns::HEADER, &number_to_db_key(<Block::Header as HeaderT>::Number::zero())).map_err(db_err)?
.map(|b| HashingFor::<Block>::hash(&b)).unwrap_or_default().into();
let meta = read_meta::<Block>(&*db, columns::HEADER)?;
Ok(BlockchainDb {
db,
meta: RwLock::new(Meta {
best_hash,
best_number,
genesis_hash,
})
meta: RwLock::new(meta)
})
}
fn read_db(&self, id: BlockId<Block>, column: Option<u32>) -> Result<Option<DBValue>, client::error::Error> {
self.id(id).and_then(|key|
match key {
Some(key) => self.db.get(column, &key).map_err(db_err),
None => Ok(None),
})
}
fn update_meta(&self, hash: Block::Hash, number: <Block::Header as HeaderT>::Number, is_best: bool) {
if is_best {
let mut meta = self.meta.write();
@@ -222,9 +145,9 @@ impl<Block: BlockT> BlockchainDb<Block> where <Block::Header as HeaderT>::Number
}
}
impl<Block: BlockT> client::blockchain::Backend<Block> for BlockchainDb<Block> where <Block::Header as HeaderT>::Number: As<u32> {
impl<Block: BlockT> client::blockchain::HeaderBackend<Block> for BlockchainDb<Block> where <Block::Header as HeaderT>::Number: As<u32> {
fn header(&self, id: BlockId<Block>) -> Result<Option<Block::Header>, client::error::Error> {
match self.read_db(id, columns::HEADER)? {
match read_db(&*self.db, columns::BLOCK_INDEX, columns::HEADER, id)? {
Some(header) => match Block::Header::decode(&mut &header[..]) {
Some(header) => Ok(Some(header)),
None => return Err(client::error::ErrorKind::Backend("Error decoding header".into()).into()),
@@ -233,26 +156,6 @@ impl<Block: BlockT> client::blockchain::Backend<Block> for BlockchainDb<Block> w
}
}
fn body(&self, id: BlockId<Block>) -> Result<Option<Vec<Block::Extrinsic>>, client::error::Error> {
match self.read_db(id, columns::BODY)? {
Some(body) => match Slicable::decode(&mut &body[..]) {
Some(body) => Ok(Some(body)),
None => return Err(client::error::ErrorKind::Backend("Error decoding body".into()).into()),
}
None => Ok(None),
}
}
fn justification(&self, id: BlockId<Block>) -> Result<Option<Justification<Block::Hash>>, client::error::Error> {
match self.read_db(id, columns::JUSTIFICATION)? {
Some(justification) => match Slicable::decode(&mut &justification[..]) {
Some(justification) => Ok(Some(justification)),
None => return Err(client::error::ErrorKind::Backend("Error decoding justification".into()).into()),
}
None => Ok(None),
}
}
fn info(&self) -> Result<client::blockchain::Info<Block>, client::error::Error> {
let meta = self.meta.read();
Ok(client::blockchain::Info {
@@ -264,7 +167,7 @@ impl<Block: BlockT> client::blockchain::Backend<Block> for BlockchainDb<Block> w
fn status(&self, id: BlockId<Block>) -> Result<client::blockchain::BlockStatus, client::error::Error> {
let exists = match id {
BlockId::Hash(_) => self.id(id)?.is_some(),
BlockId::Hash(_) => read_id(&*self.db, columns::BLOCK_INDEX, id)?.is_some(),
BlockId::Number(n) => n <= self.meta.read().best_number,
};
match exists {
@@ -274,12 +177,34 @@ impl<Block: BlockT> client::blockchain::Backend<Block> for BlockchainDb<Block> w
}
fn hash(&self, number: <Block::Header as HeaderT>::Number) -> Result<Option<Block::Hash>, client::error::Error> {
self.read_db(BlockId::Number(number), columns::HEADER).map(|x|
read_db::<Block>(&*self.db, columns::BLOCK_INDEX, columns::HEADER, BlockId::Number(number)).map(|x|
x.map(|raw| HashingFor::<Block>::hash(&raw[..])).map(Into::into)
)
}
}
impl<Block: BlockT> client::blockchain::Backend<Block> for BlockchainDb<Block> where <Block::Header as HeaderT>::Number: As<u32> {
fn body(&self, id: BlockId<Block>) -> Result<Option<Vec<Block::Extrinsic>>, client::error::Error> {
match read_db(&*self.db, columns::BLOCK_INDEX, columns::BODY, id)? {
Some(body) => match Slicable::decode(&mut &body[..]) {
Some(body) => Ok(Some(body)),
None => return Err(client::error::ErrorKind::Backend("Error decoding body".into()).into()),
}
None => Ok(None),
}
}
fn justification(&self, id: BlockId<Block>) -> Result<Option<Justification<Block::Hash>>, client::error::Error> {
match read_db(&*self.db, columns::BLOCK_INDEX, columns::JUSTIFICATION, id)? {
Some(justification) => match Slicable::decode(&mut &justification[..]) {
Some(justification) => Ok(Some(justification)),
None => return Err(client::error::ErrorKind::Backend("Error decoding justification".into()).into()),
}
None => Ok(None),
}
}
}
/// Database transaction
pub struct BlockImportOperation<Block: BlockT> {
old_state: DbState,
@@ -351,18 +276,16 @@ pub struct Backend<Block: BlockT> {
impl<Block: BlockT> Backend<Block> where <Block::Header as HeaderT>::Number: As<u32> {
/// Create a new instance of database backend.
pub fn new(config: DatabaseSettings, finalization_window: u64) -> Result<Self, client::error::Error> {
let mut db_config = DatabaseConfig::with_columns(Some(columns::NUM_COLUMNS));
db_config.memory_budget = config.cache_size;
db_config.wal = true;
let path = config.path.to_str().ok_or_else(|| client::error::ErrorKind::Backend("Invalid database path".into()))?;
let db = Arc::new(Database::open(&db_config, &path).map_err(db_err)?);
let db = open_database(&config, "full")?;
Backend::from_kvdb(db as Arc<_>, config.pruning, finalization_window)
}
#[cfg(test)]
fn new_test() -> Self {
let db = Arc::new(::kvdb_memorydb::create(columns::NUM_COLUMNS));
use utils::NUM_COLUMNS;
let db = Arc::new(::kvdb_memorydb::create(NUM_COLUMNS));
Backend::from_kvdb(db as Arc<_>, PruningMode::keep_blocks(0), 0).expect("failed to create test-db")
}
@@ -417,12 +340,12 @@ impl<Block: BlockT> client::backend::Backend<Block> for Backend<Block> where
}
fn commit_operation(&self, mut operation: Self::BlockImportOperation) -> Result<(), client::error::Error> {
use client::blockchain::Backend;
use client::blockchain::HeaderBackend;
let mut transaction = DBTransaction::new();
if let Some(pending_block) = operation.pending_block {
let hash = pending_block.header.hash();
let number = pending_block.header.number().clone();
let key = number_to_db_key(pending_block.header.number().clone());
let key = number_to_db_key(number.clone());
transaction.put(columns::HEADER, &key, &pending_block.header.encode());
if let Some(body) = pending_block.body {
transaction.put(columns::BODY, &key, &body.encode());
@@ -432,7 +355,7 @@ impl<Block: BlockT> client::backend::Backend<Block> for Backend<Block> where
}
transaction.put(columns::BLOCK_INDEX, hash.as_ref(), &key);
if pending_block.is_best {
transaction.put(columns::META, meta::BEST_BLOCK, &key);
transaction.put(columns::META, meta_keys::BEST_BLOCK, &key);
}
let mut changeset: state_db::ChangeSet<H256> = state_db::ChangeSet::default();
for (key, (val, rc)) in operation.updates.drain() {
@@ -472,7 +395,7 @@ impl<Block: BlockT> client::backend::Backend<Block> for Backend<Block> where
}
fn state_at(&self, block: BlockId<Block>) -> Result<Self::State, client::error::Error> {
use client::blockchain::Backend as BcBackend;
use client::blockchain::HeaderBackend as BcHeaderBackend;
// special case for genesis initialization
match block {
@@ -499,7 +422,7 @@ mod tests {
use super::*;
use client::backend::Backend as BTrait;
use client::backend::BlockImportOperation as Op;
use client::blockchain::Backend as BCTrait;
use client::blockchain::HeaderBackend as BlockchainHeaderBackend;
use runtime_primitives::testing::{Header, Block as RawBlock};
type Block = RawBlock<u64>;
substrate/substrate/client/db/src/light.rs
+250
View File
@@ -0,0 +1,250 @@
// Copyright 2017 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! RocksDB-based light client blockchain storage.
use std::sync::Arc;
use parking_lot::RwLock;
use kvdb::{KeyValueDB, DBTransaction};
use client::blockchain::{BlockStatus, HeaderBackend as BlockchainHeaderBackend,
Info as BlockchainInfo};
use client::error::{ErrorKind as ClientErrorKind, Result as ClientResult};
use client::light::blockchain::Storage as LightBlockchainStorage;
use codec::Slicable;
use primitives::AuthorityId;
use runtime_primitives::generic::BlockId;
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, As, Hashing, HashingFor, Zero};
use utils::{meta_keys, Meta, db_err, number_to_db_key, open_database, read_db, read_id, read_meta};
use DatabaseSettings;
pub(crate) mod columns {
pub const META: Option<u32> = ::utils::COLUMN_META;
pub const BLOCK_INDEX: Option<u32> = Some(1);
pub const HEADER: Option<u32> = Some(2);
}
/// Light blockchain storage. Stores most recent headers + CHTs for older headers.
pub struct LightStorage<Block: BlockT> {
db: Arc<KeyValueDB>,
meta: RwLock<Meta<<<Block as BlockT>::Header as HeaderT>::Number, Block::Hash>>,
}
#[derive(Clone, PartialEq, Debug)]
struct BestAuthorities<N> {
/// first block, when this set became actual
valid_from: N,
/// None means that we do not know the set starting from `valid_from` block
authorities: Option<Vec<AuthorityId>>,
}
impl<Block> LightStorage<Block>
where
Block: BlockT,
<<Block as BlockT>::Header as HeaderT>::Number: As<u32>,
{
/// Create new storage with given settings.
pub fn new(config: DatabaseSettings) -> ClientResult<Self> {
let db = open_database(&config, "light")?;
Self::from_kvdb(db as Arc<_>)
}
#[cfg(test)]
pub(crate) fn new_test() -> Self {
use utils::NUM_COLUMNS;
let db = Arc::new(::kvdb_memorydb::create(NUM_COLUMNS));
Self::from_kvdb(db as Arc<_>).expect("failed to create test-db")
}
fn from_kvdb(db: Arc<KeyValueDB>) -> ClientResult<Self> {
let meta = RwLock::new(read_meta::<Block>(&*db, columns::HEADER)?);
Ok(LightStorage {
db,
meta,
})
}
fn update_meta(&self, hash: Block::Hash, number: <<Block as BlockT>::Header as HeaderT>::Number, is_best: bool) {
if is_best {
let mut meta = self.meta.write();
if number == <<Block as BlockT>::Header as HeaderT>::Number::zero() {
meta.genesis_hash = hash;
}
meta.best_number = number;
meta.best_hash = hash;
}
}
}
impl<Block> BlockchainHeaderBackend<Block> for LightStorage<Block>
where
Block: BlockT,
<<Block as BlockT>::Header as HeaderT>::Number: As<u32>,
{
fn header(&self, id: BlockId<Block>) -> ClientResult<Option<Block::Header>> {
match read_db(&*self.db, columns::BLOCK_INDEX, columns::HEADER, id)? {
Some(header) => match Block::Header::decode(&mut &header[..]) {
Some(header) => Ok(Some(header)),
None => return Err(ClientErrorKind::Backend("Error decoding header".into()).into()),
}
None => Ok(None),
}
}
fn info(&self) -> ClientResult<BlockchainInfo<Block>> {
let meta = self.meta.read();
Ok(BlockchainInfo {
best_hash: meta.best_hash,
best_number: meta.best_number,
genesis_hash: meta.genesis_hash,
})
}
fn status(&self, id: BlockId<Block>) -> ClientResult<BlockStatus> {
let exists = match id {
BlockId::Hash(_) => read_id(&*self.db, columns::BLOCK_INDEX, id)?.is_some(),
BlockId::Number(n) => n <= self.meta.read().best_number,
};
match exists {
true => Ok(BlockStatus::InChain),
false => Ok(BlockStatus::Unknown),
}
}
fn hash(&self, number: <<Block as BlockT>::Header as HeaderT>::Number) -> ClientResult<Option<Block::Hash>> {
read_db::<Block>(&*self.db, columns::BLOCK_INDEX, columns::HEADER, BlockId::Number(number)).map(|x|
x.map(|raw| HashingFor::<Block>::hash(&raw[..])).map(Into::into)
)
}
}
impl<Block> LightBlockchainStorage<Block> for LightStorage<Block>
where
Block: BlockT,
<<Block as BlockT>::Header as HeaderT>::Number: As<u32>,
<Block as BlockT>::Hash: From<[u8; 32]> + Into<[u8; 32]>,
{
fn import_header(&self, is_new_best: bool, header: Block::Header) -> ClientResult<()> {
let mut transaction = DBTransaction::new();
let hash = header.hash();
let number = *header.number();
let key = number_to_db_key(number);
transaction.put(columns::HEADER, &key, &header.encode());
transaction.put(columns::BLOCK_INDEX, hash.as_ref(), &key);
if is_new_best {
transaction.put(columns::META, meta_keys::BEST_BLOCK, &key);
}
debug!("Light DB Commit {:?} ({})", hash, number);
self.db.write(transaction).map_err(db_err)?;
self.update_meta(hash, number, is_new_best);
Ok(())
}
}
#[cfg(test)]
pub(crate) mod tests {
use runtime_primitives::testing::{H256 as Hash, Header, Block as RawBlock};
use super::*;
type Block = RawBlock<u32>;
pub fn insert_block(db: &LightStorage<Block>, parent: &Hash, number: u32) -> Hash {
let header = Header {
number: number.into(),
parent_hash: *parent,
state_root: Default::default(),
digest: Default::default(),
extrinsics_root: Default::default(),
};
let hash = header.hash();
db.import_header(true, header).unwrap();
hash
}
#[test]
fn returns_known_header() {
let db = LightStorage::new_test();
let known_hash = insert_block(&db, &Default::default(), 0);
let header_by_hash = db.header(BlockId::Hash(known_hash)).unwrap().unwrap();
let header_by_number = db.header(BlockId::Number(0)).unwrap().unwrap();
assert_eq!(header_by_hash, header_by_number);
}
#[test]
fn does_not_return_unknown_header() {
let db = LightStorage::<Block>::new_test();
assert!(db.header(BlockId::Hash(1.into())).unwrap().is_none());
assert!(db.header(BlockId::Number(0)).unwrap().is_none());
}
#[test]
fn returns_info() {
let db = LightStorage::new_test();
let genesis_hash = insert_block(&db, &Default::default(), 0);
let info = db.info().unwrap();
assert_eq!(info.best_hash, genesis_hash);
assert_eq!(info.best_number, 0);
assert_eq!(info.genesis_hash, genesis_hash);
let best_hash = insert_block(&db, &genesis_hash, 1);
let info = db.info().unwrap();
assert_eq!(info.best_hash, best_hash);
assert_eq!(info.best_number, 1);
assert_eq!(info.genesis_hash, genesis_hash);
}
#[test]
fn returns_block_status() {
let db = LightStorage::new_test();
let genesis_hash = insert_block(&db, &Default::default(), 0);
assert_eq!(db.status(BlockId::Hash(genesis_hash)).unwrap(), BlockStatus::InChain);
assert_eq!(db.status(BlockId::Number(0)).unwrap(), BlockStatus::InChain);
assert_eq!(db.status(BlockId::Hash(1.into())).unwrap(), BlockStatus::Unknown);
assert_eq!(db.status(BlockId::Number(1)).unwrap(), BlockStatus::Unknown);
}
#[test]
fn returns_block_hash() {
let db = LightStorage::new_test();
let genesis_hash = insert_block(&db, &Default::default(), 0);
assert_eq!(db.hash(0).unwrap(), Some(genesis_hash));
assert_eq!(db.hash(1).unwrap(), None);
}
#[test]
fn import_header_works() {
let db = LightStorage::new_test();
let genesis_hash = insert_block(&db, &Default::default(), 0);
assert_eq!(db.db.iter(columns::HEADER).count(), 1);
assert_eq!(db.db.iter(columns::BLOCK_INDEX).count(), 1);
let _ = insert_block(&db, &genesis_hash, 1);
assert_eq!(db.db.iter(columns::HEADER).count(), 2);
assert_eq!(db.db.iter(columns::BLOCK_INDEX).count(), 2);
}
}
substrate/substrate/client/db/src/utils.rs
+167
View File
@@ -0,0 +1,167 @@
// Copyright 2017 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Db-based backend utility structures and functions, used by both
//! full and light storages.
use std::sync::Arc;
use kvdb::{self, KeyValueDB, DBTransaction};
use kvdb_rocksdb::{Database, DatabaseConfig};
use client;
use codec::Slicable;
use hashdb::DBValue;
use runtime_primitives::generic::BlockId;
use runtime_primitives::traits::{As, Block as BlockT, Header as HeaderT, Hashing, HashingFor, Zero};
use DatabaseSettings;
/// Number of columns in the db. Must be the same for both full && light dbs.
/// Otherwise RocksDb will fail to open database && check its type.
pub const NUM_COLUMNS: u32 = 7;
/// Meta column. Thes set of keys in the column is shared by full && light storages.
pub const COLUMN_META: Option<u32> = Some(0);
/// Keys of entries in COLUMN_META.
pub mod meta_keys {
/// Type of storage (full or light).
pub const TYPE: &[u8; 4] = b"type";
/// Best block key.
pub const BEST_BLOCK: &[u8; 4] = b"best";
}
/// Database metadata.
pub struct Meta<N, H> {
/// Hash of the best known block.
pub best_hash: H,
/// Number of the best known block.
pub best_number: N,
/// Hash of the genesis block.
pub genesis_hash: H,
}
/// Type of block key in the database (LE block number).
pub type BlockKey = [u8; 4];
/// Convert block number into key (LE representation).
pub fn number_to_db_key<N>(n: N) -> BlockKey where N: As<u32> {
let n: u32 = n.as_();
[
(n >> 24) as u8,
((n >> 16) & 0xff) as u8,
((n >> 8) & 0xff) as u8,
(n & 0xff) as u8
]
}
/// Maps database error to client error
pub fn db_err(err: kvdb::Error) -> client::error::Error {
use std::error::Error;
match err.kind() {
&kvdb::ErrorKind::Io(ref err) => client::error::ErrorKind::Backend(err.description().into()).into(),
&kvdb::ErrorKind::Msg(ref m) => client::error::ErrorKind::Backend(m.clone()).into(),
_ => client::error::ErrorKind::Backend("Unknown backend error".into()).into(),
}
}
/// Open RocksDB database.
pub fn open_database(config: &DatabaseSettings, db_type: &str) -> client::error::Result<Arc<KeyValueDB>> {
let mut db_config = DatabaseConfig::with_columns(Some(NUM_COLUMNS));
db_config.memory_budget = config.cache_size;
db_config.wal = true;
let path = config.path.to_str().ok_or_else(|| client::error::ErrorKind::Backend("Invalid database path".into()))?;
let db = Database::open(&db_config, &path).map_err(db_err)?;
// check database type
match db.get(COLUMN_META, meta_keys::TYPE).map_err(db_err)? {
Some(stored_type) => {
if db_type.as_bytes() != &*stored_type {
return Err(client::error::ErrorKind::Backend(
format!("Unexpected database type. Expected: {}", db_type)).into());
}
},
None => {
let mut transaction = DBTransaction::new();
transaction.put(COLUMN_META, meta_keys::TYPE, db_type.as_bytes());
db.write(transaction).map_err(db_err)?;
},
}
Ok(Arc::new(db))
}
/// Convert block id to block key, reading number from db if required.
pub fn read_id<Block>(db: &KeyValueDB, col_index: Option<u32>, id: BlockId<Block>) -> Result<Option<BlockKey>, client::error::Error>
where
Block: BlockT,
<<Block as BlockT>::Header as HeaderT>::Number: As<u32>,
{
match id {
BlockId::Hash(h) => db.get(col_index, h.as_ref())
.map(|v| v.map(|v| {
let mut key: [u8; 4] = [0; 4];
key.copy_from_slice(&v);
key
})).map_err(db_err),
BlockId::Number(n) => Ok(Some(number_to_db_key(n))),
}
}
/// Read database column entry for the given block.
pub fn read_db<Block>(db: &KeyValueDB, col_index: Option<u32>, col: Option<u32>, id: BlockId<Block>) -> client::error::Result<Option<DBValue>>
where
Block: BlockT,
<<Block as BlockT>::Header as HeaderT>::Number: As<u32>,
{
read_id(db, col_index, id).and_then(|key| match key {
Some(key) => db.get(col, &key).map_err(db_err),
None => Ok(None),
})
}
/// Read meta from the database.
pub fn read_meta<Block>(db: &KeyValueDB, col_header: Option<u32>) -> Result<Meta<<<Block as BlockT>::Header as HeaderT>::Number, Block::Hash>, client::error::Error>
where
Block: BlockT,
<<Block as BlockT>::Header as HeaderT>::Number: As<u32>,
{
let genesis_number = <<Block as BlockT>::Header as HeaderT>::Number::zero();
let (best_hash, best_number) = if let Some(Some(header)) = db.get(COLUMN_META, meta_keys::BEST_BLOCK).and_then(|id|
match id {
Some(id) => db.get(col_header, &id).map(|h| h.map(|b| Block::Header::decode(&mut &b[..]))),
None => Ok(None),
}).map_err(db_err)?
{
let hash = header.hash();
debug!("DB Opened blockchain db, best {:?} ({})", hash, header.number());
(hash, *header.number())
} else {
(Default::default(), genesis_number)
};
let genesis_hash = db.get(col_header, &number_to_db_key(genesis_number))
.map_err(db_err)?
.map(|raw| HashingFor::<Block>::hash(&raw[..]))
.unwrap_or_default()
.into();
Ok(Meta {
best_hash,
best_number,
genesis_hash,
})
}
substrate/substrate/client/src/blockchain.rs
+10 -6
View File
@@ -22,14 +22,10 @@ use runtime_primitives::bft::Justification;
use error::Result;
/// Blockchain database backend. Does not perform any validation.
pub trait Backend<Block: BlockT>: Send + Sync {
/// Blockchain database header backend. Does not perform any validation.
pub trait HeaderBackend<Block: BlockT>: Send + Sync {
/// Get block header. Returns `None` if block is not found.
fn header(&self, id: BlockId<Block>) -> Result<Option<<Block as BlockT>::Header>>;
/// Get block body. Returns `None` if block is not found.
fn body(&self, id: BlockId<Block>) -> Result<Option<Vec<<Block as BlockT>::Extrinsic>>>;
/// Get block justification. Returns `None` if justification does not exist.
fn justification(&self, id: BlockId<Block>) -> Result<Option<Justification<Block::Hash>>>;
/// Get blockchain info.
fn info(&self) -> Result<Info<Block>>;
/// Get block status.
@@ -38,6 +34,14 @@ pub trait Backend<Block: BlockT>: Send + Sync {
fn hash(&self, number: <<Block as BlockT>::Header as HeaderT>::Number) -> Result<Option<<<Block as BlockT>::Header as HeaderT>::Hash>>;
}
/// Blockchain database backend. Does not perform any validation.
pub trait Backend<Block: BlockT>: HeaderBackend<Block> {
/// Get block body. Returns `None` if block is not found.
fn body(&self, id: BlockId<Block>) -> Result<Option<Vec<<Block as BlockT>::Extrinsic>>>;
/// Get block justification. Returns `None` if justification does not exist.
fn justification(&self, id: BlockId<Block>) -> Result<Option<Justification<Block::Hash>>>;
}
/// Block import outcome
pub enum ImportResult<E> {
/// Imported successfully.
substrate/substrate/client/src/call_executor.rs
+4 -115
View File
@@ -15,18 +15,15 @@
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
use std::sync::Arc;
use futures::{IntoFuture, Future};
use runtime_primitives::generic::BlockId;
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT};
use runtime_primitives::traits::Block as BlockT;
use state_machine::{self, OverlayedChanges, Backend as StateBackend, CodeExecutor};
use backend;
use blockchain::Backend as ChainBackend;
use error;
use light::{Fetcher, RemoteCallRequest};
/// Information regarding the result of a call.
#[derive(Debug)]
#[derive(Debug, Clone)]
pub struct CallResult {
/// The data that was returned from the call.
pub return_data: Vec<u8>,
@@ -62,13 +59,6 @@ pub struct LocalCallExecutor<B, E> {
executor: E,
}
/// Call executor that executes methods on remote node, querying execution proof
/// and checking proof by re-executing locally.
pub struct RemoteCallExecutor<B, F> {
backend: Arc<B>,
fetcher: Arc<F>,
}
impl<B, E> LocalCallExecutor<B, E> {
/// Creates new instance of local call executor.
pub fn new(backend: Arc<B>, executor: E) -> Self {
@@ -111,7 +101,7 @@ impl<B, E, Block> CallExecutor<Block> for LocalCallExecutor<B, E>
}
fn prove_at_state<S: state_machine::Backend>(&self, state: S, changes: &mut OverlayedChanges, method: &str, call_data: &[u8]) -> Result<(Vec<u8>, Vec<Vec<u8>>), error::Error> {
state_machine::prove(
state_machine::prove_execution(
state,
changes,
&self.executor,
@@ -121,105 +111,4 @@ impl<B, E, Block> CallExecutor<Block> for LocalCallExecutor<B, E>
.map(|(result, proof, _)| (result, proof))
.map_err(Into::into)
}
}
impl<B, F> RemoteCallExecutor<B, F> {
/// Creates new instance of remote call executor.
pub fn new(backend: Arc<B>, fetcher: Arc<F>) -> Self {
RemoteCallExecutor { backend, fetcher }
}
}
impl<B, F, Block> CallExecutor<Block> for RemoteCallExecutor<B, F>
where
B: backend::RemoteBackend<Block>,
F: Fetcher<Block>,
Block: BlockT,
error::Error: From<<<B as backend::Backend<Block>>::State as StateBackend>::Error>,
{
type Error = error::Error;
fn call(&self, id: &BlockId<Block>, method: &str, call_data: &[u8]) -> error::Result<CallResult> {
let block_hash = match *id {
BlockId::Hash(hash) => hash,
BlockId::Number(number) => self.backend.blockchain().hash(number)?
.ok_or_else(|| error::ErrorKind::UnknownBlock(format!("{}", number)))?,
};
self.fetcher.remote_call(RemoteCallRequest {
block: block_hash,
method: method.into(),
call_data: call_data.to_vec(),
}).into_future().wait()
}
fn call_at_state<S: state_machine::Backend>(&self, _state: &S, _changes: &mut OverlayedChanges, _method: &str, _call_data: &[u8]) -> error::Result<(Vec<u8>, S::Transaction)> {
Err(error::ErrorKind::NotAvailableOnLightClient.into())
}
fn prove_at_state<S: state_machine::Backend>(&self, _state: S, _changes: &mut OverlayedChanges, _method: &str, _call_data: &[u8]) -> Result<(Vec<u8>, Vec<Vec<u8>>), error::Error> {
Err(error::ErrorKind::NotAvailableOnLightClient.into())
}
}
/// Check remote execution proof using given backend.
pub fn check_execution_proof<B, E, Block>(backend: &B, executor: &E, request: &RemoteCallRequest<Block::Hash>, remote_proof: Vec<Vec<u8>>) -> Result<CallResult, error::Error>
where
B: backend::RemoteBackend<Block>,
E: CodeExecutor,
Block: BlockT,
<<Block as BlockT>::Header as HeaderT>::Hash: Into<[u8; 32]>, // TODO: remove when patricia_trie generic.
error::Error: From<<<B as backend::Backend<Block>>::State as StateBackend>::Error>,
{
let local_header = backend.blockchain().header(BlockId::Hash(request.block))?;
let local_header = local_header.ok_or_else(|| error::ErrorKind::UnknownBlock(format!("{}", request.block)))?;
let local_state_root = local_header.state_root().clone();
do_check_execution_proof(local_state_root, executor, request, remote_proof)
}
/// Check remote execution proof using given state root.
fn do_check_execution_proof<E, H>(local_state_root: H, executor: &E, request: &RemoteCallRequest<H>, remote_proof: Vec<Vec<u8>>) -> Result<CallResult, error::Error>
where
E: CodeExecutor,
H: Into<[u8; 32]>, // TODO: remove when patricia_trie generic.
{
let mut changes = OverlayedChanges::default();
let (local_result, _) = state_machine::proof_check(
local_state_root.into(),
remote_proof,
&mut changes,
executor,
&request.method,
&request.call_data)?;
Ok(CallResult { return_data: local_result, changes })
}
#[cfg(test)]
mod tests {
use runtime_primitives::generic::BlockId;
use state_machine::Backend;
use test_client;
use light::RemoteCallRequest;
use super::do_check_execution_proof;
#[test]
fn execution_proof_is_generated_and_checked() {
// prepare remote client
let remote_client = test_client::new();
let remote_block_id = BlockId::Number(0);
let remote_block_storage_root = remote_client.state_at(&remote_block_id)
.unwrap().storage_root(::std::iter::empty()).0;
// 'fetch' execution proof from remote node
let remote_execution_proof = remote_client.execution_proof(&remote_block_id, "authorities", &[]).unwrap().1;
// check remote execution proof locally
let local_executor = test_client::NativeExecutor::new();
do_check_execution_proof(remote_block_storage_root, &local_executor, &RemoteCallRequest {
block: Default::default(),
method: "authorities".into(),
call_data: vec![],
}, remote_execution_proof).unwrap();
}
}
}
substrate/substrate/client/src/client.rs
+1 -1
View File
@@ -28,7 +28,7 @@ use codec::Slicable;
use state_machine::{self, Ext, OverlayedChanges, Backend as StateBackend, CodeExecutor};
use backend::{self, BlockImportOperation};
use blockchain::{self, Info as ChainInfo, Backend as ChainBackend};
use blockchain::{self, Info as ChainInfo, Backend as ChainBackend, HeaderBackend as ChainHeaderBackend};
use call_executor::{CallExecutor, LocalCallExecutor};
use {error, in_mem, block_builder, runtime_io, bft, genesis};
substrate/substrate/client/src/error.rs
+8 -2
View File
@@ -88,17 +88,23 @@ error_chain! {
display("This method is not currently available when running in light client mode"),
}
/// Invalid remote proof.
/// Invalid remote execution proof.
InvalidExecutionProof {
description("invalid execution proof"),
display("Remote node has responded with invalid execution proof"),
}
/// Invalid remote proof.
/// Remote fetch has been cancelled.
RemoteFetchCancelled {
description("remote fetch cancelled"),
display("Remote data fetch has been cancelled"),
}
/// Remote fetch has been failed.
RemoteFetchFailed {
description("remote fetch failed"),
display("Remote data fetch has been failed"),
}
}
}
substrate/substrate/client/src/in_mem.rs
+39 -32
View File
@@ -17,9 +17,11 @@
//! In memory client backend
use std::collections::HashMap;
use std::sync::Arc;
use parking_lot::RwLock;
use error;
use backend;
use light;
use runtime_primitives::generic::BlockId;
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, Zero};
use runtime_primitives::bft::Justification;
@@ -86,16 +88,9 @@ struct BlockchainStorage<Block: BlockT> {
}
/// In-memory blockchain. Supports concurrent reads.
#[derive(Clone)]
pub struct Blockchain<Block: BlockT> {
storage: RwLock<BlockchainStorage<Block>>,
}
impl<Block: BlockT + Clone> Clone for Blockchain<Block> {
fn clone(&self) -> Self {
Blockchain {
storage: RwLock::new(self.storage.read().clone()),
}
}
storage: Arc<RwLock<BlockchainStorage<Block>>>,
}
impl<Block: BlockT> Blockchain<Block> {
@@ -108,16 +103,17 @@ impl<Block: BlockT> Blockchain<Block> {
}
/// Create new in-memory blockchain storage.
pub fn new() -> Self {
pub fn new() -> Blockchain<Block> {
let storage = Arc::new(RwLock::new(
BlockchainStorage {
blocks: HashMap::new(),
hashes: HashMap::new(),
best_hash: Default::default(),
best_number: Zero::zero(),
genesis_hash: Default::default(),
}));
Blockchain {
storage: RwLock::new(
BlockchainStorage {
blocks: HashMap::new(),
hashes: HashMap::new(),
best_hash: Default::default(),
best_number: Zero::zero(),
genesis_hash: Default::default(),
})
storage: storage,
}
}
@@ -159,26 +155,13 @@ impl<Block: BlockT> Blockchain<Block> {
}
}
impl<Block: BlockT> blockchain::Backend<Block> for Blockchain<Block> {
impl<Block: BlockT> blockchain::HeaderBackend<Block> for Blockchain<Block> {
fn header(&self, id: BlockId<Block>) -> error::Result<Option<<Block as BlockT>::Header>> {
Ok(self.id(id).and_then(|hash| {
self.storage.read().blocks.get(&hash).map(|b| b.header().clone())
}))
}
fn body(&self, id: BlockId<Block>) -> error::Result<Option<Vec<<Block as BlockT>::Extrinsic>>> {
Ok(self.id(id).and_then(|hash| {
self.storage.read().blocks.get(&hash)
.and_then(|b| b.extrinsics().map(|x| x.to_vec()))
}))
}
fn justification(&self, id: BlockId<Block>) -> error::Result<Option<Justification<Block::Hash>>> {
Ok(self.id(id).and_then(|hash| self.storage.read().blocks.get(&hash).and_then(|b|
b.justification().map(|x| x.clone()))
))
}
fn info(&self) -> error::Result<blockchain::Info<Block>> {
let storage = self.storage.read();
Ok(blockchain::Info {
@@ -200,6 +183,30 @@ impl<Block: BlockT> blockchain::Backend<Block> for Blockchain<Block> {
}
}
impl<Block: BlockT> blockchain::Backend<Block> for Blockchain<Block> {
fn body(&self, id: BlockId<Block>) -> error::Result<Option<Vec<<Block as BlockT>::Extrinsic>>> {
Ok(self.id(id).and_then(|hash| {
self.storage.read().blocks.get(&hash)
.and_then(|b| b.extrinsics().map(|x| x.to_vec()))
}))
}
fn justification(&self, id: BlockId<Block>) -> error::Result<Option<Justification<Block::Hash>>> {
Ok(self.id(id).and_then(|hash| self.storage.read().blocks.get(&hash).and_then(|b|
b.justification().map(|x| x.clone()))
))
}
}
impl<Block: BlockT> light::blockchain::Storage<Block> for Blockchain<Block> {
fn import_header(&self, is_new_best: bool, header: Block::Header) -> error::Result<()> {
let hash = header.hash();
self.insert(hash, header, None, None, is_new_best);
Ok(())
}
}
/// In-memory operation.
pub struct BlockImportOperation<Block: BlockT> {
pending_block: Option<PendingBlock<Block>>,
substrate/substrate/client/src/lib.rs
+1 -3
View File
@@ -56,6 +56,4 @@ pub use client::{
ImportResult,
};
pub use blockchain::Info as ChainInfo;
pub use call_executor::{
CallResult, CallExecutor, LocalCallExecutor, RemoteCallExecutor,
};
pub use call_executor::{CallResult, CallExecutor, LocalCallExecutor};
substrate/substrate/client/src/light.rs
-256
View File
@@ -1,256 +0,0 @@
// Copyright 2017 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Light client backend. Only stores headers and justifications of blocks.
//! Everything else is requested from full nodes on demand.
use std::sync::Arc;
use futures::future::IntoFuture;
use state_machine::{CodeExecutor, TryIntoTrieBackend as TryIntoStateTrieBackend,
TrieBackend as StateTrieBackend};
use state_machine::backend::Backend as StateBackend;
use runtime_primitives::generic::BlockId;
use runtime_primitives::bft::Justification;
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT};
use runtime_primitives::BuildStorage;
use blockchain::{self, BlockStatus};
use backend;
use call_executor::{CallResult, RemoteCallExecutor, check_execution_proof};
use client::Client;
use error;
use in_mem::Blockchain as InMemBlockchain;
/// Remote call request.
pub struct RemoteCallRequest<H> {
/// Call at state of block referenced by given header hash.
pub block: H,
/// Method to call.
pub method: String,
/// Call data.
pub call_data: Vec<u8>,
}
/// Light client data fetcher. Implementations of this trait must check if remote data
/// is correct (see FetchedDataChecker) and return already checked data.
pub trait Fetcher<B: BlockT>: Send + Sync {
/// Remote call result future.
type RemoteCallResult: IntoFuture<Item=CallResult, Error=error::Error>;
/// Fetch remote call result.
fn remote_call(&self, request: RemoteCallRequest<B::Hash>) -> Self::RemoteCallResult;
}
/// Light client remote data checker.
pub trait FetchChecker<B: BlockT>: Send + Sync {
/// Check remote method execution proof.
fn check_execution_proof(&self, request: &RemoteCallRequest<B::Hash>, remote_proof: Vec<Vec<u8>>) -> error::Result<CallResult>;
}
/// Light client backend.
pub struct Backend<B: BlockT> {
blockchain: Blockchain<B>,
}
/// Light client blockchain.
pub struct Blockchain<B: BlockT> {
storage: InMemBlockchain<B>,
}
/// Block (header and justification) import operation.
pub struct BlockImportOperation<B: BlockT> {
pending_block: Option<PendingBlock<B>>,
}
/// On-demand state.
#[derive(Clone)]
pub struct OnDemandState<H> {
/// Hash of the block, state is valid for.
_block: H,
}
/// Remote data checker.
pub struct LightDataChecker<E, B: BlockT> {
/// Backend reference.
backend: Arc<Backend<B>>,
/// Executor.
executor: E,
}
struct PendingBlock<B: BlockT> {
header: B::Header,
justification: Option<Justification<B::Hash>>,
is_best: bool,
}
impl<B: BlockT> backend::Backend<B> for Backend<B> {
type BlockImportOperation = BlockImportOperation<B>;
type Blockchain = Blockchain<B>;
type State = OnDemandState<B::Hash>;
fn begin_operation(&self, _block: BlockId<B>) -> error::Result<Self::BlockImportOperation> {
Ok(BlockImportOperation {
pending_block: None,
})
}
fn commit_operation(&self, operation: Self::BlockImportOperation) -> error::Result<()> {
if let Some(pending_block) = operation.pending_block {
let hash = pending_block.header.hash();
self.blockchain.storage.insert(hash, pending_block.header, pending_block.justification, None, pending_block.is_best);
}
Ok(())
}
fn blockchain(&self) -> &Blockchain<B> {
&self.blockchain
}
fn state_at(&self, block: BlockId<B>) -> error::Result<Self::State> {
Ok(OnDemandState {
_block: self.blockchain.storage.id(block).ok_or(error::ErrorKind::UnknownBlock(format!("{:?}", block)))?,
})
}
}
impl<B: BlockT> backend::RemoteBackend<B> for Backend<B> {}
impl<B: BlockT> backend::BlockImportOperation<B> for BlockImportOperation<B> {
type State = OnDemandState<B::Hash>;
fn state(&self) -> error::Result<Option<&Self::State>> {
// None means 'locally-stateless' backend
Ok(None)
}
fn set_block_data(&mut self, header: B::Header, _body: Option<Vec<B::Extrinsic>>, justification: Option<Justification<B::Hash>>, is_new_best: bool) -> error::Result<()> {
assert!(self.pending_block.is_none(), "Only one block per operation is allowed");
self.pending_block = Some(PendingBlock {
header,
justification,
is_best: is_new_best,
});
Ok(())
}
fn update_storage(&mut self, _update: <Self::State as StateBackend>::Transaction) -> error::Result<()> {
// we're not storing anything locally => ignore changes
Ok(())
}
fn reset_storage<I: Iterator<Item=(Vec<u8>, Vec<u8>)>>(&mut self, _iter: I) -> error::Result<()> {
// we're not storing anything locally => ignore changes
Ok(())
}
}
impl<B: BlockT> blockchain::Backend<B> for Blockchain<B> {
fn header(&self, id: BlockId<B>) -> error::Result<Option<B::Header>> {
self.storage.header(id)
}
fn body(&self, _id: BlockId<B>) -> error::Result<Option<Vec<B::Extrinsic>>> {
// TODO [light]: fetch from remote node
Ok(None)
}
fn justification(&self, id: BlockId<B>) -> error::Result<Option<Justification<B::Hash>>> {
self.storage.justification(id)
}
fn info(&self) -> error::Result<blockchain::Info<B>> {
self.storage.info()
}
fn status(&self, id: BlockId<B>) -> error::Result<BlockStatus> {
self.storage.status(id)
}
fn hash(&self, number: <B::Header as HeaderT>::Number) -> error::Result<Option<B::Hash>> {
self.storage.hash(number)
}
}
impl<H: Clone> StateBackend for OnDemandState<H> {
type Error = error::Error;
type Transaction = ();
fn storage(&self, _key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
// TODO [light]: fetch from remote node
Err(error::ErrorKind::NotAvailableOnLightClient.into())
}
fn storage_root<I>(&self, _delta: I) -> ([u8; 32], Self::Transaction)
where I: IntoIterator<Item=(Vec<u8>, Option<Vec<u8>>)>
{
([0; 32], ())
}
fn pairs(&self) -> Vec<(Vec<u8>, Vec<u8>)> {
// whole state is not available on light node
Vec::new()
}
}
impl<H> TryIntoStateTrieBackend for OnDemandState<H> {
fn try_into_trie_backend(self) -> Option<StateTrieBackend> {
None
}
}
impl<E, B> FetchChecker<B> for LightDataChecker<E, B>
where
E: CodeExecutor,
B: BlockT,
<<B as BlockT>::Header as HeaderT>::Hash: Into<[u8; 32]>, // TODO: remove when patricia_trie generic.
{
fn check_execution_proof(&self, request: &RemoteCallRequest<B::Hash>, remote_proof: Vec<Vec<u8>>) -> error::Result<CallResult> {
check_execution_proof(&*self.backend, &self.executor, request, remote_proof)
}
}
/// Create an instance of light client backend.
pub fn new_light_backend<B: BlockT>() -> Arc<Backend<B>> {
let storage = InMemBlockchain::new();
let blockchain = Blockchain { storage };
Arc::new(Backend { blockchain })
}
/// Create an instance of light client.
pub fn new_light<F, S, Block>(
backend: Arc<Backend<Block>>,
fetcher: Arc<F>,
genesis_storage: S,
) -> error::Result<Client<Backend<Block>, RemoteCallExecutor<Backend<Block>, F>, Block>>
where
F: Fetcher<Block>,
S: BuildStorage,
Block: BlockT,
{
let executor = RemoteCallExecutor::new(backend.clone(), fetcher);
Client::new(backend, executor, genesis_storage)
}
/// Create an instance of fetch data checker.
pub fn new_fetch_checker<E, Block>(
backend: Arc<Backend<Block>>,
executor: E,
) -> LightDataChecker<E, Block>
where
E: CodeExecutor,
Block: BlockT,
{
LightDataChecker { backend, executor }
}
substrate/substrate/client/src/light/backend.rs
+183
View File
@@ -0,0 +1,183 @@
// Copyright 2017 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Light client backend. Only stores headers and justifications of blocks.
//! Everything else is requested from full nodes on demand.
use std::sync::{Arc, Weak};
use runtime_primitives::{bft::Justification, generic::BlockId};
use runtime_primitives::traits::Block as BlockT;
use state_machine::{Backend as StateBackend, TrieBackend as StateTrieBackend,
TryIntoTrieBackend as TryIntoStateTrieBackend};
use backend::{Backend as ClientBackend, BlockImportOperation, RemoteBackend};
use blockchain::HeaderBackend as BlockchainHeaderBackend;
use error::{Error as ClientError, ErrorKind as ClientErrorKind, Result as ClientResult};
use light::blockchain::{Blockchain, Storage as BlockchainStorage};
use light::fetcher::Fetcher;
/// Light client backend.
pub struct Backend<S, F> {
blockchain: Arc<Blockchain<S, F>>,
}
/// Light block (header and justification) import operation.
pub struct ImportOperation<Block: BlockT, F> {
is_new_best: bool,
header: Option<Block::Header>,
_phantom: ::std::marker::PhantomData<F>,
}
/// On-demand state.
pub struct OnDemandState<Block: BlockT, F> {
fetcher: Weak<F>,
block: Block::Hash,
}
impl<S, F> Backend<S, F> {
/// Create new light backend.
pub fn new(blockchain: Arc<Blockchain<S, F>>) -> Self {
Self { blockchain }
}
/// Get shared blockchain reference.
pub fn blockchain(&self) -> &Arc<Blockchain<S, F>> {
&self.blockchain
}
}
impl<S, F, Block> ClientBackend<Block> for Backend<S, F> where Block: BlockT, S: BlockchainStorage<Block>, F: Fetcher<Block> {
type BlockImportOperation = ImportOperation<Block, F>;
type Blockchain = Blockchain<S, F>;
type State = OnDemandState<Block, F>;
fn begin_operation(&self, _block: BlockId<Block>) -> ClientResult<Self::BlockImportOperation> {
Ok(ImportOperation {
is_new_best: false,
header: None,
_phantom: Default::default(),
})
}
fn commit_operation(&self, operation: Self::BlockImportOperation) -> ClientResult<()> {
let header = operation.header.expect("commit is called after set_block_data; set_block_data sets header; qed");
self.blockchain.storage().import_header(operation.is_new_best, header)
}
fn blockchain(&self) -> &Blockchain<S, F> {
&self.blockchain
}
fn state_at(&self, block: BlockId<Block>) -> ClientResult<Self::State> {
let block_hash = match block {
BlockId::Hash(h) => Some(h),
BlockId::Number(n) => self.blockchain.hash(n).unwrap_or_default(),
};
Ok(OnDemandState {
block: block_hash.ok_or_else(|| ClientErrorKind::UnknownBlock(format!("{}", block)))?,
fetcher: self.blockchain.fetcher(),
})
}
}
impl<S, F, Block> RemoteBackend<Block> for Backend<S, F> where Block: BlockT, S: BlockchainStorage<Block>, F: Fetcher<Block> {}
impl<F, Block> BlockImportOperation<Block> for ImportOperation<Block, F> where Block: BlockT, F: Fetcher<Block> {
type State = OnDemandState<Block, F>;
fn state(&self) -> ClientResult<Option<&Self::State>> {
// None means 'locally-stateless' backend
Ok(None)
}
fn set_block_data(
&mut self,
header: Block::Header,
_body: Option<Vec<Block::Extrinsic>>,
_justification: Option<Justification<Block::Hash>>,
is_new_best: bool
) -> ClientResult<()> {
self.is_new_best = is_new_best;
self.header = Some(header);
Ok(())
}
fn update_storage(&mut self, _update: <Self::State as StateBackend>::Transaction) -> ClientResult<()> {
// we're not storing anything locally => ignore changes
Ok(())
}
fn reset_storage<I: Iterator<Item=(Vec<u8>, Vec<u8>)>>(&mut self, _iter: I) -> ClientResult<()> {
// we're not storing anything locally => ignore changes
Ok(())
}
}
impl<Block: BlockT, F> Clone for OnDemandState<Block, F> {
fn clone(&self) -> Self {
OnDemandState {
fetcher: self.fetcher.clone(),
block: self.block,
}
}
}
impl<Block, F> StateBackend for OnDemandState<Block, F> where Block: BlockT, F: Fetcher<Block> {
type Error = ClientError;
type Transaction = ();
fn storage(&self, _key: &[u8]) -> ClientResult<Option<Vec<u8>>> {
Err(ClientErrorKind::NotAvailableOnLightClient.into()) // TODO: fetch from remote node
}
fn storage_root<I>(&self, _delta: I) -> ([u8; 32], Self::Transaction)
where I: IntoIterator<Item=(Vec<u8>, Option<Vec<u8>>)> {
([0; 32], ())
}
fn pairs(&self) -> Vec<(Vec<u8>, Vec<u8>)> {
// whole state is not available on light node
Vec::new()
}
}
impl<Block, F> TryIntoStateTrieBackend for OnDemandState<Block, F> where Block: BlockT, F: Fetcher<Block> {
fn try_into_trie_backend(self) -> Option<StateTrieBackend> {
None
}
}
#[cfg(test)]
pub mod tests {
use futures::future::{ok, FutureResult};
use parking_lot::Mutex;
use call_executor::CallResult;
use error::Error as ClientError;
use test_client::runtime::{Hash, Block};
use light::fetcher::{Fetcher, RemoteCallRequest};
pub type OkCallFetcher = Mutex<CallResult>;
impl Fetcher<Block> for OkCallFetcher {
type RemoteCallResult = FutureResult<CallResult, ClientError>;
fn remote_call(&self, _request: RemoteCallRequest<Hash>) -> Self::RemoteCallResult {
ok((*self.lock()).clone())
}
}
}
substrate/substrate/client/src/light/blockchain.rs
+95
View File
@@ -0,0 +1,95 @@
// Copyright 2017 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Light client blockchin backend. Only stores headers and justifications of recent
//! blocks. CHT roots are stored for headers of ancient blocks.
use std::sync::Weak;
use parking_lot::Mutex;
use runtime_primitives::{bft::Justification, generic::BlockId};
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT};
use blockchain::{Backend as BlockchainBackend, BlockStatus,
HeaderBackend as BlockchainHeaderBackend, Info as BlockchainInfo};
use error::Result as ClientResult;
use light::fetcher::Fetcher;
/// Light client blockchain storage.
pub trait Storage<Block: BlockT>: BlockchainHeaderBackend<Block> {
/// Store new header.
fn import_header(&self, is_new_best: bool, header: Block::Header) -> ClientResult<()>;
}
/// Light client blockchain.
pub struct Blockchain<S, F> {
fetcher: Mutex<Weak<F>>,
storage: S,
}
impl<S, F> Blockchain<S, F> {
/// Create new light blockchain backed with given storage.
pub fn new(storage: S) -> Self {
Self {
fetcher: Mutex::new(Default::default()),
storage,
}
}
/// Sets fetcher reference.
pub fn set_fetcher(&self, fetcher: Weak<F>) {
*self.fetcher.lock() = fetcher;
}
/// Get fetcher weak reference.
pub fn fetcher(&self) -> Weak<F> {
self.fetcher.lock().clone()
}
/// Get storage reference.
pub fn storage(&self) -> &S {
&self.storage
}
}
impl<S, F, Block> BlockchainHeaderBackend<Block> for Blockchain<S, F> where Block: BlockT, S: Storage<Block>, F: Fetcher<Block> {
fn header(&self, id: BlockId<Block>) -> ClientResult<Option<Block::Header>> {
self.storage.header(id)
}
fn info(&self) -> ClientResult<BlockchainInfo<Block>> {
self.storage.info()
}
fn status(&self, id: BlockId<Block>) -> ClientResult<BlockStatus> {
self.storage.status(id)
}
fn hash(&self, number: <<Block as BlockT>::Header as HeaderT>::Number) -> ClientResult<Option<Block::Hash>> {
self.storage.hash(number)
}
}
impl<S, F, Block> BlockchainBackend<Block> for Blockchain<S, F> where Block: BlockT, S: Storage<Block>, F: Fetcher<Block> {
fn body(&self, _id: BlockId<Block>) -> ClientResult<Option<Vec<Block::Extrinsic>>> {
// TODO [light]: fetch from remote node
Ok(None)
}
fn justification(&self, _id: BlockId<Block>) -> ClientResult<Option<Justification<Block::Hash>>> {
Ok(None)
}
}
substrate/substrate/client/src/light/call_executor.rs
+143
View File
@@ -0,0 +1,143 @@
// Copyright 2017 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Light client call exector. Executes methods on remote full nodes, fetching
//! execution proof and checking it locally.
use std::sync::Arc;
use futures::{IntoFuture, Future};
use runtime_primitives::generic::BlockId;
use runtime_primitives::traits::{Block as BlockT, Header as HeaderT};
use state_machine::{Backend as StateBackend, CodeExecutor, OverlayedChanges, execution_proof_check};
use blockchain::Backend as ChainBackend;
use call_executor::{CallExecutor, CallResult};
use error::{Error as ClientError, ErrorKind as ClientErrorKind, Result as ClientResult};
use light::fetcher::{Fetcher, RemoteCallRequest};
/// Call executor that executes methods on remote node, querying execution proof
/// and checking proof by re-executing locally.
pub struct RemoteCallExecutor<B, F> {
blockchain: Arc<B>,
fetcher: Arc<F>,
}
impl<B, F> RemoteCallExecutor<B, F> {
/// Creates new instance of remote call executor.
pub fn new(blockchain: Arc<B>, fetcher: Arc<F>) -> Self {
RemoteCallExecutor { blockchain, fetcher }
}
}
impl<B, F, Block> CallExecutor<Block> for RemoteCallExecutor<B, F>
where
Block: BlockT,
B: ChainBackend<Block>,
F: Fetcher<Block>,
{
type Error = ClientError;
fn call(&self, id: &BlockId<Block>, method: &str, call_data: &[u8]) -> ClientResult<CallResult> {
let block_hash = match *id {
BlockId::Hash(hash) => hash,
BlockId::Number(number) => self.blockchain.hash(number)?
.ok_or_else(|| ClientErrorKind::UnknownBlock(format!("{}", number)))?,
};
self.fetcher.remote_call(RemoteCallRequest {
block: block_hash.clone(),
method: method.into(),
call_data: call_data.to_vec(),
}).into_future().wait()
}
fn call_at_state<S: StateBackend>(&self, _state: &S, _changes: &mut OverlayedChanges, _method: &str, _call_data: &[u8]) -> ClientResult<(Vec<u8>, S::Transaction)> {
Err(ClientErrorKind::NotAvailableOnLightClient.into())
}
fn prove_at_state<S: StateBackend>(&self, _state: S, _changes: &mut OverlayedChanges, _method: &str, _call_data: &[u8]) -> ClientResult<(Vec<u8>, Vec<Vec<u8>>)> {
Err(ClientErrorKind::NotAvailableOnLightClient.into())
}
}
/// Check remote execution proof using given backend.
pub fn check_execution_proof<Block, B, E>(
blockchain: &B,
executor: &E,
request: &RemoteCallRequest<Block::Hash>,
remote_proof: Vec<Vec<u8>>
) -> ClientResult<CallResult>
where
Block: BlockT,
<Block as BlockT>::Hash: Into<[u8; 32]>, // TODO: remove when patricia_trie generic.
B: ChainBackend<Block>,
E: CodeExecutor,
{
let local_header = blockchain.header(BlockId::Hash(request.block))?;
let local_header = local_header.ok_or_else(|| ClientErrorKind::UnknownBlock(format!("{}", request.block)))?;
let local_state_root = *local_header.state_root();
do_check_execution_proof(local_state_root.into(), executor, request, remote_proof)
}
/// Check remote execution proof using given state root.
fn do_check_execution_proof<Hash, E>(
local_state_root: [u8; 32],
executor: &E,
request: &RemoteCallRequest<Hash>,
remote_proof: Vec<Vec<u8>>,
) -> ClientResult<CallResult>
where
Hash: ::std::fmt::Display,
E: CodeExecutor,
{
let mut changes = OverlayedChanges::default();
let (local_result, _) = execution_proof_check(
local_state_root,
remote_proof,
&mut changes,
executor,
&request.method,
&request.call_data)?;
Ok(CallResult { return_data: local_result, changes })
}
#[cfg(test)]
mod tests {
use test_client;
use super::*;
#[test]
fn execution_proof_is_generated_and_checked() {
// prepare remote client
let remote_client = test_client::new();
let remote_block_id = BlockId::Number(0);
let remote_block_storage_root = remote_client.state_at(&remote_block_id)
.unwrap().storage_root(::std::iter::empty()).0;
// 'fetch' execution proof from remote node
let remote_execution_proof = remote_client.execution_proof(&remote_block_id, "authorities", &[]).unwrap().1;
// check remote execution proof locally
let local_executor = test_client::NativeExecutor::new();
do_check_execution_proof(remote_block_storage_root.into(), &local_executor, &RemoteCallRequest {
block: test_client::runtime::Hash::default(),
method: "authorities".into(),
call_data: vec![],
}, remote_execution_proof).unwrap();
}
}
substrate/substrate/client/src/light/fetcher.rs
+90
View File
@@ -0,0 +1,90 @@
// Copyright 2017 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Light client data fetcher. Fetches requested data from remote full nodes.
use std::sync::Arc;
use futures::IntoFuture;
use runtime_primitives::traits::{As, Block as BlockT, Header as HeaderT};
use state_machine::CodeExecutor;
use call_executor::CallResult;
use error::{Error as ClientError, Result as ClientResult};
use light::blockchain::{Blockchain, Storage as BlockchainStorage};
use light::call_executor::check_execution_proof;
/// Remote call request.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct RemoteCallRequest<Hash: ::std::fmt::Display> {
/// Call at state of given block.
pub block: Hash,
/// Method to call.
pub method: String,
/// Call data.
pub call_data: Vec<u8>,
}
/// Light client data fetcher. Implementations of this trait must check if remote data
/// is correct (see FetchedDataChecker) and return already checked data.
pub trait Fetcher<Block: BlockT>: Send + Sync {
/// Remote call result future.
type RemoteCallResult: IntoFuture<Item=CallResult, Error=ClientError>;
/// Fetch remote call result.
fn remote_call(&self, request: RemoteCallRequest<Block::Hash>) -> Self::RemoteCallResult;
}
/// Light client remote data checker.
pub trait FetchChecker<Block: BlockT>: Send + Sync {
/// Check remote method execution proof.
fn check_execution_proof(&self, request: &RemoteCallRequest<Block::Hash>, remote_proof: Vec<Vec<u8>>) -> ClientResult<CallResult>;
}
/// Remote data checker.
pub struct LightDataChecker<S, E, F> {
blockchain: Arc<Blockchain<S, F>>,
executor: E,
}
impl<S, E, F> LightDataChecker<S, E, F> {
/// Create new light data checker.
pub fn new(blockchain: Arc<Blockchain<S, F>>, executor: E) -> Self {
Self {
blockchain,
executor,
}
}
/// Get blockchain reference.
pub fn blockchain(&self) -> &Arc<Blockchain<S, F>> {
&self.blockchain
}
}
impl<S, E, F, Block> FetchChecker<Block> for LightDataChecker<S, E, F>
where
Block: BlockT,
<Block as BlockT>::Hash: From<[u8; 32]> + Into<[u8; 32]>, // TODO: remove when patricia_trie generic.
<<Block as BlockT>::Header as HeaderT>::Number: As<u32>,
S: BlockchainStorage<Block>,
E: CodeExecutor,
F: Fetcher<Block>,
{
fn check_execution_proof(&self, request: &RemoteCallRequest<Block::Hash>, remote_proof: Vec<Vec<u8>>) -> ClientResult<CallResult> {
check_execution_proof(&*self.blockchain, &self.executor, request, remote_proof)
}
}
substrate/substrate/client/src/light/mod.rs
+75
View File
@@ -0,0 +1,75 @@
// Copyright 2017 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! Light client components.
pub mod backend;
pub mod blockchain;
pub mod call_executor;
pub mod fetcher;
use std::sync::Arc;
use runtime_primitives::BuildStorage;
use runtime_primitives::traits::Block as BlockT;
use state_machine::CodeExecutor;
use client::Client;
use error::Result as ClientResult;
use light::backend::Backend;
use light::blockchain::{Blockchain, Storage as BlockchainStorage};
use light::call_executor::RemoteCallExecutor;
use light::fetcher::{Fetcher, LightDataChecker};
/// Create an instance of light client blockchain backend.
pub fn new_light_blockchain<B: BlockT, S: BlockchainStorage<B>, F>(storage: S) -> Arc<Blockchain<S, F>> {
Arc::new(Blockchain::new(storage))
}
/// Create an instance of light client backend.
pub fn new_light_backend<B: BlockT, S: BlockchainStorage<B>, F: Fetcher<B>>(blockchain: Arc<Blockchain<S, F>>, fetcher: Arc<F>) -> Arc<Backend<S, F>> {
blockchain.set_fetcher(Arc::downgrade(&fetcher));
Arc::new(Backend::new(blockchain))
}
/// Create an instance of light client.
pub fn new_light<B, S, F, GS>(
backend: Arc<Backend<S, F>>,
fetcher: Arc<F>,
genesis_storage: GS,
) -> ClientResult<Client<Backend<S, F>, RemoteCallExecutor<Blockchain<S, F>, F>, B>>
where
B: BlockT,
S: BlockchainStorage<B>,
F: Fetcher<B>,
GS: BuildStorage,
{
let executor = RemoteCallExecutor::new(backend.blockchain().clone(), fetcher);
Client::new(backend, executor, genesis_storage)
}
/// Create an instance of fetch data checker.
pub fn new_fetch_checker<B, S, E, F>(
blockchain: Arc<Blockchain<S, F>>,
executor: E,
) -> LightDataChecker<S, E, F>
where
B: BlockT,
S: BlockchainStorage<B>,
E: CodeExecutor,
{
LightDataChecker::new(blockchain, executor)
}
substrate/substrate/network/src/lib.rs
+1 -1
View File
@@ -68,4 +68,4 @@ pub use network::{NonReservedPeerMode, NetworkConfiguration, ConnectionFilter, C
pub use message::{generic as generic_message, BftMessage, LocalizedBftMessage, ConsensusVote, SignedConsensusVote, SignedConsensusMessage, SignedConsensusProposal};
pub use error::Error;
pub use config::{Role, ProtocolConfig};
pub use on_demand::{OnDemand, OnDemandService, Response as OnDemandResponse};
pub use on_demand::{OnDemand, OnDemandService, RemoteCallResponse};
substrate/substrate/network/src/on_demand.rs
+99 -68
View File
@@ -19,13 +19,13 @@
use std::collections::VecDeque;
use std::sync::{Arc, Weak};
use std::time::{Instant, Duration};
use futures::{Future, Poll};
use futures::{Async, Future, Poll};
use futures::sync::oneshot::{channel, Receiver, Sender};
use linked_hash_map::LinkedHashMap;
use linked_hash_map::Entry;
use parking_lot::Mutex;
use client;
use client::light::{Fetcher, FetchChecker, RemoteCallRequest};
use client::light::fetcher::{Fetcher, FetchChecker, RemoteCallRequest};
use io::SyncIo;
use message;
use network::PeerId;
@@ -36,7 +36,7 @@ use runtime_primitives::traits::{Block as BlockT, Header as HeaderT};
const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
/// On-demand service API.
pub trait OnDemandService: Send + Sync {
pub trait OnDemandService<Block: BlockT>: Send + Sync {
/// When new node is connected.
fn on_connect(&self, peer: PeerId, role: service::Role);
@@ -46,8 +46,8 @@ pub trait OnDemandService: Send + Sync {
/// Maintain peers requests.
fn maintain_peers(&self, io: &mut SyncIo);
/// When response is received from remote node.
fn on_remote_response(&self, io: &mut SyncIo, peer: PeerId, response: message::RemoteCallResponse);
/// When call response is received from remote node.
fn on_remote_call_response(&self, io: &mut SyncIo, peer: PeerId, response: message::RemoteCallResponse);
}
/// On-demand requests service. Dispatches requests to appropriate peers.
@@ -56,34 +56,47 @@ pub struct OnDemand<B: BlockT, E: service::ExecuteInContext<B>> {
checker: Arc<FetchChecker<B>>,
}
/// On-demand response.
pub struct Response {
receiver: Receiver<client::CallResult>,
/// On-demand remote call response.
pub struct RemoteCallResponse {
receiver: Receiver<Result<client::CallResult, client::error::Error>>,
}
#[derive(Default)]
struct OnDemandCore<B: BlockT, E: service::ExecuteInContext<B>> {
service: Weak<E>,
next_request_id: u64,
pending_requests: VecDeque<Request<B::Hash>>,
active_peers: LinkedHashMap<PeerId, Request<B::Hash>>,
pending_requests: VecDeque<Request<B>>,
active_peers: LinkedHashMap<PeerId, Request<B>>,
idle_peers: VecDeque<PeerId>,
}
struct Request<H> {
struct Request<Block: BlockT> {
id: u64,
timestamp: Instant,
sender: Sender<client::CallResult>,
request: RemoteCallRequest<H>,
data: RequestData<Block>,
}
impl Future for Response {
enum RequestData<Block: BlockT> {
RemoteCall(RemoteCallRequest<Block::Hash>, Sender<Result<client::CallResult, client::error::Error>>),
}
enum Accept<Block: BlockT> {
Ok,
CheckFailed(client::error::Error, RequestData<Block>),
}
impl Future for RemoteCallResponse {
type Item = client::CallResult;
type Error = client::error::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.receiver.poll()
.map_err(|_| client::error::ErrorKind::RemoteFetchCancelled.into())
.and_then(|r| match r {
Async::Ready(Ok(ready)) => Ok(Async::Ready(ready)),
Async::Ready(Err(error)) => Err(error),
Async::NotReady => Ok(Async::NotReady),
})
}
}
@@ -110,24 +123,47 @@ impl<B: BlockT, E> OnDemand<B, E> where
self.core.lock().service = service;
}
/// Execute method call on remote node, returning execution result and proof.
pub fn remote_call(&self, request: RemoteCallRequest<B::Hash>) -> Response {
let (sender, receiver) = channel();
let result = Response {
receiver: receiver,
/// Schedule && dispatch all scheduled requests.
fn schedule_request<R>(&self, data: RequestData<B>, result: R) -> R {
let mut core = self.core.lock();
core.insert(data);
core.dispatch();
result
}
/// Try to accept response from given peer.
fn accept_response<F: FnOnce(Request<B>) -> Accept<B>>(&self, rtype: &str, io: &mut SyncIo, peer: PeerId, request_id: u64, try_accept: F) {
let mut core = self.core.lock();
let request = match core.remove(peer, request_id) {
Some(request) => request,
None => {
trace!(target: "sync", "Invalid remote {} response from peer {}", rtype, peer);
io.disconnect_peer(peer);
core.remove_peer(peer);
return;
},
};
{
let mut core = self.core.lock();
core.insert(sender, request);
core.dispatch();
let retry_request_data = match try_accept(request) {
Accept::Ok => None,
Accept::CheckFailed(error, retry_request_data) => {
trace!(target: "sync", "Failed to check remote {} response from peer {}: {}", rtype, peer, error);
io.disconnect_peer(peer);
core.remove_peer(peer);
Some(retry_request_data)
},
};
if let Some(request_data) = retry_request_data {
core.insert(request_data);
}
result
core.dispatch();
}
}
impl<B, E> OnDemandService for OnDemand<B, E> where
impl<B, E> OnDemandService<B> for OnDemand<B, E> where
B: BlockT,
E: service::ExecuteInContext<B>,
B::Header: HeaderT<Number=u64>,
@@ -157,29 +193,17 @@ impl<B, E> OnDemandService for OnDemand<B, E> where
core.dispatch();
}
fn on_remote_response(&self, io: &mut SyncIo, peer: PeerId, response: message::RemoteCallResponse) {
let mut core = self.core.lock();
match core.remove(peer, response.id) {
Some(request) => match self.checker.check_execution_proof(&request.request, response.proof) {
fn on_remote_call_response(&self, io: &mut SyncIo, peer: PeerId, response: message::RemoteCallResponse) {
self.accept_response("call", io, peer, response.id, |request| match request.data {
RequestData::RemoteCall(request, sender) => match self.checker.check_execution_proof(&request, response.proof) {
Ok(response) => {
// we do not bother if receiver has been dropped already
let _ = request.sender.send(response);
},
Err(error) => {
trace!(target: "sync", "Failed to check remote response from peer {}: {}", peer, error);
io.disconnect_peer(peer);
core.remove_peer(peer);
core.insert(request.sender, request.request);
let _ = sender.send(Ok(response));
Accept::Ok
},
Err(error) => Accept::CheckFailed(error, RequestData::RemoteCall(request, sender)),
},
None => {
trace!(target: "sync", "Invalid remote response from peer {}", peer);
io.disconnect_peer(peer);
core.remove_peer(peer);
},
}
core.dispatch();
})
}
}
@@ -188,10 +212,12 @@ impl<B, E> Fetcher<B> for OnDemand<B, E> where
E: service::ExecuteInContext<B>,
B::Header: HeaderT<Number=u64>,
{
type RemoteCallResult = Response;
type RemoteCallResult = RemoteCallResponse;
fn remote_call(&self, request: RemoteCallRequest<B::Hash>) -> Self::RemoteCallResult {
OnDemand::remote_call(self, request)
let (sender, receiver) = channel();
self.schedule_request(RequestData::RemoteCall(request, sender),
RemoteCallResponse { receiver })
}
}
@@ -230,19 +256,18 @@ impl<B, E> OnDemandCore<B, E> where
}
}
pub fn insert(&mut self, sender: Sender<client::CallResult>, request: RemoteCallRequest<B::Hash>) {
pub fn insert(&mut self, data: RequestData<B>) {
let request_id = self.next_request_id;
self.next_request_id += 1;
self.pending_requests.push_back(Request {
id: request_id,
timestamp: Instant::now(),
sender,
request,
data,
});
}
pub fn remove(&mut self, peer: PeerId, id: u64) -> Option<Request<B::Hash>> {
pub fn remove(&mut self, peer: PeerId, id: u64) -> Option<Request<B>> {
match self.active_peers.entry(peer) {
Entry::Occupied(entry) => match entry.get().id == id {
true => {
@@ -272,20 +297,26 @@ impl<B, E> OnDemandCore<B, E> where
trace!(target: "sync", "Dispatching remote request {} to peer {}", request.id, peer);
service.execute_in_context(|ctx, protocol| {
let message = message::RemoteCallRequest {
id: request.id,
block: request.request.block,
method: request.request.method.clone(),
data: request.request.call_data.clone(),
};
protocol.send_message(ctx, peer, message::generic::Message::RemoteCallRequest(message))
protocol.send_message(ctx, peer, request.message())
});
self.active_peers.insert(peer, request);
}
}
}
impl<Block: BlockT> Request<Block> {
pub fn message(&self) -> message::Message<Block> {
match self.data {
RequestData::RemoteCall(ref data, _) => message::generic::Message::RemoteCallRequest(message::RemoteCallRequest {
id: self.id,
block: data.block,
method: data.method.clone(),
data: data.call_data.clone(),
}),
}
}
}
#[cfg(test)]
mod tests {
use std::collections::VecDeque;
@@ -294,7 +325,7 @@ mod tests {
use futures::Future;
use parking_lot::RwLock;
use client;
use client::light::{FetchChecker, RemoteCallRequest};
use client::light::fetcher::{Fetcher, FetchChecker, RemoteCallRequest};
use io::NetSyncIo;
use message;
use network::PeerId;
@@ -335,8 +366,8 @@ mod tests {
core.idle_peers.len() + core.active_peers.len()
}
fn receive_response(on_demand: &OnDemand<Block, DummyExecutor>, network: &mut TestIo, peer: PeerId, id: message::RequestId) {
on_demand.on_remote_response(network, peer, message::RemoteCallResponse {
fn receive_call_response(on_demand: &OnDemand<Block, DummyExecutor>, network: &mut TestIo, peer: PeerId, id: message::RequestId) {
on_demand.on_remote_call_response(network, peer, message::RemoteCallResponse {
id: id,
proof: vec![vec![2]],
});
@@ -391,7 +422,7 @@ mod tests {
on_demand.on_connect(0, Role::FULL);
on_demand.remote_call(RemoteCallRequest { block: Default::default(), method: "test".into(), call_data: vec![] });
receive_response(&*on_demand, &mut network, 0, 1);
receive_call_response(&*on_demand, &mut network, 0, 1);
assert!(network.to_disconnect.contains(&0));
assert_eq!(on_demand.core.lock().pending_requests.len(), 1);
}
@@ -401,10 +432,10 @@ mod tests {
let (_x, on_demand) = dummy(false);
let queue = RwLock::new(VecDeque::new());
let mut network = TestIo::new(&queue, None);
on_demand.on_connect(0, Role::FULL);
on_demand.remote_call(RemoteCallRequest { block: Default::default(), method: "test".into(), call_data: vec![] });
receive_response(&*on_demand, &mut network, 0, 0);
on_demand.on_connect(0, Role::FULL);
receive_call_response(&*on_demand, &mut network, 0, 0);
assert!(network.to_disconnect.contains(&0));
assert_eq!(on_demand.core.lock().pending_requests.len(), 1);
}
@@ -416,7 +447,7 @@ mod tests {
let mut network = TestIo::new(&queue, None);
on_demand.on_connect(0, Role::FULL);
receive_response(&*on_demand, &mut network, 0, 0);
receive_call_response(&*on_demand, &mut network, 0, 0);
assert!(network.to_disconnect.contains(&0));
}
@@ -433,7 +464,7 @@ mod tests {
assert_eq!(result.return_data, vec![42]);
});
receive_response(&*on_demand, &mut network, 0, 0);
receive_call_response(&*on_demand, &mut network, 0, 0);
thread.join().unwrap();
}
}
}
substrate/substrate/network/src/protocol.rs
+7 -7
View File
@@ -45,7 +45,7 @@ const MAX_BLOCK_DATA_RESPONSE: u32 = 128;
pub struct Protocol<B: BlockT> {
config: ProtocolConfig,
chain: Arc<Client<B>>,
on_demand: Option<Arc<OnDemandService>>,
on_demand: Option<Arc<OnDemandService<B>>>,
genesis_hash: B::Hash,
sync: RwLock<ChainSync<B>>,
consensus: Mutex<Consensus<B>>,
@@ -108,7 +108,7 @@ impl<B: BlockT> Protocol<B> where
pub fn new(
config: ProtocolConfig,
chain: Arc<Client<B>>,
on_demand: Option<Arc<OnDemandService>>,
on_demand: Option<Arc<OnDemandService<B>>>,
transaction_pool: Arc<TransactionPool<B>>
) -> error::Result<Self> {
let info = chain.info()?;
@@ -182,7 +182,7 @@ impl<B: BlockT> Protocol<B> where
GenericMessage::BftMessage(m) => self.on_bft_message(io, peer_id, m, HashingFor::<B>::hash(data)),
GenericMessage::Transactions(m) => self.on_transactions(io, peer_id, m),
GenericMessage::RemoteCallRequest(request) => self.on_remote_call_request(io, peer_id, request),
GenericMessage::RemoteCallResponse(response) => self.on_remote_call_response(io, peer_id, response)
GenericMessage::RemoteCallResponse(response) => self.on_remote_call_response(io, peer_id, response),
}
}
@@ -512,11 +512,11 @@ impl<B: BlockT> Protocol<B> where
}
fn on_remote_call_request(&self, io: &mut SyncIo, peer_id: PeerId, request: message::RemoteCallRequest<B::Hash>) {
trace!(target: "sync", "Remote request {} from {} ({} at {})", request.id, peer_id, request.method, request.block);
trace!(target: "sync", "Remote call request {} from {} ({} at {})", request.id, peer_id, request.method, request.block);
let proof = match self.chain.execution_proof(&request.block, &request.method, &request.data) {
Ok((_, proof)) => proof,
Err(error) => {
trace!(target: "sync", "Remote request {} from {} ({} at {}) failed with: {}",
trace!(target: "sync", "Remote call request {} from {} ({} at {}) failed with: {}",
request.id, peer_id, request.method, request.block, error);
Default::default()
},
@@ -528,8 +528,8 @@ impl<B: BlockT> Protocol<B> where
}
fn on_remote_call_response(&self, io: &mut SyncIo, peer_id: PeerId, response: message::RemoteCallResponse) {
trace!(target: "sync", "Remote response {} from {}", response.id, peer_id);
self.on_demand.as_ref().map(|s| s.on_remote_response(io, peer_id, response));
trace!(target: "sync", "Remote call response {} from {}", response.id, peer_id);
self.on_demand.as_ref().map(|s| s.on_remote_call_response(io, peer_id, response));
}
pub fn chain(&self) -> &Client<B> {
substrate/substrate/network/src/service.rs
+1 -1
View File
@@ -133,7 +133,7 @@ pub struct Params<B: BlockT> {
/// Polkadot relay chain access point.
pub chain: Arc<Client<B>>,
/// On-demand service reference.
pub on_demand: Option<Arc<OnDemandService>>,
pub on_demand: Option<Arc<OnDemandService<B>>>,
/// Transaction pool.
pub transaction_pool: Arc<TransactionPool<B>>,
}
substrate/substrate/network/src/test/sync.rs
+1 -1
View File
@@ -15,7 +15,7 @@
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
use client::backend::Backend;
use client::blockchain::Backend as BlockchainBackend;
use client::blockchain::HeaderBackend as BlockchainHeaderBackend;
use sync::SyncState;
use {Role};
use super::*;
substrate/substrate/state-machine/src/lib.rs
+6 -6
View File
@@ -209,7 +209,7 @@ pub fn execute<B: backend::Backend, Exec: CodeExecutor>(
///
/// Note: changes to code will be in place if this call is made again. For running partial
/// blocks (e.g. a transaction at a time), ensure a different method is used.
pub fn prove<B: TryIntoTrieBackend, Exec: CodeExecutor>(
pub fn prove_execution<B: TryIntoTrieBackend, Exec: CodeExecutor>(
backend: B,
overlay: &mut OverlayedChanges,
exec: &Exec,
@@ -225,8 +225,8 @@ pub fn prove<B: TryIntoTrieBackend, Exec: CodeExecutor>(
Ok((result, proof, transaction))
}
/// Check execution proof, generated by `prove` call.
pub fn proof_check<Exec: CodeExecutor>(
/// Check execution proof, generated by `prove_execution` call.
pub fn execution_proof_check<Exec: CodeExecutor>(
root: [u8; 32],
proof: Vec<Vec<u8>>,
overlay: &mut OverlayedChanges,
@@ -327,15 +327,15 @@ mod tests {
}
#[test]
fn prove_and_proof_check_works() {
fn prove_execution_and_proof_check_works() {
// fetch execution proof from 'remote' full node
let remote_backend = trie_backend::tests::test_trie();
let remote_root = remote_backend.storage_root(::std::iter::empty()).0;
let (remote_result, remote_proof, _) = prove(remote_backend,
let (remote_result, remote_proof, _) = prove_execution(remote_backend,
&mut Default::default(), &DummyCodeExecutor, "test", &[]).unwrap();
// check proof locally
let (local_result, _) = proof_check(remote_root, remote_proof,
let (local_result, _) = execution_proof_check(remote_root, remote_proof,
&mut Default::default(), &DummyCodeExecutor, "test", &[]).unwrap();
// check that both results are correct