From 3f15f7f2f97bf94a43df4127fb22edb75403384a Mon Sep 17 00:00:00 2001 From: Antioch Peverell <30642645+antiochp@users.noreply.github.com> Date: Wed, 31 Jan 2018 15:39:55 -0500 Subject: [PATCH] Request compact blocks (#667) * first pass at allow/deny lists for hard-coded peers (not just seeds) * commit * add peers_allow and peers_deny examples and comments to grin.toml * always ask for compact block always fail to hydrate always fallback to requesting full block * decide to send full (empty) block over compact (empty) block * add some randomness to the decision around broadcasting an empty block as a block or as a compact block (so we can exercise more code paths easily) --- chain/src/pipe.rs | 5 +-- grin/src/adapters.rs | 81 +++++++++++++++++++++++++++++++++++++------- grin/src/sync.rs | 1 - p2p/src/msg.rs | 34 ++++++++++--------- p2p/src/peer.rs | 25 ++++++++++++++ p2p/src/peers.rs | 34 +++++++++++++++++++ p2p/src/protocol.rs | 61 +++++++++++++++++++++++++++++++++ p2p/src/server.rs | 1 + p2p/src/types.rs | 8 ++++- 9 files changed, 217 insertions(+), 33 deletions(-) diff --git a/chain/src/pipe.rs b/chain/src/pipe.rs index 13cdb844..f29dd6d7 100644 --- a/chain/src/pipe.rs +++ b/chain/src/pipe.rs @@ -53,11 +53,12 @@ pub fn process_block(b: &Block, mut ctx: BlockContext) -> Result, Er debug!( LOGGER, - "pipe: process_block {} at {} with {} inputs and {} outputs.", + "pipe: process_block {} at {} with {} inputs, {} outputs, {} kernels", b.hash(), b.header.height, b.inputs.len(), - b.outputs.len() + b.outputs.len(), + b.kernels.len(), ); check_known(b.hash(), &mut ctx)?; diff --git a/grin/src/adapters.rs b/grin/src/adapters.rs index 85967f10..41013745 100644 --- a/grin/src/adapters.rs +++ b/grin/src/adapters.rs @@ -15,6 +15,8 @@ use std::net::SocketAddr; use std::sync::{Arc, RwLock}; use std::sync::atomic::{AtomicBool, Ordering}; +use rand; +use rand::Rng; use chain::{self, ChainAdapter, Options, MINE}; use core::core; @@ -88,6 +90,26 @@ impl p2p::ChainAdapter for NetToChainAdapter { true } + fn compact_block_received(&self, bh: core::CompactBlock, addr: SocketAddr) -> bool { + let bhash = bh.hash(); + debug!( + LOGGER, + "Received compact_block {} at {} from {}, going to process.", + bhash, + bh.header.height, + addr, + ); + + debug!( + LOGGER, + "*** cannot hydrate compact block (not yet implemented), falling back to requesting full block", + ); + + self.request_block(&bh.header, &addr); + + true + } + fn header_received(&self, bh: core::BlockHeader, addr: SocketAddr) -> bool { let bhash = bh.hash(); debug!( @@ -117,7 +139,7 @@ impl p2p::ChainAdapter for NetToChainAdapter { // we have successfully processed a block header // so we can go request the block itself - self.request_block(&bh, &addr); + self.request_compact_block(&bh, &addr); // done receiving the header true @@ -276,28 +298,42 @@ impl NetToChainAdapter { } } - // After we have received a block header in "header first" propagation - // we need to go request the block from the same peer that gave us the header - // (unless we have already accepted the block) + // After receiving a compact block if we cannot successfully hydrate + // it into a full block then fallback to requesting the full block + // from the same peer that gave us the compact block // // TODO - currently only request block from a single peer // consider additional peers for redundancy? fn request_block(&self, bh: &BlockHeader, addr: &SocketAddr) { if let None = self.peers.borrow().adapter.get_block(bh.hash()) { - if let Some(_) = self.peers.borrow().adapter.get_block(bh.previous) { - if let Some(peer) = self.peers.borrow().get_connected_peer(addr) { - if let Ok(peer) = peer.read() { - let _ = peer.send_block_request(bh.hash()); - } + if let Some(peer) = self.peers.borrow().get_connected_peer(addr) { + if let Ok(peer) = peer.read() { + let _ = peer.send_block_request(bh.hash()); } - } else { - debug!(LOGGER, "request_block: prev block {} missing, skipping", bh.previous); } } else { debug!(LOGGER, "request_block: block {} already known", bh.hash()); } } + // After we have received a block header in "header first" propagation + // we need to go request the block (compact representation) from the + // same peer that gave us the header (unless we have already accepted the block) + // + // TODO - currently only request block from a single peer + // consider additional peers for redundancy? + fn request_compact_block(&self, bh: &BlockHeader, addr: &SocketAddr) { + if let None = self.peers.borrow().adapter.get_block(bh.hash()) { + if let Some(peer) = self.peers.borrow().get_connected_peer(addr) { + if let Ok(peer) = peer.read() { + let _ = peer.send_compact_block_request(bh.hash()); + } + } + } else { + debug!(LOGGER, "request_compact_block: block {} already known", bh.hash()); + } + } + /// Prepare options for the chain pipeline fn chain_opts(&self) -> chain::Options { let opts = if self.currently_syncing.load(Ordering::Relaxed) { @@ -331,10 +367,29 @@ impl ChainAdapter for ChainToPoolAndNetAdapter { } // If we mined the block then we want to broadcast the block itself. - // But if we received the block from another node then broadcast "header first" + // If block is empty then broadcast the block. + // If block contains txs then broadcast the compact block. + // If we received the block from another node then broadcast "header first" // to minimize network traffic. if opts.contains(MINE) { - self.peers.borrow().broadcast_block(b); + // propagate compact block out if we mined the block + // but broadcast full block if we have no txs + let cb = b.as_compact_block(); + if cb.kern_ids.is_empty() { + + // in the interest of testing all code paths + // randomly decide how we send an empty block out + // TODO - lock this down once we are comfortable it works... + + let mut rng = rand::thread_rng(); + if rng.gen() { + self.peers.borrow().broadcast_block(&b); + } else { + self.peers.borrow().broadcast_compact_block(&cb); + } + } else { + self.peers.borrow().broadcast_compact_block(&cb); + } } else { // "header first" propagation if we are not the originator of this block self.peers.borrow().broadcast_header(&b.header); diff --git a/grin/src/sync.rs b/grin/src/sync.rs index cf9e9c6f..91b61aa3 100644 --- a/grin/src/sync.rs +++ b/grin/src/sync.rs @@ -158,7 +158,6 @@ pub fn header_sync(peers: Peers, chain: Arc) { if let Some(peer) = peers.most_work_peer() { if let Ok(p) = peer.try_read() { let peer_difficulty = p.info.total_difficulty.clone(); - debug!(LOGGER, "sync: header_sync: {}, {}", difficulty, peer_difficulty); if peer_difficulty > difficulty { let _ = request_headers( peer.clone(), diff --git a/p2p/src/msg.rs b/p2p/src/msg.rs index 31308007..b41a2847 100644 --- a/p2p/src/msg.rs +++ b/p2p/src/msg.rs @@ -49,22 +49,24 @@ pub enum ErrCodes { /// Types of messages enum_from_primitive! { - #[derive(Debug, Clone, Copy, PartialEq)] - pub enum Type { - Error, - Hand, - Shake, - Ping, - Pong, - GetPeerAddrs, - PeerAddrs, - GetHeaders, - Header, - Headers, - GetBlock, - Block, - Transaction, - } + #[derive(Debug, Clone, Copy, PartialEq)] + pub enum Type { + Error, + Hand, + Shake, + Ping, + Pong, + GetPeerAddrs, + PeerAddrs, + GetHeaders, + Header, + Headers, + GetBlock, + Block, + GetCompactBlock, + CompactBlock, + Transaction, + } } /// Future combinator to read any message where the body is a Readable. Reads diff --git a/p2p/src/peer.rs b/p2p/src/peer.rs index 6cb93274..59441efc 100644 --- a/p2p/src/peer.rs +++ b/p2p/src/peer.rs @@ -184,6 +184,21 @@ impl Peer { } } + pub fn send_compact_block(&self, b: &core::CompactBlock) -> Result<(), Error> { + if !self.tracking_adapter.has(b.hash()) { + debug!(LOGGER, "Send compact block {} to {}", b.hash(), self.info.addr); + self.proto.send_compact_block(b) + } else { + debug!( + LOGGER, + "Suppress compact block send {} to {} (already seen)", + b.hash(), + self.info.addr, + ); + Ok(()) + } + } + pub fn send_header(&self, bh: &core::BlockHeader) -> Result<(), Error> { if !self.tracking_adapter.has(bh.hash()) { debug!(LOGGER, "Send header {} to {}", bh.hash(), self.info.addr); @@ -220,6 +235,11 @@ impl Peer { self.proto.send_block_request(h) } + pub fn send_compact_block_request(&self, h: Hash) -> Result<(), Error> { + debug!(LOGGER, "Requesting compact block {} from {}", h, self.info.addr); + self.proto.send_compact_block_request(h) + } + pub fn send_peer_request(&self, capab: Capabilities) -> Result<(), Error> { debug!(LOGGER, "Asking {} for more peers.", self.info.addr); self.proto.send_peer_request(capab) @@ -281,6 +301,11 @@ impl ChainAdapter for TrackingAdapter { self.adapter.block_received(b, addr) } + fn compact_block_received(&self, cb: core::CompactBlock, addr: SocketAddr) -> bool { + self.push(cb.hash()); + self.adapter.compact_block_received(cb, addr) + } + fn header_received(&self, bh: core::BlockHeader, addr: SocketAddr) -> bool { self.push(bh.hash()); self.adapter.header_received(bh, addr) diff --git a/p2p/src/peers.rs b/p2p/src/peers.rs index 366ac438..d4481342 100644 --- a/p2p/src/peers.rs +++ b/p2p/src/peers.rs @@ -247,6 +247,30 @@ impl Peers { ); } + pub fn broadcast_compact_block(&self, b: &core::CompactBlock) { + let peers = self.connected_peers(); + let preferred_peers = 8; + let mut count = 0; + for p in peers.iter().take(preferred_peers) { + let p = p.read().unwrap(); + if p.is_connected() { + if let Err(e) = p.send_compact_block(b) { + debug!(LOGGER, "Error sending compact block to peer: {:?}", e); + } else { + count += 1; + } + } + } + debug!( + LOGGER, + "broadcast_compact_block: {}, {} at {}, to {} peers, done.", + b.hash(), + b.header.total_difficulty, + b.header.height, + count, + ); + } + /// Broadcasts the provided block to PEER_PREFERRED_COUNT of our peers. /// We may be connected to PEER_MAX_COUNT peers so we only /// want to broadcast to a random subset of peers. @@ -429,6 +453,16 @@ impl ChainAdapter for Peers { true } } + fn compact_block_received(&self, cb: core::CompactBlock, peer_addr: SocketAddr) -> bool { + if !self.adapter.compact_block_received(cb, peer_addr) { + // if the peer sent us a block that's intrinsically bad + // they are either mistaken or manevolent, both of which require a ban + self.ban_peer(&peer_addr); + false + } else { + true + } + } fn header_received(&self, bh: core::BlockHeader, peer_addr: SocketAddr) -> bool { if !self.adapter.header_received(bh, peer_addr) { // if the peer sent us a block header that's intrinsically bad diff --git a/p2p/src/protocol.rs b/p2p/src/protocol.rs index 283f1efb..d455b4af 100644 --- a/p2p/src/protocol.rs +++ b/p2p/src/protocol.rs @@ -18,6 +18,8 @@ use std::net::SocketAddr; use futures::Future; use futures::sync::mpsc::UnboundedSender; use futures_cpupool::CpuPool; +use rand; +use rand::Rng; use tokio_core::net::TcpStream; use core::core; @@ -83,6 +85,10 @@ impl Protocol for ProtocolV1 { self.send_msg(Type::Block, b) } + fn send_compact_block(&self, cb: &core::CompactBlock) -> Result<(), Error> { + self.send_msg(Type::CompactBlock, cb) + } + /// Serializes and sends a block header to our remote peer ("header first" propagation) fn send_header(&self, bh: &core::BlockHeader) -> Result<(), Error> { self.send_msg(Type::Header, bh) @@ -106,6 +112,10 @@ impl Protocol for ProtocolV1 { self.send_request(Type::GetBlock, Type::Block, &h, Some(h)) } + fn send_compact_block_request(&self, h: Hash) -> Result<(), Error> { + self.send_request(Type::GetCompactBlock, Type::CompactBlock, &h, Some(h)) + } + fn send_peer_request(&self, capab: Capabilities) -> Result<(), Error> { self.send_request( Type::GetPeerAddrs, @@ -211,6 +221,57 @@ fn handle_payload( adapter.block_received(b, addr); Ok(Some(bh)) } + Type::GetCompactBlock => { + let h = ser::deserialize::(&mut &buf[..])?; + debug!(LOGGER, "handle_payload: GetCompactBlock: {}", h); + + if let Some(b) = adapter.get_block(h) { + let cb = b.as_compact_block(); + + // serialize and send the block over in compact representation + let mut body_data = vec![]; + let mut data = vec![]; + + // if we have txs in the block send a compact block + // but if block is empty - + // to allow us to test all code paths, randomly choose to send + // either the block or the compact block + let mut rng = rand::thread_rng(); + + if cb.kern_ids.is_empty() && rng.gen() { + debug!( + LOGGER, + "handle_payload: GetCompactBlock: empty block, sending full block", + ); + + try!(ser::serialize(&mut body_data, &b)); + try!(ser::serialize( + &mut data, + &MsgHeader::new(Type::Block, body_data.len() as u64), + )); + } else { + try!(ser::serialize(&mut body_data, &cb)); + try!(ser::serialize( + &mut data, + &MsgHeader::new(Type::CompactBlock, body_data.len() as u64), + )); + } + + data.append(&mut body_data); + if let Err(e) = sender.unbounded_send(data) { + debug!(LOGGER, "handle_payload: GetCompactBlock, error sending: {:?}", e); + } + } + Ok(None) + } + Type::CompactBlock => { + let b = ser::deserialize::(&mut &buf[..])?; + let bh = b.hash(); + debug!(LOGGER, "handle_payload: CompactBlock: {}", bh); + + adapter.compact_block_received(b, addr); + Ok(Some(bh)) + } // A peer is asking us for some headers via a locator Type::GetHeaders => { let loc = ser::deserialize::(&mut &buf[..])?; diff --git a/p2p/src/server.rs b/p2p/src/server.rs index 2157452c..09d0232d 100644 --- a/p2p/src/server.rs +++ b/p2p/src/server.rs @@ -50,6 +50,7 @@ impl ChainAdapter for DummyAdapter { } fn transaction_received(&self, _tx: core::Transaction) {} fn block_received(&self, _b: core::Block, _addr: SocketAddr) -> bool { true } + fn compact_block_received(&self, _cb: core::CompactBlock, _addr: SocketAddr) -> bool { true } fn header_received(&self, _bh: core::BlockHeader, _addr: SocketAddr) -> bool { true } fn headers_received(&self, _bh: Vec, _addr:SocketAddr) {} fn locate_headers(&self, _loc: Vec) -> Vec { diff --git a/p2p/src/types.rs b/p2p/src/types.rs index 4a7fa3b6..154f54eb 100644 --- a/p2p/src/types.rs +++ b/p2p/src/types.rs @@ -147,6 +147,8 @@ pub trait Protocol { /// Relays a block to the remote peer. fn send_block(&self, b: &core::Block) -> Result<(), Error>; + fn send_compact_block(&self, cb: &core::CompactBlock) -> Result<(), Error>; + /// Relays a block header to the remote peer ("header first" propagation). fn send_header(&self, bh: &core::BlockHeader) -> Result<(), Error>; @@ -159,6 +161,8 @@ pub trait Protocol { /// Sends a request for a block from its hash. fn send_block_request(&self, h: Hash) -> Result<(), Error>; + fn send_compact_block_request(&self, h: Hash) -> Result<(), Error>; + /// Sends a request for some peer addresses. fn send_peer_request(&self, capab: Capabilities) -> Result<(), Error>; @@ -188,7 +192,9 @@ pub trait ChainAdapter: Sync + Send { /// may result in the peer being banned. fn block_received(&self, b: core::Block, addr: SocketAddr) -> bool; - fn header_received(&self, b: core::BlockHeader, addr: SocketAddr) -> bool; + fn compact_block_received(&self, cb: core::CompactBlock, addr: SocketAddr) -> bool; + + fn header_received(&self, bh: core::BlockHeader, addr: SocketAddr) -> bool; /// A set of block header has been received, typically in response to a /// block