byteflow/scheduler/
link.rs

1//! Bidirectional links (`A <──────────> B`).
2//!
3//! Distinct from monitors: a monitor delivers `DOWN` to the owner; a link
4//! **propagates abnormal exit** to the peer (BEAM: `normal` does not kill).
5//! Propagation is cooperative — see [`super::finalize`].
6
7use std::collections::HashMap;
8use std::sync::atomic::{AtomicU64, Ordering};
9
10use super::error::{LifecycleError, RuntimeError};
11use super::process::FlowId;
12use super::sync_lock;
13
14/// Opaque link identifier returned by [`LinkTable::link`].
15///
16/// Inner id is runtime-minted (`pub(crate)`), same rule as [`crate::CapId`].
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
18pub struct LinkId(pub(crate) u64);
19
20static NEXT_LINK: AtomicU64 = AtomicU64::new(1);
21
22impl LinkId {
23    #[inline]
24    pub fn as_u64(self) -> u64 {
25        self.0
26    }
27
28    #[inline]
29    pub(crate) fn from_u64(raw: u64) -> Self {
30        Self(raw)
31    }
32}
33
34impl std::fmt::Display for LinkId {
35    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
36        write!(f, "link#{}", self.0)
37    }
38}
39
40#[derive(Debug, Clone, Copy)]
41struct Link {
42    a: FlowId,
43    b: FlowId,
44}
45
46/// Bidirectional flow links. Lives on [`super::runtime::Shared`].
47pub struct LinkTable {
48    links: HashMap<LinkId, Link>,
49}
50
51impl LinkTable {
52    pub fn new() -> Self {
53        Self {
54            links: HashMap::new(),
55        }
56    }
57
58    pub fn link(&mut self, a: FlowId, b: FlowId) -> Result<LinkId, LifecycleError> {
59        if a == b {
60            return Err(LifecycleError::SelfRelation);
61        }
62        if self
63            .links
64            .values()
65            .any(|link| (link.a == a && link.b == b) || (link.a == b && link.b == a))
66        {
67            return Err(LifecycleError::AlreadyLinked);
68        }
69        let id = LinkId(NEXT_LINK.fetch_add(1, Ordering::Relaxed));
70        self.links.insert(id, Link { a, b });
71        Ok(id)
72    }
73
74    pub fn unlink_owned(&mut self, owner: FlowId, id: LinkId) -> Result<(), LifecycleError> {
75        match self.links.get(&id) {
76            Some(link) if link.a == owner || link.b == owner => {
77                self.links.remove(&id);
78                Ok(())
79            }
80            Some(_) => Err(LifecycleError::NotOwner),
81            None => Err(LifecycleError::InvalidLink),
82        }
83    }
84
85    /// Peers of `target` and the link ids, then drop those links.
86    pub fn remove_links_of(&mut self, target: FlowId) -> Vec<(LinkId, FlowId)> {
87        let mut result = Vec::new();
88        self.links.retain(|id, link| {
89            if link.a == target {
90                result.push((*id, link.b));
91                false
92            } else if link.b == target {
93                result.push((*id, link.a));
94                false
95            } else {
96                true
97            }
98        });
99        result
100    }
101}
102
103impl Default for LinkTable {
104    fn default() -> Self {
105        Self::new()
106    }
107}
108
109pub struct LinkStore {
110    inner: std::sync::Mutex<LinkTable>,
111}
112
113impl LinkStore {
114    pub fn new() -> Self {
115        Self {
116            inner: std::sync::Mutex::new(LinkTable::new()),
117        }
118    }
119
120    pub fn link(&self, a: FlowId, b: FlowId) -> Result<Result<LinkId, LifecycleError>, RuntimeError> {
121        Ok(sync_lock::lock(&self.inner, "LinkStore::link")?.link(a, b))
122    }
123
124    pub fn unlink_owned(
125        &self,
126        owner: FlowId,
127        id: LinkId,
128    ) -> Result<Result<(), LifecycleError>, RuntimeError> {
129        Ok(sync_lock::lock(&self.inner, "LinkStore::unlink_owned")?.unlink_owned(owner, id))
130    }
131
132    pub fn remove_links_of(&self, target: FlowId) -> Result<Vec<(LinkId, FlowId)>, RuntimeError> {
133        Ok(sync_lock::lock(&self.inner, "LinkStore::remove_links_of")?.remove_links_of(target))
134    }
135}
136
137impl Default for LinkStore {
138    fn default() -> Self {
139        Self::new()
140    }
141}
142
143#[cfg(test)]
144mod tests {
145    use super::*;
146    use crate::scheduler::process::next_flow_id;
147
148    #[test]
149    fn remove_links_returns_both_directions() {
150        let mut table = LinkTable::new();
151        let a = next_flow_id();
152        let b = next_flow_id();
153        let c = next_flow_id();
154        table.link(a, b).expect("link a-b");
155        table.link(c, a).expect("link c-a");
156        let peers: Vec<FlowId> = table.remove_links_of(a).into_iter().map(|(_, p)| p).collect();
157        assert_eq!(peers.len(), 2);
158        assert!(peers.contains(&b));
159        assert!(peers.contains(&c));
160        assert!(table.remove_links_of(a).is_empty());
161    }
162
163    #[test]
164    fn rejects_self_and_duplicate() {
165        let mut table = LinkTable::new();
166        let a = next_flow_id();
167        let b = next_flow_id();
168        assert_eq!(table.link(a, a), Err(LifecycleError::SelfRelation));
169        assert!(table.link(a, b).is_ok());
170        assert_eq!(table.link(b, a), Err(LifecycleError::AlreadyLinked));
171    }
172}
byteflow/scheduler/link.rs

byteflow/scheduler/
link.rs
