pub mod bid;
pub mod consensus;
use std::collections::BTreeMap;
pub use bid::{BidVector, BID_BYTES_PER_AGENT, BID_BYTES_PER_TASK};
pub use consensus::TableAction;
use crate::geo::Ned;
pub const CBBA_DISCOUNT: f64 = 0.95;
pub const CBBA_SPEED_MPS: f64 = 8.0;
pub const CBBA_BUNDLE_CAPACITY: usize = 5;
#[derive(Debug, Clone, PartialEq)]
pub struct CbbaTask {
pub id: String,
pub pos: Ned,
pub reward: f32,
}
impl CbbaTask {
pub fn new(id: impl Into<String>, pos: Ned, reward: f32) -> Self {
Self {
id: id.into(),
pos,
reward,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct TaskAssignment {
pub task_id: Option<String>,
pub bundle_position: Option<usize>,
}
#[derive(Debug, Clone)]
pub struct CbbaAgent {
pub slot: u8,
pub pos: Ned,
pub speed: f64,
pub capacity: usize,
pub discount: f64,
bundle: Vec<usize>,
path: Vec<usize>,
vector: BidVector,
scratch: Vec<usize>,
}
impl CbbaAgent {
pub fn new(slot: u8, pos: Ned, n_tasks: usize, n_agents: usize) -> Self {
Self {
slot,
pos,
speed: CBBA_SPEED_MPS,
capacity: CBBA_BUNDLE_CAPACITY,
discount: CBBA_DISCOUNT,
bundle: Vec::new(),
path: Vec::new(),
vector: BidVector::new(n_tasks, n_agents),
scratch: Vec::new(),
}
}
pub fn with_capacity(mut self, capacity: usize) -> Self {
self.capacity = capacity.max(1);
self
}
pub fn vector(&self) -> &BidVector {
&self.vector
}
pub fn path(&self) -> &[usize] {
&self.path
}
pub fn bundle(&self) -> &[usize] {
&self.bundle
}
pub fn assignment(&self, tasks: &[CbbaTask]) -> TaskAssignment {
let Some(&next) = self.path.first() else {
return TaskAssignment::default();
};
TaskAssignment {
task_id: tasks.get(next).map(|t| t.id.clone()),
bundle_position: self.bundle.iter().position(|&j| j == next),
}
}
pub fn build_bundle(&mut self, tasks: &[CbbaTask]) -> bool {
let mut changed = false;
while self.bundle.len() < self.capacity && self.path.len() < tasks.len() {
let base = path_score(self.pos, self.speed, self.discount, tasks, &self.path);
let mut best: Option<(usize, usize, f32)> = None;
for j in 0..tasks.len() {
if self.path.contains(&j) {
continue;
}
let mut here: Option<(usize, f64)> = None;
for at in 0..=self.path.len() {
self.scratch.clear();
self.scratch.extend_from_slice(&self.path);
self.scratch.insert(at, j);
let marginal =
path_score(self.pos, self.speed, self.discount, tasks, &self.scratch)
- base;
if here.is_none_or(|(_, m)| marginal > m) {
here = Some((at, marginal));
}
}
let Some((at, marginal)) = here else { continue };
let marginal = marginal as f32;
if marginal > self.vector.y[j] && best.is_none_or(|(_, _, m)| marginal > m) {
best = Some((j, at, marginal));
}
}
let Some((j, at, marginal)) = best else { break };
self.path.insert(at, j);
self.bundle.push(j);
self.vector.y[j] = marginal;
self.vector.z[j] = Some(self.slot);
changed = true;
}
changed
}
pub fn receive(&mut self, sender: u8, theirs: &BidVector, roster: &[u8], now: u16) -> bool {
if theirs.y.len() != self.vector.y.len() || theirs.s.len() != self.vector.s.len() {
return false;
}
let mut changed = false;
for j in 0..self.vector.y.len() {
let action = consensus::decide(j, sender, self.slot, theirs, &self.vector, roster);
changed |= consensus::apply(action, j, theirs, &mut self.vector);
}
consensus::merge_stamps(&mut self.vector, theirs, roster, sender, now);
changed
}
pub fn release_lost(&mut self) -> bool {
let Some(from) = self
.bundle
.iter()
.position(|&j| self.vector.z[j] != Some(self.slot))
else {
return false;
};
for n in (from..self.bundle.len()).rev() {
let j = self.bundle[n];
if n > from {
self.vector.y[j] = 0.0;
self.vector.z[j] = None;
}
if let Some(at) = self.path.iter().position(|&t| t == j) {
self.path.remove(at);
}
self.bundle.pop();
}
true
}
}
fn path_score(start: Ned, speed: f64, discount: f64, tasks: &[CbbaTask], path: &[usize]) -> f64 {
let speed = if speed.is_finite() && speed > 0.0 {
speed
} else {
CBBA_SPEED_MPS
};
let mut at = start;
let mut t = 0.0;
let mut score = 0.0;
for &j in path {
let Some(task) = tasks.get(j) else { continue };
t += (task.pos - at).norm() / speed;
score += discount.powf(t) * task.reward as f64;
at = task.pos;
}
score
}
#[derive(Debug, Clone, PartialEq)]
pub struct ConvergenceReport {
pub rounds: usize,
pub converged: bool,
pub consensus: bool,
pub conflict_free: bool,
pub assignment: BTreeMap<u8, Vec<usize>>,
}
pub fn converge_broadcast(
agents: &mut [CbbaAgent],
tasks: &[CbbaTask],
roster: &[u8],
) -> ConvergenceReport {
let bound = tasks.len() * agents.len() + 2;
let mut rounds = 0;
let mut converged = false;
for round in 1..=bound {
let mut changed = false;
for a in agents.iter_mut() {
changed |= a.build_bundle(tasks);
}
let snapshots: Vec<(u8, BidVector)> =
agents.iter().map(|a| (a.slot, a.vector.clone())).collect();
for a in agents.iter_mut() {
for (slot, v) in &snapshots {
if *slot == a.slot {
continue;
}
changed |= a.receive(*slot, v, roster, round as u16);
}
changed |= a.release_lost();
}
if !changed {
converged = true;
break;
}
rounds = round;
}
ConvergenceReport {
rounds,
converged,
consensus: consensus_reached(agents),
conflict_free: conflicts(agents).is_empty(),
assignment: agents.iter().map(|a| (a.slot, a.path.to_vec())).collect(),
}
}
pub fn conflicts(agents: &[CbbaAgent]) -> Vec<usize> {
let mut count: BTreeMap<usize, usize> = BTreeMap::new();
for a in agents {
for &j in &a.path {
*count.entry(j).or_default() += 1;
}
}
count
.into_iter()
.filter(|(_, n)| *n > 1)
.map(|(j, _)| j)
.collect()
}
pub fn consensus_reached(agents: &[CbbaAgent]) -> bool {
let mut iter = agents.iter();
let Some(first) = iter.next() else {
return true;
};
iter.all(|a| a.vector.z == first.vector.z)
}
#[cfg(test)]
mod tests;