From 9b4f90718f7ee49cd5bd8932f763c6efbe432d92 Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Tue, 27 Dec 2016 13:54:51 +0100 Subject: [PATCH 01/15] Display impl for ReqId --- ethcore/light/src/net/mod.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/ethcore/light/src/net/mod.rs b/ethcore/light/src/net/mod.rs index e27ba8612..f1585643f 100644 --- a/ethcore/light/src/net/mod.rs +++ b/ethcore/light/src/net/mod.rs @@ -30,6 +30,7 @@ use util::{Bytes, Mutex, RwLock, U256}; use time::{Duration, SteadyTime}; use std::collections::HashMap; +use std::fmt; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -122,6 +123,12 @@ mod timeout { #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct ReqId(usize); +impl fmt::Display for ReqId { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "Request #{}", self.0) + } +} + // A pending peer: one we've sent our status to but // may not have received one for. struct PendingPeer { From 428d5db9d1a764eade06efa81ae5f6b88032fb37 Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Tue, 27 Dec 2016 16:43:28 +0100 Subject: [PATCH 02/15] beginning of on_demand --- Cargo.lock | 1 + ethcore/light/Cargo.toml | 1 + ethcore/light/src/lib.rs | 2 + ethcore/light/src/on_demand/mod.rs | 125 +++++++++++++++++++++++++++++ 4 files changed, 129 insertions(+) create mode 100644 ethcore/light/src/on_demand/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 80c8810fe..43f6c7af2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -509,6 +509,7 @@ dependencies = [ "ethcore-ipc-codegen 1.5.0", "ethcore-network 1.5.0", "ethcore-util 1.5.0", + "futures 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)", "rlp 0.1.0", "smallvec 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", diff --git a/ethcore/light/Cargo.toml b/ethcore/light/Cargo.toml index 67ae166dc..20f5f93bd 100644 --- a/ethcore/light/Cargo.toml +++ b/ethcore/light/Cargo.toml @@ -20,6 +20,7 @@ ethcore-ipc = { path = "../../ipc/rpc", optional = true } rlp = { path = "../../util/rlp" } time = "0.1" smallvec = "0.3.1" +futures = "0.1" [features] default = [] diff --git a/ethcore/light/src/lib.rs b/ethcore/light/src/lib.rs index 5cdc3addc..60984d9c2 100644 --- a/ethcore/light/src/lib.rs +++ b/ethcore/light/src/lib.rs @@ -34,6 +34,7 @@ pub mod client; pub mod net; +pub mod on_demand; #[cfg(not(feature = "ipc"))] pub mod provider; @@ -64,6 +65,7 @@ extern crate ethcore_io as io; extern crate rlp; extern crate smallvec; extern crate time; +extern crate futures; #[cfg(feature = "ipc")] extern crate ethcore_ipc as ipc; diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs new file mode 100644 index 000000000..0bf25d68b --- /dev/null +++ b/ethcore/light/src/on_demand/mod.rs @@ -0,0 +1,125 @@ +// Copyright 2015, 2016 Parity Technologies (UK) Ltd. +// This file is part of Parity. + +// Parity is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// Parity is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with Parity. If not, see . + +//! On-demand chain requests over LES. This is a major building block for RPCs. +//! The request service is implemented using Futures. Higher level request handlers +//! will take the raw data received here and extract meaningful results from it. + +use std::collections::HashMap; + +use ethcore::ids::BlockId; +use ethcore::block::Block; +use ethcore::header::Header; +use ethcore::receipt::Receipt; + +use futures::Future; +use futures::sync::oneshot::{self, Sender, Receiver}; +use network::PeerId; + +use client::Client; +use net::{Handler, Status, Capabilities, Announcement, EventContext, BasicContext, ReqId}; +use util::{Address, H256, RwLock}; + +struct Account; + +// relevant peer info. +struct Peer { + status: Status, + capabilities: Capabilities, +} + +// request info and where to send the result. +enum Request { + HeaderByNumber(u64, H256, Sender
), // num + CHT root + HeaderByHash(H256, Sender
), + Block(Header, Sender), + BlockReceipts(Header, Sender>), + Account(Header, Address, Sender), + Storage(Header, Address, H256, Sender), +} + +/// On demand request service. See module docs for more details. +/// Accumulates info about all peers' capabilities and dispatches +/// requests to them accordingly. +pub struct OnDemand { + peers: RwLock>, + pending_requests: RwLock>, +} + +impl Handler for OnDemand { + fn on_connect(&self, ctx: &EventContext, status: &Status, capabilities: &Capabilities) { + self.peers.write().insert(ctx.peer(), Peer { status: status.clone(), capabilities: capabilities.clone() }) + } + + fn on_disconnect(&self, ctx: &EventContext, unfulfilled: &[ReqId]) { + self.peers.write().remove(&ctx.peer()); + } +} + +impl OnDemand { + /// Request a header by block number and CHT root hash. + pub fn header_by_number(&self, ctx: &BasicContext, num: u64, cht_root: H256) -> Receiver
{ + let (sender, receiver) = oneshot::channel(); + self.dispatch_request(ctx, Request::HeaderByNumber(num, cht_root, sender)); + receiver + } + + /// Request a header by hash. This is less accurate than by-number because we don't know + /// where in the chain this header lies, and therefore can't find a peer who is supposed to have + /// it as easily. + pub fn header_by_hash(&self, ctx: &BasicContext, hash: H256) -> Receiver
{ + let (sender, receiver) = oneshot::channel(); + self.dispatch_request(ctx, Request::HeaderByHash(hash, sender)); + receiver + } + + /// Request a block, given its header. Block bodies are requestable by hash only, + /// and the header is required anyway to verify and complete the block body + /// -- this just doesn't obscure the network query. + pub fn block(&self, ctx: &BasicContext, header: Header) -> Receiver { + let (sender, receiver) = oneshot::channel(); + self.dispatch_request(ctx, Request::Block(header, sender)); + receiver + } + + /// Request the receipts for a block. The header serves two purposes: + /// provide the block hash to fetch receipts for, and for verification of the receipts root. + pub fn block_receipts(&self, ctx: &BasicContext, header: Header) -> Receiver> { + let (sender, receiver) = oneshot::channel(); + self.dispatch_request(ctx, Request::BlockReceipts(header, sender)); + receiver + } + + /// Request an account by address and block header -- which gives a hash to query and a state root + /// to verify against. + pub fn account(&self, ctx: &BasicContext, header: Header, address: Address) -> Receiver { + let (sender, receiver) = oneshot::channel(); + self.dispatch_request(ctx, Request::Account(header, address, sender)); + receiver + } + + /// Request account storage value by block header, address, and key. + pub fn storage(&self, ctx: &BasicContext, header: Header, address: Address, key: H256) -> Receiver { + let (sender, receiver) = oneshot::channel(); + self.dispatch_request(ctx, Request::Storage(header, address, key, sender)); + receiver + } + + // dispatch a request to a suitable peer. + fn dispatch_request(&self, ctx: &BasicContext, request: Request) { + unimplemented!() + } +} From 5309ba45a8ba6d623fa71061bd4b3d9728f49d13 Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 28 Dec 2016 16:20:35 +0100 Subject: [PATCH 03/15] update peer status from announcement --- ethcore/light/src/net/status.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/ethcore/light/src/net/status.rs b/ethcore/light/src/net/status.rs index d058bc2f2..a11ed5b11 100644 --- a/ethcore/light/src/net/status.rs +++ b/ethcore/light/src/net/status.rs @@ -158,6 +158,16 @@ pub struct Status { pub last_head: Option<(H256, u64)>, } +impl Status { + /// Update the status from an announcement. + pub fn update_from(&mut self, announcement: &Announcement) { + self.last_head = Some((self.head_hash, announcement.reorg_depth)); + self.head_td = announcement.head_td; + self.head_hash = announcement.head_hash; + self.head_num = announcement.head_num; + } +} + /// Peer capabilities. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Capabilities { From 4daa64578948db6cff9d79c71586174ddfe01617 Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 28 Dec 2016 16:20:46 +0100 Subject: [PATCH 04/15] dispatch header requests --- Cargo.lock | 1 + ethcore/light/Cargo.toml | 1 + ethcore/light/src/lib.rs | 1 + ethcore/light/src/on_demand/mod.rs | 171 +++++++++++++++++++++++++---- 4 files changed, 153 insertions(+), 21 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 27d737568..e9f54c918 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -522,6 +522,7 @@ dependencies = [ "ethcore-util 1.5.0", "futures 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)", + "rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)", "rlp 0.1.0", "smallvec 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", "time 0.1.35 (registry+https://github.com/rust-lang/crates.io-index)", diff --git a/ethcore/light/Cargo.toml b/ethcore/light/Cargo.toml index 20f5f93bd..e0a1b581f 100644 --- a/ethcore/light/Cargo.toml +++ b/ethcore/light/Cargo.toml @@ -21,6 +21,7 @@ rlp = { path = "../../util/rlp" } time = "0.1" smallvec = "0.3.1" futures = "0.1" +rand = "0.3" [features] default = [] diff --git a/ethcore/light/src/lib.rs b/ethcore/light/src/lib.rs index 60984d9c2..0e650f992 100644 --- a/ethcore/light/src/lib.rs +++ b/ethcore/light/src/lib.rs @@ -66,6 +66,7 @@ extern crate rlp; extern crate smallvec; extern crate time; extern crate futures; +extern crate rand; #[cfg(feature = "ipc")] extern crate ethcore_ipc as ipc; diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index f932dc024..09b477c20 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -20,22 +20,65 @@ use std::collections::HashMap; -use ethcore::ids::BlockId; -use ethcore::block::Block; use ethcore::encoded; -use ethcore::header::Header; use ethcore::receipt::Receipt; -use futures::Future; -use futures::sync::oneshot::{self, Sender, Receiver}; +use futures::{Async, Poll, Future}; +use futures::sync::oneshot; use network::PeerId; use client::Client; use net::{Handler, Status, Capabilities, Announcement, EventContext, BasicContext, ReqId}; -use util::{Address, H256, RwLock}; +use util::{Address, H256, U256, RwLock}; +use request as les_request; -/// Dummy type for "basic account" -pub struct Account; +/// Basic account data. +// TODO: [rob] unify with similar struct in `snapshot`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Account { + /// Balance in Wei + pub balance: U256, + /// Storage trie root. + pub storage_root: H256, + /// Code hash + pub code_hash: H256, + /// Nonce + pub nonce: U256, +} + +/// Errors which can occur while trying to fulfill a request. +pub enum Error { + /// Request was canceled. + Canceled, + /// No suitable peers available to serve the request. + NoPeersAvailable, + /// Request timed out. + TimedOut, +} + +impl From for Error { + fn from(_: oneshot::Canceled) -> Self { + Error::Canceled + } +} + +/// Future which awaits a response to an on-demand request. +pub struct Response(oneshot::Receiver>); + +impl Future for Response { + type Item = T; + type Error = Error; + + fn poll(&mut self) -> Poll { + match self.0.poll().map_err(Into::into) { + Ok(Async::Ready(val)) => val.map(Async::Ready), + Ok(Async::NotReady) => Ok(Async::NotReady), + Err(e) => Err(e), + } + } +} + +type Sender = oneshot::Sender>; // relevant peer info. struct Peer { @@ -68,60 +111,146 @@ impl Handler for OnDemand { fn on_disconnect(&self, ctx: &EventContext, unfulfilled: &[ReqId]) { self.peers.write().remove(&ctx.peer()); + + for unfulfilled in unfulfilled { + if let Some(pending) = self.pending_requests.write().remove(unfulfilled) { + trace!(target: "on_demand", "Attempting to reassign dropped request"); + self.dispatch_request(ctx.as_basic(), pending); + } + } + } + + fn on_announcement(&self, ctx: &EventContext, announcement: &Announcement) { + let mut peers = self.peers.write(); + if let Some(ref mut peer) = peers.get_mut(&ctx.peer()) { + peer.status.update_from(&announcement); + peer.capabilities.update_from(&announcement); + } } } impl OnDemand { /// Request a header by block number and CHT root hash. - pub fn header_by_number(&self, ctx: &BasicContext, num: u64, cht_root: H256) -> Receiver { + pub fn header_by_number(&self, ctx: &BasicContext, num: u64, cht_root: H256) -> Response { let (sender, receiver) = oneshot::channel(); self.dispatch_request(ctx, Request::HeaderByNumber(num, cht_root, sender)); - receiver + Response(receiver) } /// Request a header by hash. This is less accurate than by-number because we don't know /// where in the chain this header lies, and therefore can't find a peer who is supposed to have /// it as easily. - pub fn header_by_hash(&self, ctx: &BasicContext, hash: H256) -> Receiver { + pub fn header_by_hash(&self, ctx: &BasicContext, hash: H256) -> Response { let (sender, receiver) = oneshot::channel(); self.dispatch_request(ctx, Request::HeaderByHash(hash, sender)); - receiver + Response(receiver) } /// Request a block, given its header. Block bodies are requestable by hash only, /// and the header is required anyway to verify and complete the block body /// -- this just doesn't obscure the network query. - pub fn block(&self, ctx: &BasicContext, header: encoded::Header) -> Receiver { + pub fn block(&self, ctx: &BasicContext, header: encoded::Header) -> Response { let (sender, receiver) = oneshot::channel(); self.dispatch_request(ctx, Request::Block(header, sender)); - receiver + Response(receiver) } /// Request the receipts for a block. The header serves two purposes: /// provide the block hash to fetch receipts for, and for verification of the receipts root. - pub fn block_receipts(&self, ctx: &BasicContext, header: encoded::Header) -> Receiver> { + pub fn block_receipts(&self, ctx: &BasicContext, header: encoded::Header) -> Response> { let (sender, receiver) = oneshot::channel(); self.dispatch_request(ctx, Request::BlockReceipts(header, sender)); - receiver + Response(receiver) } /// Request an account by address and block header -- which gives a hash to query and a state root /// to verify against. - pub fn account(&self, ctx: &BasicContext, header: encoded::Header, address: Address) -> Receiver { + pub fn account(&self, ctx: &BasicContext, header: encoded::Header, address: Address) -> Response { let (sender, receiver) = oneshot::channel(); self.dispatch_request(ctx, Request::Account(header, address, sender)); - receiver + Response(receiver) } /// Request account storage value by block header, address, and key. - pub fn storage(&self, ctx: &BasicContext, header: encoded::Header, address: Address, key: H256) -> Receiver { + pub fn storage(&self, ctx: &BasicContext, header: encoded::Header, address: Address, key: H256) -> Response { let (sender, receiver) = oneshot::channel(); self.dispatch_request(ctx, Request::Storage(header, address, key, sender)); - receiver + Response(receiver) } // dispatch a request to a suitable peer. fn dispatch_request(&self, ctx: &BasicContext, request: Request) { - unimplemented!() + match request { + Request::HeaderByNumber(num, cht_hash, sender) => { + let cht_num = ::client::cht::block_to_cht_number(num); + let req = les_request::Request::HeaderProofs(les_request::HeaderProofs { + requests: vec![les_request::HeaderProof { + cht_number: cht_num, + block_number: num, + from_level: 0, + }], + }); + + // we're looking for a peer with serveHeaders who's far enough along in the + // chain. + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_headers && peer.status.head_num >= num { + match ctx.request_from(*id, req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Request::HeaderByNumber(num, cht_hash, sender) + ); + return; + }, + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + } + + // TODO: retrying. + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); + } + Request::HeaderByHash(hash, sender) => { + let req = les_request::Request::Headers(les_request::Headers { + start: hash.into(), + max: 1, + skip: 0, + reverse: false, + }); + + // all we've got is a hash, so we'll just guess at peers who might have + // it randomly. + let mut potential_peers = self.peers.read().iter() + .filter(|&(_, peer)| peer.capabilities.serve_headers) + .map(|(id, _)| *id) + .collect::>(); + + let mut rng = ::rand::thread_rng(); + + ::rand::Rng::shuffle(&mut rng, &mut potential_peers); + + for id in potential_peers { + match ctx.request_from(id, req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Request::HeaderByHash(hash, sender), + ); + return; + } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + + sender.complete(Err(Error::NoPeersAvailable)); + } + _ => unimplemented!() + } } } From ddf2b944b5dd3f0e9c9d9a69b5d4f7c7a2b92c1c Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 28 Dec 2016 21:46:55 +0100 Subject: [PATCH 05/15] on demand request dispatch --- ethcore/light/src/on_demand/mod.rs | 144 ++++++++++++++++++++++++++--- 1 file changed, 133 insertions(+), 11 deletions(-) diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index 09b477c20..f6efb95a2 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -27,10 +27,9 @@ use futures::{Async, Poll, Future}; use futures::sync::oneshot; use network::PeerId; -use client::Client; use net::{Handler, Status, Capabilities, Announcement, EventContext, BasicContext, ReqId}; use util::{Address, H256, U256, RwLock}; -use request as les_request; +use request::{self as les_request, Request as LesRequest}; /// Basic account data. // TODO: [rob] unify with similar struct in `snapshot`. @@ -90,7 +89,7 @@ struct Peer { enum Request { HeaderByNumber(u64, H256, Sender), // num + CHT root HeaderByHash(H256, Sender), - Block(encoded::Header, Sender), + Block(encoded::Header, H256, Sender), BlockReceipts(encoded::Header, Sender>), Account(encoded::Header, Address, Sender), Storage(encoded::Header, Address, H256, Sender), @@ -151,7 +150,8 @@ impl OnDemand { /// -- this just doesn't obscure the network query. pub fn block(&self, ctx: &BasicContext, header: encoded::Header) -> Response { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Request::Block(header, sender)); + let hash = header.hash(); + self.dispatch_request(ctx, Request::Block(header, hash, sender)); Response(receiver) } @@ -183,7 +183,7 @@ impl OnDemand { match request { Request::HeaderByNumber(num, cht_hash, sender) => { let cht_num = ::client::cht::block_to_cht_number(num); - let req = les_request::Request::HeaderProofs(les_request::HeaderProofs { + let req = LesRequest::HeaderProofs(les_request::HeaderProofs { requests: vec![les_request::HeaderProof { cht_number: cht_num, block_number: num, @@ -202,7 +202,7 @@ impl OnDemand { req_id, Request::HeaderByNumber(num, cht_hash, sender) ); - return; + return }, Err(e) => trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), @@ -210,12 +210,12 @@ impl OnDemand { } } - // TODO: retrying. + // TODO: retrying trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); } Request::HeaderByHash(hash, sender) => { - let req = les_request::Request::Headers(les_request::Headers { + let req = LesRequest::Headers(les_request::Headers { start: hash.into(), max: 1, skip: 0, @@ -241,16 +241,138 @@ impl OnDemand { req_id, Request::HeaderByHash(hash, sender), ); - return; + return } Err(e) => trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), } } - + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); + } + Request::Block(header, hash, sender) => { + let num = header.number(); + let req = LesRequest::Bodies(les_request::Bodies { + block_hashes: vec![hash], + }); + + // we're looking for a peer with serveChainSince(num) + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_chain_since.as_ref().map_or(false, |x| *x >= num) { + match ctx.request_from(*id, req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Request::Block(header, hash, sender) + ); + return + } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); + } + Request::BlockReceipts(header, sender) => { + let num = header.number(); + let req = LesRequest::Receipts(les_request::Receipts { + block_hashes: vec![header.hash()], + }); + + // we're looking for a peer with serveChainSince(num) + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_chain_since.as_ref().map_or(false, |x| *x >= num) { + match ctx.request_from(*id, req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Request::BlockReceipts(header, sender) + ); + return + } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); + } + Request::Account(header, address, sender) => { + let num = header.number(); + let req = LesRequest::StateProofs(les_request::StateProofs { + requests: vec![les_request::StateProof { + block: header.hash(), + key1: ::util::Hashable::sha3(&address), + key2: None, + from_level: 0, + }], + }); + + // we're looking for a peer with serveStateSince(num) + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_state_since.as_ref().map_or(false, |x| *x >= num) { + match ctx.request_from(*id, req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Request::Account(header, address, sender) + ); + return + } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); + } + Request::Storage(header, address, key, sender) => { + let num = header.number(); + let req = LesRequest::StateProofs(les_request::StateProofs { + requests: vec![les_request::StateProof { + block: header.hash(), + key1: ::util::Hashable::sha3(&address), + key2: Some(::util::Hashable::sha3(&key)), + from_level: 0, + }], + }); + + // we're looking for a peer with serveStateSince(num) + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_state_since.as_ref().map_or(false, |x| *x >= num) { + match ctx.request_from(*id, req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Request::Storage(header, address, key, sender) + ); + return + } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); } - _ => unimplemented!() } } } From 4dbbc3bc88779c094c60fb08979b1daa09365fa5 Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Tue, 3 Jan 2017 16:18:46 +0100 Subject: [PATCH 06/15] beginnings of on_demand request module + verification --- ethcore/light/src/net/mod.rs | 4 +- ethcore/light/src/on_demand/mod.rs | 118 ++++++++++-------- ethcore/light/src/on_demand/request.rs | 158 +++++++++++++++++++++++++ 3 files changed, 231 insertions(+), 49 deletions(-) create mode 100644 ethcore/light/src/on_demand/request.rs diff --git a/ethcore/light/src/net/mod.rs b/ethcore/light/src/net/mod.rs index aa63aca3c..863382ecc 100644 --- a/ethcore/light/src/net/mod.rs +++ b/ethcore/light/src/net/mod.rs @@ -192,12 +192,12 @@ pub trait Handler: Send + Sync { fn on_block_headers(&self, _ctx: &EventContext, _req_id: ReqId, _headers: &[Bytes]) { } /// Called when a peer responds with block receipts. fn on_receipts(&self, _ctx: &EventContext, _req_id: ReqId, _receipts: &[Vec]) { } - /// Called when a peer responds with state proofs. Each proof is a series of trie + /// Called when a peer responds with state proofs. Each proof should be a series of trie /// nodes in ascending order by distance from the root. fn on_state_proofs(&self, _ctx: &EventContext, _req_id: ReqId, _proofs: &[Vec]) { } /// Called when a peer responds with contract code. fn on_code(&self, _ctx: &EventContext, _req_id: ReqId, _codes: &[Bytes]) { } - /// Called when a peer responds with header proofs. Each proof is a block header coupled + /// Called when a peer responds with header proofs. Each proof should be a block header coupled /// with a series of trie nodes is ascending order by distance from the root. fn on_header_proofs(&self, _ctx: &EventContext, _req_id: ReqId, _proofs: &[(Bytes, Vec)]) { } /// Called to "tick" the handler periodically. diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index f6efb95a2..d020ec8da 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -29,7 +29,9 @@ use network::PeerId; use net::{Handler, Status, Capabilities, Announcement, EventContext, BasicContext, ReqId}; use util::{Address, H256, U256, RwLock}; -use request::{self as les_request, Request as LesRequest}; +use types::les_request::{self as les_request, Request as LesRequest}; + +pub mod request; /// Basic account data. // TODO: [rob] unify with similar struct in `snapshot`. @@ -85,14 +87,13 @@ struct Peer { capabilities: Capabilities, } -// request info and where to send the result. -enum Request { - HeaderByNumber(u64, H256, Sender), // num + CHT root - HeaderByHash(H256, Sender), - Block(encoded::Header, H256, Sender), - BlockReceipts(encoded::Header, Sender>), - Account(encoded::Header, Address, Sender), - Storage(encoded::Header, Address, H256, Sender), +// Attempted request info and sender to put received value. +enum Attempted { + HeaderByNumber(request::HeaderByNumber, Sender), // num + CHT root + HeaderByHash(request::HeaderByHash, Sender), + Block(request::Block, Sender), + BlockReceipts(request::BlockReceipts, Sender>), + Account(request::Account, Sender), } /// On demand request service. See module docs for more details. @@ -103,36 +104,11 @@ pub struct OnDemand { pending_requests: RwLock>, } -impl Handler for OnDemand { - fn on_connect(&self, ctx: &EventContext, status: &Status, capabilities: &Capabilities) { - self.peers.write().insert(ctx.peer(), Peer { status: status.clone(), capabilities: capabilities.clone() }); - } - - fn on_disconnect(&self, ctx: &EventContext, unfulfilled: &[ReqId]) { - self.peers.write().remove(&ctx.peer()); - - for unfulfilled in unfulfilled { - if let Some(pending) = self.pending_requests.write().remove(unfulfilled) { - trace!(target: "on_demand", "Attempting to reassign dropped request"); - self.dispatch_request(ctx.as_basic(), pending); - } - } - } - - fn on_announcement(&self, ctx: &EventContext, announcement: &Announcement) { - let mut peers = self.peers.write(); - if let Some(ref mut peer) = peers.get_mut(&ctx.peer()) { - peer.status.update_from(&announcement); - peer.capabilities.update_from(&announcement); - } - } -} - impl OnDemand { /// Request a header by block number and CHT root hash. pub fn header_by_number(&self, ctx: &BasicContext, num: u64, cht_root: H256) -> Response { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Request::HeaderByNumber(num, cht_root, sender)); + self.dispatch_request(ctx, Pending::HeaderByNumber(num, cht_root, sender)); Response(receiver) } @@ -141,7 +117,7 @@ impl OnDemand { /// it as easily. pub fn header_by_hash(&self, ctx: &BasicContext, hash: H256) -> Response { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Request::HeaderByHash(hash, sender)); + self.dispatch_request(ctx, Pending::HeaderByHash(hash, sender)); Response(receiver) } @@ -151,7 +127,7 @@ impl OnDemand { pub fn block(&self, ctx: &BasicContext, header: encoded::Header) -> Response { let (sender, receiver) = oneshot::channel(); let hash = header.hash(); - self.dispatch_request(ctx, Request::Block(header, hash, sender)); + self.dispatch_request(ctx, Pending::Block(header, hash, sender)); Response(receiver) } @@ -159,7 +135,7 @@ impl OnDemand { /// provide the block hash to fetch receipts for, and for verification of the receipts root. pub fn block_receipts(&self, ctx: &BasicContext, header: encoded::Header) -> Response> { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Request::BlockReceipts(header, sender)); + self.dispatch_request(ctx, Pending::BlockReceipts(header, sender)); Response(receiver) } @@ -167,21 +143,23 @@ impl OnDemand { /// to verify against. pub fn account(&self, ctx: &BasicContext, header: encoded::Header, address: Address) -> Response { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Request::Account(header, address, sender)); + self.dispatch_request(ctx, Pending::Account(header, address, sender)); Response(receiver) } /// Request account storage value by block header, address, and key. pub fn storage(&self, ctx: &BasicContext, header: encoded::Header, address: Address, key: H256) -> Response { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Request::Storage(header, address, key, sender)); + self.dispatch_request(ctx, Pending::Storage(header, address, key, sender)); Response(receiver) } // dispatch a request to a suitable peer. + // + // TODO: most of this will become obsolete with a PeerSearch utility (#3987) fn dispatch_request(&self, ctx: &BasicContext, request: Request) { match request { - Request::HeaderByNumber(num, cht_hash, sender) => { + Pending::HeaderByNumber(request::HeaderByNumber(num, cht_hash), sender) => { let cht_num = ::client::cht::block_to_cht_number(num); let req = LesRequest::HeaderProofs(les_request::HeaderProofs { requests: vec![les_request::HeaderProof { @@ -200,7 +178,7 @@ impl OnDemand { trace!(target: "on_demand", "Assigning request to peer {}", id); self.pending_requests.write().insert( req_id, - Request::HeaderByNumber(num, cht_hash, sender) + Pending::HeaderByNumber(num, cht_hash, sender) ); return }, @@ -214,7 +192,7 @@ impl OnDemand { trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); } - Request::HeaderByHash(hash, sender) => { + Pending::HeaderByHash(hash, sender) => { let req = LesRequest::Headers(les_request::Headers { start: hash.into(), max: 1, @@ -251,7 +229,7 @@ impl OnDemand { trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); } - Request::Block(header, hash, sender) => { + Pending::Block(header, hash, sender) => { let num = header.number(); let req = LesRequest::Bodies(les_request::Bodies { block_hashes: vec![hash], @@ -279,7 +257,7 @@ impl OnDemand { trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); } - Request::BlockReceipts(header, sender) => { + Pending::BlockReceipts(header, sender) => { let num = header.number(); let req = LesRequest::Receipts(les_request::Receipts { block_hashes: vec![header.hash()], @@ -307,7 +285,7 @@ impl OnDemand { trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); } - Request::Account(header, address, sender) => { + Pending::Account(header, address, sender) => { let num = header.number(); let req = LesRequest::StateProofs(les_request::StateProofs { requests: vec![les_request::StateProof { @@ -340,7 +318,7 @@ impl OnDemand { trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); } - Request::Storage(header, address, key, sender) => { + Pending::Storage(header, address, key, sender) => { let num = header.number(); let req = LesRequest::StateProofs(les_request::StateProofs { requests: vec![les_request::StateProof { @@ -376,3 +354,49 @@ impl OnDemand { } } } + +impl Handler for OnDemand { + fn on_connect(&self, ctx: &EventContext, status: &Status, capabilities: &Capabilities) { + self.peers.write().insert(ctx.peer(), Peer { status: status.clone(), capabilities: capabilities.clone() }); + } + + fn on_disconnect(&self, ctx: &EventContext, unfulfilled: &[ReqId]) { + self.peers.write().remove(&ctx.peer()); + + for unfulfilled in unfulfilled { + if let Some(pending) = self.pending_requests.write().remove(unfulfilled) { + trace!(target: "on_demand", "Attempting to reassign dropped request"); + self.dispatch_request(ctx.as_basic(), pending); + } + } + } + + fn on_announcement(&self, ctx: &EventContext, announcement: &Announcement) { + let mut peers = self.peers.write(); + if let Some(ref mut peer) = peers.get_mut(&ctx.peer()) { + peer.status.update_from(&announcement); + peer.capabilities.update_from(&announcement); + } + } + + fn on_header_proofs(&self, ctx: &EventContext, req_id: ReqId, proofs: &[(Bytes, Vec)]) { + let peer = ctx.peer(); + let req = match self.pending_requests.write().remove(&req_id) { + Some(req) => req, + None => return, + }; + + match req { + Request::HeaderByNumber(num, cht_root, sender) => { + let (ref header, ref proof) = match proofs.get(0) { + Some(ref x) => x, + None => { + ctx.disconnect_peer(peer); + return + } + }; + } + _ => panic!("Only header by number request fetches header proofs; qed"), + } + } +} diff --git a/ethcore/light/src/on_demand/request.rs b/ethcore/light/src/on_demand/request.rs new file mode 100644 index 000000000..b0bb8ab18 --- /dev/null +++ b/ethcore/light/src/on_demand/request.rs @@ -0,0 +1,158 @@ +use ethcore::encoded; +use ethcore::receipt::Receipt; + +use rlp::{RlpStream, Stream}; +use util::{Address, Bytes, HashDB, H256, U256}; +use util::memorydb::MemoryDB; +use util::sha3::Hashable; +use util::trie::{Trie, TrieDB, TrieError}; + +use super::Account as BasicAccount; + +/// Errors in verification. +#[derive(Debug, PartialEq)] +pub enum Error { + /// RLP decoder error. + Decoder(::rlp::DecoderError), + /// Trie lookup error (result of bad proof) + Trie(TrieError), + /// Bad inclusion proof + BadProof, + /// Wrong header number. + WrongNumber(u64, u64), + /// Wrong header hash. + WrongHash(H256, H256), + /// Wrong trie root. + WrongTrieRoot(H256, H256), +} + +impl From<::rlp::DecoderError> for Error { + fn from(err: ::rlp::DecoderError) -> Self { + Error::Decoder(err) + } +} + +impl From> for Error { + fn from(err: Box) -> Self { + Error::Trie(*err) + } +} + +/// Request for a header by number. +pub struct HeaderByNumber { + /// The header's number. + pub num: u64 + /// The root of the CHT containing this header. + pub cht_root: H256, +} + +impl HeaderByNumber { + /// Check a response with a header and cht proof. + pub fn check_response(&self, header: &[u8], proof: &[Bytes]) -> Result { + use util::trie::{Trie, TrieDB}; + use rlp::{UntrustedRlp, View}; + + // check the proof + let mut db = MemoryDB::new(); + + for node in proof { db.insert(&node[..]) } + let key = ::rlp::encode(&self.num); + + let expected_hash: H256 = match TrieDB::new(&db, &self.cht_root).and_then(|t| t.get(&*key))? { + Some(val) => ::rlp::decode(&val), + None => return Err(Error::BadProof) + }; + + // and compare the hash to the found header. + let found_hash = header.sha3(); + match expected_hash == found_hash { + true => Ok(encoded::Header::new(header.to_vec())), + false => Err(Error::WrongHash(expected_hash, found_hash)), + } + } +} + +/// Request for a header by hash. +pub struct HeaderByHash(pub H256); + +impl HeaderByHash { + /// Check a response for the header. + pub fn check_response(&self, header: &[u8]) -> Result { + let hash = header.sha3(); + match hash == self.0 { + true => Ok(encoded::Header::new(header.to_vec())), + false => Err(Error::WrongHash(self.0, hash)), + } + } +} + +/// Request for a block, with header and precomputed hash. +pub struct Body { + /// The block's header. + pub header: encoded::Header, + /// The block's hash. + pub hash: H256, +} + +impl Body { + /// Check a response for this block body. + pub fn check_response(&self, body: &[u8]) -> Result { + let body_view = UntrustedRlp::new(&body); + + // check the integrity of the the body against the header + let tx_root = ::util::triehash::ordered_trie_root(body_view.at(0)?.iter().map(|r| r.as_raw().to_vec())); + if tx_root != self.header.transactions_root() { + return Err(Error::WrongTrieRoot(self.header.transactions_root(), tx_root)); + } + + let uncles_hash = body_view.at(1)?.as_raw().sha3(); + if uncles_hash != self.header.uncles_hash() { + return Err(Error::WrongHash(self.header.uncles_hash(), uncles_hash); + } + + // concatenate the header and the body. + let mut stream = RlpStream::new_list(3); + stream.append_raw(header.rlp().as_raw(), 1); + stream.append_raw(body, 2); + + Ok(encoded::Block::new(stream.out())) + } +} + +/// Request for a block's receipts with header for verification. +pub struct BlockReceipts(pub encoded::Header); + +impl BlockReceipts { + /// Check a response with receipts against the stored header. + pub fn check_response(&self, receipts: &[Receipt]) -> Result, Error> { + let receipts_root = self.0.receipts_root(); + let found_root = ::util::triehash::ordered_trie_root(receipts.iter().map(|r| ::rlp::encode(r).to_vec())); + + match receipts_root == found_root { + true => Ok(receipts.to_vec()), + false => Err(Error::WrongTrieRoot(receipts_root, found_root)), + } + } +} + +/// Request for an account structure. +pub struct Account { + /// Header for verification. + pub header: encoded::Header, + /// Address requested. + pub address: Address, +} + +impl Account { + /// Check a response with an account against the stored header. + pub fn check_response(&self, proof: &[Bytes]) -> Result, Error> { + let state_root = header.state_root(); + + let mut db = MemoryDB::new(); + for node in proof { db.insert(&*node) } + + match TrieDB::new(&db, &state_root).and_then(|t| t.get(&address.sha3()))? { + + } + } +} From 01977e60aa6a8b771ce50201de6abef09386204a Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Tue, 3 Jan 2017 19:13:40 +0100 Subject: [PATCH 07/15] finish request module, basic dispatch --- ethcore/light/src/on_demand/mod.rs | 381 +++++++++++-------------- ethcore/light/src/on_demand/request.rs | 50 +++- 2 files changed, 198 insertions(+), 233 deletions(-) diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index d020ec8da..d6e2b59c0 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -28,7 +28,7 @@ use futures::sync::oneshot; use network::PeerId; use net::{Handler, Status, Capabilities, Announcement, EventContext, BasicContext, ReqId}; -use util::{Address, H256, U256, RwLock}; +use util::{Bytes, H256, U256, RwLock}; use types::les_request::{self as les_request, Request as LesRequest}; pub mod request; @@ -48,6 +48,7 @@ pub struct Account { } /// Errors which can occur while trying to fulfill a request. +#[derive(Debug, Clone, Copy)] pub enum Error { /// Request was canceled. Canceled, @@ -88,10 +89,10 @@ struct Peer { } // Attempted request info and sender to put received value. -enum Attempted { +enum Pending { HeaderByNumber(request::HeaderByNumber, Sender), // num + CHT root HeaderByHash(request::HeaderByHash, Sender), - Block(request::Block, Sender), + Block(request::Body, Sender), BlockReceipts(request::BlockReceipts, Sender>), Account(request::Account, Sender), } @@ -101,257 +102,195 @@ enum Attempted { /// requests to them accordingly. pub struct OnDemand { peers: RwLock>, - pending_requests: RwLock>, + pending_requests: RwLock>, } impl OnDemand { /// Request a header by block number and CHT root hash. - pub fn header_by_number(&self, ctx: &BasicContext, num: u64, cht_root: H256) -> Response { + pub fn header_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber) -> Response { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Pending::HeaderByNumber(num, cht_root, sender)); + let num = req.num; + let cht_num = ::client::cht::block_to_cht_number(req.num); + let les_req = LesRequest::HeaderProofs(les_request::HeaderProofs { + requests: vec![les_request::HeaderProof { + cht_number: cht_num, + block_number: req.num, + from_level: 0, + }], + }); + + // we're looking for a peer with serveHeaders who's far enough along in the + // chain. + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_headers && peer.status.head_num >= num { + match ctx.request_from(*id, les_req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Pending::HeaderByNumber(req, sender) + ); + return Response(receiver); + }, + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); Response(receiver) } /// Request a header by hash. This is less accurate than by-number because we don't know /// where in the chain this header lies, and therefore can't find a peer who is supposed to have /// it as easily. - pub fn header_by_hash(&self, ctx: &BasicContext, hash: H256) -> Response { + pub fn header_by_hash(&self, ctx: &BasicContext, req: request::HeaderByHash) -> Response { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Pending::HeaderByHash(hash, sender)); + let les_req = LesRequest::Headers(les_request::Headers { + start: req.0.into(), + max: 1, + skip: 0, + reverse: false, + }); + + // all we've got is a hash, so we'll just guess at peers who might have + // it randomly. + let mut potential_peers = self.peers.read().iter() + .filter(|&(_, peer)| peer.capabilities.serve_headers) + .map(|(id, _)| *id) + .collect::>(); + + let mut rng = ::rand::thread_rng(); + + ::rand::Rng::shuffle(&mut rng, &mut potential_peers); + + for id in potential_peers { + match ctx.request_from(id, les_req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Pending::HeaderByHash(req, sender), + ); + return Response(receiver); + } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); Response(receiver) } /// Request a block, given its header. Block bodies are requestable by hash only, /// and the header is required anyway to verify and complete the block body /// -- this just doesn't obscure the network query. - pub fn block(&self, ctx: &BasicContext, header: encoded::Header) -> Response { + pub fn block(&self, ctx: &BasicContext, req: request::Body) -> Response { let (sender, receiver) = oneshot::channel(); - let hash = header.hash(); - self.dispatch_request(ctx, Pending::Block(header, hash, sender)); + let num = req.header.number(); + let les_req = LesRequest::Bodies(les_request::Bodies { + block_hashes: vec![req.hash], + }); + + // we're looking for a peer with serveChainSince(num) + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_chain_since.as_ref().map_or(false, |x| *x >= num) { + match ctx.request_from(*id, les_req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Pending::Block(req, sender) + ); + return Response(receiver) + } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); Response(receiver) } /// Request the receipts for a block. The header serves two purposes: /// provide the block hash to fetch receipts for, and for verification of the receipts root. - pub fn block_receipts(&self, ctx: &BasicContext, header: encoded::Header) -> Response> { + pub fn block_receipts(&self, ctx: &BasicContext, req: request::BlockReceipts) -> Response> { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Pending::BlockReceipts(header, sender)); + let num = req.0.number(); + let les_req = LesRequest::Receipts(les_request::Receipts { + block_hashes: vec![req.0.hash()], + }); + + // we're looking for a peer with serveChainSince(num) + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_chain_since.as_ref().map_or(false, |x| *x >= num) { + match ctx.request_from(*id, les_req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Pending::BlockReceipts(req, sender) + ); + return Response(receiver) + } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), + } + } + } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); Response(receiver) } /// Request an account by address and block header -- which gives a hash to query and a state root /// to verify against. - pub fn account(&self, ctx: &BasicContext, header: encoded::Header, address: Address) -> Response { + pub fn account(&self, ctx: &BasicContext, req: request::Account) -> Response { let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Pending::Account(header, address, sender)); - Response(receiver) - } + let num = req.header.number(); + let les_req = LesRequest::StateProofs(les_request::StateProofs { + requests: vec![les_request::StateProof { + block: req.header.hash(), + key1: ::util::Hashable::sha3(&req.address), + key2: None, + from_level: 0, + }], + }); - /// Request account storage value by block header, address, and key. - pub fn storage(&self, ctx: &BasicContext, header: encoded::Header, address: Address, key: H256) -> Response { - let (sender, receiver) = oneshot::channel(); - self.dispatch_request(ctx, Pending::Storage(header, address, key, sender)); - Response(receiver) - } - - // dispatch a request to a suitable peer. - // - // TODO: most of this will become obsolete with a PeerSearch utility (#3987) - fn dispatch_request(&self, ctx: &BasicContext, request: Request) { - match request { - Pending::HeaderByNumber(request::HeaderByNumber(num, cht_hash), sender) => { - let cht_num = ::client::cht::block_to_cht_number(num); - let req = LesRequest::HeaderProofs(les_request::HeaderProofs { - requests: vec![les_request::HeaderProof { - cht_number: cht_num, - block_number: num, - from_level: 0, - }], - }); - - // we're looking for a peer with serveHeaders who's far enough along in the - // chain. - for (id, peer) in self.peers.read().iter() { - if peer.capabilities.serve_headers && peer.status.head_num >= num { - match ctx.request_from(*id, req.clone()) { - Ok(req_id) => { - trace!(target: "on_demand", "Assigning request to peer {}", id); - self.pending_requests.write().insert( - req_id, - Pending::HeaderByNumber(num, cht_hash, sender) - ); - return - }, - Err(e) => - trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), - } + // we're looking for a peer with serveStateSince(num) + for (id, peer) in self.peers.read().iter() { + if peer.capabilities.serve_state_since.as_ref().map_or(false, |x| *x >= num) { + match ctx.request_from(*id, les_req.clone()) { + Ok(req_id) => { + trace!(target: "on_demand", "Assigning request to peer {}", id); + self.pending_requests.write().insert( + req_id, + Pending::Account(req, sender) + ); + return Response(receiver) } + Err(e) => + trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), } - - // TODO: retrying - trace!(target: "on_demand", "No suitable peer for request"); - sender.complete(Err(Error::NoPeersAvailable)); - } - Pending::HeaderByHash(hash, sender) => { - let req = LesRequest::Headers(les_request::Headers { - start: hash.into(), - max: 1, - skip: 0, - reverse: false, - }); - - // all we've got is a hash, so we'll just guess at peers who might have - // it randomly. - let mut potential_peers = self.peers.read().iter() - .filter(|&(_, peer)| peer.capabilities.serve_headers) - .map(|(id, _)| *id) - .collect::>(); - - let mut rng = ::rand::thread_rng(); - - ::rand::Rng::shuffle(&mut rng, &mut potential_peers); - - for id in potential_peers { - match ctx.request_from(id, req.clone()) { - Ok(req_id) => { - trace!(target: "on_demand", "Assigning request to peer {}", id); - self.pending_requests.write().insert( - req_id, - Request::HeaderByHash(hash, sender), - ); - return - } - Err(e) => - trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), - } - } - // TODO: retrying - trace!(target: "on_demand", "No suitable peer for request"); - sender.complete(Err(Error::NoPeersAvailable)); - } - Pending::Block(header, hash, sender) => { - let num = header.number(); - let req = LesRequest::Bodies(les_request::Bodies { - block_hashes: vec![hash], - }); - - // we're looking for a peer with serveChainSince(num) - for (id, peer) in self.peers.read().iter() { - if peer.capabilities.serve_chain_since.as_ref().map_or(false, |x| *x >= num) { - match ctx.request_from(*id, req.clone()) { - Ok(req_id) => { - trace!(target: "on_demand", "Assigning request to peer {}", id); - self.pending_requests.write().insert( - req_id, - Request::Block(header, hash, sender) - ); - return - } - Err(e) => - trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), - } - } - } - - // TODO: retrying - trace!(target: "on_demand", "No suitable peer for request"); - sender.complete(Err(Error::NoPeersAvailable)); - } - Pending::BlockReceipts(header, sender) => { - let num = header.number(); - let req = LesRequest::Receipts(les_request::Receipts { - block_hashes: vec![header.hash()], - }); - - // we're looking for a peer with serveChainSince(num) - for (id, peer) in self.peers.read().iter() { - if peer.capabilities.serve_chain_since.as_ref().map_or(false, |x| *x >= num) { - match ctx.request_from(*id, req.clone()) { - Ok(req_id) => { - trace!(target: "on_demand", "Assigning request to peer {}", id); - self.pending_requests.write().insert( - req_id, - Request::BlockReceipts(header, sender) - ); - return - } - Err(e) => - trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), - } - } - } - - // TODO: retrying - trace!(target: "on_demand", "No suitable peer for request"); - sender.complete(Err(Error::NoPeersAvailable)); - } - Pending::Account(header, address, sender) => { - let num = header.number(); - let req = LesRequest::StateProofs(les_request::StateProofs { - requests: vec![les_request::StateProof { - block: header.hash(), - key1: ::util::Hashable::sha3(&address), - key2: None, - from_level: 0, - }], - }); - - // we're looking for a peer with serveStateSince(num) - for (id, peer) in self.peers.read().iter() { - if peer.capabilities.serve_state_since.as_ref().map_or(false, |x| *x >= num) { - match ctx.request_from(*id, req.clone()) { - Ok(req_id) => { - trace!(target: "on_demand", "Assigning request to peer {}", id); - self.pending_requests.write().insert( - req_id, - Request::Account(header, address, sender) - ); - return - } - Err(e) => - trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), - } - } - } - - // TODO: retrying - trace!(target: "on_demand", "No suitable peer for request"); - sender.complete(Err(Error::NoPeersAvailable)); - } - Pending::Storage(header, address, key, sender) => { - let num = header.number(); - let req = LesRequest::StateProofs(les_request::StateProofs { - requests: vec![les_request::StateProof { - block: header.hash(), - key1: ::util::Hashable::sha3(&address), - key2: Some(::util::Hashable::sha3(&key)), - from_level: 0, - }], - }); - - // we're looking for a peer with serveStateSince(num) - for (id, peer) in self.peers.read().iter() { - if peer.capabilities.serve_state_since.as_ref().map_or(false, |x| *x >= num) { - match ctx.request_from(*id, req.clone()) { - Ok(req_id) => { - trace!(target: "on_demand", "Assigning request to peer {}", id); - self.pending_requests.write().insert( - req_id, - Request::Storage(header, address, key, sender) - ); - return - } - Err(e) => - trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), - } - } - } - - // TODO: retrying - trace!(target: "on_demand", "No suitable peer for request"); - sender.complete(Err(Error::NoPeersAvailable)); } } + + // TODO: retrying + trace!(target: "on_demand", "No suitable peer for request"); + sender.complete(Err(Error::NoPeersAvailable)); + Response(receiver) } } @@ -366,7 +305,7 @@ impl Handler for OnDemand { for unfulfilled in unfulfilled { if let Some(pending) = self.pending_requests.write().remove(unfulfilled) { trace!(target: "on_demand", "Attempting to reassign dropped request"); - self.dispatch_request(ctx.as_basic(), pending); + unimplemented!() } } } @@ -387,8 +326,8 @@ impl Handler for OnDemand { }; match req { - Request::HeaderByNumber(num, cht_root, sender) => { - let (ref header, ref proof) = match proofs.get(0) { + Pending::HeaderByNumber(req, sender) => { + let &&(ref header, ref proof) = match proofs.get(0) { Some(ref x) => x, None => { ctx.disconnect_peer(peer); diff --git a/ethcore/light/src/on_demand/request.rs b/ethcore/light/src/on_demand/request.rs index b0bb8ab18..09846c2bf 100644 --- a/ethcore/light/src/on_demand/request.rs +++ b/ethcore/light/src/on_demand/request.rs @@ -1,8 +1,26 @@ +// Copyright 2015, 2016 Parity Technologies (UK) Ltd. +// This file is part of Parity. + +// Parity is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// Parity is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with Parity. If not, see . + +//! Request types, verification, and verification errors. + use ethcore::encoded; use ethcore::receipt::Receipt; -use rlp::{RlpStream, Stream}; -use util::{Address, Bytes, HashDB, H256, U256}; +use rlp::{RlpStream, Stream, UntrustedRlp, View}; +use util::{Address, Bytes, HashDB, H256}; use util::memorydb::MemoryDB; use util::sha3::Hashable; use util::trie::{Trie, TrieDB, TrieError}; @@ -41,7 +59,7 @@ impl From> for Error { /// Request for a header by number. pub struct HeaderByNumber { /// The header's number. - pub num: u64 + pub num: u64, /// The root of the CHT containing this header. pub cht_root: H256, } @@ -50,12 +68,11 @@ impl HeaderByNumber { /// Check a response with a header and cht proof. pub fn check_response(&self, header: &[u8], proof: &[Bytes]) -> Result { use util::trie::{Trie, TrieDB}; - use rlp::{UntrustedRlp, View}; // check the proof let mut db = MemoryDB::new(); - for node in proof { db.insert(&node[..]) } + for node in proof { db.insert(&node[..]); } let key = ::rlp::encode(&self.num); let expected_hash: H256 = match TrieDB::new(&db, &self.cht_root).and_then(|t| t.get(&*key))? { @@ -107,12 +124,12 @@ impl Body { let uncles_hash = body_view.at(1)?.as_raw().sha3(); if uncles_hash != self.header.uncles_hash() { - return Err(Error::WrongHash(self.header.uncles_hash(), uncles_hash); + return Err(Error::WrongHash(self.header.uncles_hash(), uncles_hash)); } // concatenate the header and the body. let mut stream = RlpStream::new_list(3); - stream.append_raw(header.rlp().as_raw(), 1); + stream.append_raw(self.header.rlp().as_raw(), 1); stream.append_raw(body, 2); Ok(encoded::Block::new(stream.out())) @@ -145,14 +162,23 @@ pub struct Account { impl Account { /// Check a response with an account against the stored header. - pub fn check_response(&self, proof: &[Bytes]) -> Result, Error> { - let state_root = header.state_root(); + pub fn check_response(&self, proof: &[Bytes]) -> Result { + let state_root = self.header.state_root(); let mut db = MemoryDB::new(); - for node in proof { db.insert(&*node) } - - match TrieDB::new(&db, &state_root).and_then(|t| t.get(&address.sha3()))? { + for node in proof { db.insert(&node[..]); } + match TrieDB::new(&db, &state_root).and_then(|t| t.get(&self.address.sha3()))? { + Some(val) => { + let rlp = UntrustedRlp::new(&val); + Ok(BasicAccount { + nonce: rlp.val_at(0)?, + balance: rlp.val_at(1)?, + storage_root: rlp.val_at(2)?, + code_hash: rlp.val_at(3)?, + }) + }, + None => Err(Error::BadProof) } } } From 4e94f43644c3e823ed550085c98fc4b4576054ff Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 4 Jan 2017 12:48:07 +0100 Subject: [PATCH 08/15] tests for request module --- ethcore/light/src/on_demand/request.rs | 152 +++++++++++++++++++++++++ 1 file changed, 152 insertions(+) diff --git a/ethcore/light/src/on_demand/request.rs b/ethcore/light/src/on_demand/request.rs index 09846c2bf..7ffde2b02 100644 --- a/ethcore/light/src/on_demand/request.rs +++ b/ethcore/light/src/on_demand/request.rs @@ -57,6 +57,7 @@ impl From> for Error { } /// Request for a header by number. +#[derive(Debug, Clone, PartialEq, Eq)] pub struct HeaderByNumber { /// The header's number. pub num: u64, @@ -90,6 +91,7 @@ impl HeaderByNumber { } /// Request for a header by hash. +#[derive(Debug, Clone, PartialEq, Eq)] pub struct HeaderByHash(pub H256); impl HeaderByHash { @@ -104,6 +106,7 @@ impl HeaderByHash { } /// Request for a block, with header and precomputed hash. +#[derive(Debug, Clone, PartialEq, Eq)] pub struct Body { /// The block's header. pub header: encoded::Header, @@ -137,6 +140,7 @@ impl Body { } /// Request for a block's receipts with header for verification. +#[derive(Debug, Clone, PartialEq, Eq)] pub struct BlockReceipts(pub encoded::Header); impl BlockReceipts { @@ -153,6 +157,7 @@ impl BlockReceipts { } /// Request for an account structure. +#[derive(Debug, Clone, PartialEq, Eq)] pub struct Account { /// Header for verification. pub header: encoded::Header, @@ -182,3 +187,150 @@ impl Account { } } } + +#[cfg(test)] +mod tests { + use super::*; + use util::{MemoryDB, Address, H256, FixedHash}; + use util::trie::{Trie, TrieMut, TrieDB, SecTrieDB, TrieDBMut, SecTrieDBMut}; + use util::trie::recorder::{BasicRecorder, Recorder}; + + use ethcore::header::Header; + use ethcore::encoded; + use ethcore::receipt::Receipt; + + #[test] + fn check_header_by_number() { + let mut root = H256::default(); + let mut db = MemoryDB::new(); + let mut header = Header::new(); + header.set_number(10_000); + header.set_extra_data(b"test_header".to_vec()); + + { + let mut trie = TrieDBMut::new(&mut db, &mut root); + for i in (0..2048u64).map(|x| x + 8192) { + let hash = if i == 10_000 { + header.hash() + } else { + H256::random() + }; + trie.insert(&*::rlp::encode(&i), &*::rlp::encode(&hash)).unwrap(); + } + } + + let proof = { + let trie = TrieDB::new(&db, &root).unwrap(); + let key = ::rlp::encode(&10_000u64); + let mut recorder = BasicRecorder::new(); + + trie.get_recorded(&*key, &mut recorder).unwrap().unwrap(); + + recorder.drain().into_iter().map(|r| r.data).collect::>() + }; + + let req = HeaderByNumber { + num: 10_000, + cht_root: root, + }; + + let raw_header = ::rlp::encode(&header); + + assert!(req.check_response(&*raw_header, &proof[..]).is_ok()); + } + + #[test] + fn check_header_by_hash() { + let mut header = Header::new(); + header.set_number(10_000); + header.set_extra_data(b"test_header".to_vec()); + let hash = header.hash(); + let raw_header = ::rlp::encode(&header); + + assert!(HeaderByHash(hash).check_response(&*raw_header).is_ok()) + } + + #[test] + fn check_body() { + use rlp::{RlpStream, Stream}; + + let header = Header::new(); + let mut body_stream = RlpStream::new_list(2); + body_stream.begin_list(0).begin_list(0); + + let req = Body { + header: encoded::Header::new(::rlp::encode(&header).to_vec()), + hash: header.hash(), + }; + + assert!(req.check_response(&*body_stream.drain()).is_ok()) + } + + #[test] + fn check_receipts() { + let receipts = (0..5).map(|_| Receipt { + state_root: H256::random(), + gas_used: 21_000u64.into(), + log_bloom: Default::default(), + logs: Vec::new(), + }).collect::>(); + + let mut header = Header::new(); + let receipts_root = ::util::triehash::ordered_trie_root( + receipts.iter().map(|x| ::rlp::encode(x).to_vec()) + ); + + header.set_receipts_root(receipts_root); + + let req = BlockReceipts(encoded::Header::new(::rlp::encode(&header).to_vec())); + + assert!(req.check_response(&receipts).is_ok()) + } + + #[test] + fn check_state_proof() { + let mut root = H256::default(); + let mut db = MemoryDB::new(); + let mut header = Header::new(); + header.set_number(123_456); + header.set_extra_data(b"test_header".to_vec()); + + let addr = Address::random(); + let rand_acc = || { + let mut stream = RlpStream::new_list(4); + stream.append(&2u64) + .append(&100_000_000u64) + .append(&H256::random()) + .append(&H256::random()); + + stream.out() + }; + { + let mut trie = SecTrieDBMut::new(&mut db, &mut root); + for _ in 0..100 { + let address = Address::random(); + trie.insert(&*address, &rand_acc()).unwrap(); + } + + trie.insert(&*addr, &rand_acc()).unwrap(); + } + + let proof = { + let trie = SecTrieDB::new(&db, &root).unwrap(); + let mut recorder = BasicRecorder::new(); + + trie.get_recorded(&*addr, &mut recorder).unwrap().unwrap(); + + recorder.drain().into_iter().map(|r| r.data).collect::>() + }; + + header.set_state_root(root.clone()); + + let req = Account { + header: encoded::Header::new(::rlp::encode(&header).to_vec()), + address: addr, + }; + + assert!(req.check_response(&proof[..]).is_ok()); + } +} From 1d51b6f7e57c0094549461cf0a9e733cb0c2df78 Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 4 Jan 2017 13:34:50 +0100 Subject: [PATCH 09/15] handle incoming LES requests --- ethcore/light/src/on_demand/mod.rs | 131 +++++++++++++++++++++++++++-- 1 file changed, 125 insertions(+), 6 deletions(-) diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index d6e2b59c0..d36902bc5 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -327,15 +327,134 @@ impl Handler for OnDemand { match req { Pending::HeaderByNumber(req, sender) => { - let &&(ref header, ref proof) = match proofs.get(0) { - Some(ref x) => x, - None => { - ctx.disconnect_peer(peer); - return + if let Some(&(ref header, ref proof)) = proofs.get(0) { + match req.check_response(header, proof) { + Ok(header) => { + sender.complete(Ok(header)); + return + } + Err(e) => { + warn!("Error handling response for header request: {:?}", e); + ctx.disable_peer(peer); + } } - }; + } + + // attempt reassign. } _ => panic!("Only header by number request fetches header proofs; qed"), } } + + fn on_block_headers(&self, ctx: &EventContext, req_id: ReqId, headers: &[Bytes]) { + let peer = ctx.peer(); + let req = match self.pending_requests.write().remove(&req_id) { + Some(req) => req, + None => return, + }; + + match req { + Pending::HeaderByHash(req, sender) => { + if let Some(ref header) = headers.get(0) { + match req.check_response(header) { + Ok(header) => { + sender.complete(Ok(header)); + return + } + Err(e) => { + warn!("Error handling response for header request: {:?}", e); + ctx.disable_peer(peer); + } + } + } + + // attempt reassign. + } + _ => panic!("Only header by hash request fetches headers; qed"), + } + } + + fn on_block_bodies(&self, ctx: &EventContext, req_id: ReqId, bodies: &[Bytes]) { + let peer = ctx.peer(); + let req = match self.pending_requests.write().remove(&req_id) { + Some(req) => req, + None => return, + }; + + match req { + Pending::Block(req, sender) => { + if let Some(ref block) = bodies.get(0) { + match req.check_response(block) { + Ok(block) => { + sender.complete(Ok(block)); + return + } + Err(e) => { + warn!("Error handling response for block request: {:?}", e); + ctx.disable_peer(peer); + } + } + } + + // attempt reassign. + } + _ => panic!("Only block request fetches bodies; qed"), + } + } + + fn on_receipts(&self, ctx: &EventContext, req_id: ReqId, receipts: &[Vec]) { + let peer = ctx.peer(); + let req = match self.pending_requests.write().remove(&req_id) { + Some(req) => req, + None => return, + }; + + match req { + Pending::BlockReceipts(req, sender) => { + if let Some(ref receipts) = receipts.get(0) { + match req.check_response(receipts) { + Ok(receipts) => { + sender.complete(Ok(receipts)); + return + } + Err(e) => { + warn!("Error handling response for receipts request: {:?}", e); + ctx.disable_peer(peer); + } + } + } + + // attempt reassign. + } + _ => panic!("Only receipts request fetches receipts; qed"), + } + } + + fn on_state_proofs(&self, ctx: &EventContext, req_id: ReqId, proofs: &[Vec]) { + let peer = ctx.peer(); + let req = match self.pending_requests.write().remove(&req_id) { + Some(req) => req, + None => return, + }; + + match req { + Pending::Account(req, sender) => { + if let Some(ref proof) = proofs.get(0) { + match req.check_response(proof) { + Ok(proof) => { + sender.complete(Ok(proof)); + return + } + Err(e) => { + warn!("Error handling response for state request: {:?}", e); + ctx.disable_peer(peer); + } + } + } + + // attempt reassign. + } + _ => panic!("Only account request fetches state proof; qed"), + } + } } From ca35b345ca788a5b5775afbd8b43c467fc949025 Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 4 Jan 2017 13:58:26 +0100 Subject: [PATCH 10/15] separate request dispatch from creation --- ethcore/light/src/on_demand/mod.rs | 55 ++++++++++++++++++++++++------ 1 file changed, 44 insertions(+), 11 deletions(-) diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index d36902bc5..0fcd4a4e9 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -109,6 +109,12 @@ impl OnDemand { /// Request a header by block number and CHT root hash. pub fn header_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber) -> Response { let (sender, receiver) = oneshot::channel(); + self.dispatch_header_by_number(ctx, req, sender); + Response(receiver) + } + + // dispatch the request, completing the request if no peers available. + fn dispatch_header_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber, sender: Sender) { let num = req.num; let cht_num = ::client::cht::block_to_cht_number(req.num); let les_req = LesRequest::HeaderProofs(les_request::HeaderProofs { @@ -130,7 +136,7 @@ impl OnDemand { req_id, Pending::HeaderByNumber(req, sender) ); - return Response(receiver); + return }, Err(e) => trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), @@ -141,7 +147,6 @@ impl OnDemand { // TODO: retrying trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); - Response(receiver) } /// Request a header by hash. This is less accurate than by-number because we don't know @@ -149,6 +154,11 @@ impl OnDemand { /// it as easily. pub fn header_by_hash(&self, ctx: &BasicContext, req: request::HeaderByHash) -> Response { let (sender, receiver) = oneshot::channel(); + self.dispatch_header_by_hash(ctx, req, sender); + Response(receiver) + } + + fn dispatch_header_by_hash(&self, ctx: &BasicContext, req: request::HeaderByHash, sender: Sender) { let les_req = LesRequest::Headers(les_request::Headers { start: req.0.into(), max: 1, @@ -175,7 +185,7 @@ impl OnDemand { req_id, Pending::HeaderByHash(req, sender), ); - return Response(receiver); + return } Err(e) => trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), @@ -185,7 +195,6 @@ impl OnDemand { // TODO: retrying trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); - Response(receiver) } /// Request a block, given its header. Block bodies are requestable by hash only, @@ -193,6 +202,11 @@ impl OnDemand { /// -- this just doesn't obscure the network query. pub fn block(&self, ctx: &BasicContext, req: request::Body) -> Response { let (sender, receiver) = oneshot::channel(); + self.dispatch_block(ctx, req, sender); + Response(receiver) + } + + fn dispatch_block(&self, ctx: &BasicContext, req: request::Body, sender: Sender) { let num = req.header.number(); let les_req = LesRequest::Bodies(les_request::Bodies { block_hashes: vec![req.hash], @@ -208,7 +222,7 @@ impl OnDemand { req_id, Pending::Block(req, sender) ); - return Response(receiver) + return } Err(e) => trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), @@ -219,13 +233,17 @@ impl OnDemand { // TODO: retrying trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); - Response(receiver) } /// Request the receipts for a block. The header serves two purposes: /// provide the block hash to fetch receipts for, and for verification of the receipts root. pub fn block_receipts(&self, ctx: &BasicContext, req: request::BlockReceipts) -> Response> { let (sender, receiver) = oneshot::channel(); + self.dispatch_block_receipts(ctx, req, sender); + Response(receiver) + } + + fn dispatch_block_receipts(&self, ctx: &BasicContext, req: request::BlockReceipts, sender: Sender>) { let num = req.0.number(); let les_req = LesRequest::Receipts(les_request::Receipts { block_hashes: vec![req.0.hash()], @@ -241,7 +259,7 @@ impl OnDemand { req_id, Pending::BlockReceipts(req, sender) ); - return Response(receiver) + return } Err(e) => trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), @@ -252,13 +270,17 @@ impl OnDemand { // TODO: retrying trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); - Response(receiver) } /// Request an account by address and block header -- which gives a hash to query and a state root /// to verify against. pub fn account(&self, ctx: &BasicContext, req: request::Account) -> Response { let (sender, receiver) = oneshot::channel(); + self.dispatch_account(ctx, req, sender); + Response(receiver) + } + + fn dispatch_account(&self, ctx: &BasicContext, req: request::Account, sender: Sender) { let num = req.header.number(); let les_req = LesRequest::StateProofs(les_request::StateProofs { requests: vec![les_request::StateProof { @@ -279,7 +301,7 @@ impl OnDemand { req_id, Pending::Account(req, sender) ); - return Response(receiver) + return } Err(e) => trace!(target: "on_demand", "Failed to make request of peer {}: {:?}", id, e), @@ -290,7 +312,6 @@ impl OnDemand { // TODO: retrying trace!(target: "on_demand", "No suitable peer for request"); sender.complete(Err(Error::NoPeersAvailable)); - Response(receiver) } } @@ -301,11 +322,23 @@ impl Handler for OnDemand { fn on_disconnect(&self, ctx: &EventContext, unfulfilled: &[ReqId]) { self.peers.write().remove(&ctx.peer()); + let ctx = ctx.as_basic(); for unfulfilled in unfulfilled { if let Some(pending) = self.pending_requests.write().remove(unfulfilled) { trace!(target: "on_demand", "Attempting to reassign dropped request"); - unimplemented!() + match pending { + Pending::HeaderByNumber(req, sender) + => self.dispatch_header_by_number(ctx, req, sender), + Pending::HeaderByHash(req, sender) + => self.dispatch_header_by_hash(ctx, req, sender), + Pending::Block(req, sender) + => self.dispatch_block(ctx, req, sender), + Pending::BlockReceipts(req, sender) + => self.dispatch_block_receipts(ctx, req, sender), + Pending::Account(req, sender) + => self.dispatch_account(ctx, req, sender), + } } } } From 38ac84f0c7fd0cea7b626fd715b7223e57fd772a Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 4 Jan 2017 14:33:32 +0100 Subject: [PATCH 11/15] reassign requests on failure --- ethcore/light/src/on_demand/mod.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index 0fcd4a4e9..fc3d4d30e 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -373,7 +373,7 @@ impl Handler for OnDemand { } } - // attempt reassign. + self.dispatch_header_by_number(ctx.as_basic(), req, sender); } _ => panic!("Only header by number request fetches header proofs; qed"), } @@ -401,7 +401,7 @@ impl Handler for OnDemand { } } - // attempt reassign. + self.dispatch_header_by_hash(ctx.as_basic(), req, sender); } _ => panic!("Only header by hash request fetches headers; qed"), } @@ -429,7 +429,7 @@ impl Handler for OnDemand { } } - // attempt reassign. + self.dispatch_block(ctx.as_basic(), req, sender); } _ => panic!("Only block request fetches bodies; qed"), } @@ -457,7 +457,7 @@ impl Handler for OnDemand { } } - // attempt reassign. + self.dispatch_block_receipts(ctx.as_basic(), req, sender); } _ => panic!("Only receipts request fetches receipts; qed"), } @@ -485,7 +485,7 @@ impl Handler for OnDemand { } } - // attempt reassign. + self.dispatch_account(ctx.as_basic(), req, sender); } _ => panic!("Only account request fetches state proof; qed"), } From ed98c389ca96f9d129f38d6e017fc392bf1c131d Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 4 Jan 2017 14:54:50 +0100 Subject: [PATCH 12/15] no_peers test --- ethcore/light/src/on_demand/mod.rs | 43 +++++++++++++++++++++++++++++- 1 file changed, 42 insertions(+), 1 deletion(-) diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index fc3d4d30e..4a99c4a39 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -48,7 +48,7 @@ pub struct Account { } /// Errors which can occur while trying to fulfill a request. -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq)] pub enum Error { /// Request was canceled. Canceled, @@ -105,6 +105,15 @@ pub struct OnDemand { pending_requests: RwLock>, } +impl Default for OnDemand { + fn default() -> Self { + OnDemand { + peers: RwLock::new(HashMap::new()), + pending_requests: RwLock::new(HashMap::new()), + } + } +} + impl OnDemand { /// Request a header by block number and CHT root hash. pub fn header_by_number(&self, ctx: &BasicContext, req: request::HeaderByNumber) -> Response { @@ -491,3 +500,35 @@ impl Handler for OnDemand { } } } + +#[cfg(test)] +mod tests { + use super::*; + use on_demand::request; + use net::{BasicContext, ReqId, Error as LesError}; + use request::{Request as LesRequest, Kind as LesRequestKind}; + use network::{PeerId, NodeId}; + use futures::Future; + use util::H256; + + struct FakeContext; + + impl BasicContext for FakeContext { + fn persistent_peer_id(&self, _: PeerId) -> Option { None } + fn request_from(&self, _: PeerId, _: LesRequest) -> Result { + unimplemented!() + } + fn make_announcement(&self, _: Announcement) { } + fn max_requests(&self, _: PeerId, _: LesRequestKind) -> usize { 0 } + fn disconnect_peer(&self, _: PeerId) { } + fn disable_peer(&self, _: PeerId) { } + } + + #[test] + fn no_peers() { + let on_demand = OnDemand::default(); + let result = on_demand.header_by_hash(&FakeContext, request::HeaderByHash(H256::default())); + + assert_eq!(result.wait().unwrap_err(), Error::NoPeersAvailable); + } +} From 6ad0a0baaa6cba8fbcc240a09f48b00ec1fae58f Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 4 Jan 2017 16:12:58 +0100 Subject: [PATCH 13/15] fix test imports --- ethcore/light/src/on_demand/mod.rs | 3 +-- ethcore/light/src/on_demand/request.rs | 2 ++ 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index 4a99c4a39..218a866a3 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -504,8 +504,7 @@ impl Handler for OnDemand { #[cfg(test)] mod tests { use super::*; - use on_demand::request; - use net::{BasicContext, ReqId, Error as LesError}; + use net::{Announcement, BasicContext, ReqId, Error as LesError}; use request::{Request as LesRequest, Kind as LesRequestKind}; use network::{PeerId, NodeId}; use futures::Future; diff --git a/ethcore/light/src/on_demand/request.rs b/ethcore/light/src/on_demand/request.rs index 7ffde2b02..52383e721 100644 --- a/ethcore/light/src/on_demand/request.rs +++ b/ethcore/light/src/on_demand/request.rs @@ -289,6 +289,8 @@ mod tests { #[test] fn check_state_proof() { + use rlp::{RlpStream, Stream}; + let mut root = H256::default(); let mut db = MemoryDB::new(); let mut header = Header::new(); From 75b5acf21dff78fe03aaec71bca6748bcc87bffe Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Thu, 5 Jan 2017 13:26:14 +0100 Subject: [PATCH 14/15] use common BasicAccount type --- ethcore/light/src/on_demand/mod.rs | 23 +++++------------------ ethcore/light/src/on_demand/request.rs | 3 +-- 2 files changed, 6 insertions(+), 20 deletions(-) diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs index 218a866a3..aadb43c0e 100644 --- a/ethcore/light/src/on_demand/mod.rs +++ b/ethcore/light/src/on_demand/mod.rs @@ -20,6 +20,7 @@ use std::collections::HashMap; +use ethcore::basic_account::BasicAccount; use ethcore::encoded; use ethcore::receipt::Receipt; @@ -28,25 +29,11 @@ use futures::sync::oneshot; use network::PeerId; use net::{Handler, Status, Capabilities, Announcement, EventContext, BasicContext, ReqId}; -use util::{Bytes, H256, U256, RwLock}; +use util::{Bytes, RwLock}; use types::les_request::{self as les_request, Request as LesRequest}; pub mod request; -/// Basic account data. -// TODO: [rob] unify with similar struct in `snapshot`. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct Account { - /// Balance in Wei - pub balance: U256, - /// Storage trie root. - pub storage_root: H256, - /// Code hash - pub code_hash: H256, - /// Nonce - pub nonce: U256, -} - /// Errors which can occur while trying to fulfill a request. #[derive(Debug, Clone, Copy, PartialEq)] pub enum Error { @@ -94,7 +81,7 @@ enum Pending { HeaderByHash(request::HeaderByHash, Sender), Block(request::Body, Sender), BlockReceipts(request::BlockReceipts, Sender>), - Account(request::Account, Sender), + Account(request::Account, Sender), } /// On demand request service. See module docs for more details. @@ -283,13 +270,13 @@ impl OnDemand { /// Request an account by address and block header -- which gives a hash to query and a state root /// to verify against. - pub fn account(&self, ctx: &BasicContext, req: request::Account) -> Response { + pub fn account(&self, ctx: &BasicContext, req: request::Account) -> Response { let (sender, receiver) = oneshot::channel(); self.dispatch_account(ctx, req, sender); Response(receiver) } - fn dispatch_account(&self, ctx: &BasicContext, req: request::Account, sender: Sender) { + fn dispatch_account(&self, ctx: &BasicContext, req: request::Account, sender: Sender) { let num = req.header.number(); let les_req = LesRequest::StateProofs(les_request::StateProofs { requests: vec![les_request::StateProof { diff --git a/ethcore/light/src/on_demand/request.rs b/ethcore/light/src/on_demand/request.rs index 52383e721..f483c6fdd 100644 --- a/ethcore/light/src/on_demand/request.rs +++ b/ethcore/light/src/on_demand/request.rs @@ -16,6 +16,7 @@ //! Request types, verification, and verification errors. +use ethcore::basic_account::BasicAccount; use ethcore::encoded; use ethcore::receipt::Receipt; @@ -25,8 +26,6 @@ use util::memorydb::MemoryDB; use util::sha3::Hashable; use util::trie::{Trie, TrieDB, TrieError}; -use super::Account as BasicAccount; - /// Errors in verification. #[derive(Debug, PartialEq)] pub enum Error { From 78e670812c2e9818717b07e8d0587dea853db6cb Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Wed, 11 Jan 2017 18:57:38 +0100 Subject: [PATCH 15/15] fix test compilation --- ethcore/light/src/on_demand/request.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/ethcore/light/src/on_demand/request.rs b/ethcore/light/src/on_demand/request.rs index f483c6fdd..86fcc5d05 100644 --- a/ethcore/light/src/on_demand/request.rs +++ b/ethcore/light/src/on_demand/request.rs @@ -192,7 +192,7 @@ mod tests { use super::*; use util::{MemoryDB, Address, H256, FixedHash}; use util::trie::{Trie, TrieMut, TrieDB, SecTrieDB, TrieDBMut, SecTrieDBMut}; - use util::trie::recorder::{BasicRecorder, Recorder}; + use util::trie::recorder::Recorder; use ethcore::header::Header; use ethcore::encoded; @@ -221,9 +221,9 @@ mod tests { let proof = { let trie = TrieDB::new(&db, &root).unwrap(); let key = ::rlp::encode(&10_000u64); - let mut recorder = BasicRecorder::new(); + let mut recorder = Recorder::new(); - trie.get_recorded(&*key, &mut recorder).unwrap().unwrap(); + trie.get_with(&*key, &mut recorder).unwrap().unwrap(); recorder.drain().into_iter().map(|r| r.data).collect::>() }; @@ -318,9 +318,9 @@ mod tests { let proof = { let trie = SecTrieDB::new(&db, &root).unwrap(); - let mut recorder = BasicRecorder::new(); + let mut recorder = Recorder::new(); - trie.get_recorded(&*addr, &mut recorder).unwrap().unwrap(); + trie.get_with(&*addr, &mut recorder).unwrap().unwrap(); recorder.drain().into_iter().map(|r| r.data).collect::>() };