rutis 0.6.1

Cordis core paradigms in idiomatic Rust: plugin assembly, fiber lifecycle, typed service registry, dependency-driven reload
Documentation 
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

//! 派发尾链并发回归测试(D31):两个任务并发 emit 同一事件类型,
//! 链不得分叉——监听器调用永不重叠(max_in_flight 恒 1)。
//!
//! 背景:emit 的"取 prev → spawn → 存尾"必须单次持锁原子完成;
//! 若 remove/insert 分两段锁,并发同类型 emit 会各自拿到 prev,
//! 派发任务并发执行,顺序保证失效(修复前实测 max_in_flight=4)。

use std::sync::{
    atomic::{AtomicUsize, Ordering},
    Arc,
};

use rutis::{Ctx, Event, Listener};

#[derive(Debug)]
struct ProbeEvent {}

impl Event for ProbeEvent {
    const NAME: &'static str = "probe/fork";
    type Value = ();
}

/// 监听器持 Arc 原子句柄(注册移走监听器,断言侧仍可读)。
struct ChainProbe {
    in_flight: Arc<AtomicUsize>,
    max_in_flight: Arc<AtomicUsize>,
    completed: Arc<AtomicUsize>,
}

impl Listener<ProbeEvent> for ChainProbe {
    fn call<'a>(
        &'a self,
        _ctx: &'a Ctx,
        _e: &'a ProbeEvent,
    ) -> rutis::BoxFuture<'a, Result<Option<()>, rutis::CordisError>> {
        let in_flight = self.in_flight.clone();
        let max_in_flight = self.max_in_flight.clone();
        let completed = self.completed.clone();
        Box::pin(async move {
            let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
            max_in_flight.fetch_max(now, Ordering::SeqCst);
            // 一点工作量,放大重叠窗口:若链分叉,并发派发在此明显重叠
            std::thread::sleep(std::time::Duration::from_micros(50));
            in_flight.fetch_sub(1, Ordering::SeqCst);
            completed.fetch_add(1, Ordering::SeqCst);
            Ok(None)
        })
    }
}

const EMITS_PER_EMITTER: usize = 2_000;

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn dispatch_chain_does_not_fork_under_concurrent_emit() {
    let root = Ctx::root().unwrap();
    let in_flight = Arc::new(AtomicUsize::new(0));
    let max_in_flight = Arc::new(AtomicUsize::new(0));
    let completed = Arc::new(AtomicUsize::new(0));
    root.events()
        .on(
            &root,
            &rutis::EventKey::of(),
            ChainProbe {
                in_flight: in_flight.clone(),
                max_in_flight: max_in_flight.clone(),
                completed: completed.clone(),
            },
        )
        .unwrap();

    let a = {
        let root = root.clone();
        tokio::spawn(async move {
            for _ in 0..EMITS_PER_EMITTER {
                root.events()
                    .emit(&root, &rutis::EventKey::of(), Arc::new(ProbeEvent {}))
                    .expect("default event dispatch");
            }
        })
    };
    let b = {
        let root = root.clone();
        tokio::spawn(async move {
            for _ in 0..EMITS_PER_EMITTER {
                root.events()
                    .emit(&root, &rutis::EventKey::of(), Arc::new(ProbeEvent {}))
                    .expect("default event dispatch");
            }
        })
    };
    let (_, _) = tokio::join!(a, b);

    // 确定性排干:全部派发完成(completed == 总 emit 数)后读终值
    let total = EMITS_PER_EMITTER * 2;
    tokio::time::timeout(std::time::Duration::from_secs(60), async {
        while completed.load(Ordering::SeqCst) < total {
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
    })
    .await
    .expect("chain drains");

    assert_eq!(
        max_in_flight.load(Ordering::SeqCst),
        1,
        "同事件类型的派发任务必须串行:链分叉了"
    );
}