Struct PeerSet

pub struct PeerSet<D, C>
where
    D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin,
    D::Error: Into<BoxError>,
    C: ChainTip,
{

    discover: D,
    demand_signal: Sender<MorePeers>,
    ready_services: HashMap<D::Key, D::Service>,
    inventory_registry: InventoryRegistry,
    unready_services: FuturesUnordered<UnreadyService<D::Key, D::Service, Request>>,
    cancel_handles: HashMap<D::Key, Sender<CancelClientWork>>,
    minimum_peer_version: MinimumPeerVersion<C>,
    peerset_total_connection_limit: usize,
    handle_rx: Receiver<Vec<JoinHandle<Result<(), BoxError>>>>,
    guards: FuturesUnordered<JoinHandle<Result<(), BoxError>>>,
    address_metrics: Receiver<AddressMetrics>,
    last_peer_log: Option<Instant>,
    max_conns_per_ip: usize,
}

A [tower::Service] that abstractly represents “the rest of the network”.

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.

Fields

discover: D

Provides new and deleted peer [Change]s to the peer set, via the [Discover] trait implementation.

demand_signal: Sender<MorePeers>

A channel that asks the peer crawler task to connect to more peers.

ready_services: HashMap<D::Key, D::Service>

Connected peers that are ready to receive requests from Zebra, or send requests to Zebra.

inventory_registry: InventoryRegistry

Stores gossiped inventory hashes from connected peers.

Used to route inventory requests to peers that are likely to have it.

unready_services: FuturesUnordered<UnreadyService<D::Key, D::Service, Request>>

Connected peers that are handling a Zebra request, or Zebra is handling one of their requests.

cancel_handles: HashMap<D::Key, Sender<CancelClientWork>>

Channels used to cancel the request that an unready service is doing.

minimum_peer_version: MinimumPeerVersion<C>

An endpoint to see the minimum peer protocol version in real time.

The minimum version depends on the block height, and MinimumPeerVersion listens for height changes and determines the correct minimum version.

peerset_total_connection_limit: usize

The configured limit for inbound and outbound connections.

The peer set panics if this size is exceeded. If that happens, our connection limit code has a bug.

handle_rx: Receiver<Vec<JoinHandle<Result<(), BoxError>>>>

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

guards: FuturesUnordered<JoinHandle<Result<(), BoxError>>>

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

address_metrics: Receiver<AddressMetrics>

Address book metrics watch channel.

Used for logging diagnostics.

last_peer_log: Option<Instant>

The last time we logged a message about the peer set size

max_conns_per_ip: usize

The configured maximum number of peers that can be in the peer set per IP, defaults to crate::constants::DEFAULT_MAX_CONNS_PER_IP

Implementations

impl<D, C> PeerSet<D, C>

where D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin, D::Error: Into<BoxError>, C: ChainTip,

pub fn new( config: &Config, discover: D, demand_signal: Sender<MorePeers>, handle_rx: Receiver<Vec<JoinHandle<Result<(), BoxError>>>>, inv_stream: Receiver<InventoryResponse<(AtLeastOne<InventoryHash>, PeerSocketAddr), (AtLeastOne<InventoryHash>, PeerSocketAddr)>>, address_metrics: Receiver<AddressMetrics>, minimum_peer_version: MinimumPeerVersion<C>, max_conns_per_ip: Option<usize>, ) -> Self

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 changes from peers, allowing the peer set to direct inventory requests;
• address_book: when peer set is busy, it logs address book diagnostics.
• minimum_peer_version: endpoint to see the minimum peer protocol version in real time.
• max_conns_per_ip: configured maximum number of peers that can be in the peer set per IP, defaults to the config value or to crate::constants::DEFAULT_MAX_CONNS_PER_IP.

