use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, Ordering};
use std::sync::{Arc, Mutex};
use crate::config::{BackendProtocol, BackendState, BackendTarget, ServiceConfig};
#[derive(Debug)]
pub struct BackendRuntime {
target: BackendTarget,
state: Mutex<BackendState>,
healthy: AtomicBool,
inflight: AtomicU32,
current_weight: AtomicI64,
}
impl BackendRuntime {
#[must_use]
pub fn new(target: BackendTarget) -> Self {
Self {
target,
state: Mutex::new(BackendState::Active),
healthy: AtomicBool::new(true),
inflight: AtomicU32::new(0),
current_weight: AtomicI64::new(0),
}
}
#[must_use]
pub fn target(&self) -> &BackendTarget {
&self.target
}
#[must_use]
pub fn url(&self) -> &str {
&self.target.url
}
#[must_use]
pub fn protocol(&self) -> BackendProtocol {
self.target.protocol()
}
#[must_use]
pub fn state(&self) -> BackendState {
*self.state.lock().expect("backend state mutex poisoned")
}
pub fn set_state(&self, state: BackendState) {
*self.state.lock().expect("backend state mutex poisoned") = state;
}
#[must_use]
pub fn is_healthy(&self) -> bool {
self.healthy.load(Ordering::SeqCst)
}
pub fn set_healthy(&self, healthy: bool) {
self.healthy.store(healthy, Ordering::SeqCst);
}
#[must_use]
pub fn inflight(&self) -> u32 {
self.inflight.load(Ordering::SeqCst)
}
#[must_use]
pub fn is_eligible(&self) -> bool {
self.state() == BackendState::Active
&& self.is_healthy()
&& self.inflight() < self.target.max_connections
}
fn try_reserve(self: &Arc<Self>) -> Option<BackendLease> {
let cap = self.target.max_connections;
loop {
let current = self.inflight.load(Ordering::SeqCst);
if current >= cap {
return None;
}
if self
.inflight
.compare_exchange(current, current + 1, Ordering::SeqCst, Ordering::SeqCst)
.is_ok()
{
return Some(BackendLease {
backend: Arc::clone(self),
});
}
}
}
}
#[derive(Debug)]
pub struct BackendLease {
backend: Arc<BackendRuntime>,
}
impl BackendLease {
#[must_use]
pub fn backend(&self) -> &Arc<BackendRuntime> {
&self.backend
}
}
impl Drop for BackendLease {
fn drop(&mut self) {
let prev = self.backend.inflight.fetch_sub(1, Ordering::SeqCst);
if prev == 0 {
self.backend.inflight.store(0, Ordering::SeqCst);
tracing::error!(url = %self.backend.url(), "backend lease dropped with zero in-flight");
}
}
}
#[derive(Debug)]
pub struct ServiceRuntime {
config: ServiceConfig,
backends: Vec<Arc<BackendRuntime>>,
select_lock: Mutex<()>,
}
pub const STICKY_COOKIE: &str = "__proxy_sticky";
impl ServiceRuntime {
#[must_use]
pub fn new(config: ServiceConfig) -> Self {
let backends = config
.backends
.iter()
.cloned()
.map(|t| Arc::new(BackendRuntime::new(t)))
.collect();
Self {
config,
backends,
select_lock: Mutex::new(()),
}
}
#[must_use]
pub fn config(&self) -> &ServiceConfig {
&self.config
}
#[must_use]
pub fn backends(&self) -> &[Arc<BackendRuntime>] {
&self.backends
}
#[must_use]
pub fn pick_sticky(&self, sticky_cookie: Option<usize>) -> Option<(BackendLease, usize)> {
let _guard = self.select_lock.lock().expect("select mutex poisoned");
if let Some(idx) = sticky_cookie {
if let Some(preferred) = self.backends.get(idx) {
if preferred.is_eligible() {
if let Some(lease) = preferred.try_reserve() {
return Some((lease, idx));
}
}
}
}
let (lease, idx) = self.pick_weighted_round_robin()?;
Some((lease, idx))
}
#[must_use]
pub fn pick(&self) -> Option<BackendLease> {
self.pick_sticky(None).map(|(lease, _)| lease)
}
fn pick_weighted_round_robin(&self) -> Option<(BackendLease, usize)> {
let mut total: i64 = 0;
let mut best_idx: Option<usize> = None;
let mut best_weight = i64::MIN;
for (i, backend) in self.backends.iter().enumerate() {
if !backend.is_eligible() {
continue;
}
let weight = i64::from(backend.target().weight.max(1));
total += weight;
let updated = backend.current_weight.fetch_add(weight, Ordering::SeqCst) + weight;
if updated > best_weight {
best_weight = updated;
best_idx = Some(i);
}
}
let idx = best_idx?;
let chosen = &self.backends[idx];
chosen.current_weight.fetch_sub(total, Ordering::SeqCst);
if let Some(lease) = chosen.try_reserve() {
return Some((lease, idx));
}
self.backends.iter().enumerate().find_map(|(i, b)| {
if b.is_eligible() { b.try_reserve().map(|l| (l, i)) } else { None }
})
}
}