bacnet_server/server/
request_admission.rs

1//! Local fail-fast handler policy, not a protocol or throughput guarantee.
2use std::collections::HashMap;
3use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
4use std::sync::{Arc, Mutex};
5
6use super::request_peer::CanonicalRequester;
7
8use bacnet_types::error::Error;
9use tokio::sync::{OwnedSemaphorePermit, Semaphore};
10
11#[path = "recovery_classifier.rs"]
12mod recovery_classifier;
13pub(super) use recovery_classifier::confirmed_class;
14
15/// Global and logical-peer top-level handler limits with a strict recovery reserve.
16/// Defaults are provisional owner policy, not benchmark-derived values.
17#[derive(Clone, Copy, Debug, PartialEq, Eq)]
18pub struct RequestAdmissionPolicy {
19    /// Maximum concurrent confirmed handlers (default 64).
20    pub max_confirmed_in_flight: usize,
21    /// Maximum concurrent unconfirmed handlers (default 32).
22    pub max_unconfirmed_in_flight: usize,
23    /// Maximum concurrent ordinary confirmed handlers per logical peer (default 16).
24    /// Clamped to global capacity minus the reserve; independent of recovery.
25    pub max_confirmed_in_flight_per_peer: usize,
26    /// Maximum concurrent unconfirmed handlers per logical peer (default 8).
27    /// This partitions capacity by identity; it does not guarantee fairness.
28    pub max_unconfirmed_in_flight_per_peer: usize,
29    /// Strict DCC ENABLE partition inside the confirmed global limit (default 4).
30    /// Zero disables protection; otherwise must be smaller than the global limit.
31    pub confirmed_recovery_reserve: usize,
32    /// Independent protected per-peer limit (default 1), always positive.
33    /// Clamped only to the reserve; zero reserve uses ordinary accounting instead.
34    pub max_recovery_in_flight_per_peer: usize,
35}
36
37impl Default for RequestAdmissionPolicy {
38    fn default() -> Self {
39        Self {
40            max_confirmed_in_flight: 64,
41            max_unconfirmed_in_flight: 32,
42            max_confirmed_in_flight_per_peer: 16,
43            max_unconfirmed_in_flight_per_peer: 8,
44            confirmed_recovery_reserve: 4,
45            max_recovery_in_flight_per_peer: 1,
46        }
47    }
48}
49
50impl RequestAdmissionPolicy {
51    /// Reject zero and limits that the underlying semaphore cannot represent.
52    pub fn validate(&self) -> Result<(), Error> {
53        for (name, value) in [
54            ("max_confirmed_in_flight", self.max_confirmed_in_flight),
55            ("max_unconfirmed_in_flight", self.max_unconfirmed_in_flight),
56            (
57                "max_recovery_in_flight_per_peer",
58                self.max_recovery_in_flight_per_peer,
59            ),
60            (
61                "max_confirmed_in_flight_per_peer",
62                self.max_confirmed_in_flight_per_peer,
63            ),
64            (
65                "max_unconfirmed_in_flight_per_peer",
66                self.max_unconfirmed_in_flight_per_peer,
67            ),
68        ] {
69            if value == 0 || value > Semaphore::MAX_PERMITS {
70                return Err(Error::Encoding(format!(
71                    "{name} must be in 1..={}",
72                    Semaphore::MAX_PERMITS
73                )));
74            }
75        }
76        if self.confirmed_recovery_reserve >= self.max_confirmed_in_flight {
77            return Err(Error::Encoding(
78                "confirmed_recovery_reserve must be smaller than max_confirmed_in_flight".into(),
79            ));
80        }
81        Ok(())
82    }
83}
84
85/// Independently sampled counters for one server lifetime, not an atomic aggregate.
86/// Admitted totals count registrations, not successful responses. Active counts
87/// include registered futures not yet polled and exclude completed unreaped tasks.
88#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
89pub struct RequestAdmissionCounters {
90    /// Live protected handlers, also included in confirmed_active.
91    pub recovery_active: usize,
92    /// Protected registrations, also included in confirmed_admitted_total.
93    pub recovery_admitted_total: u64,
94    /// Protected capacity rejections, also included in confirmed_overloaded_total.
95    pub recovery_overloaded_total: u64,
96    /// Live confirmed handlers.
97    pub confirmed_active: usize,
98    /// Confirmed handlers registered since startup.
99    pub confirmed_admitted_total: u64,
100    /// Confirmed requests rejected for exhausted handler capacity.
101    pub confirmed_overloaded_total: u64,
102    /// Confirmed capacity rejections classified partition-first (ordinary or protected).
103    pub confirmed_global_overloaded_total: u64,
104    /// Confirmed rejections with a partition permit but no slot in that peer partition.
105    pub confirmed_peer_overloaded_total: u64,
106    /// Confirmed registrations rejected after shutdown sealed the owner.
107    pub confirmed_shutdown_rejected_total: u64,
108    /// Live unconfirmed handlers.
109    pub unconfirmed_active: usize,
110    /// Unconfirmed handlers registered since startup.
111    pub unconfirmed_admitted_total: u64,
112    /// Unconfirmed requests dropped for exhausted handler capacity.
113    pub unconfirmed_overloaded_total: u64,
114    /// Unconfirmed capacity rejections classified global-first.
115    pub unconfirmed_global_overloaded_total: u64,
116    /// Unconfirmed capacity rejections with a global permit but no peer slot.
117    pub unconfirmed_peer_overloaded_total: u64,
118    /// Unconfirmed registrations rejected after shutdown sealed the owner.
119    pub unconfirmed_shutdown_rejected_total: u64,
120    /// Live overload Abort send workers (private fixed capacity 8).
121    pub abort_active: usize,
122    /// Overload Abort workers registered, not necessarily successfully sent.
123    pub abort_admitted_total: u64,
124    /// Confirmed overloads silently dropped because all Abort workers were busy.
125    pub confirmed_fallback_dropped_total: u64,
126    /// Abort registrations rejected after shutdown sealed the owner.
127    pub abort_shutdown_rejected_total: u64,
128}
129
130#[derive(Clone, Copy)]
131pub(super) enum Class {
132    Confirmed = 0,
133    Unconfirmed = 1,
134    Abort = 2,
135    Recovery = 3,
136}
137
138#[derive(Debug, PartialEq, Eq)]
139pub(super) enum Rejection {
140    Closed,
141    Overloaded,
142}
143
144pub(super) struct Admission {
145    pools: [Arc<Pool>; 3],
146    recovery: Arc<Pool>,
147    recovery_enabled: bool,
148}
149
150struct Pool {
151    permits: Arc<Semaphore>,
152    peer_limit: usize,
153    peers: Mutex<HashMap<CanonicalRequester, usize>>,
154    active: AtomicUsize,
155    admitted: AtomicU64,
156    overloaded: AtomicU64,
157    global_overloaded: AtomicU64,
158    peer_overloaded: AtomicU64,
159    closed: AtomicU64,
160}
161
162impl Pool {
163    fn new(limit: usize, peer_limit: usize) -> Arc<Self> {
164        Arc::new(Self {
165            permits: Arc::new(Semaphore::new(limit)),
166            peer_limit,
167            peers: Mutex::new(HashMap::new()),
168            active: AtomicUsize::new(0),
169            admitted: AtomicU64::new(0),
170            overloaded: AtomicU64::new(0),
171            global_overloaded: AtomicU64::new(0),
172            peer_overloaded: AtomicU64::new(0),
173            closed: AtomicU64::new(0),
174        })
175    }
176}
177
178pub(super) struct Guard {
179    pool: Arc<Pool>,
180    // Counters only: recovery never registers in the ordinary peer map.
181    aggregate: Option<Arc<Pool>>,
182    peer: Option<CanonicalRequester>,
183    _permit: OwnedSemaphorePermit,
184}
185
186impl Drop for Guard {
187    fn drop(&mut self) {
188        // Remove registration before Rust drops _permit and releases global
189        // capacity. This lock never acquires the task owner's lock.
190        if let Some(peer) = &self.peer {
191            let mut peers = self.pool.peers.lock().unwrap_or_else(|e| e.into_inner());
192            let count = peers.get_mut(peer).expect("live peer registration");
193            *count -= 1;
194            if *count == 0 {
195                peers.remove(peer);
196            }
197        }
198        for pool in std::iter::once(&self.pool).chain(self.aggregate.iter()) {
199            pool.active.fetch_sub(1, Ordering::Relaxed);
200        }
201    }
202}
203
204impl Admission {
205    #[cfg(test)]
206    pub(super) fn peer_entries(&self) -> [usize; 3] {
207        let mut entries = std::array::from_fn(|i| self.pools[i].peers.lock().unwrap().len());
208        // Count entries in both confirmed maps, not distinct confirmed identities.
209        entries[0] += self.recovery.peers.lock().unwrap().len();
210        entries
211    }
212
213    pub(super) fn new(policy: RequestAdmissionPolicy) -> Result<Self, Error> {
214        policy.validate()?;
215        Ok(Self {
216            pools: [
217                Pool::new(
218                    policy.max_confirmed_in_flight - policy.confirmed_recovery_reserve,
219                    policy
220                        .max_confirmed_in_flight_per_peer
221                        .min(policy.max_confirmed_in_flight - policy.confirmed_recovery_reserve),
222                ),
223                Pool::new(
224                    policy.max_unconfirmed_in_flight,
225                    policy
226                        .max_unconfirmed_in_flight_per_peer
227                        .min(policy.max_unconfirmed_in_flight),
228                ),
229                Pool::new(8, 8),
230            ],
231            recovery: Pool::new(
232                policy.confirmed_recovery_reserve,
233                policy
234                    .max_recovery_in_flight_per_peer
235                    .min(policy.confirmed_recovery_reserve),
236            ),
237            recovery_enabled: policy.confirmed_recovery_reserve != 0,
238        })
239    }
240
241    // Called under the task owner's spawn/close lock. No capacity waiters exist.
242    pub(super) fn try_enter(
243        &self,
244        class: Class,
245        peer: CanonicalRequester,
246        closed: bool,
247    ) -> Result<Guard, Rejection> {
248        let (pool, aggregate) = if matches!(class, Class::Recovery) {
249            if self.recovery_enabled {
250                (&self.recovery, Some(&self.pools[0]))
251            } else {
252                (&self.pools[0], None)
253            }
254        } else {
255            (&self.pools[class as usize], None)
256        };
257        if closed {
258            pool.closed.fetch_add(1, Ordering::Relaxed);
259            if let Some(total) = aggregate {
260                total.closed.fetch_add(1, Ordering::Relaxed);
261            }
262            return Err(Rejection::Closed);
263        }
264        let permit = Arc::clone(&pool.permits).try_acquire_owned().map_err(|_| {
265            pool.overloaded.fetch_add(1, Ordering::Relaxed);
266            pool.global_overloaded.fetch_add(1, Ordering::Relaxed);
267            if let Some(total) = aggregate {
268                total.overloaded.fetch_add(1, Ordering::Relaxed);
269                total.global_overloaded.fetch_add(1, Ordering::Relaxed);
270            }
271            Rejection::Overloaded
272        })?;
273        // Spawn serialization belongs to RequestTasks. Peer registrations are
274        // partition-local, while confirmed counters remain inclusive. Only one
275        // map is locked, so no cross-partition registration or rollback is needed.
276        // Abort workers are global-only, with no peer registration or charge.
277        let peer = if matches!(class, Class::Abort) {
278            None
279        } else {
280            let mut peers = pool.peers.lock().unwrap_or_else(|e| e.into_inner());
281            if peers.get(&peer).copied().unwrap_or(0) >= pool.peer_limit {
282                pool.overloaded.fetch_add(1, Ordering::Relaxed);
283                pool.peer_overloaded.fetch_add(1, Ordering::Relaxed);
284                if let Some(total) = aggregate {
285                    total.overloaded.fetch_add(1, Ordering::Relaxed);
286                    total.peer_overloaded.fetch_add(1, Ordering::Relaxed);
287                }
288                return Err(Rejection::Overloaded);
289            }
290            *peers.entry(peer.clone()).or_default() += 1;
291            Some(peer)
292        };
293        pool.active.fetch_add(1, Ordering::Relaxed);
294        pool.admitted.fetch_add(1, Ordering::Relaxed);
295        if let Some(total) = aggregate {
296            total.active.fetch_add(1, Ordering::Relaxed);
297            total.admitted.fetch_add(1, Ordering::Relaxed);
298        }
299        Ok(Guard {
300            pool: Arc::clone(pool),
301            aggregate: aggregate.map(Arc::clone),
302            peer,
303            _permit: permit,
304        })
305    }
306
307    pub(super) fn snapshot(&self) -> RequestAdmissionCounters {
308        let [c, u, a] = &self.pools;
309        RequestAdmissionCounters {
310            recovery_active: self.recovery.active.load(Ordering::Relaxed),
311            recovery_admitted_total: self.recovery.admitted.load(Ordering::Relaxed),
312            recovery_overloaded_total: self.recovery.overloaded.load(Ordering::Relaxed),
313            confirmed_active: c.active.load(Ordering::Relaxed),
314            confirmed_admitted_total: c.admitted.load(Ordering::Relaxed),
315            confirmed_overloaded_total: c.overloaded.load(Ordering::Relaxed),
316            confirmed_global_overloaded_total: c.global_overloaded.load(Ordering::Relaxed),
317            confirmed_peer_overloaded_total: c.peer_overloaded.load(Ordering::Relaxed),
318            confirmed_shutdown_rejected_total: c.closed.load(Ordering::Relaxed),
319            unconfirmed_active: u.active.load(Ordering::Relaxed),
320            unconfirmed_admitted_total: u.admitted.load(Ordering::Relaxed),
321            unconfirmed_overloaded_total: u.overloaded.load(Ordering::Relaxed),
322            unconfirmed_global_overloaded_total: u.global_overloaded.load(Ordering::Relaxed),
323            unconfirmed_peer_overloaded_total: u.peer_overloaded.load(Ordering::Relaxed),
324            unconfirmed_shutdown_rejected_total: u.closed.load(Ordering::Relaxed),
325            abort_active: a.active.load(Ordering::Relaxed),
326            abort_admitted_total: a.admitted.load(Ordering::Relaxed),
327            confirmed_fallback_dropped_total: a.overloaded.load(Ordering::Relaxed),
328            abort_shutdown_rejected_total: a.closed.load(Ordering::Relaxed),
329        }
330    }
331}
332
333impl<T: super::TransportPort + 'static> super::ServerBuilder<T> {
334    /// Set positive global and peer handler limits, validated before startup.
335    pub fn request_admission_policy(mut self, policy: RequestAdmissionPolicy) -> Self {
336        self.config.request_admission_policy = policy;
337        self
338    }
339}
340
341impl super::BipServerBuilder {
342    /// Set positive global and peer handler limits, validated before startup.
343    pub fn request_admission_policy(mut self, policy: RequestAdmissionPolicy) -> Self {
344        self.config.request_admission_policy = policy;
345        self
346    }
347}
348
349#[cfg(feature = "sc-tls")]
350impl super::ScServerBuilder {
351    /// Set positive global and peer handler limits, validated before TLS dialing.
352    pub fn request_admission_policy(mut self, policy: RequestAdmissionPolicy) -> Self {
353        self.config.request_admission_policy = policy;
354        self
355    }
356}
357
358impl<T: super::TransportPort + 'static> super::BACnetServer<T> {
359    /// Sample admission counters. Available after stop; active counts then are zero.
360    pub fn request_admission_counters(&self) -> RequestAdmissionCounters {
361        self.request_tasks.counters()
362    }
363}
bacnet_server/server/request_admission.rs

bacnet_server/server/
request_admission.rs