fn poll_background_errors( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<(), BoxError>>

Check background task handles to make sure they’re still running.

Never returns Ok.

If any background task exits, shuts down all other background tasks, and returns an error. Otherwise, returns Pending, and registers a wakeup for receiving the background tasks, or the background tasks exiting.

fn receive_tasks_if_needed( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<(), BoxError>>

Receive background tasks, if they’ve been sent on the channel, but not consumed yet.

Returns a result representing the current task state, or Poll::Pending if the background tasks should be polled again to check their state.

fn shut_down_tasks_and_channels(&mut self, cx: &mut Context<'_>)

Shut down:

• services by dropping the service lists
• background tasks via their join handles or cancel handles
• channels by closing the channel

fn poll_unready( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<Option<()>, BoxError>>

Check busy peer services for request completion or errors.

Move newly ready services to the ready list if they are for peers with supported protocol versions, otherwise they are dropped. Also drop failed services.

Never returns an error.

Returns Ok(Some(()) if at least one peer became ready, Poll::Pending if there are unready peers, but none became ready, and Ok(None) if the unready peers were empty.

If there are any remaining unready peers, registers a wakeup for the next time one becomes ready. If there are no unready peers, doesn’t register any wakeups. (Since wakeups come from peers, there needs to be at least one peer to register a wakeup.)

fn poll_ready_peer_errors(&mut self, cx: &mut Context<'_>) -> Poll<()>

Checks previously ready peer services for errors.

The only way these peer Clients can become unready is when we send them a request, because the peer set has exclusive access to send requests to each peer. (If an inbound request is in progress, it will be handled, then our request will be sent by the connection task.)

Returns Poll::Ready if there are some ready peers, and Poll::Pending if there are no ready peers. Registers a wakeup if any peer has failed due to a disconnection, hang, or protocol error.

Panics

If any peers somehow became unready without being sent a request. This indicates a bug in the peer set, where requests are sent to peers without putting them in unready_peers.

fn num_peers_with_ip(&self, ip: IpAddr) -> usize

Returns the number of peer connections Zebra already has with the provided IP address

Performance

This method is O(connected peers), so it should not be called from a loop that is already iterating through the peer set.

fn has_peer_with_addr(&self, addr: PeerSocketAddr) -> bool

Returns true if Zebra is already connected to the IP and port in addr.

fn poll_discover(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>>

Processes the entire list of newly inserted or removed services.

Puts inserted services in the unready list. Drops removed services, after cancelling any pending requests.

If the peer connector channel is closed, returns an error.

Otherwise, returns Ok if it discovered at least one peer, or Poll::Pending if it didn’t discover any peers. Always registers a wakeup for new peers, even when it returns Ok.

fn disconnect_from_outdated_peers(&mut self)

Checks if the minimum peer version has changed, and disconnects from outdated peers.

fn take_ready_service(&mut self, key: &D::Key) -> Option<D::Service>

Takes a ready service by key.

fn remove(&mut self, key: &D::Key)

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.

fn push_ready(&mut self, was_unready: bool, key: D::Key, svc: D::Service)

Adds a ready service to the ready list if it’s for a peer with a supported version. If was_unready is true, also removes the peer’s cancel handle.

If the service is for a connection to an outdated peer, the service is dropped.

fn push_unready(&mut self, key: D::Key, svc: D::Service)

Adds a busy service to the unready list if it’s for a peer with a supported version, and adds a cancel handle for the service’s current request.

If the service is for a connection to an outdated peer, the request is cancelled and the service is dropped.

fn select_ready_p2c_peer(&self) -> Option<D::Key>

Performs P2C on self.ready_services to randomly select a less-loaded ready service.

fn select_p2c_peer_from_list( &self, ready_service_list: &HashSet<D::Key>, ) -> Option<D::Key>

Performs P2C on ready_service_list to randomly select a less-loaded ready service.

fn select_random_ready_peers(&self, max_peers: usize) -> Vec<D::Key>

Randomly chooses max_peers ready services, ignoring service load.

The chosen peers are unique, but their order is not fully random.

fn query_load(&self, key: &D::Key) -> Option<<D::Service as Load>::Metric>

Accesses a ready endpoint by key and returns its current load.

Returns None if the service is not in the ready service list.

fn route_p2c(&mut self, req: Request) -> <Self as Service<Request>>::Future

Routes a request using P2C load-balancing.

fn route_inv( &mut self, req: Request, hash: InventoryHash, ) -> <Self as Service<Request>>::Future

Tries to route a request to a ready peer that advertised that inventory, falling back to a ready peer that isn’t missing the inventory.

If all ready peers are missing the inventory, returns a synthetic NotFoundRegistry error.

Uses P2C to route requests to the least loaded peer in each list.

fn route_multiple( &mut self, req: Request, max_peers: usize, ) -> <Self as Service<Request>>::Future

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_broadcast( &mut self, req: Request, ) -> <Self as Service<Request>>::Future

Broadcasts the same request to lots of ready peers, ignoring return values.

pub(crate) fn number_of_peers_to_broadcast(&self) -> usize

Given a number of ready peers calculate to how many of them Zebra will actually send the request to. Return this number.

fn peer_set_addresses(&self) -> Vec<PeerSocketAddr>

Returns the list of addresses in the peer set.

fn log_peer_set_size(&mut self)

Logs the peer set size, and any potential connectivity issues.

fn update_metrics(&self)

Updates the peer set metrics.

Panics

If the peer set size exceeds the connection limit.

Trait Implementations

impl<D, C> Drop for PeerSet<D, C>

where D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin, D::Error: Into<BoxError>, C: ChainTip,

fn drop(&mut self)

Executes the destructor for this type.

impl<D, C> Service<Request> for PeerSet<D, C>

where D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin, D::Error: Into<BoxError>, C: ChainTip,

type Response = Response

Responses given by the service.

type Error = Box<dyn Error + Sync + Send>

Errors produced by the service.

type Future = Pin<Box<dyn Future<Output = Result<<PeerSet<D, C> as Service<Request>>::Response, <PeerSet<D, C> as Service<Request>>::Error>> + Send>>

The future response value.

fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, req: Request) -> Self::Future

Process the request and return the response asynchronously.