2021-10-21 18:26:04 -07:00
|
|
|
//! Abstractions that represent "the rest of the network".
|
|
|
|
//!
|
|
|
|
//! # Implementation
|
|
|
|
//!
|
2021-11-23 16:31:42 -08:00
|
|
|
//! The [`PeerSet`] implementation is adapted from the one in the [Tower Balance][tower-balance] crate.
|
|
|
|
//! As described in that crate's documentation, it:
|
2021-10-21 18:26:04 -07:00
|
|
|
//!
|
|
|
|
//! > Distributes requests across inner services using the [Power of Two Choices][p2c].
|
|
|
|
//! >
|
|
|
|
//! > As described in the [Finagle Guide][finagle]:
|
|
|
|
//! >
|
|
|
|
//! > > The algorithm randomly picks two services from the set of ready endpoints and
|
|
|
|
//! > > selects the least loaded of the two. By repeatedly using this strategy, we can
|
|
|
|
//! > > expect a manageable upper bound on the maximum load of any server.
|
|
|
|
//! > >
|
|
|
|
//! > > The maximum load variance between any two servers is bound by `ln(ln(n))` where
|
|
|
|
//! > > `n` is the number of servers in the cluster.
|
|
|
|
//!
|
2021-11-23 16:31:42 -08:00
|
|
|
//! The Power of Two Choices should work well for many network requests, but not all of them.
|
|
|
|
//! Some requests should only be made to a subset of connected peers.
|
|
|
|
//! For example, a request for a particular inventory item
|
|
|
|
//! should be made to a peer that has recently advertised that inventory hash.
|
|
|
|
//! Other requests require broadcasts, such as transaction diffusion.
|
2021-10-21 18:26:04 -07:00
|
|
|
//!
|
|
|
|
//! Implementing this specialized routing logic inside the `PeerSet` -- so that
|
|
|
|
//! it continues to abstract away "the rest of the network" into one endpoint --
|
|
|
|
//! is not a problem, as the `PeerSet` can simply maintain more information on
|
|
|
|
//! its peers and route requests appropriately. However, there is a problem with
|
|
|
|
//! maintaining accurate backpressure information, because the `Service` trait
|
|
|
|
//! requires that service readiness is independent of the data in the request.
|
|
|
|
//!
|
|
|
|
//! For this reason, in the future, this code will probably be refactored to
|
|
|
|
//! address this backpressure mismatch. One possibility is to refactor the code
|
|
|
|
//! so that one entity holds and maintains the peer set and metadata on the
|
|
|
|
//! peers, and each "backpressure category" of request is assigned to different
|
|
|
|
//! `Service` impls with specialized `poll_ready()` implementations. Another
|
|
|
|
//! less-elegant solution (which might be useful as an intermediate step for the
|
|
|
|
//! inventory case) is to provide a way to borrow a particular backing service,
|
|
|
|
//! say by address.
|
|
|
|
//!
|
|
|
|
//! [finagle]: https://twitter.github.io/finagle/guide/Clients.html#power-of-two-choices-p2c-least-loaded
|
|
|
|
//! [p2c]: http://www.eecs.harvard.edu/~michaelm/postscripts/handbook2001.pdf
|
|
|
|
//! [tower-balance]: https://crates.io/crates/tower-balance
|
|
|
|
|
2019-10-10 18:15:24 -07:00
|
|
|
use std::{
|
2021-11-23 16:31:42 -08:00
|
|
|
collections::{HashMap, HashSet},
|
2021-11-18 05:28:25 -08:00
|
|
|
convert,
|
2019-10-10 18:15:24 -07:00
|
|
|
fmt::Debug,
|
2019-12-13 14:25:14 -08:00
|
|
|
future::Future,
|
2019-10-10 18:15:24 -07:00
|
|
|
marker::PhantomData,
|
2021-10-21 18:26:04 -07:00
|
|
|
net::SocketAddr,
|
2019-10-10 18:15:24 -07:00
|
|
|
pin::Pin,
|
2021-04-18 23:04:24 -07:00
|
|
|
sync::Arc,
|
2019-10-10 18:15:24 -07:00
|
|
|
task::{Context, Poll},
|
2020-11-29 22:41:14 -08:00
|
|
|
time::Instant,
|
2019-10-10 18:15:24 -07:00
|
|
|
};
|
|
|
|
|
2019-10-21 15:24:17 -07:00
|
|
|
use futures::{
|
|
|
|
channel::{mpsc, oneshot},
|
2021-11-30 13:04:32 -08:00
|
|
|
future::{FutureExt, TryFutureExt},
|
2019-12-13 14:25:14 -08:00
|
|
|
prelude::*,
|
2019-10-21 15:24:17 -07:00
|
|
|
stream::FuturesUnordered,
|
|
|
|
};
|
2021-10-21 18:26:04 -07:00
|
|
|
use tokio::{
|
|
|
|
sync::{broadcast, oneshot::error::TryRecvError},
|
|
|
|
task::JoinHandle,
|
|
|
|
};
|
2019-10-10 18:15:24 -07:00
|
|
|
use tower::{
|
|
|
|
discover::{Change, Discover},
|
2020-09-21 14:00:20 -07:00
|
|
|
load::Load,
|
2019-10-10 18:15:24 -07:00
|
|
|
Service,
|
|
|
|
};
|
|
|
|
|
|
|
|
use crate::{
|
2021-10-21 18:26:04 -07:00
|
|
|
peer_set::{
|
|
|
|
unready_service::{Error as UnreadyError, UnreadyService},
|
|
|
|
InventoryRegistry,
|
|
|
|
},
|
2020-09-01 14:28:54 -07:00
|
|
|
protocol::{
|
|
|
|
external::InventoryHash,
|
|
|
|
internal::{Request, Response},
|
|
|
|
},
|
2021-10-27 18:49:31 -07:00
|
|
|
AddressBook, BoxError, Config,
|
2019-10-10 18:15:24 -07:00
|
|
|
};
|
|
|
|
|
2021-10-21 18:26:04 -07:00
|
|
|
/// A signal sent by the [`PeerSet`] when it has no ready peers, and gets a request from Zebra.
|
|
|
|
///
|
|
|
|
/// In response to this signal, the crawler tries to open more peer connections.
|
|
|
|
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
|
|
|
|
pub struct MorePeers;
|
|
|
|
|
|
|
|
/// A signal sent by the [`PeerSet`] to cancel a [`Client`]'s current request or response.
|
|
|
|
///
|
|
|
|
/// When it receives this signal, the [`Client`] stops processing and exits.
|
|
|
|
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
|
|
|
|
pub struct CancelClientWork;
|
2019-10-10 18:15:24 -07:00
|
|
|
|
|
|
|
/// A [`tower::Service`] that abstractly represents "the rest of the network".
|
|
|
|
///
|
2021-05-06 17:50:04 -07:00
|
|
|
/// # Security
|
|
|
|
///
|
|
|
|
/// The `Discover::Key` must be the transient remote address of each peer. This
|
|
|
|
/// address may only be valid for the duration of a single connection. (For
|
|
|
|
/// example, inbound connections have an ephemeral remote port, and proxy
|
|
|
|
/// connections have an ephemeral local or proxy port.)
|
|
|
|
///
|
|
|
|
/// Otherwise, malicious peers could interfere with other peers' `PeerSet` state.
|
2019-10-10 18:15:24 -07:00
|
|
|
pub struct PeerSet<D>
|
|
|
|
where
|
2021-11-22 17:29:34 -08:00
|
|
|
D: Discover<Key = SocketAddr> + Unpin,
|
|
|
|
D::Service: Service<Request, Response = Response> + Load,
|
|
|
|
D::Error: Into<BoxError>,
|
|
|
|
<D::Service as Service<Request>>::Error: Into<BoxError> + 'static,
|
|
|
|
<D::Service as Service<Request>>::Future: Send + 'static,
|
|
|
|
<D::Service as Load>::Metric: Debug,
|
2019-10-10 18:15:24 -07:00
|
|
|
{
|
2021-11-22 17:29:34 -08:00
|
|
|
/// Provides new and deleted peer [`Change`]s to the peer set,
|
|
|
|
/// via the [`Discover`] trait implementation.
|
2019-10-10 18:15:24 -07:00
|
|
|
discover: D,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
|
|
|
/// Connected peers that are ready to receive requests from Zebra,
|
|
|
|
/// or send requests to Zebra.
|
2021-11-23 16:31:42 -08:00
|
|
|
ready_services: HashMap<D::Key, D::Service>,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// A preselected ready service.
|
|
|
|
///
|
|
|
|
/// # Correctness
|
|
|
|
///
|
|
|
|
/// If this is `Some(addr)`, `addr` must be a key for a peer in `ready_services`.
|
|
|
|
/// If that peer is removed from `ready_services`, we must set the preselected peer to `None`.
|
|
|
|
///
|
2021-11-30 13:04:32 -08:00
|
|
|
/// This is handled by [`PeerSet::take_ready_service`].
|
2021-11-23 16:31:42 -08:00
|
|
|
preselected_p2c_peer: Option<D::Key>,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Stores gossiped inventory hashes from connected peers.
|
|
|
|
///
|
2021-11-22 17:29:34 -08:00
|
|
|
/// Used to route inventory requests to peers that are likely to have it.
|
|
|
|
inventory_registry: InventoryRegistry,
|
|
|
|
|
|
|
|
/// Connected peers that are handling a Zebra request,
|
|
|
|
/// or Zebra is handling one of their requests.
|
2019-10-10 18:15:24 -07:00
|
|
|
unready_services: FuturesUnordered<UnreadyService<D::Key, D::Service, Request>>,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
|
|
|
/// Channels used to cancel the request that an unready service is doing.
|
|
|
|
cancel_handles: HashMap<D::Key, oneshot::Sender<CancelClientWork>>,
|
|
|
|
|
|
|
|
/// A channel that asks the peer crawler task to connect to more peers.
|
2021-10-21 18:26:04 -07:00
|
|
|
demand_signal: mpsc::Sender<MorePeers>,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
2020-06-09 12:24:28 -07:00
|
|
|
/// Channel for passing ownership of tokio JoinHandles from PeerSet's background tasks
|
|
|
|
///
|
|
|
|
/// The join handles passed into the PeerSet are used populate the `guards` member
|
2020-09-18 11:20:55 -07:00
|
|
|
handle_rx: tokio::sync::oneshot::Receiver<Vec<JoinHandle<Result<(), BoxError>>>>,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
2020-06-09 12:24:28 -07:00
|
|
|
/// Unordered set of handles to background tasks associated with the `PeerSet`
|
|
|
|
///
|
|
|
|
/// These guards are checked for errors as part of `poll_ready` which lets
|
|
|
|
/// the `PeerSet` propagate errors from background tasks back to the user
|
2020-09-18 11:20:55 -07:00
|
|
|
guards: futures::stream::FuturesUnordered<JoinHandle<Result<(), BoxError>>>,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
2021-03-15 05:02:12 -07:00
|
|
|
/// A shared list of peer addresses.
|
|
|
|
///
|
|
|
|
/// Used for logging diagnostics.
|
2021-04-18 23:04:24 -07:00
|
|
|
address_book: Arc<std::sync::Mutex<AddressBook>>,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
|
|
|
/// The last time we logged a message about the peer set size
|
|
|
|
last_peer_log: Option<Instant>,
|
|
|
|
|
2021-10-27 18:49:31 -07:00
|
|
|
/// The configured limit for inbound and outbound connections.
|
2021-11-22 17:29:34 -08:00
|
|
|
///
|
|
|
|
/// The peer set panics if this size is exceeded.
|
|
|
|
/// If that happens, our connection limit code has a bug.
|
2021-10-27 18:49:31 -07:00
|
|
|
peerset_total_connection_limit: usize,
|
2019-10-10 18:15:24 -07:00
|
|
|
}
|
|
|
|
|
2021-11-22 17:29:34 -08:00
|
|
|
impl<D> Drop for PeerSet<D>
|
|
|
|
where
|
|
|
|
D: Discover<Key = SocketAddr> + Unpin,
|
|
|
|
D::Service: Service<Request, Response = Response> + Load,
|
|
|
|
D::Error: Into<BoxError>,
|
|
|
|
<D::Service as Service<Request>>::Error: Into<BoxError> + 'static,
|
|
|
|
<D::Service as Service<Request>>::Future: Send + 'static,
|
|
|
|
<D::Service as Load>::Metric: Debug,
|
|
|
|
{
|
|
|
|
fn drop(&mut self) {
|
|
|
|
self.shut_down_tasks_and_channels()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-10-10 18:15:24 -07:00
|
|
|
impl<D> PeerSet<D>
|
|
|
|
where
|
2020-09-01 14:28:54 -07:00
|
|
|
D: Discover<Key = SocketAddr> + Unpin,
|
2019-10-10 18:15:24 -07:00
|
|
|
D::Service: Service<Request, Response = Response> + Load,
|
2020-09-18 11:20:55 -07:00
|
|
|
D::Error: Into<BoxError>,
|
|
|
|
<D::Service as Service<Request>>::Error: Into<BoxError> + 'static,
|
2019-10-10 18:15:24 -07:00
|
|
|
<D::Service as Service<Request>>::Future: Send + 'static,
|
|
|
|
<D::Service as Load>::Metric: Debug,
|
|
|
|
{
|
2021-11-18 05:28:25 -08:00
|
|
|
/// Construct a peerset which uses `discover` to manage peer connections.
|
|
|
|
///
|
|
|
|
/// Arguments:
|
|
|
|
/// - `config`: configures the peer set connection limit;
|
|
|
|
/// - `discover`: handles peer connects and disconnects;
|
|
|
|
/// - `demand_signal`: requests more peers when all peers are busy (unready);
|
|
|
|
/// - `handle_rx`: receives background task handles,
|
|
|
|
/// monitors them to make sure they're still running,
|
|
|
|
/// and shuts down all the tasks as soon as one task exits;
|
|
|
|
/// - `inv_stream`: receives inventory advertisements for peers,
|
|
|
|
/// allowing the peer set to direct inventory requests;
|
|
|
|
/// - `address_book`: when peer set is busy, it logs address book diagnostics.
|
2020-06-09 12:24:28 -07:00
|
|
|
pub fn new(
|
2021-10-27 18:49:31 -07:00
|
|
|
config: &Config,
|
2020-06-09 12:24:28 -07:00
|
|
|
discover: D,
|
2021-10-21 18:26:04 -07:00
|
|
|
demand_signal: mpsc::Sender<MorePeers>,
|
2020-09-18 11:20:55 -07:00
|
|
|
handle_rx: tokio::sync::oneshot::Receiver<Vec<JoinHandle<Result<(), BoxError>>>>,
|
2020-09-01 14:28:54 -07:00
|
|
|
inv_stream: broadcast::Receiver<(InventoryHash, SocketAddr)>,
|
2021-04-18 23:04:24 -07:00
|
|
|
address_book: Arc<std::sync::Mutex<AddressBook>>,
|
2020-06-09 12:24:28 -07:00
|
|
|
) -> Self {
|
2019-10-10 18:15:24 -07:00
|
|
|
Self {
|
2021-11-22 17:29:34 -08:00
|
|
|
// Ready peers
|
2019-10-10 18:15:24 -07:00
|
|
|
discover,
|
2021-11-23 16:31:42 -08:00
|
|
|
ready_services: HashMap::new(),
|
|
|
|
preselected_p2c_peer: None,
|
2021-11-22 17:29:34 -08:00
|
|
|
inventory_registry: InventoryRegistry::new(inv_stream),
|
|
|
|
|
|
|
|
// Unready peers
|
2019-10-10 18:15:24 -07:00
|
|
|
unready_services: FuturesUnordered::new(),
|
2021-11-22 17:29:34 -08:00
|
|
|
cancel_handles: HashMap::new(),
|
2019-10-21 15:24:17 -07:00
|
|
|
demand_signal,
|
2021-11-22 17:29:34 -08:00
|
|
|
|
|
|
|
// Background tasks
|
2020-06-09 12:24:28 -07:00
|
|
|
handle_rx,
|
2021-11-22 17:29:34 -08:00
|
|
|
guards: futures::stream::FuturesUnordered::new(),
|
|
|
|
|
|
|
|
// Metrics
|
2020-11-29 22:41:14 -08:00
|
|
|
last_peer_log: None,
|
2021-03-15 05:02:12 -07:00
|
|
|
address_book,
|
2021-10-27 18:49:31 -07:00
|
|
|
peerset_total_connection_limit: config.peerset_total_connection_limit(),
|
2019-10-10 18:15:24 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-18 05:28:25 -08:00
|
|
|
/// Check background task handles to make sure they're still running.
|
|
|
|
///
|
|
|
|
/// If any background task exits, shuts down all other background tasks,
|
|
|
|
/// and returns an error.
|
2020-11-23 23:52:40 -08:00
|
|
|
fn poll_background_errors(&mut self, cx: &mut Context) -> Result<(), BoxError> {
|
2021-11-22 17:29:34 -08:00
|
|
|
if let Some(result) = self.receive_tasks_if_needed() {
|
|
|
|
return result;
|
2020-06-09 12:24:28 -07:00
|
|
|
}
|
|
|
|
|
2021-11-22 17:29:34 -08:00
|
|
|
match Pin::new(&mut self.guards).poll_next(cx) {
|
|
|
|
// All background tasks are still running.
|
|
|
|
Poll::Pending => Ok(()),
|
|
|
|
|
2021-11-18 05:28:25 -08:00
|
|
|
Poll::Ready(Some(res)) => {
|
|
|
|
info!(
|
|
|
|
background_tasks = %self.guards.len(),
|
|
|
|
"a peer set background task exited, shutting down other peer set tasks"
|
|
|
|
);
|
|
|
|
|
2021-11-22 17:29:34 -08:00
|
|
|
self.shut_down_tasks_and_channels();
|
|
|
|
|
2021-11-18 05:28:25 -08:00
|
|
|
// Flatten the join result and inner result,
|
|
|
|
// then turn Ok() task exits into errors.
|
|
|
|
res.map_err(Into::into)
|
2021-11-22 17:29:34 -08:00
|
|
|
// TODO: replace with Result::flatten when it stabilises (#70142)
|
2021-11-18 05:28:25 -08:00
|
|
|
.and_then(convert::identity)
|
|
|
|
.and(Err("a peer set background task exited".into()))
|
|
|
|
}
|
|
|
|
|
2021-11-22 17:29:34 -08:00
|
|
|
Poll::Ready(None) => {
|
|
|
|
self.shut_down_tasks_and_channels();
|
|
|
|
Err("all peer set background tasks have exited".into())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Receive background tasks, if they've been sent on the channel,
|
|
|
|
/// but not consumed yet.
|
|
|
|
///
|
|
|
|
/// Returns a result representing the current task state,
|
|
|
|
/// or `None` if the background tasks should be polled to check their state.
|
|
|
|
fn receive_tasks_if_needed(&mut self) -> Option<Result<(), BoxError>> {
|
|
|
|
if self.guards.is_empty() {
|
|
|
|
match self.handle_rx.try_recv() {
|
|
|
|
// The tasks haven't been sent yet.
|
|
|
|
Err(TryRecvError::Empty) => Some(Ok(())),
|
|
|
|
|
|
|
|
// The tasks have been sent, but not consumed.
|
|
|
|
Ok(handles) => {
|
|
|
|
// Currently, the peer set treats an empty background task set as an error.
|
|
|
|
//
|
|
|
|
// TODO: refactor `handle_rx` and `guards` into an enum
|
|
|
|
// for the background task state: Waiting/Running/Shutdown.
|
|
|
|
assert!(
|
|
|
|
!handles.is_empty(),
|
|
|
|
"the peer set requires at least one background task"
|
|
|
|
);
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
self.guards.extend(handles);
|
2021-11-22 17:29:34 -08:00
|
|
|
|
|
|
|
None
|
|
|
|
}
|
|
|
|
|
|
|
|
// The tasks have been sent and consumed, but then they exited.
|
|
|
|
//
|
|
|
|
// Correctness: the peer set must receive at least one task.
|
|
|
|
//
|
|
|
|
// TODO: refactor `handle_rx` and `guards` into an enum
|
|
|
|
// for the background task state: Waiting/Running/Shutdown.
|
|
|
|
Err(TryRecvError::Closed) => {
|
|
|
|
Some(Err("all peer set background tasks have exited".into()))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
None
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Shut down:
|
|
|
|
/// - services by dropping the service lists
|
|
|
|
/// - background tasks via their join handles or cancel handles
|
|
|
|
/// - channels by closing the channel
|
|
|
|
fn shut_down_tasks_and_channels(&mut self) {
|
|
|
|
// Drop services and cancel their background tasks.
|
2021-11-23 16:31:42 -08:00
|
|
|
self.preselected_p2c_peer = None;
|
|
|
|
self.ready_services = HashMap::new();
|
2021-11-22 17:29:34 -08:00
|
|
|
|
|
|
|
for (_peer_key, handle) in self.cancel_handles.drain() {
|
|
|
|
let _ = handle.send(CancelClientWork);
|
|
|
|
}
|
|
|
|
self.unready_services = FuturesUnordered::new();
|
|
|
|
|
|
|
|
// Close the MorePeers channel for all senders,
|
|
|
|
// so we don't add more peers to a shut down peer set.
|
|
|
|
self.demand_signal.close_channel();
|
|
|
|
|
|
|
|
// Shut down background tasks.
|
|
|
|
self.handle_rx.close();
|
|
|
|
self.receive_tasks_if_needed();
|
2021-11-18 05:28:25 -08:00
|
|
|
for guard in self.guards.iter() {
|
|
|
|
guard.abort();
|
2020-06-09 12:24:28 -07:00
|
|
|
}
|
|
|
|
|
2021-11-22 17:29:34 -08:00
|
|
|
// TODO: implement graceful shutdown for InventoryRegistry (#1678)
|
2020-06-09 12:24:28 -07:00
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Check busy peer services for request completion or errors.
|
|
|
|
///
|
|
|
|
/// Move newly ready services to the ready list, and drop failed services.
|
2019-10-10 18:15:24 -07:00
|
|
|
fn poll_unready(&mut self, cx: &mut Context<'_>) {
|
|
|
|
loop {
|
|
|
|
match Pin::new(&mut self.unready_services).poll_next(cx) {
|
2021-11-23 16:31:42 -08:00
|
|
|
// No unready service changes, or empty unready services
|
2019-10-10 18:15:24 -07:00
|
|
|
Poll::Pending | Poll::Ready(None) => return,
|
2021-11-23 16:31:42 -08:00
|
|
|
|
|
|
|
// Unready -> Ready
|
2019-10-10 18:15:24 -07:00
|
|
|
Poll::Ready(Some(Ok((key, svc)))) => {
|
|
|
|
trace!(?key, "service became ready");
|
2021-11-23 16:31:42 -08:00
|
|
|
let cancel = self.cancel_handles.remove(&key);
|
|
|
|
assert!(cancel.is_some(), "missing cancel handle");
|
2019-10-10 18:15:24 -07:00
|
|
|
self.ready_services.insert(key, svc);
|
|
|
|
}
|
2021-11-23 16:31:42 -08:00
|
|
|
|
|
|
|
// Unready -> Canceled
|
2019-10-10 18:15:24 -07:00
|
|
|
Poll::Ready(Some(Err((key, UnreadyError::Canceled)))) => {
|
2021-11-23 16:31:42 -08:00
|
|
|
// A service be canceled because we've connected to the same service twice.
|
|
|
|
// In that case, there is a cancel handle for the peer address,
|
|
|
|
// but it belongs to the service for the newer connection.
|
|
|
|
trace!(
|
|
|
|
?key,
|
|
|
|
duplicate_connection = self.cancel_handles.contains_key(&key),
|
|
|
|
"service was canceled, dropping service"
|
|
|
|
);
|
2019-10-10 18:15:24 -07:00
|
|
|
}
|
2021-11-23 16:31:42 -08:00
|
|
|
|
|
|
|
// Unready -> Errored
|
2019-10-10 18:15:24 -07:00
|
|
|
Poll::Ready(Some(Err((key, UnreadyError::Inner(e))))) => {
|
|
|
|
let error = e.into();
|
2021-11-23 16:31:42 -08:00
|
|
|
debug!(%error, "service failed while unready, dropping service");
|
|
|
|
|
|
|
|
let cancel = self.cancel_handles.remove(&key);
|
|
|
|
assert!(cancel.is_some(), "missing cancel handle");
|
2019-10-10 18:15:24 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Checks for newly inserted or removed services.
|
|
|
|
///
|
|
|
|
/// Puts inserted services in the unready list.
|
|
|
|
/// Drops removed services, after cancelling any pending requests.
|
2020-11-23 23:52:40 -08:00
|
|
|
fn poll_discover(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>> {
|
|
|
|
use futures::ready;
|
|
|
|
loop {
|
|
|
|
match ready!(Pin::new(&mut self.discover).poll_discover(cx))
|
|
|
|
.ok_or("discovery stream closed")?
|
|
|
|
.map_err(Into::into)?
|
|
|
|
{
|
|
|
|
Change::Remove(key) => {
|
|
|
|
trace!(?key, "got Change::Remove from Discover");
|
|
|
|
self.remove(&key);
|
|
|
|
}
|
|
|
|
Change::Insert(key, svc) => {
|
|
|
|
trace!(?key, "got Change::Insert from Discover");
|
|
|
|
self.remove(&key);
|
|
|
|
self.push_unready(key, svc);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Takes a ready service by key, invalidating `preselected_p2c_peer` if needed.
|
|
|
|
fn take_ready_service(&mut self, key: &D::Key) -> Option<D::Service> {
|
|
|
|
if let Some(svc) = self.ready_services.remove(key) {
|
|
|
|
if Some(*key) == self.preselected_p2c_peer {
|
|
|
|
self.preselected_p2c_peer = None;
|
|
|
|
}
|
|
|
|
|
|
|
|
assert!(
|
|
|
|
!self.cancel_handles.contains_key(key),
|
|
|
|
"cancel handles are only used for unready service work"
|
|
|
|
);
|
|
|
|
|
|
|
|
Some(svc)
|
2020-11-23 23:52:40 -08:00
|
|
|
} else {
|
|
|
|
None
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Remove the service corresponding to `key` from the peer set.
|
|
|
|
///
|
|
|
|
/// Drops the service, cancelling any pending request or response to that peer.
|
|
|
|
/// If the peer does not exist, does nothing.
|
2020-11-23 23:52:40 -08:00
|
|
|
fn remove(&mut self, key: &D::Key) {
|
2021-11-23 16:31:42 -08:00
|
|
|
if let Some(ready_service) = self.take_ready_service(key) {
|
|
|
|
// A ready service has no work to cancel, so just drop it.
|
|
|
|
std::mem::drop(ready_service);
|
2020-11-23 23:52:40 -08:00
|
|
|
} else if let Some(handle) = self.cancel_handles.remove(key) {
|
2021-11-23 16:31:42 -08:00
|
|
|
// Cancel the work, implicitly dropping the cancel handle.
|
|
|
|
// The service future returns a `Canceled` error,
|
|
|
|
// making `poll_unready` drop the service.
|
2021-10-21 18:26:04 -07:00
|
|
|
let _ = handle.send(CancelClientWork);
|
2020-11-23 23:52:40 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Adds a busy service to the unready list,
|
|
|
|
/// and adds a cancel handle for the service's current request.
|
2020-11-23 23:52:40 -08:00
|
|
|
fn push_unready(&mut self, key: D::Key, svc: D::Service) {
|
|
|
|
let (tx, rx) = oneshot::channel();
|
|
|
|
self.cancel_handles.insert(key, tx);
|
|
|
|
self.unready_services.push(UnreadyService {
|
|
|
|
key: Some(key),
|
|
|
|
service: Some(svc),
|
|
|
|
cancel: rx,
|
|
|
|
_req: PhantomData,
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Performs P2C on `self.ready_services` to randomly select a less-loaded ready service.
|
|
|
|
fn preselect_p2c_peer(&self) -> Option<D::Key> {
|
2021-11-30 13:04:32 -08:00
|
|
|
self.select_p2c_peer_from_list(&self.ready_services.keys().copied().collect())
|
2021-11-23 16:31:42 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
/// Performs P2C on `ready_service_list` to randomly select a less-loaded ready service.
|
2021-11-30 13:04:32 -08:00
|
|
|
fn select_p2c_peer_from_list(&self, ready_service_list: &HashSet<D::Key>) -> Option<D::Key> {
|
2021-11-23 16:31:42 -08:00
|
|
|
match ready_service_list.len() {
|
2019-10-10 18:15:24 -07:00
|
|
|
0 => None,
|
2021-11-23 16:31:42 -08:00
|
|
|
1 => Some(
|
2021-11-30 13:04:32 -08:00
|
|
|
*ready_service_list
|
|
|
|
.iter()
|
2021-11-23 16:31:42 -08:00
|
|
|
.next()
|
|
|
|
.expect("just checked there is one service"),
|
|
|
|
),
|
2019-10-10 18:15:24 -07:00
|
|
|
len => {
|
2021-11-23 16:31:42 -08:00
|
|
|
// If there are only 2 peers, randomise their order.
|
|
|
|
// Otherwise, choose 2 random peers in a random order.
|
2019-10-10 18:15:24 -07:00
|
|
|
let (a, b) = {
|
|
|
|
let idxs = rand::seq::index::sample(&mut rand::thread_rng(), len, 2);
|
2021-11-23 16:31:42 -08:00
|
|
|
let a = idxs.index(0);
|
|
|
|
let b = idxs.index(1);
|
|
|
|
|
|
|
|
let a = *ready_service_list
|
|
|
|
.iter()
|
|
|
|
.nth(a)
|
|
|
|
.expect("sample returns valid indexes");
|
|
|
|
let b = *ready_service_list
|
|
|
|
.iter()
|
|
|
|
.nth(b)
|
|
|
|
.expect("sample returns valid indexes");
|
|
|
|
|
|
|
|
(a, b)
|
2019-10-10 18:15:24 -07:00
|
|
|
};
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
let a_load = self.query_load(&a).expect("supplied services are ready");
|
|
|
|
let b_load = self.query_load(&b).expect("supplied services are ready");
|
2019-10-10 18:15:24 -07:00
|
|
|
|
|
|
|
let selected = if a_load <= b_load { a } else { b };
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
trace!(
|
|
|
|
a.key = ?a,
|
|
|
|
a.load = ?a_load,
|
|
|
|
b.key = ?b,
|
|
|
|
b.load = ?b_load,
|
|
|
|
selected = ?selected,
|
|
|
|
?len,
|
|
|
|
"selected service by p2c"
|
|
|
|
);
|
2019-10-10 18:15:24 -07:00
|
|
|
|
|
|
|
Some(selected)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-30 13:04:32 -08:00
|
|
|
/// Randomly chooses `max_peers` ready services, ignoring service load.
|
|
|
|
///
|
|
|
|
/// The chosen peers are unique, but their order is not fully random.
|
|
|
|
fn select_random_ready_peers(&self, max_peers: usize) -> Vec<D::Key> {
|
|
|
|
use rand::seq::IteratorRandom;
|
|
|
|
|
|
|
|
self.ready_services
|
|
|
|
.keys()
|
|
|
|
.copied()
|
|
|
|
.choose_multiple(&mut rand::thread_rng(), max_peers)
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Accesses a ready endpoint by `key` and returns its current load.
|
|
|
|
///
|
|
|
|
/// Returns `None` if the service is not in the ready service list.
|
|
|
|
fn query_load(&self, key: &D::Key) -> Option<<D::Service as Load>::Metric> {
|
|
|
|
let svc = self.ready_services.get(key);
|
|
|
|
svc.map(|svc| svc.load())
|
2019-10-10 18:15:24 -07:00
|
|
|
}
|
2020-09-01 14:28:54 -07:00
|
|
|
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
/// Routes a request using P2C load-balancing.
|
|
|
|
fn route_p2c(&mut self, req: Request) -> <Self as tower::Service<Request>>::Future {
|
2021-11-23 16:31:42 -08:00
|
|
|
let preselected_key = self
|
|
|
|
.preselected_p2c_peer
|
|
|
|
.expect("ready peer service must have a preselected peer");
|
|
|
|
|
|
|
|
tracing::trace!(?preselected_key, "routing based on p2c");
|
2020-09-01 14:28:54 -07:00
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
let mut svc = self
|
|
|
|
.take_ready_service(&preselected_key)
|
|
|
|
.expect("ready peer set must have preselected a ready peer");
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
|
|
|
|
let fut = svc.call(req);
|
2021-11-23 16:31:42 -08:00
|
|
|
self.push_unready(preselected_key, svc);
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
fut.map_err(Into::into).boxed()
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Tries to route a request to a peer that advertised that inventory,
|
|
|
|
/// falling back to P2C if there is no ready peer.
|
|
|
|
fn route_inv(
|
|
|
|
&mut self,
|
|
|
|
req: Request,
|
|
|
|
hash: InventoryHash,
|
|
|
|
) -> <Self as tower::Service<Request>>::Future {
|
2021-11-23 16:31:42 -08:00
|
|
|
let inventory_peer_list = self
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
.inventory_registry
|
|
|
|
.peers(&hash)
|
2021-11-23 16:31:42 -08:00
|
|
|
.filter(|&key| self.ready_services.contains_key(key))
|
|
|
|
.copied()
|
|
|
|
.collect();
|
|
|
|
|
|
|
|
// # Security
|
|
|
|
//
|
|
|
|
// Choose a random, less-loaded peer with the inventory.
|
|
|
|
//
|
|
|
|
// If we chose the first peer in HashMap order,
|
|
|
|
// peers would be able to influence our choice by switching addresses.
|
|
|
|
// But we need the choice to be random,
|
|
|
|
// so that a peer can't provide all our inventory responses.
|
2021-11-30 13:04:32 -08:00
|
|
|
let peer = self.select_p2c_peer_from_list(&inventory_peer_list);
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
|
2020-11-23 23:52:40 -08:00
|
|
|
match peer.and_then(|key| self.take_ready_service(&key)) {
|
2021-11-23 16:31:42 -08:00
|
|
|
Some(mut svc) => {
|
|
|
|
let peer = peer.expect("just checked peer is Some");
|
|
|
|
tracing::trace!(?hash, ?peer, "routing based on inventory");
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
let fut = svc.call(req);
|
2021-11-23 16:31:42 -08:00
|
|
|
self.push_unready(peer, svc);
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
fut.map_err(Into::into).boxed()
|
|
|
|
}
|
|
|
|
None => {
|
2021-11-23 16:31:42 -08:00
|
|
|
tracing::trace!(?hash, "no ready peer for inventory, falling back to p2c");
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
self.route_p2c(req)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-30 13:04:32 -08:00
|
|
|
/// Routes the same request to up to `max_peers` ready peers, ignoring return values.
|
|
|
|
///
|
|
|
|
/// `max_peers` must be at least one, and at most the number of ready peers.
|
|
|
|
fn route_multiple(
|
|
|
|
&mut self,
|
|
|
|
req: Request,
|
|
|
|
max_peers: usize,
|
|
|
|
) -> <Self as tower::Service<Request>>::Future {
|
|
|
|
assert!(
|
|
|
|
max_peers > 0,
|
|
|
|
"requests must be routed to at least one peer"
|
|
|
|
);
|
|
|
|
assert!(
|
|
|
|
max_peers <= self.ready_services.len(),
|
|
|
|
"requests can only be routed to ready peers"
|
|
|
|
);
|
|
|
|
|
|
|
|
// # Security
|
|
|
|
//
|
|
|
|
// We choose peers randomly, ignoring load.
|
|
|
|
// This avoids favouring malicious peers, because peers can influence their own load.
|
|
|
|
//
|
|
|
|
// The order of peers isn't completely random,
|
|
|
|
// but peer request order is not security-sensitive.
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
|
|
|
|
let futs = FuturesUnordered::new();
|
2021-11-30 13:04:32 -08:00
|
|
|
for key in self.select_random_ready_peers(max_peers) {
|
|
|
|
let mut svc = self
|
|
|
|
.take_ready_service(&key)
|
|
|
|
.expect("selected peers are ready");
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
futs.push(svc.call(req.clone()).map_err(|_| ()));
|
|
|
|
self.push_unready(key, svc);
|
|
|
|
}
|
|
|
|
|
|
|
|
async move {
|
|
|
|
let results = futs.collect::<Vec<Result<_, _>>>().await;
|
|
|
|
tracing::debug!(
|
|
|
|
ok.len = results.iter().filter(|r| r.is_ok()).count(),
|
|
|
|
err.len = results.iter().filter(|r| r.is_err()).count(),
|
2021-11-30 13:04:32 -08:00
|
|
|
"sent peer request to multiple peers"
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
);
|
|
|
|
Ok(Response::Nil)
|
|
|
|
}
|
|
|
|
.boxed()
|
2020-09-01 14:28:54 -07:00
|
|
|
}
|
2020-11-23 23:52:40 -08:00
|
|
|
|
2021-11-30 13:04:32 -08:00
|
|
|
/// Broadcasts the same request to lots of ready peers, ignoring return values.
|
|
|
|
fn route_broadcast(&mut self, req: Request) -> <Self as tower::Service<Request>>::Future {
|
|
|
|
// Round up, so that if we have one ready peer, it gets the request
|
|
|
|
let half_ready_peers = (self.ready_services.len() + 1) / 2;
|
|
|
|
|
|
|
|
// Broadcasts ignore the response
|
|
|
|
self.route_multiple(req, half_ready_peers)
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Logs the peer set size.
|
2020-11-29 22:41:14 -08:00
|
|
|
fn log_peer_set_size(&mut self) {
|
|
|
|
let ready_services_len = self.ready_services.len();
|
|
|
|
let unready_services_len = self.unready_services.len();
|
|
|
|
trace!(ready_peers = ?ready_services_len, unready_peers = ?unready_services_len);
|
|
|
|
|
|
|
|
if ready_services_len > 0 {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
// These logs are designed to be human-readable in a terminal, at the
|
|
|
|
// default Zebra log level. If you need to know the peer set size for
|
|
|
|
// every request, use the trace-level logs, or the metrics exporter.
|
|
|
|
if let Some(last_peer_log) = self.last_peer_log {
|
|
|
|
// Avoid duplicate peer set logs
|
|
|
|
if Instant::now().duration_since(last_peer_log).as_secs() < 60 {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// Suppress initial logs until the peer set has started up.
|
|
|
|
// There can be multiple initial requests before the first peer is
|
|
|
|
// ready.
|
|
|
|
self.last_peer_log = Some(Instant::now());
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
self.last_peer_log = Some(Instant::now());
|
2021-04-18 23:04:24 -07:00
|
|
|
|
|
|
|
// # Correctness
|
|
|
|
//
|
2021-03-15 05:02:12 -07:00
|
|
|
// Only log address metrics in exceptional circumstances, to avoid lock contention.
|
2021-04-18 23:04:24 -07:00
|
|
|
//
|
|
|
|
// TODO: replace with a watch channel that is updated in `AddressBook::update_metrics()`,
|
|
|
|
// or turn the address book into a service (#1976)
|
2021-03-15 05:02:12 -07:00
|
|
|
let address_metrics = self.address_book.lock().unwrap().address_metrics();
|
2020-11-29 22:41:14 -08:00
|
|
|
if unready_services_len == 0 {
|
2021-03-15 05:02:12 -07:00
|
|
|
warn!(
|
|
|
|
?address_metrics,
|
|
|
|
"network request with no peer connections. Hint: check your network connection"
|
|
|
|
);
|
2020-11-29 22:41:14 -08:00
|
|
|
} else {
|
2021-03-15 05:02:12 -07:00
|
|
|
info!(?address_metrics, "network request with no ready peers: finding more peers, waiting for {} peers to answer requests",
|
2020-11-29 22:41:14 -08:00
|
|
|
unready_services_len);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
/// Updates the peer set metrics.
|
|
|
|
///
|
|
|
|
/// # Panics
|
|
|
|
///
|
|
|
|
/// If the peer set size exceeds the connection limit.
|
2020-11-23 23:52:40 -08:00
|
|
|
fn update_metrics(&self) {
|
|
|
|
let num_ready = self.ready_services.len();
|
|
|
|
let num_unready = self.unready_services.len();
|
|
|
|
let num_peers = num_ready + num_unready;
|
2021-01-11 18:28:56 -08:00
|
|
|
metrics::gauge!("pool.num_ready", num_ready as f64);
|
|
|
|
metrics::gauge!("pool.num_unready", num_unready as f64);
|
2021-03-13 18:03:13 -08:00
|
|
|
metrics::gauge!("zcash.net.peers", num_peers as f64);
|
2021-10-27 18:49:31 -07:00
|
|
|
|
|
|
|
// Security: make sure we haven't exceeded the connection limit
|
|
|
|
if num_peers > self.peerset_total_connection_limit {
|
|
|
|
let address_metrics = self.address_book.lock().unwrap().address_metrics();
|
|
|
|
panic!(
|
|
|
|
"unexpectedly exceeded configured peer set connection limit: \n\
|
|
|
|
peers: {:?}, ready: {:?}, unready: {:?}, \n\
|
|
|
|
address_metrics: {:?}",
|
|
|
|
num_peers, num_ready, num_unready, address_metrics,
|
|
|
|
);
|
|
|
|
}
|
2020-11-23 23:52:40 -08:00
|
|
|
}
|
2019-10-10 18:15:24 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
impl<D> Service<Request> for PeerSet<D>
|
|
|
|
where
|
2020-09-01 14:28:54 -07:00
|
|
|
D: Discover<Key = SocketAddr> + Unpin,
|
2019-10-10 18:15:24 -07:00
|
|
|
D::Service: Service<Request, Response = Response> + Load,
|
2020-09-18 11:20:55 -07:00
|
|
|
D::Error: Into<BoxError>,
|
|
|
|
<D::Service as Service<Request>>::Error: Into<BoxError> + 'static,
|
2019-10-10 18:15:24 -07:00
|
|
|
<D::Service as Service<Request>>::Future: Send + 'static,
|
|
|
|
<D::Service as Load>::Metric: Debug,
|
|
|
|
{
|
|
|
|
type Response = Response;
|
2020-09-18 11:20:55 -07:00
|
|
|
type Error = BoxError;
|
2019-10-16 17:06:21 -07:00
|
|
|
type Future =
|
|
|
|
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
|
2019-10-10 18:15:24 -07:00
|
|
|
|
|
|
|
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
2020-11-23 23:52:40 -08:00
|
|
|
self.poll_background_errors(cx)?;
|
2021-11-23 16:31:42 -08:00
|
|
|
|
|
|
|
// Update peer statuses
|
2019-10-10 18:15:24 -07:00
|
|
|
let _ = self.poll_discover(cx)?;
|
2020-09-01 14:28:54 -07:00
|
|
|
self.inventory_registry.poll_inventory(cx)?;
|
2019-10-10 18:15:24 -07:00
|
|
|
self.poll_unready(cx);
|
2020-11-23 23:52:40 -08:00
|
|
|
|
2020-11-29 22:41:14 -08:00
|
|
|
self.log_peer_set_size();
|
2020-11-23 23:52:40 -08:00
|
|
|
self.update_metrics();
|
2019-10-10 18:15:24 -07:00
|
|
|
|
|
|
|
loop {
|
|
|
|
// Re-check that the pre-selected service is ready, in case
|
|
|
|
// something has happened since (e.g., it failed, peer closed
|
|
|
|
// connection, ...)
|
2021-11-23 16:31:42 -08:00
|
|
|
if let Some(key) = self.preselected_p2c_peer {
|
|
|
|
trace!(preselected_key = ?key);
|
|
|
|
let mut service = self
|
|
|
|
.take_ready_service(&key)
|
|
|
|
.expect("preselected peer must be in the ready list");
|
2019-10-10 18:15:24 -07:00
|
|
|
match service.poll_ready(cx) {
|
2021-11-23 16:31:42 -08:00
|
|
|
Poll::Ready(Ok(())) => {
|
|
|
|
trace!("preselected service is still ready, keeping it selected");
|
|
|
|
self.preselected_p2c_peer = Some(key);
|
|
|
|
self.ready_services.insert(key, service);
|
|
|
|
return Poll::Ready(Ok(()));
|
|
|
|
}
|
2019-10-10 18:15:24 -07:00
|
|
|
Poll::Pending => {
|
2021-11-23 16:31:42 -08:00
|
|
|
trace!("preselected service is no longer ready, moving to unready list");
|
2019-10-10 18:15:24 -07:00
|
|
|
self.push_unready(key, service);
|
|
|
|
}
|
|
|
|
Poll::Ready(Err(e)) => {
|
|
|
|
let error = e.into();
|
|
|
|
trace!(%error, "preselected service failed, dropping it");
|
2021-11-23 16:31:42 -08:00
|
|
|
std::mem::drop(service);
|
2019-10-10 18:15:24 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
trace!("preselected service was not ready, preselecting another ready service");
|
|
|
|
self.preselected_p2c_peer = self.preselect_p2c_peer();
|
2021-02-17 13:06:59 -08:00
|
|
|
self.update_metrics();
|
2019-10-10 18:15:24 -07:00
|
|
|
|
2021-11-23 16:31:42 -08:00
|
|
|
if self.preselected_p2c_peer.is_none() {
|
2021-04-01 19:50:17 -07:00
|
|
|
// CORRECTNESS
|
|
|
|
//
|
|
|
|
// If the channel is full, drop the demand signal rather than waiting.
|
|
|
|
// If we waited here, the crawler could deadlock sending a request to
|
|
|
|
// fetch more peers, because it also empties the channel.
|
2019-10-21 15:24:17 -07:00
|
|
|
trace!("no ready services, sending demand signal");
|
2021-10-21 18:26:04 -07:00
|
|
|
let _ = self.demand_signal.try_send(MorePeers);
|
2021-04-01 19:50:17 -07:00
|
|
|
|
2021-03-26 20:50:46 -07:00
|
|
|
// CORRECTNESS
|
|
|
|
//
|
|
|
|
// The current task must be scheduled for wakeup every time we
|
|
|
|
// return `Poll::Pending`.
|
|
|
|
//
|
|
|
|
// As long as there are unready or new peers, this task will run,
|
|
|
|
// because:
|
|
|
|
// - `poll_discover` schedules this task for wakeup when new
|
|
|
|
// peers arrive.
|
|
|
|
// - if there are unready peers, `poll_unready` schedules this
|
|
|
|
// task for wakeup when peer services become ready.
|
|
|
|
// - if the preselected peer is not ready, `service.poll_ready`
|
|
|
|
// schedules this task for wakeup when that service becomes
|
|
|
|
// ready.
|
|
|
|
//
|
|
|
|
// To avoid peers blocking on a full background error channel:
|
|
|
|
// - if no background tasks have exited since the last poll,
|
|
|
|
// `poll_background_errors` schedules this task for wakeup when
|
|
|
|
// the next task exits.
|
2019-10-10 18:15:24 -07:00
|
|
|
return Poll::Pending;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
fn call(&mut self, req: Request) -> Self::Future {
|
2021-02-17 13:06:59 -08:00
|
|
|
let fut = match req {
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
// Only do inventory-aware routing on individual items.
|
|
|
|
Request::BlocksByHash(ref hashes) if hashes.len() == 1 => {
|
|
|
|
let hash = InventoryHash::from(*hashes.iter().next().unwrap());
|
|
|
|
self.route_inv(req, hash)
|
|
|
|
}
|
2021-08-18 15:55:24 -07:00
|
|
|
Request::TransactionsById(ref hashes) if hashes.len() == 1 => {
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
let hash = InventoryHash::from(*hashes.iter().next().unwrap());
|
|
|
|
self.route_inv(req, hash)
|
|
|
|
}
|
2021-08-18 15:55:24 -07:00
|
|
|
|
2021-11-30 13:04:32 -08:00
|
|
|
// Broadcast advertisements to lots of peers
|
|
|
|
Request::AdvertiseTransactionIds(_) => self.route_broadcast(req),
|
|
|
|
Request::AdvertiseBlock(_) => self.route_broadcast(req),
|
2021-08-18 15:55:24 -07:00
|
|
|
|
|
|
|
// Choose a random less-loaded peer for all other requests
|
network: implement transaction request handling. (#1016)
This commit makes several related changes to the network code:
- adds a `TransactionsByHash(HashSet<transaction::Hash>)` request and
`Transactions(Vec<Arc<Transaction>>)` response pair that allows
fetching transactions from a remote peer;
- adds a `PushTransaction(Arc<Transaction>)` request that pushes an
unsolicited transaction to a remote peer;
- adds an `AdvertiseTransactions(HashSet<transaction::Hash>)` request
that advertises transactions by hash to a remote peer;
- adds an `AdvertiseBlock(block::Hash)` request that advertises a block
by hash to a remote peer;
Then, it modifies the connection state machine so that outbound
requests to remote peers are handled properly:
- `TransactionsByHash` generates a `getdata` message and collects the
results, like the existing `BlocksByHash` request.
- `PushTransaction` generates a `tx` message, and returns `Nil` immediately.
- `AdvertiseTransactions` and `AdvertiseBlock` generate an `inv`
message, and return `Nil` immediately.
Next, it modifies the connection state machine so that messages
from remote peers generate requests to the inbound service:
- `getdata` messages generate `BlocksByHash` or `TransactionsByHash`
requests, depending on the content of the message;
- `tx` messages generate `PushTransaction` requests;
- `inv` messages generate `AdvertiseBlock` or `AdvertiseTransactions`
requests.
Finally, it refactors the request routing logic for the peer set to
handle advertisement messages, providing three routing methods:
- `route_p2c`, which uses p2c as normal (default);
- `route_inv`, which uses the inventory registry and falls back to p2c
(used for `BlocksByHash` or `TransactionsByHash`);
- `route_all`, which broadcasts a request to all ready peers (used for
`AdvertiseBlock` and `AdvertiseTransactions`).
2020-09-08 10:16:29 -07:00
|
|
|
_ => self.route_p2c(req),
|
2021-02-17 13:06:59 -08:00
|
|
|
};
|
|
|
|
self.update_metrics();
|
|
|
|
|
|
|
|
fut
|
2019-10-10 18:15:24 -07:00
|
|
|
}
|
|
|
|
}
|