Rayhaan JaufeerallyandClaude Fable 5.1
277b973f1a
The attention strip's names lead to the peer's row, which lights up for a moment. The IPv4 and IPv6 sessions of one neighbour (NAME-v4, NAME-v6) are listed together, joined down the side, the name said once in full. A route count that moved by more than 2% between refreshes carries ▲ or ▼ for two minutes; a row whose state or count changed while the tab was in the background carries ● until it is clicked. A compact mode drops the next hop and session columns, and is remembered by the browser. Colours are variables, with a second set for prefers-color-scheme: dark. A new per-peer `expected_routes` says about how many routes a peer sends when all is well. The daemon only passes it to the API; the page lists an established peer with under half of it as needing attention, a minute after the session came up, with the expected count beside the actual one. An established session with next to no routes looked, until now, exactly like one that was fine. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
730 lines
27 KiB
Rust
730 lines
27 KiB
Rust
//! The server: listeners, the tasks that run for as long as it does, and
|
|
//! what happens when one of them stops.
|
|
|
|
use std::collections::HashMap;
|
|
use std::io;
|
|
use std::net::SocketAddr;
|
|
use std::pin::Pin;
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use futures::stream::{FuturesUnordered, StreamExt};
|
|
use tokio::net::{TcpListener, TcpSocket, TcpStream};
|
|
use tokio::task::{JoinError, JoinHandle};
|
|
use tracing::{error, info, warn};
|
|
|
|
use bgp_fib::{Fib, FibError};
|
|
use bgp_rib::table::RibTable;
|
|
use bgp_rib::{LocalRib, RibStopped};
|
|
use bgp_rpki::Rpki;
|
|
|
|
use crate::address::{PeerAddress, socket_text};
|
|
use crate::advertise::spawn_advertiser;
|
|
use crate::collision::{CollisionTable, Direction};
|
|
use crate::config::{Config, PeerConfig};
|
|
use crate::connection::{PeerSettings, ServerContext, off_thread, run_session};
|
|
use crate::dialer::spawn_dialers;
|
|
use crate::error::ServerError;
|
|
use crate::holddown::HoldDowns;
|
|
use crate::metrics::{CONNECTIONS_REFUSED, PEER_MAX_PREFIXES};
|
|
use crate::panic;
|
|
use crate::policy::ExportLimits;
|
|
use crate::reach;
|
|
use crate::rejections::RejectionLog;
|
|
use crate::restart::{Leave, Restarts};
|
|
use crate::rpki::spawn_revalidator;
|
|
use crate::session::{ConfiguredPeer, DialState, SessionRegistry};
|
|
use crate::sync::IgnorePoison;
|
|
use crate::throttle::Throttle;
|
|
|
|
/// How long sessions get to send their Cease when the server stops.
|
|
const SHUTDOWN_GRACE: Duration = Duration::from_secs(2);
|
|
|
|
/// How long taking our routes out of the kernel may take at shutdown. A
|
|
/// full table is some seconds.
|
|
const KERNEL_GRACE: Duration = Duration::from_secs(30);
|
|
|
|
/// How long to wait before accepting again after an accept failed. What
|
|
/// makes it fail (no file descriptors left, usually) does not pass at once,
|
|
/// and trying again in a loop would only fill the log.
|
|
const ACCEPT_RETRY: Duration = Duration::from_millis(200);
|
|
|
|
/// How often attribute sets that no route uses any more are dropped. They
|
|
/// are not dropped as their last route goes (see `AttrStore::sweep`), and a
|
|
/// table whose routes keep changing attributes collects them.
|
|
const SWEEP_INTERVAL: Duration = Duration::from_secs(60);
|
|
|
|
/// How often it is looked at whether they have piled up, and how many
|
|
/// more than twice what the last sweep left count as that.
|
|
const SWEEP_CHECK_INTERVAL: Duration = Duration::from_secs(1);
|
|
const SWEEP_SLACK: usize = 100_000;
|
|
|
|
/// How many connections may wait to be accepted on a listener.
|
|
const LISTEN_BACKLOG: u32 = 1024;
|
|
|
|
/// How many connections one address may have open to us at a time. A peer
|
|
/// needs one, a second while a collision is resolved, and now and then one
|
|
/// more while an old one is being closed. Each connection costs a task and
|
|
/// a socket for up to four minutes before it has to show an OPEN, so
|
|
/// without a limit a single host could use up either.
|
|
const MAX_CONNECTIONS_PER_ADDRESS: usize = 4;
|
|
|
|
/// At most one line in the log about refused connections in this long: a
|
|
/// host that keeps trying must not be able to fill the log.
|
|
const REFUSAL_LOG_INTERVAL: Duration = Duration::from_secs(10);
|
|
|
|
/// A BGP speaker that has bound its listeners and is ready to accept.
|
|
/// Call [`BgpServer::run`] to start it.
|
|
pub struct BgpServer {
|
|
config: Config,
|
|
listeners: Vec<(SocketAddr, TcpListener)>,
|
|
/// The routing table: every route learned from peers, and the best one
|
|
/// for each prefix. Created at bind time so that callers can share it
|
|
/// before the accept loop starts.
|
|
pub rib: Arc<RibTable>,
|
|
/// Routes this router originates; announced to every established peer.
|
|
pub local_rib: LocalRib,
|
|
/// Live view of established sessions.
|
|
pub registry: SessionRegistry,
|
|
/// What peers announced that was not taken.
|
|
pub rejections: RejectionLog,
|
|
/// The data for route origin validation, where that is configured.
|
|
pub rpki: Option<Rpki>,
|
|
/// The task that keeps that data up to date.
|
|
rpki_client: Option<JoinHandle<()>>,
|
|
/// What keeps the best routes in the kernel, where that is configured.
|
|
pub kernel: Option<Fib>,
|
|
kernel_task: Option<JoinHandle<Result<(), FibError>>>,
|
|
/// Graceful restart: the routes kept of restarting peers, and how far
|
|
/// the peers are with sending theirs since the start.
|
|
pub restarts: Arc<Restarts>,
|
|
}
|
|
|
|
impl BgpServer {
|
|
/// Binds all configured listen endpoints. Separate from
|
|
/// [`BgpServer::run`] so that callers (tests, say) can learn the
|
|
/// addresses actually bound before any connection is accepted.
|
|
pub async fn bind(config: Config) -> Result<Self, ServerError> {
|
|
let mut listeners = Vec::with_capacity(config.listen_endpoints.len());
|
|
for &addr in &config.listen_endpoints {
|
|
let listener = listen(addr).map_err(|source| ServerError::Bind { addr, source })?;
|
|
let bound_addr = listener.local_addr()?;
|
|
info!(%bound_addr, "Listening for BGP connections");
|
|
listeners.push((bound_addr, listener));
|
|
}
|
|
// Routes from the configuration go in before any session exists;
|
|
// each session is sent the whole table when it comes up.
|
|
let local_rib = LocalRib::spawn();
|
|
for route in &config.originate {
|
|
local_rib.announce(route.clone()).await?;
|
|
}
|
|
let configured = config
|
|
.peers
|
|
.iter()
|
|
.map(|p| ConfiguredPeer {
|
|
name: p.name.clone(),
|
|
endpoint: p.endpoint.clone(),
|
|
remote_asn: p.remote_asn,
|
|
role: p.role,
|
|
passive: p.passive,
|
|
expected_routes: p.expected_routes,
|
|
})
|
|
.collect();
|
|
// The client starts now: with luck the data is there before the
|
|
// first routes are.
|
|
let (rpki, rpki_client) = config.rpki.clone().map(Rpki::spawn).unzip();
|
|
let rib = Arc::new(RibTable::new(table_shards()));
|
|
// Started now, so that a machine on which it cannot be done stops
|
|
// the daemon from starting. There is nothing to install yet.
|
|
let (kernel, kernel_task) = match config.kernel {
|
|
Some(kernel) => Some(Fib::spawn(Arc::clone(&rib), kernel)?),
|
|
None => None,
|
|
}
|
|
.unzip();
|
|
// Whether the forwarding by our routes outlived the last run, which
|
|
// peers are told in our OPENs: so it has to be known before any.
|
|
// Where the kernel's routes are not ours, it is not ours to lose.
|
|
let forwarding_preserved = match &kernel {
|
|
Some(kernel) => kernel.adopted().await > 0,
|
|
None => true,
|
|
};
|
|
let restarts = Restarts::new(
|
|
Arc::clone(&rib),
|
|
forwarding_preserved,
|
|
config.peers.iter().map(|peer| peer.name.clone()),
|
|
config.graceful_restart.restart_time,
|
|
);
|
|
Ok(BgpServer {
|
|
restarts,
|
|
rpki,
|
|
rpki_client,
|
|
kernel,
|
|
kernel_task,
|
|
listeners,
|
|
rib,
|
|
local_rib,
|
|
registry: SessionRegistry::with_configured(configured),
|
|
rejections: RejectionLog::default(),
|
|
config,
|
|
})
|
|
}
|
|
|
|
pub fn config(&self) -> &Config {
|
|
&self.config
|
|
}
|
|
|
|
/// The addresses actually bound (useful when the config uses port 0).
|
|
pub fn local_addrs(&self) -> Vec<SocketAddr> {
|
|
self.listeners.iter().map(|(addr, _)| *addr).collect()
|
|
}
|
|
|
|
/// Runs the server. Returns only if something it cannot do without has
|
|
/// failed, after closing its sessions.
|
|
pub async fn run(mut self) -> Result<(), ServerError> {
|
|
crate::metrics::init();
|
|
log_startup(&self.config);
|
|
let context = Arc::new(ServerContext {
|
|
router_id: self.config.router_id,
|
|
local_asn: self.config.local_asn,
|
|
rib: Arc::clone(&self.rib),
|
|
registry: self.registry.clone(),
|
|
collisions: CollisionTable::new(self.config.collision_detect_established_state),
|
|
hold_downs: HoldDowns::default(),
|
|
rejections: self.rejections.clone(),
|
|
rpki: self.rpki.clone(),
|
|
interfaces: self.kernel.as_ref().map(Fib::interfaces),
|
|
restarts: Arc::clone(&self.restarts),
|
|
});
|
|
// What an earlier run left in the kernel and no peer has announced
|
|
// again goes once every peer has sent its routes.
|
|
if let Some(kernel) = self.kernel.clone() {
|
|
let mut converged = self.restarts.convergence().watch();
|
|
tokio::spawn(async move {
|
|
if converged.wait_for(|done| *done).await.is_ok() {
|
|
kernel.release_kept();
|
|
}
|
|
});
|
|
}
|
|
let mut tasks = self.spawn_tasks(&context);
|
|
|
|
let mut kernel_task = self.kernel_task;
|
|
let connections = ConnectionCounts::default();
|
|
let mut refusals = Refusals::default();
|
|
let mut accepting: FuturesUnordered<_> = self
|
|
.listeners
|
|
.into_iter()
|
|
.map(|(addr, listener)| accept_one(addr, listener))
|
|
.collect();
|
|
|
|
loop {
|
|
let (listen_addr, result, listener) = tokio::select! {
|
|
Some(accepted) = accepting.next() => accepted,
|
|
failure = failed(&self.local_rib, &mut tasks) => {
|
|
error!(err = %failure, "Shutting down: the server cannot continue");
|
|
shut_down(&self.registry, self.kernel.as_ref()).await;
|
|
return Err(failure);
|
|
}
|
|
failure = kernel_failed(&mut kernel_task) => {
|
|
error!(err = %failure, "Shutting down: the server cannot continue");
|
|
close_sessions(&self.registry).await;
|
|
return Err(failure);
|
|
}
|
|
};
|
|
match result {
|
|
Ok((stream, peer_addr)) => {
|
|
match admit(&self.config, &context, peer_addr.into(), &connections) {
|
|
Ok((peer, slot)) => {
|
|
let context = Arc::clone(&context);
|
|
tokio::spawn(serve(stream, peer_addr, context, peer, slot));
|
|
}
|
|
// Closed by being dropped, with nothing read and
|
|
// nothing said: a host we do not know learns no
|
|
// more than that the port is open.
|
|
Err(refusal) => refusals.note(peer_addr, refusal),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
error!(%listen_addr, err = %e, "Accept error");
|
|
tokio::time::sleep(ACCEPT_RETRY).await;
|
|
}
|
|
}
|
|
accepting.push(accept_one(listen_addr, listener));
|
|
}
|
|
}
|
|
|
|
/// Starts the tasks the server cannot do without. None of them ends
|
|
/// while the server runs. The advertiser comes first: it has to be
|
|
/// listening before a dialer brings up the first session.
|
|
fn spawn_tasks(&mut self, context: &Arc<ServerContext>) -> FuturesUnordered<NamedTask> {
|
|
let tasks = FuturesUnordered::new();
|
|
tasks.push(named(
|
|
"advertiser",
|
|
spawn_advertiser(
|
|
self.local_rib.clone(),
|
|
Arc::clone(&self.rib),
|
|
self.registry.clone(),
|
|
self.config.local_asn,
|
|
self.config.control,
|
|
self.config.cluster_id,
|
|
self.restarts.convergence().watch(),
|
|
),
|
|
));
|
|
tasks.push(named("sweeper", spawn_sweeper(Arc::clone(&self.rib))));
|
|
if let (Some(rpki), Some(client)) = (self.rpki.clone(), self.rpki_client.take()) {
|
|
tasks.push(named("RPKI client", client));
|
|
tasks.push(named(
|
|
"revalidator",
|
|
spawn_revalidator(rpki, Arc::clone(&self.rib), self.registry.clone()),
|
|
));
|
|
}
|
|
if let Some(interfaces) = context.interfaces.clone() {
|
|
tasks.push(named(
|
|
"next hop revalidator",
|
|
reach::spawn_revalidator(interfaces, Arc::clone(&self.rib), self.registry.clone()),
|
|
));
|
|
}
|
|
for peer in self.config.peers.iter().filter(|peer| peer.passive) {
|
|
info!(peer = %peer.name, endpoint = %peer.endpoint, "Waiting for the peer to connect (passive)");
|
|
}
|
|
for dialer in spawn_dialers(&self.config.peers, context, self.config.connect_retry) {
|
|
tasks.push(named("dialer", dialer));
|
|
}
|
|
tasks
|
|
}
|
|
}
|
|
|
|
/// Resolves when the task that keeps the kernel's routes has ended, which
|
|
/// while the server runs it does only by failing. Never, if there is no
|
|
/// such task.
|
|
async fn kernel_failed(task: &mut Option<JoinHandle<Result<(), FibError>>>) -> ServerError {
|
|
let Some(running) = task else {
|
|
return std::future::pending().await;
|
|
};
|
|
let ended = running.await;
|
|
*task = None;
|
|
match ended {
|
|
Ok(Err(e)) => e.into(),
|
|
Ok(Ok(())) => ServerError::TaskStopped {
|
|
task: "kernel routes",
|
|
reason: "it was stopped".into(),
|
|
},
|
|
Err(e) => ServerError::TaskStopped {
|
|
task: "kernel routes",
|
|
reason: e.to_string(),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Leaves in good order: closes every session (see [`close_sessions`]),
|
|
/// and then takes our routes out of the kernel, if we put any there. In
|
|
/// that order: a peer goes on sending us packets for our routes until it
|
|
/// is told we are going.
|
|
pub async fn shut_down(registry: &SessionRegistry, kernel: Option<&Fib>) {
|
|
close_sessions(registry).await;
|
|
if let Some(kernel) = kernel
|
|
&& tokio::time::timeout(KERNEL_GRACE, kernel.stop())
|
|
.await
|
|
.is_err()
|
|
{
|
|
warn!("Our routes were not all taken out of the kernel in time");
|
|
}
|
|
}
|
|
|
|
/// Leaves to come back at once, as for an upgrade (graceful restart, RFC
|
|
/// 4724): our routes stay in the kernel, so that the machine goes on
|
|
/// forwarding, and the sessions are closed in a way that has peers that
|
|
/// agreed to graceful restart keep our routes until we are back. The
|
|
/// kernel is let go of first: the routes of the sessions that close must
|
|
/// not be taken out of it.
|
|
pub async fn leave_for_restart(registry: &SessionRegistry, kernel: Option<&Fib>) {
|
|
if let Some(kernel) = kernel
|
|
&& tokio::time::timeout(KERNEL_GRACE, kernel.leave())
|
|
.await
|
|
.is_err()
|
|
{
|
|
warn!("Letting go of the kernel's routes took too long");
|
|
}
|
|
leave_sessions(registry, Leave::Restart).await;
|
|
}
|
|
|
|
/// Closes every session politely and waits a moment for them to go: each
|
|
/// peer is told we are going (Cease / Administrative Shutdown), so that it
|
|
/// stops using our routes now and not when its hold timer expires.
|
|
pub async fn close_sessions(registry: &SessionRegistry) {
|
|
leave_sessions(registry, Leave::Stop).await;
|
|
}
|
|
|
|
/// Closes every session as `leave` says, and waits a moment for them to
|
|
/// go.
|
|
async fn leave_sessions(registry: &SessionRegistry, leave: Leave) {
|
|
let left = registry.leave(leave, SHUTDOWN_GRACE).await;
|
|
if left > 0 {
|
|
warn!(sessions = left, "Some sessions did not close in time");
|
|
}
|
|
}
|
|
|
|
/// Says what the configuration is, and what it leaves undone.
|
|
fn log_startup(config: &Config) {
|
|
info!(
|
|
router_id = %config.router_id,
|
|
local_asn = config.local_asn,
|
|
peers = config.peers.len(),
|
|
"BGP server started"
|
|
);
|
|
for peer in &config.peers {
|
|
if let Some(limit) = peer.import.max_prefixes {
|
|
PEER_MAX_PREFIXES
|
|
.with_label_values(&[&peer.name])
|
|
.set(limit.into());
|
|
} else {
|
|
warn!(
|
|
peer = %peer.name,
|
|
"No limit on the routes taken from this peer (import.max_prefixes)"
|
|
);
|
|
}
|
|
}
|
|
warn_about_export_limits(config);
|
|
if config.accept_unconfigured_peers {
|
|
warn!("Taking sessions and routes from any host (accept_unconfigured_peers)");
|
|
}
|
|
}
|
|
|
|
/// Says what the export limits leave undone: the peers they are for that
|
|
/// have none, and the routes of our own that they leave out, which is
|
|
/// likely a mistake in one or the other.
|
|
fn warn_about_export_limits(config: &Config) {
|
|
let limited: Vec<&PeerConfig> = config
|
|
.peers
|
|
.iter()
|
|
.filter(|peer| ExportLimits::apply_to(peer.role, peer.remote_asn == config.local_asn))
|
|
.collect();
|
|
let unlimited: Vec<&str> = limited
|
|
.iter()
|
|
.filter(|peer| peer.export_limits.is_none())
|
|
.map(|peer| peer.name.as_str())
|
|
.collect();
|
|
if !unlimited.is_empty() {
|
|
warn!(
|
|
peers = unlimited.join(", "),
|
|
"Nothing limits what these peers are told (export_limits)"
|
|
);
|
|
}
|
|
// Each distinct set of limits once, with the peers that have it.
|
|
let mut distinct: Vec<(&Arc<ExportLimits>, Vec<&str>)> = Vec::new();
|
|
for peer in &limited {
|
|
let Some(limits) = &peer.export_limits else {
|
|
continue;
|
|
};
|
|
match distinct
|
|
.iter_mut()
|
|
.find(|(seen, _)| Arc::ptr_eq(seen, limits))
|
|
{
|
|
Some((_, peers)) => peers.push(&peer.name),
|
|
None => distinct.push((limits, vec![&peer.name])),
|
|
}
|
|
}
|
|
for route in &config.originate {
|
|
for (limits, peers) in &distinct {
|
|
if !limits.allows(&route.prefix) {
|
|
warn!(
|
|
prefix = %route.prefix,
|
|
peers = peers.join(", "),
|
|
"An originated route is not announced to these peers: export_limits.prefixes \
|
|
leaves it out"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Opens a listening socket on `addr`.
|
|
fn listen(addr: SocketAddr) -> io::Result<TcpListener> {
|
|
let socket = match addr {
|
|
SocketAddr::V4(_) => TcpSocket::new_v4()?,
|
|
SocketAddr::V6(_) => {
|
|
let socket = TcpSocket::new_v6()?;
|
|
// IPv6 only. Where a system's default is to take IPv4
|
|
// connections on such a socket too, listening on both 0.0.0.0
|
|
// and [::] would fail on the second, and IPv4 peers would come
|
|
// in under IPv4-mapped addresses that match no configured peer.
|
|
socket2::SockRef::from(&socket).set_only_v6(true)?;
|
|
socket
|
|
}
|
|
};
|
|
socket.set_reuseaddr(true)?;
|
|
socket.bind(addr)?;
|
|
socket.listen(LISTEN_BACKLOG)
|
|
}
|
|
|
|
/// Decides whether to take a connection from `address`, and if so with what
|
|
/// settings.
|
|
fn admit(
|
|
config: &Config,
|
|
context: &ServerContext,
|
|
address: PeerAddress,
|
|
connections: &ConnectionCounts,
|
|
) -> Result<(PeerSettings, ConnectionSlot), Refusal> {
|
|
if context.hold_downs.remaining(address).is_some() {
|
|
return Err(Refusal::HeldDown);
|
|
}
|
|
// A connection from a configured peer is run with what is configured
|
|
// for that peer.
|
|
let peer = match config.peer_at(address) {
|
|
Some(peer) => PeerSettings::from(peer),
|
|
None if config.accept_unconfigured_peers => PeerSettings::unconfigured(
|
|
config.hold_time_secs,
|
|
Arc::clone(&config.export),
|
|
config.export_limits.clone(),
|
|
config.graceful_restart,
|
|
),
|
|
None => return Err(Refusal::NotAPeer),
|
|
};
|
|
let slot = connections.take(address).ok_or(Refusal::TooMany)?;
|
|
Ok((peer, slot))
|
|
}
|
|
|
|
/// Runs a session on a connection a peer opened, and logs how it ended.
|
|
/// `_slot` is the connection's place among those its address may have.
|
|
async fn serve(
|
|
stream: TcpStream,
|
|
peer_addr: SocketAddr,
|
|
context: Arc<ServerContext>,
|
|
peer: PeerSettings,
|
|
_slot: ConnectionSlot,
|
|
) {
|
|
// Nobody dials a passive peer, so how its session ended is noted here.
|
|
let passive = if peer.passive {
|
|
peer.name.clone()
|
|
} else {
|
|
None
|
|
};
|
|
let registry = context.registry.clone();
|
|
let end = run_session(stream, peer_addr, Direction::Inbound, context, peer).await;
|
|
if let Err(e) = &end.result {
|
|
warn!(peer_addr = %socket_text(peer_addr), err = %e, "BGP session terminated with error");
|
|
}
|
|
if let Some(name) = passive {
|
|
let state = DialState::Passive {
|
|
last_error: Some(end.reason()),
|
|
};
|
|
registry.set_dial_state(&name, state);
|
|
}
|
|
}
|
|
|
|
/// Why an incoming connection was closed without a session.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
enum Refusal {
|
|
/// It is not from the address of a configured peer.
|
|
NotAPeer,
|
|
/// Its address has as many connections open as one may.
|
|
TooMany,
|
|
/// The peer is being kept down: it went over its limit of routes, we
|
|
/// were to send it more than its export limit allows, or its sessions
|
|
/// keep failing.
|
|
HeldDown,
|
|
}
|
|
|
|
impl Refusal {
|
|
fn as_str(self) -> &'static str {
|
|
match self {
|
|
Self::NotAPeer => "not a configured peer",
|
|
Self::TooMany => "too many connections from the address",
|
|
Self::HeldDown => "peer is held down",
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Counts refused connections, and says so in the log now and then.
|
|
struct Refusals {
|
|
log: Throttle,
|
|
}
|
|
|
|
impl Default for Refusals {
|
|
fn default() -> Self {
|
|
Self {
|
|
log: Throttle::every(REFUSAL_LOG_INTERVAL),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Refusals {
|
|
fn note(&mut self, peer_addr: SocketAddr, refusal: Refusal) {
|
|
CONNECTIONS_REFUSED
|
|
.with_label_values(&[refusal.as_str()])
|
|
.inc();
|
|
if let Some(unsaid) = self.log.pass() {
|
|
warn!(
|
|
peer_addr = %socket_text(peer_addr),
|
|
reason = refusal.as_str(),
|
|
others_since_last_report = unsaid,
|
|
"Refused a connection"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// How many connections each address has open to us.
|
|
#[derive(Clone, Default)]
|
|
struct ConnectionCounts(Arc<Mutex<HashMap<PeerAddress, usize>>>);
|
|
|
|
impl ConnectionCounts {
|
|
/// A place for one more connection from `address`, unless it has as
|
|
/// many as it may. The place is given back when the slot is dropped.
|
|
fn take(&self, address: PeerAddress) -> Option<ConnectionSlot> {
|
|
let mut counts = self.0.locked();
|
|
let count = counts.entry(address).or_insert(0);
|
|
if *count >= MAX_CONNECTIONS_PER_ADDRESS {
|
|
return None;
|
|
}
|
|
*count += 1;
|
|
Some(ConnectionSlot {
|
|
counts: self.clone(),
|
|
address,
|
|
})
|
|
}
|
|
}
|
|
|
|
struct ConnectionSlot {
|
|
counts: ConnectionCounts,
|
|
address: PeerAddress,
|
|
}
|
|
|
|
impl Drop for ConnectionSlot {
|
|
fn drop(&mut self) {
|
|
let mut counts = self.counts.0.locked();
|
|
if let Some(count) = counts.get_mut(&self.address) {
|
|
*count -= 1;
|
|
if *count == 0 {
|
|
counts.remove(&self.address);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Drops, every so often, the attribute sets no route uses any more; and
|
|
/// sooner if they pile up. A peer that announces one prefix over and over
|
|
/// with different attributes each time has one route, well within any
|
|
/// limit, and leaves a set behind with every UPDATE.
|
|
fn spawn_sweeper(rib: Arc<RibTable>) -> JoinHandle<()> {
|
|
tokio::spawn(async move {
|
|
let mut look = tokio::time::interval(SWEEP_CHECK_INTERVAL);
|
|
let mut swept_at = Instant::now();
|
|
// How many sets there were after the last sweep.
|
|
let mut kept = rib.attrs().len();
|
|
loop {
|
|
look.tick().await;
|
|
let piled_up = rib.attrs().len() > 2 * kept + SWEEP_SLACK;
|
|
if !piled_up && swept_at.elapsed() < SWEEP_INTERVAL {
|
|
continue;
|
|
}
|
|
// Takes each of the store's locks in turn and looks at every
|
|
// set: not for a thread that sessions run on.
|
|
if let Err(e) = off_thread(&rib, |_| ()).await {
|
|
error!(err = %e, "Sweeping the attribute sets failed");
|
|
}
|
|
swept_at = Instant::now();
|
|
kept = rib.attrs().len();
|
|
}
|
|
})
|
|
}
|
|
|
|
/// How many shards the routing table has. Sessions write to it from every
|
|
/// core at once; a good many more shards than cores keeps two of them from
|
|
/// wanting the same one.
|
|
fn table_shards() -> usize {
|
|
16 * std::thread::available_parallelism().map_or(1, |cores| cores.get())
|
|
}
|
|
|
|
type NamedTask = Pin<Box<dyn Future<Output = (&'static str, Result<(), JoinError>)> + Send>>;
|
|
|
|
fn named(name: &'static str, task: JoinHandle<()>) -> NamedTask {
|
|
Box::pin(async move { (name, task.await) })
|
|
}
|
|
|
|
/// Resolves when something the server depends on has stopped: the local
|
|
/// RIB's actor or one of `tasks`. In a healthy server it never resolves.
|
|
///
|
|
/// Nothing else would notice. A stopped RIB is only found out by whoever
|
|
/// next uses it, and a stopped task by nobody at all: the server would go
|
|
/// on accepting sessions while storing or announcing nothing.
|
|
async fn failed(local_rib: &LocalRib, tasks: &mut FuturesUnordered<NamedTask>) -> ServerError {
|
|
tokio::select! {
|
|
() = local_rib.stopped() => RibStopped { rib: "local RIB" }.into(),
|
|
Some((task, result)) = tasks.next() => ServerError::TaskStopped {
|
|
task,
|
|
reason: match result {
|
|
Ok(()) => "it returned".into(),
|
|
Err(e) if e.is_panic() => {
|
|
format!("it panicked: {}", panic::message(e.into_panic().as_ref()))
|
|
}
|
|
Err(e) => e.to_string(),
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
async fn accept_one(
|
|
addr: SocketAddr,
|
|
listener: TcpListener,
|
|
) -> (SocketAddr, io::Result<(TcpStream, SocketAddr)>, TcpListener) {
|
|
let result = listener.accept().await;
|
|
(addr, result, listener)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// The advertiser and the dialers run for as long as the server does.
|
|
/// If one of them panics, nothing else would ever find out.
|
|
#[tokio::test]
|
|
async fn a_task_that_panics_is_a_failure_of_the_server() {
|
|
let local_rib = LocalRib::spawn();
|
|
let mut tasks: FuturesUnordered<NamedTask> = FuturesUnordered::new();
|
|
tasks.push(named("dialer", tokio::spawn(std::future::pending())));
|
|
tasks.push(named(
|
|
"advertiser",
|
|
tokio::spawn(async { panic!("no route to announce") }),
|
|
));
|
|
|
|
let failure = failed(&local_rib, &mut tasks).await;
|
|
assert_eq!(
|
|
failure.to_string(),
|
|
"the advertiser task stopped unexpectedly: it panicked: no route to announce"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_stopped_rib_is_a_failure_of_the_server() {
|
|
let local_rib = LocalRib::spawn();
|
|
let mut tasks: FuturesUnordered<NamedTask> = FuturesUnordered::new();
|
|
tasks.push(named("advertiser", tokio::spawn(std::future::pending())));
|
|
|
|
local_rib.stop_for_testing();
|
|
let failure = failed(&local_rib, &mut tasks).await;
|
|
assert!(
|
|
matches!(failure, ServerError::RibStopped(RibStopped { rib }) if rib == "local RIB")
|
|
);
|
|
}
|
|
|
|
/// One address may hold only so many connections, and a connection
|
|
/// that ends makes room for another.
|
|
#[test]
|
|
fn an_address_may_have_only_so_many_connections() {
|
|
let counts = ConnectionCounts::default();
|
|
let address = PeerAddress::from("192.0.2.1".parse::<std::net::IpAddr>().unwrap());
|
|
let other = PeerAddress::from("192.0.2.2".parse::<std::net::IpAddr>().unwrap());
|
|
|
|
let mut slots: Vec<_> = (0..MAX_CONNECTIONS_PER_ADDRESS)
|
|
.map(|_| counts.take(address).expect("room"))
|
|
.collect();
|
|
assert!(counts.take(address).is_none());
|
|
assert!(counts.take(other).is_some(), "each address has its own");
|
|
|
|
slots.pop();
|
|
assert!(counts.take(address).is_some());
|
|
drop(slots);
|
|
assert!(counts.0.locked().is_empty(), "nothing is kept for nobody");
|
|
}
|
|
}
|