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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
//! Kept tap records are published and failed records retained.
use super::{
ConsoleError, ConsoleFailure, ConsoleHub, PumpReport, ServerPublisher, TAP_SUBJECT_PREFIX, add,
canonical_bytes, failed, len_u64, lock, pending_total, set, tap_batch_record,
};
impl ConsoleHub {
pub(super) fn pump_taps(
&self,
out: &dyn ServerPublisher,
report: &mut PumpReport,
) -> Result<(), ConsoleError> {
let mut due = Vec::new();
// R9: one tap's failure is kept and returned after every other tap
// has been drained and published; it never drops what they took.
let mut first_error = None;
let mut taps = lock(&self.taps, "console taps")?;
for (subject, entry) in &mut taps.live {
if entry.pending.is_empty() {
let target = subject.strip_prefix(TAP_SUBJECT_PREFIX).unwrap_or(subject);
#[cfg(test)]
let taken = self.take_batches(subject, &entry.subscription);
#[cfg(not(test))]
let taken = entry.subscription.take_batches();
match taken {
Ok(batches) => {
for batch in batches {
let encoded =
tap_batch_record(&self.device, target, &batch, &self.wall)
.and_then(|record| canonical_bytes(&record, "tap.batch"));
match encoded {
Ok(bytes) => entry.pending.push_back(bytes),
Err(error) => {
// Taken from the ring and not encodable:
// the one loss besides eviction, counted.
failed(
&self.counts.console_records_lost,
1,
ConsoleFailure::ConsoleRecordsLost,
);
first_error.get_or_insert(error);
}
}
}
}
Err(error) => {
first_error.get_or_insert_with(|| error.into());
}
}
}
if !entry.pending.is_empty() {
due.push((
subject.clone(),
entry.generation,
std::mem::take(&mut entry.pending),
));
}
}
drop(taps);
let mut unsent = Vec::new();
for (subject, generation, mut records) in due {
while let Some(bytes) = records.front() {
if !self.send(out, &subject, bytes) {
report.publish_failures += 1;
unsent.push((subject, generation, records));
break;
}
records.pop_front();
add(&self.counts.tap_records_published, 1);
report.tap_records += 1;
}
}
let mut taps = lock(&self.taps, "console taps")?;
for (subject, generation, records) in unsent {
// Only to the tap they were taken from: one that detached
// meanwhile, or was attached anew, has no reader for them.
match taps.live.get_mut(&subject) {
Some(entry) if entry.generation == generation => entry.pending = records,
_ => add(&self.counts.tap_records_abandoned, len_u64(records.len())),
}
}
let pending = pending_total(&taps.live);
drop(taps);
set(&self.counts.console_pending_records, pending);
first_error.map_or(Ok(()), Err)
}
pub(super) fn retire(
&self,
out: &dyn ServerPublisher,
report: &mut PumpReport,
) -> Result<(), ConsoleError> {
let mut retiring = std::mem::take(&mut lock(&self.taps, "console taps")?.retiring);
while let Some(subject) = retiring.first() {
// The check and the roster removal share the taps lock, taken
// before the roster's, so no subscribe can reserve or attach the
// tap between them and find its channel gone (SRV-012 R6).
let mut taps = lock(&self.taps, "console taps")?;
// Attached or reserved again since it was queued: it stays.
if taps.live.contains_key(subject) || taps.reserved.contains_key(subject) {
drop(taps);
retiring.remove(0);
continue;
}
if let Err(error) = out.retire_console_channel(subject) {
// Kept for the next pump unless it was attached again since.
retiring.retain(|subject| !taps.live.contains_key(subject));
retiring.append(&mut taps.retiring);
taps.retiring = retiring;
drop(taps);
failed(
&self.counts.tap_detach_failures,
1,
ConsoleFailure::TapDetach,
);
return Err(error);
}
drop(taps);
retiring.remove(0);
report.retired += 1;
}
Ok(())
}
}