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
//! The service directory (SRV-027 R1): which connection serves a subject.
//!
//! A worker's registration names the one subject it serves, its
//! `task_queue`; the directory indexes that subject to the connection, in
//! registration order, so a request on the subject can be routed to a worker
//! by one keyed lookup. The directory holds subject and connection ids only:
//! the registration itself stays on the connection record, and the two are
//! kept under the one registry lock so an entry never outlives its record.
//! Two registrations on one subject are both kept and the earlier one
//! serves until it closes; sharing load between them is not done here.
use std::collections::{BTreeMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
/// Subject to the connections serving it, earliest registration first.
#[derive(Debug, Default)]
pub(super) struct ServiceDirectory {
serving: BTreeMap<String, VecDeque<u64>>,
/// Keyed lookups made, for the count pins (R4): one per `serving` call.
lookups: AtomicU64,
}
impl ServiceDirectory {
/// Indexes `subject` to connection `pid`, after any connection already
/// serving it. A pid already listed under `subject` is not listed twice.
pub(super) fn register(&mut self, pid: u64, subject: &str) {
let line = self.serving.entry(subject.to_owned()).or_default();
if !line.contains(&pid) {
line.push_back(pid);
}
}
/// Removes connection `pid` from `subject`'s line; the subject leaves the
/// directory with its last connection. Unknown pairs are a no-op.
pub(super) fn forget(&mut self, pid: u64, subject: &str) {
if let Some(line) = self.serving.get_mut(subject) {
line.retain(|serving| *serving != pid);
if line.is_empty() {
self.serving.remove(subject);
}
}
}
/// The connection serving `subject`: the earliest registration still
/// listed. One keyed lookup; no record is visited.
pub(super) fn serving(&self, subject: &str) -> Option<u64> {
self.lookups.fetch_add(1, Ordering::Relaxed);
self.serving
.get(subject)
.and_then(|line| line.front().copied())
}
/// How many lookups `serving` has made.
pub(super) fn lookups(&self) -> u64 {
self.lookups.load(Ordering::Relaxed)
}
/// How many subjects are served.
#[cfg(test)]
pub(super) fn subjects(&self) -> usize {
self.serving.len()
}
}
#[cfg(test)]
#[path = "service_directory_tests.rs"]
mod tests;