1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
//! Subscription drains and release preserve their bounded sampling state.
use super::{Admitted, NonZeroU32, Ordering, TapBatch, TapError, TapFilter, TapSubscription, lock};
impl TapSubscription {
/// The filter this subscription was made with.
#[must_use]
pub fn filter(&self) -> &TapFilter {
&self.subscriber.filter
}
/// Takes every held message as batches, one per run of messages admitted
/// at the same divisor, oldest first, then halves the divisor (the
/// subscriber has drained). The evictions since the previous take ride
/// the first batch; with no messages and a nonzero eviction count, one
/// empty batch carries it. With neither, nothing is returned.
///
/// # Errors
/// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
pub fn take_batches(&self) -> Result<Vec<TapBatch>, TapError> {
let mut guard = lock(&self.subscriber.state, "tap subscriber")?;
let state = &mut *guard;
let (drained, dropped) = state.ring.take();
let mut batches: Vec<TapBatch> = Vec::new();
for Admitted { one_in, message } in drained {
match batches.last_mut() {
Some(open) if open.sampled == one_in => open.messages.push(message),
_ => batches.push(TapBatch {
seq: 0,
sampled: one_in,
dropped: 0,
messages: vec![message],
}),
}
}
match batches.first_mut() {
Some(first) => first.dropped = dropped,
None if dropped > 0 => batches.push(TapBatch {
seq: 0,
sampled: state.one_in,
dropped,
messages: Vec::new(),
}),
None => {}
}
for batch in &mut batches {
batch.seq = state.next_seq;
state.next_seq = state.next_seq.saturating_add(1);
}
// Halving 1 stays 1: the divisor never reaches zero.
state.one_in = NonZeroU32::new(state.one_in.get() / 2).unwrap_or(NonZeroU32::MIN);
drop(guard);
Ok(batches)
}
/// The current sampling divisor.
///
/// # Errors
/// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
pub fn sample_one_in(&self) -> Result<NonZeroU32, TapError> {
Ok(lock(&self.subscriber.state, "tap subscriber")?.one_in)
}
/// Messages held right now (never above the capacity).
///
/// # Errors
/// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
pub fn held(&self) -> Result<usize, TapError> {
Ok(lock(&self.subscriber.state, "tap subscriber")?.ring.len())
}
}
impl Drop for TapSubscription {
fn drop(&mut self) {
// A poisoned list is refused, not recovered: the entry stays, the
// failure is counted, and every later `observe` answers the poison.
let Ok(mut subscribers) = lock(&self.shared.subscribers, "tap subscribers") else {
self.shared
.unsubscribe_failures
.fetch_add(1, Ordering::Relaxed);
crate::metrics::console_failure(
crate::console::counts::ConsoleFailure::TapUnsubscribe,
1,
);
tracing::warn!("TAP_UNSUBSCRIBE_FAILED: the tap subscriber list is poisoned");
return;
};
let id = self.subscriber.id;
subscribers.retain(|subscriber| subscriber.id != id);
self.shared
.active
.store(subscribers.len(), Ordering::Release);
}
}