2019-10-17 15:42:19 -07:00
|
|
|
//! The timestamp collector collects liveness information from peers.
|
2019-10-07 15:36:16 -07:00
|
|
|
|
2021-06-22 14:59:06 -07:00
|
|
|
use std::{net::SocketAddr, sync::Arc};
|
2019-10-07 15:36:16 -07:00
|
|
|
|
2021-11-18 04:34:51 -08:00
|
|
|
use thiserror::Error;
|
2021-12-19 16:44:43 -08:00
|
|
|
use tokio::{
|
|
|
|
sync::{mpsc, watch},
|
|
|
|
task::JoinHandle,
|
|
|
|
};
|
2022-01-28 14:12:19 -08:00
|
|
|
use tracing::Span;
|
2019-10-07 15:36:16 -07:00
|
|
|
|
2021-12-19 16:44:43 -08:00
|
|
|
use crate::{
|
|
|
|
address_book::AddressMetrics, meta_addr::MetaAddrChange, AddressBook, BoxError, Config,
|
|
|
|
};
|
2019-10-07 15:36:16 -07:00
|
|
|
|
2021-11-04 04:34:00 -07:00
|
|
|
/// The `AddressBookUpdater` hooks into incoming message streams for each peer
|
|
|
|
/// and lets the owner of the sender handle update the address book. For
|
|
|
|
/// example, it can be used to record per-connection last-seen timestamps, or
|
|
|
|
/// add new initial peers to the address book.
|
2021-11-18 04:34:51 -08:00
|
|
|
#[derive(Debug, Eq, PartialEq)]
|
|
|
|
pub struct AddressBookUpdater;
|
|
|
|
|
|
|
|
#[derive(Copy, Clone, Debug, Error, Eq, PartialEq, Hash)]
|
|
|
|
#[error("all address book updater senders are closed")]
|
|
|
|
pub struct AllAddressBookUpdaterSendersClosed;
|
2019-10-07 15:36:16 -07:00
|
|
|
|
2021-11-04 04:34:00 -07:00
|
|
|
impl AddressBookUpdater {
|
|
|
|
/// Spawn a new [`AddressBookUpdater`] task, updating a new [`AddressBook`]
|
2021-11-18 04:34:51 -08:00
|
|
|
/// configured with Zebra's actual `local_listener` address.
|
2021-06-22 14:59:06 -07:00
|
|
|
///
|
2021-11-18 04:34:51 -08:00
|
|
|
/// Returns handles for:
|
2021-12-19 16:44:43 -08:00
|
|
|
/// - the address book,
|
2021-11-18 04:34:51 -08:00
|
|
|
/// - the transmission channel for address book update events,
|
2021-12-19 16:44:43 -08:00
|
|
|
/// - a watch channel for address book metrics, and
|
|
|
|
/// - the address book updater task join handle.
|
|
|
|
///
|
|
|
|
/// The task exits with an error when the returned [`mpsc::Sender`] is closed.
|
2021-06-14 20:31:16 -07:00
|
|
|
pub fn spawn(
|
2021-11-18 04:34:51 -08:00
|
|
|
config: &Config,
|
2021-06-22 14:59:06 -07:00
|
|
|
local_listener: SocketAddr,
|
2021-06-14 20:31:16 -07:00
|
|
|
) -> (
|
|
|
|
Arc<std::sync::Mutex<AddressBook>>,
|
|
|
|
mpsc::Sender<MetaAddrChange>,
|
2021-12-19 16:44:43 -08:00
|
|
|
watch::Receiver<AddressMetrics>,
|
2021-11-18 04:34:51 -08:00
|
|
|
JoinHandle<Result<(), BoxError>>,
|
2021-06-14 20:31:16 -07:00
|
|
|
) {
|
2019-10-21 15:56:16 -07:00
|
|
|
use tracing::Level;
|
2021-11-18 04:34:51 -08:00
|
|
|
|
|
|
|
// Create an mpsc channel for peerset address book updates,
|
|
|
|
// based on the maximum number of inbound and outbound peers.
|
|
|
|
let (worker_tx, mut worker_rx) = mpsc::channel(config.peerset_total_connection_limit());
|
|
|
|
|
2022-06-13 21:58:37 -07:00
|
|
|
let address_book = AddressBook::new(
|
|
|
|
local_listener,
|
|
|
|
config.network,
|
|
|
|
span!(Level::TRACE, "address book"),
|
|
|
|
);
|
2021-12-19 16:44:43 -08:00
|
|
|
let address_metrics = address_book.address_metrics_watcher();
|
|
|
|
let address_book = Arc::new(std::sync::Mutex::new(address_book));
|
2019-10-07 15:36:16 -07:00
|
|
|
|
2021-12-19 16:44:43 -08:00
|
|
|
let worker_address_book = address_book.clone();
|
2022-01-28 14:12:19 -08:00
|
|
|
let span = Span::current();
|
2021-12-19 16:44:43 -08:00
|
|
|
let worker = move || {
|
2022-01-28 14:12:19 -08:00
|
|
|
span.in_scope(|| {
|
|
|
|
info!("starting the address book updater");
|
2022-01-04 15:43:30 -08:00
|
|
|
|
2022-01-28 14:12:19 -08:00
|
|
|
while let Some(event) = worker_rx.blocking_recv() {
|
|
|
|
trace!(?event, "got address book change");
|
2022-01-04 15:43:30 -08:00
|
|
|
|
2022-01-28 14:12:19 -08:00
|
|
|
// # Correctness
|
|
|
|
//
|
|
|
|
// Briefly hold the address book threaded mutex, to update the
|
|
|
|
// state for a single address.
|
|
|
|
worker_address_book
|
|
|
|
.lock()
|
|
|
|
.expect("mutex should be unpoisoned")
|
|
|
|
.update(event);
|
|
|
|
}
|
2021-11-18 04:34:51 -08:00
|
|
|
|
2022-01-28 14:12:19 -08:00
|
|
|
let error = Err(AllAddressBookUpdaterSendersClosed.into());
|
|
|
|
info!(?error, "stopping address book updater");
|
|
|
|
error
|
|
|
|
})
|
2019-10-07 15:36:16 -07:00
|
|
|
};
|
|
|
|
|
2021-12-19 16:44:43 -08:00
|
|
|
// Correctness: spawn address book accesses on a blocking thread,
|
|
|
|
// to avoid deadlocks (see #1976)
|
|
|
|
let address_book_updater_task_handle = tokio::task::spawn_blocking(worker);
|
2021-11-18 04:34:51 -08:00
|
|
|
|
2021-12-19 16:44:43 -08:00
|
|
|
(
|
|
|
|
address_book,
|
|
|
|
worker_tx,
|
|
|
|
address_metrics,
|
|
|
|
address_book_updater_task_handle,
|
|
|
|
)
|
2019-10-07 15:36:16 -07:00
|
|
|
}
|
|
|
|
}
|