diff --git a/server/Cargo.lock b/server/Cargo.lock index 8ac49a0..5d62337 100644 --- a/server/Cargo.lock +++ b/server/Cargo.lock @@ -1009,6 +1009,25 @@ version = "1.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba467056f1b547ed52077911161fc86985becbc60e8e1857c8a144dab0def891" +[[package]] +name = "sn-exit" +version = "0.1.0" +dependencies = [ + "clap", + "thiserror", + "tokio", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "sn-fair" +version = "0.1.0" +dependencies = [ + "clap", + "thiserror", +] + [[package]] name = "sn-fetch" version = "0.1.0" @@ -1024,6 +1043,16 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "sn-node" +version = "0.1.0" +dependencies = [ + "clap", + "serde_json", + "thiserror", + "tokio", +] + [[package]] name = "socket2" version = "0.6.5" diff --git a/server/Cargo.toml b/server/Cargo.toml index ce411f9..81a4e1f 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -1,6 +1,6 @@ [workspace] resolver = "2" -members = ["crates/sn-fetch", "crates/sn-node", "crates/sn-exit"] +members = ["crates/sn-fetch", "crates/sn-node", "crates/sn-exit", "crates/sn-fair"] [workspace.package] version = "0.1.0" diff --git a/server/crates/sn-fair/Cargo.toml b/server/crates/sn-fair/Cargo.toml new file mode 100644 index 0000000..cd67e5e --- /dev/null +++ b/server/crates/sn-fair/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "sn-fair" +description = "Weighted max-min fairness scheduler for SpiderNet neighborhood networks" +version.workspace = true +edition.workspace = true +license.workspace = true + +[dependencies] +thiserror = { workspace = true } +clap = { workspace = true } + +[lints] +workspace = true diff --git a/server/crates/sn-fair/src/error.rs b/server/crates/sn-fair/src/error.rs new file mode 100644 index 0000000..8bf2aff --- /dev/null +++ b/server/crates/sn-fair/src/error.rs @@ -0,0 +1,24 @@ +//! Error type for the fairness scheduler. + +use crate::types::ParticipantId; + +#[derive(Debug, thiserror::Error)] +pub enum FairnessError { + #[error("capacity must be greater than zero")] + ZeroCapacity, + + #[error("invalid participant id `{0}`: expected a decimal number greater than zero")] + InvalidParticipantId(String), + + #[error("demand references unknown participant {0}")] + UnknownParticipant(ParticipantId), + + #[error("duplicate demand for participant {0}")] + DuplicateDemand(ParticipantId), + + #[error("duplicate participant {0}")] + DuplicateParticipant(ParticipantId), + + #[error("conflicting participation classes for participant {0}")] + ConflictingParticipation(ParticipantId), +} diff --git a/server/crates/sn-fair/src/fair.rs b/server/crates/sn-fair/src/fair.rs new file mode 100644 index 0000000..6ab0141 --- /dev/null +++ b/server/crates/sn-fair/src/fair.rs @@ -0,0 +1,578 @@ +//! Two-class weighted max-min fairness allocation. +//! +//! Participants are split into two priority classes: +//! +//! 1. **Contributors** ([`Participation::Contributing`]) — households that +//! offer their own uplink to the pool. They are served first, with +//! shares shaped by their [`Weight`] via weighted max-min fairness +//! over the *full* capacity. +//! 2. **Leeches** ([`Participation::Leeching`]) — receive-only +//! households. Leech mode is explicitly allowed (the network exists +//! under neighbors, not against them). Leeches have the lowest but +//! still fair priority: when contributors saturate the capacity, +//! leeches get (near) nothing; when contributors are undersubscribed, +//! leeches benefit from the leftovers. Nobody is dropped from the +//! system. +//! +//! See the crate root for the phase overview. + +use std::collections::{BTreeMap, BTreeSet}; + +use crate::error::FairnessError; +use crate::types::{Capacity, ParticipantId, Rate, Weight}; + +/// How a household takes part in the neighborhood network. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Participation { + /// The household offers its own uplink to the pool; `weight` shapes + /// its share of the capacity (1 = equal share). + Contributing { weight: Weight }, + /// The household only receives (leech mode, explicitly allowed). + /// Leeches share whatever contributors leave unused, with implicit + /// equal weights, capped by their demand. + Leeching, +} + +/// A participant of the allocation: an id plus its participation class. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Participant { + id: ParticipantId, + participation: Participation, +} + +impl Participant { + pub const fn new(id: ParticipantId, participation: Participation) -> Self { + Self { id, participation } + } + + pub const fn id(&self) -> ParticipantId { + self.id + } + + pub const fn participation(&self) -> Participation { + self.participation + } +} + +/// Requested rate of a participant. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Demand { + participant: ParticipantId, + rate: Rate, +} + +impl Demand { + pub const fn new(participant: ParticipantId, rate: Rate) -> Self { + Self { participant, rate } + } + + pub const fn participant(&self) -> ParticipantId { + self.participant + } + + pub const fn rate(&self) -> Rate { + self.rate + } +} + +/// Granted rate for a participant. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Allocation { + participant: ParticipantId, + rate: Rate, +} + +impl Allocation { + pub const fn new(participant: ParticipantId, rate: Rate) -> Self { + Self { participant, rate } + } + + pub const fn participant(&self) -> ParticipantId { + self.participant + } + + pub const fn rate(&self) -> Rate { + self.rate + } +} + +/// Result of [`allocate`]: per-participant rates plus the totals. +/// +/// Allocations are keyed by participant id in a `BTreeMap`, so iteration +/// is always in ascending id order (deterministic output). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FairShare { + allocations: BTreeMap, + total_allocated: Rate, + unused: Rate, +} + +impl FairShare { + /// Allocations ordered by participant id. Participants without a + /// demand at all are absent; use [`FairShare::rate_of`] to read a + /// single participant's rate. + pub fn allocations(&self) -> &BTreeMap { + &self.allocations + } + + /// Allocations as [`Allocation`] values, in participant-id order. + pub fn iter(&self) -> impl Iterator + '_ { + self.allocations + .iter() + .map(|(&participant, &rate)| Allocation::new(participant, rate)) + } + + /// Allocated rate of one participant (zero if it received nothing). + pub fn rate_of(&self, participant: ParticipantId) -> Rate { + self.allocations + .get(&participant) + .copied() + .unwrap_or(Rate::new(0)) + } + + pub const fn total_allocated(&self) -> Rate { + self.total_allocated + } + + pub const fn unused(&self) -> Rate { + self.unused + } +} + +/// Allocates `capacity` among `participants` according to their `demands`. +/// +/// Phase 1 runs weighted max-min fairness among contributors over the +/// full capacity; phase 2 runs weighted max-min among leeches (implicit +/// equal weights) over the capacity the contributors leave unused. +/// +/// Empty `demands` is valid: nothing is allocated, everything stays +/// unused. Participants without a demand (or with demand 0) receive +/// nothing. +/// +/// Errors: +/// - [`FairnessError::ZeroCapacity`] — capacity is zero (cannot happen +/// with a validated [`Capacity`]) +/// - [`FairnessError::UnknownParticipant`] — a demand references an id +/// that is not among `participants` +/// - [`FairnessError::DuplicateDemand`] — two demands for the same id +/// - [`FairnessError::DuplicateParticipant`] — the same id appears twice +/// among `participants` with the same participation class +/// - [`FairnessError::ConflictingParticipation`] — the same id appears +/// twice among `participants` with different participation classes +pub fn allocate( + capacity: Capacity, + participants: &[Participant], + demands: &[Demand], +) -> Result { + let mut participation_of: BTreeMap = BTreeMap::new(); + for participant in participants { + if let Some(previous) = + participation_of.insert(participant.id(), participant.participation()) + { + if previous == participant.participation() { + return Err(FairnessError::DuplicateParticipant(participant.id())); + } + return Err(FairnessError::ConflictingParticipation(participant.id())); + } + } + + for demand in demands { + if !participation_of.contains_key(&demand.participant()) { + return Err(FairnessError::UnknownParticipant(demand.participant())); + } + } + + let mut demand_of: BTreeMap = BTreeMap::new(); + for demand in demands { + if demand_of + .insert(demand.participant(), demand.rate().value()) + .is_some() + { + return Err(FairnessError::DuplicateDemand(demand.participant())); + } + } + + let mut contributors = Vec::new(); + let mut leeches = Vec::new(); + for (&id, &participation) in &participation_of { + let demand = demand_of.get(&id).copied().unwrap_or(0); + match participation { + Participation::Contributing { weight } => contributors.push(Entry { + id, + demand, + weight: weight.value(), + }), + Participation::Leeching => leeches.push(Entry { + id, + demand, + weight: 1, + }), + } + } + + let phase_one = weighted_max_min(contributors, capacity.value()); + let allocated: u128 = phase_one.values().map(|rate| u128::from(*rate)).sum(); + let leftover = capacity.value().saturating_sub(to_u64(allocated)); + + let phase_two = weighted_max_min(leeches, leftover); + + let mut allocations = phase_one; + allocations.extend(phase_two); + + let total: u128 = allocations.values().map(|rate| u128::from(*rate)).sum(); + let total = to_u64(total); + let unused = capacity.value().saturating_sub(total); + + Ok(FairShare { + allocations: allocations + .into_iter() + .map(|(id, rate)| (id, Rate::new(rate))) + .collect(), + total_allocated: Rate::new(total), + unused: Rate::new(unused), + }) +} + +/// Internal per-participant input for the max-min helper. +struct Entry { + id: ParticipantId, + demand: u64, + weight: u32, +} + +/// Weighted max-min fairness over `entries` with (at most) `capacity` to +/// distribute. Deterministic: participants are processed in participant-id +/// order and the result is a `BTreeMap`. +/// +/// Loop: each round hands out proportional shares +/// `share_i = capacity * weight_i / total_weight` (computed in `u128`, +/// floored). Participants whose demand fits their share are satisfied and +/// leave the active set together with their allocation; the rest re-share +/// the remaining capacity. When nobody's demand fits, everyone gets its +/// floor share and the integer-rounding remainder is distributed +/// deterministically in participant-id order, one unit at a time, only to +/// participants still below their demand. +/// +/// Guarantees (asserted by tests): the sum of allocations never exceeds +/// `capacity`; when every remaining demand is at least its computed +/// share, the capacity is distributed in full. +fn weighted_max_min(entries: Vec, capacity: u64) -> BTreeMap { + let mut result = BTreeMap::new(); + let mut active: Vec = Vec::new(); + + // Zero demand is trivially satisfied with an allocation of zero. + for entry in entries { + if entry.demand == 0 { + result.insert(entry.id, 0); + } else { + active.push(entry); + } + } + active.sort_by_key(|entry| entry.id); + + let mut capacity = capacity; + while capacity > 0 && !active.is_empty() { + let total_weight: u128 = active.iter().map(|entry| u128::from(entry.weight)).sum(); + let shares: Vec = active + .iter() + .map(|entry| u128::from(capacity) * u128::from(entry.weight) / total_weight) + .collect(); + + let mut satisfied = Vec::new(); + for (index, entry) in active.iter().enumerate() { + if u128::from(entry.demand) <= shares[index] { + satisfied.push(index); + } + } + + if satisfied.is_empty() { + // Every demand is above its share: hand out the floor shares, + // then spread the rounding remainder one unit at a time. The + // remainder is smaller than the number of active participants + // and every one of them still has room, so it is distributed + // completely. + for (index, entry) in active.iter().enumerate() { + result.insert(entry.id, to_u64(shares[index])); + } + let distributed: u128 = shares.iter().sum(); + let mut remainder = u128::from(capacity) - distributed; + while remainder > 0 { + let mut handed_out = 0; + for entry in &active { + if remainder == 0 { + break; + } + if let Some(allocation) = result.get_mut(&entry.id) { + if *allocation < entry.demand { + *allocation += 1; + remainder -= 1; + handed_out += 1; + } + } + } + if handed_out == 0 { + break; + } + } + active.clear(); + break; + } + + let allocated: u64 = satisfied.iter().map(|&index| active[index].demand).sum(); + let satisfied_set: BTreeSet = satisfied.into_iter().collect(); + let mut remaining = Vec::new(); + for (index, entry) in active.into_iter().enumerate() { + if satisfied_set.contains(&index) { + result.insert(entry.id, entry.demand); + } else { + remaining.push(entry); + } + } + active = remaining; + capacity = capacity.saturating_sub(allocated); + } + + // Capacity exhausted (or nothing to distribute): whoever is still + // active receives nothing. + for entry in active { + result.insert(entry.id, 0); + } + result +} + +/// All values in the allocation arithmetic are bounded by the `u64` +/// capacity, so this conversion cannot truncate; `u64::MAX` is a safe +/// fallback that keeps the code panic-free. +fn to_u64(value: u128) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used)] + + use std::num::{NonZeroU16, NonZeroU32}; + + use super::*; + + fn id(value: u16) -> ParticipantId { + ParticipantId::new(NonZeroU16::new(value).unwrap()) + } + + fn contributing(participant: u16, weight: u32) -> Participant { + Participant::new( + id(participant), + Participation::Contributing { + weight: Weight::new(NonZeroU32::new(weight).unwrap()), + }, + ) + } + + fn leeching(participant: u16) -> Participant { + Participant::new(id(participant), Participation::Leeching) + } + + fn demand(participant: u16, rate: u64) -> Demand { + Demand::new(id(participant), Rate::new(rate)) + } + + fn capacity(value: u64) -> Capacity { + Capacity::try_new(value).unwrap() + } + + fn rate_of(share: &FairShare, participant: u16) -> u64 { + share.rate_of(id(participant)).value() + } + + #[test] + fn equal_demands_equal_weights_split_equally() { + let share = allocate( + capacity(800), + &[contributing(1, 1), contributing(2, 1)], + &[demand(1, 800), demand(2, 800)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 400); + assert_eq!(rate_of(&share, 2), 400); + assert_eq!(share.total_allocated().value(), 800); + assert_eq!(share.unused().value(), 0); + } + + #[test] + fn small_demand_is_satisfied_first_then_remainder_is_redistributed() { + let share = allocate( + capacity(900), + &[contributing(1, 1), contributing(2, 1)], + &[demand(1, 100), demand(2, 900)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 100); + assert_eq!(rate_of(&share, 2), 800); + } + + #[test] + fn weights_shape_the_split_two_to_one() { + let share = allocate( + capacity(900), + &[contributing(1, 2), contributing(2, 1)], + &[demand(1, 900), demand(2, 900)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 600); + assert_eq!(rate_of(&share, 2), 300); + } + + #[test] + fn rounding_remainder_is_distributed_in_id_order() { + let share = allocate( + capacity(1001), + &[contributing(1, 1), contributing(2, 1)], + &[demand(1, 1000), demand(2, 1000)], + ) + .unwrap(); + // Shares are 500 + 500, the extra unit goes to the lower id. + assert_eq!(rate_of(&share, 1), 501); + assert_eq!(rate_of(&share, 2), 500); + assert_eq!(share.total_allocated().value(), 1001); + } + + #[test] + fn saturated_contributor_leaves_nothing_for_the_leech() { + let share = allocate( + capacity(100), + &[contributing(1, 1), leeching(2)], + &[demand(1, 200), demand(2, 50)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 100); + assert_eq!(rate_of(&share, 2), 0); + } + + #[test] + fn undersubscribed_contributor_leaves_leftover_for_the_leech() { + let share = allocate( + capacity(100), + &[contributing(1, 1), leeching(2)], + &[demand(1, 60), demand(2, 100)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 60); + assert_eq!(rate_of(&share, 2), 40); + } + + #[test] + fn two_leeches_split_leftover_equally_capped_by_demand() { + let share = allocate( + capacity(100), + &[contributing(1, 1), leeching(2), leeching(3)], + &[demand(1, 50), demand(2, 100), demand(3, 20)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 50); + // Leftover 50: leech 3 (demand 20) is satisfied first, leech 2 + // gets the rest. + assert_eq!(rate_of(&share, 2), 30); + assert_eq!(rate_of(&share, 3), 20); + } + + #[test] + fn capacity_below_all_demands_yields_proportional_weighted_split() { + let share = allocate( + capacity(30), + &[contributing(1, 2), contributing(2, 1)], + &[demand(1, 100), demand(2, 100)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 20); + assert_eq!(rate_of(&share, 2), 10); + assert_eq!(share.unused().value(), 0); + } + + #[test] + fn zero_demand_receives_zero_without_consuming_capacity() { + let share = allocate( + capacity(800), + &[contributing(1, 1), contributing(2, 1)], + &[demand(1, 0), demand(2, 500)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 0); + assert_eq!(share.allocations().get(&id(1)), Some(&Rate::new(0))); + assert_eq!(rate_of(&share, 2), 500); + assert_eq!(share.unused().value(), 300); + } + + #[test] + fn empty_demands_allocates_nothing_and_everything_is_unused() { + let share = allocate(capacity(500), &[contributing(1, 1), leeching(2)], &[]).unwrap(); + assert_eq!(share.total_allocated().value(), 0); + assert_eq!(share.unused().value(), 500); + assert!(share.allocations().values().all(|rate| rate.value() == 0)); + } + + #[test] + fn unknown_participant_in_demand_is_rejected() { + let error = allocate(capacity(100), &[contributing(1, 1)], &[demand(9, 50)]).unwrap_err(); + assert!(matches!(error, FairnessError::UnknownParticipant(_))); + } + + #[test] + fn duplicate_demand_is_rejected() { + let error = allocate( + capacity(100), + &[contributing(1, 1)], + &[demand(1, 50), demand(1, 20)], + ) + .unwrap_err(); + assert!(matches!(error, FairnessError::DuplicateDemand(_))); + } + + #[test] + fn duplicate_participant_with_same_class_is_rejected() { + let error = allocate( + capacity(100), + &[contributing(1, 1), contributing(1, 1)], + &[], + ) + .unwrap_err(); + assert!(matches!(error, FairnessError::DuplicateParticipant(_))); + } + + #[test] + fn duplicate_participant_with_different_class_is_rejected() { + let error = allocate(capacity(100), &[contributing(1, 1), leeching(1)], &[]).unwrap_err(); + assert!(matches!(error, FairnessError::ConflictingParticipation(_))); + } + + #[test] + fn all_demands_at_least_share_distributes_capacity_in_full() { + let share = allocate( + capacity(1000), + &[contributing(1, 1), contributing(2, 1), contributing(3, 2)], + &[demand(1, 1000), demand(2, 1000), demand(3, 1000)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 250); + assert_eq!(rate_of(&share, 2), 250); + assert_eq!(rate_of(&share, 3), 500); + assert_eq!(share.total_allocated().value(), 1000); + assert_eq!(share.unused().value(), 0); + } + + /// Roadmap pilot scenario: two household downloads (contributing, + /// equal) plus one leech competing for the pooled uplinks. + #[test] + fn pilot_scenario_two_households_and_a_leech() { + let share = allocate( + capacity(600), + &[contributing(1, 1), contributing(2, 1), leeching(3)], + &[demand(1, 250), demand(2, 250), demand(3, 150)], + ) + .unwrap(); + assert_eq!(rate_of(&share, 1), 250); + assert_eq!(rate_of(&share, 2), 250); + assert_eq!(rate_of(&share, 3), 100); + assert_eq!(share.total_allocated().value(), 600); + assert_eq!(share.unused().value(), 0); + } +} diff --git a/server/crates/sn-fair/src/lib.rs b/server/crates/sn-fair/src/lib.rs new file mode 100644 index 0000000..0cfb8ba --- /dev/null +++ b/server/crates/sn-fair/src/lib.rs @@ -0,0 +1,38 @@ +//! sn-fair — weighted max-min fairness scheduler for `SpiderNet`. +//! +//! Allocates a shared capacity (e.g. the pooled uplink bandwidth of a +//! neighborhood) among participants with a two-class weighted max-min +//! fairness scheme: +//! +//! 1. **Contributors** (households that offer capacity to the network, +//! with a configurable weight) are served first, via weighted max-min +//! fairness over the full capacity. +//! 2. **Leeches** (receive-only households; explicitly allowed — the +//! network exists under neighbors, not against them) share whatever +//! the contributors leave unused, weighted max-min with implicit equal +//! weights, capped by their demand. +//! +//! Rationale: leeches have the lowest but still fair priority. When +//! contributors saturate the capacity, leeches get (near) nothing; +//! otherwise they benefit from the leftovers. No one is dropped from the +//! system. +//! +//! Pure computation: no I/O, no async runtime. +//! +//! `SpiderNet` software carries no warranty and accepts no responsibility +//! for its usage or for transported content (AGPL-3.0, sections 15/16 — +//! responsibility for usage lies with the users). + +#![forbid(unsafe_code)] +#![warn(clippy::pedantic)] +// deny is the goal before the first release +// PoC scope: doc-annotation noise, not correctness. Revisit at release. +#![allow(clippy::must_use_candidate, clippy::missing_errors_doc)] + +pub mod error; +pub mod fair; +pub mod types; + +pub use error::FairnessError; +pub use fair::{allocate, Allocation, Demand, FairShare, Participant, Participation}; +pub use types::{Capacity, ParticipantId, Rate, Weight}; diff --git a/server/crates/sn-fair/src/main.rs b/server/crates/sn-fair/src/main.rs new file mode 100644 index 0000000..ebf5ce1 --- /dev/null +++ b/server/crates/sn-fair/src/main.rs @@ -0,0 +1,212 @@ +//! CLI for the sn-fair fairness scheduler. +//! +//! KEINE GARANTIE — this software is provided "as is" without warranty of +//! any kind (AGPL-3.0, sections 15/16). Users are solely responsible for +//! what they transmit or receive with it. +//! +//! Pilot scope: simulates how pooled neighborhood capacity is split — +//! e.g. two household downloads plus a leech competing for the pooled +//! uplinks. + +#![forbid(unsafe_code)] +#![warn(clippy::pedantic)] +// PoC scope: doc-annotation noise, not correctness. Revisit at release. +#![allow(clippy::must_use_candidate, clippy::missing_errors_doc)] + +use std::collections::BTreeMap; +use std::num::NonZeroU32; +use std::process::ExitCode; + +use clap::Parser; +use sn_fair::fair::{Demand, Participant, Participation}; +use sn_fair::types::{Capacity, ParticipantId, Rate, Weight}; +use sn_fair::{allocate, FairShare}; + +/// Parses `id=contributing[:weight]` or `id=leeching`. +fn parse_participant_spec(spec: &str) -> Result<(ParticipantId, Participation), String> { + let (id_raw, class_raw) = spec.split_once('=').ok_or_else(|| { + format!("invalid participant spec `{spec}`, expected `id=contributing[:weight]` or `id=leeching`") + })?; + let participant = id_raw + .trim() + .parse::() + .map_err(|err| format!("invalid participant spec `{spec}`: {err}"))?; + let class = class_raw.trim(); + let participation = match class.strip_prefix("contributing") { + Some(rest) => { + let weight = if rest.is_empty() { + Weight::default() + } else if let Some(weight_raw) = rest.strip_prefix(':') { + weight_raw + .trim() + .parse::() + .ok() + .and_then(NonZeroU32::new) + .map(Weight::new) + .ok_or_else(|| { + format!("invalid weight `{weight_raw}` in `{spec}` (must be a non-zero number)") + })? + } else { + return Err(format!( + "invalid participation `{class}` in `{spec}`, expected `contributing[:weight]` or `leeching`" + )); + }; + Participation::Contributing { weight } + } + None if class == "leeching" => Participation::Leeching, + None => { + return Err(format!( + "invalid participation `{class}` in `{spec}`, expected `contributing[:weight]` or `leeching`" + )) + } + }; + Ok((participant, participation)) +} + +/// Parses `id=rate`. +fn parse_demand_spec(spec: &str) -> Result<(ParticipantId, Rate), String> { + let (id_raw, rate_raw) = spec + .split_once('=') + .ok_or_else(|| format!("invalid demand spec `{spec}`, expected `id=rate`"))?; + let participant = id_raw + .trim() + .parse::() + .map_err(|err| format!("invalid demand spec `{spec}`: {err}"))?; + let rate = rate_raw.trim().parse::().map(Rate::new).map_err(|_| { + format!("invalid rate `{rate_raw}` in `{spec}` (must be a non-negative number)") + })?; + Ok((participant, rate)) +} + +#[derive(Debug, Parser)] +#[command( + name = "sn-fair", + version, + about = "SpiderNet weighted max-min fairness scheduler (pilot)" +)] +struct Args { + /// Total pooled capacity, in Mbit/s. + #[arg(long)] + capacity: u64, + + /// Participant: `id=contributing[:weight]` (offers its uplink, + /// weight shapes the share, default weight 1) or `id=leeching` + /// (receive-only). Repeatable. + #[arg( + long = "participant", + value_name = "ID=CLASS[:WEIGHT]", + value_parser = parse_participant_spec + )] + participants: Vec<(ParticipantId, Participation)>, + + /// Demand: `id=rate` in Mbit/s. Repeatable; participants without a + /// demand receive nothing. + #[arg( + long = "demand", + value_name = "ID=RATE", + value_parser = parse_demand_spec + )] + demands: Vec<(ParticipantId, Rate)>, +} + +fn run() -> Result<(), String> { + let args = Args::parse(); + let capacity = Capacity::try_new(args.capacity).map_err(|err| err.to_string())?; + + let participants: Vec = args + .participants + .into_iter() + .map(|(id, participation)| Participant::new(id, participation)) + .collect(); + let demand_of: BTreeMap = args + .demands + .iter() + .map(|(participant, rate)| (*participant, rate.value())) + .collect(); + let demands: Vec = args + .demands + .into_iter() + .map(|(participant, rate)| Demand::new(participant, rate)) + .collect(); + + let share = allocate(capacity, &participants, &demands).map_err(|err| err.to_string())?; + print_report(capacity, &participants, &demand_of, &share); + Ok(()) +} + +fn print_report( + capacity: Capacity, + participants: &[Participant], + demand_of: &BTreeMap, + share: &FairShare, +) { + println!("NO WARRANTY — SpiderNet sn-fair (AGPL-3.0, no liability)"); + println!(); + println!("pooled capacity: {} Mbit/s", capacity.value()); + println!(); + + let headers = ["participant", "class", "weight", "demand", "allocation"]; + let mut rows = Vec::new(); + for participant in participants { + let (class, weight) = match participant.participation() { + Participation::Contributing { weight } => ("contributing", weight.value().to_string()), + Participation::Leeching => ("leeching", "-".to_string()), + }; + let demand = demand_of + .get(&participant.id()) + .map_or_else(|| "0".to_string(), ToString::to_string); + rows.push([ + participant.id().to_string(), + class.to_string(), + weight, + demand, + share.rate_of(participant.id()).to_string(), + ]); + } + + let mut widths = [0usize; 5]; + for (index, header) in headers.iter().enumerate() { + widths[index] = header.len(); + } + for row in &rows { + for (index, cell) in row.iter().enumerate() { + widths[index] = widths[index].max(cell.len()); + } + } + + print_row(&headers.map(ToString::to_string), &widths); + for row in &rows { + print_row(row, &widths); + } + + println!(); + println!("total allocated: {}", share.total_allocated()); + println!("unused: {}", share.unused()); +} + +fn print_row(cells: &[String], widths: &[usize]) { + println!( + "{:w2$} {:>w3$} {:>w4$}", + cells[0], + cells[1], + cells[2], + cells[3], + cells[4], + w0 = widths[0], + w1 = widths[1], + w2 = widths[2], + w3 = widths[3], + w4 = widths[4], + ); +} + +fn main() -> ExitCode { + match run() { + Ok(()) => ExitCode::SUCCESS, + Err(report) => { + eprintln!("NO WARRANTY — SpiderNet sn-fair (AGPL-3.0, no liability)"); + eprintln!("error: {report}"); + ExitCode::FAILURE + } + } +} diff --git a/server/crates/sn-fair/src/types.rs b/server/crates/sn-fair/src/types.rs new file mode 100644 index 0000000..99c6233 --- /dev/null +++ b/server/crates/sn-fair/src/types.rs @@ -0,0 +1,133 @@ +//! Domain types: newtypes over raw primitives, validated at construction. + +use std::fmt; +use std::num::{NonZeroU16, NonZeroU32, NonZeroU64}; +use std::str::FromStr; + +use crate::error::FairnessError; + +/// Identifier of a participant (a household) in the fairness allocation. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ParticipantId(NonZeroU16); + +impl ParticipantId { + pub const fn new(value: NonZeroU16) -> Self { + Self(value) + } + + pub const fn value(self) -> u16 { + self.0.get() + } +} + +impl fmt::Display for ParticipantId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0.get()) + } +} + +/// Parses a decimal id; rejects zero and non-numeric input. +impl FromStr for ParticipantId { + type Err = FairnessError; + + fn from_str(s: &str) -> Result { + let parsed = s.trim().parse::().ok().and_then(NonZeroU16::new); + parsed + .map(Self::new) + .ok_or_else(|| FairnessError::InvalidParticipantId(s.to_string())) + } +} + +/// Relative share shaping of a contributing participant. 1 means equal +/// share; a participant with weight 2 gets twice the share of one with +/// weight 1 (as far as demands allow). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Weight(NonZeroU32); + +impl Weight { + pub const fn new(value: NonZeroU32) -> Self { + Self(value) + } + + pub const fn value(self) -> u32 { + self.0.get() + } +} + +impl Default for Weight { + fn default() -> Self { + Self(NonZeroU32::MIN) + } +} + +/// A traffic rate, unit-agnostic: bytes per second, Mbit/s — whatever +/// unit the caller works with (the CLI treats it as Mbit/s). +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct Rate(u64); + +impl Rate { + pub const fn new(value: u64) -> Self { + Self(value) + } + + pub const fn value(self) -> u64 { + self.0 + } +} + +impl fmt::Display for Rate { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +/// Total poolable capacity shared by all participants; strictly positive. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct Capacity(NonZeroU64); + +impl Capacity { + pub fn try_new(value: u64) -> Result { + NonZeroU64::new(value) + .map(Self) + .ok_or(FairnessError::ZeroCapacity) + } + + pub const fn value(self) -> u64 { + self.0.get() + } +} + +impl fmt::Display for Capacity { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0.get()) + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used)] + + use super::*; + + #[test] + fn participant_id_parses_decimal_and_rejects_zero_and_garbage() { + assert_eq!("7".parse::().unwrap().value(), 7); + assert_eq!(" 42 ".parse::().unwrap().value(), 42); + assert!("0".parse::().is_err()); + assert!("abc".parse::().is_err()); + assert!("-1".parse::().is_err()); + assert!("70000".parse::().is_err()); + assert!("".parse::().is_err()); + } + + #[test] + fn weight_defaults_to_one() { + assert_eq!(Weight::default().value(), 1); + } + + #[test] + fn capacity_rejects_zero() { + assert!(Capacity::try_new(0).is_err()); + assert_eq!(Capacity::try_new(5).unwrap().value(), 5); + } +}