monoloop_loop/transaction/
capacity.rs

1//! Global and per-Channel active-transaction capacity (reserved at admission in WP-04).
2
3use monoloop_contracts::ChannelId;
4use std::collections::HashMap;
5use std::sync::atomic::{AtomicUsize, Ordering};
6use std::sync::Arc;
7
8/// Capacity counters installed at startup; admission acquires permits later.
9#[derive(Debug)]
10pub struct CapacityManagers {
11    max_global: usize,
12    global_active: AtomicUsize,
13    per_channel: HashMap<ChannelId, ChannelCapacity>,
14}
15
16#[derive(Debug)]
17struct ChannelCapacity {
18    max: usize,
19    active: AtomicUsize,
20}
21
22impl CapacityManagers {
23    /// Build managers from runtime and Channel limits.
24    pub fn new(max_global: usize, channels: impl IntoIterator<Item = (ChannelId, usize)>) -> Self {
25        let mut per_channel = HashMap::new();
26        for (id, max) in channels {
27            per_channel.insert(
28                id,
29                ChannelCapacity {
30                    max,
31                    active: AtomicUsize::new(0),
32                },
33            );
34        }
35        Self {
36            max_global,
37            global_active: AtomicUsize::new(0),
38            per_channel,
39        }
40    }
41
42    /// Global active count (observability / tests).
43    pub fn global_active(&self) -> usize {
44        self.global_active.load(Ordering::SeqCst)
45    }
46
47    /// Per-Channel active count.
48    pub fn channel_active(&self, id: &ChannelId) -> Option<usize> {
49        self.per_channel
50            .get(id)
51            .map(|c| c.active.load(Ordering::SeqCst))
52    }
53
54    /// Configured global max.
55    pub fn max_global(&self) -> usize {
56        self.max_global
57    }
58
59    /// Try reserve one global + channel slot (WP-04 will use; available for tests).
60    pub fn try_reserve(self: &Arc<Self>, channel: &ChannelId) -> bool {
61        let Some(ch) = self.per_channel.get(channel) else {
62            return false;
63        };
64        // Optimistic CAS loops.
65        loop {
66            let g = self.global_active.load(Ordering::SeqCst);
67            if g >= self.max_global {
68                return false;
69            }
70            if self
71                .global_active
72                .compare_exchange(g, g + 1, Ordering::SeqCst, Ordering::SeqCst)
73                .is_ok()
74            {
75                break;
76            }
77        }
78        loop {
79            let c = ch.active.load(Ordering::SeqCst);
80            if c >= ch.max {
81                self.global_active.fetch_sub(1, Ordering::SeqCst);
82                return false;
83            }
84            if ch
85                .active
86                .compare_exchange(c, c + 1, Ordering::SeqCst, Ordering::SeqCst)
87                .is_ok()
88            {
89                return true;
90            }
91        }
92    }
93
94    /// Release a prior reservation.
95    pub fn release(&self, channel: &ChannelId) {
96        if let Some(ch) = self.per_channel.get(channel) {
97            ch.active.fetch_sub(1, Ordering::SeqCst);
98        }
99        self.global_active.fetch_sub(1, Ordering::SeqCst);
100    }
101}
monoloop_loop/transaction/capacity.rs

monoloop_loop/transaction/
capacity.rs
