subetha_core/
migration.rs

1//! Dual-stack migration protocol.
2//!
3//! Migration steps:
4//! 1. Allocate new representation alongside old.
5//! 2. Initialize new from a snapshot of old.
6//! 3. Bump generation, swap strategy tag - both representations now live.
7//! 4. Wait for old generation's in-flight count to drain to zero.
8//! 5. Drop old representation.
9//!
10//! The [`MigrationGuard`] enforces steps 3-5 in scope.
11
12use crate::handshake::HandshakeHeader;
13
14/// Wraps the current generation captured at op entry.
15///
16/// Acts as a witness that the caller holds an in-flight slot. Drop
17/// releases the slot.
18#[must_use = "Generation captures an in-flight slot; drop or pass to exit_op"]
19pub struct Generation<'a> {
20    pub(crate) header: &'a HandshakeHeader,
21    pub(crate) value: u32,
22}
23
24impl<'a> Generation<'a> {
25    pub fn enter(header: &'a HandshakeHeader) -> Self {
26        let value = header.enter_op();
27        Self { header, value }
28    }
29
30    #[inline(always)]
31    pub fn value(&self) -> u32 {
32        self.value
33    }
34}
35
36impl<'a> Drop for Generation<'a> {
37    fn drop(&mut self) {
38        self.header.exit_op(self.value);
39    }
40}
41
42/// Guard for the migration coordinator. Wraps the migrate-then-drain
43/// sequence: `begin` bumps the generation and swaps the tag,
44/// `wait_quiescent` drains the old generation. The drain is explicit -
45/// the guard does NOT drain on drop.
46pub struct MigrationGuard<'a> {
47    header: &'a HandshakeHeader,
48    old_value: u32,
49}
50
51impl<'a> MigrationGuard<'a> {
52    /// Begin a migration. `new_tag` is the strategy tag to install.
53    pub fn begin(header: &'a HandshakeHeader, new_tag: u32) -> Self {
54        let old_value = header.migrate(new_tag);
55        Self { header, old_value }
56    }
57
58    /// Wait for in-flight ops on the old generation to complete.
59    pub fn wait_quiescent(&self) {
60        self.header.drain(self.old_value);
61    }
62
63    pub fn old_generation(&self) -> u32 {
64        self.old_value
65    }
66}
67
68#[cfg(test)]
69mod tests {
70    use super::*;
71
72    #[test]
73    fn generation_captures_and_releases() {
74        let h = HandshakeHeader::new();
75        {
76            let guard = Generation::enter(&h);
77            assert_eq!(guard.value(), 0);
78        }
79        h.drain(0);
80    }
81
82    #[test]
83    fn migration_drains_old_generation() {
84        let h = HandshakeHeader::new();
85        let m = MigrationGuard::begin(&h, 1);
86        m.wait_quiescent();
87        assert_eq!(m.old_generation(), 0);
88        assert_eq!(h.tag(), 1);
89    }
90}
subetha_core/migration.rs

subetha_core/
migration.rs
