moirai-executor 0.4.0

Hybrid executor implementation for Moirai concurrency library
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

use moirai_core::Priority;

use super::{
    AcceleratorCounts, AcceleratorRoutePolicy, AsyncLanesPerProcess, HybridRoutePolicy,
    HybridRouter, ProcessCount, RouteSummary, RouteTopology, SchedulerRoute, ServerCount,
    ServerRoutePolicy, ThreadRoutePolicy, WorkerCount,
};
use crate::schedule::{AsyncTask, BlockingTask, SyncTask};

fn topology() -> RouteTopology {
    RouteTopology::new(
        WorkerCount::new(4),
        ProcessCount::new(3),
        AsyncLanesPerProcess::new(2),
        ServerCount::new(2),
    )
}

#[test]
fn route_policies_are_zero_sized() {
    assert_eq!(core::mem::size_of::<ThreadRoutePolicy>(), 0);
    assert_eq!(core::mem::size_of::<HybridRoutePolicy>(), 0);
    assert_eq!(core::mem::size_of::<ServerRoutePolicy>(), 0);
    assert_eq!(core::mem::size_of::<AcceleratorRoutePolicy>(), 0);
}

#[test]
fn async_work_routes_to_process_async_lanes() {
    let router = HybridRouter::<HybridRoutePolicy>::new(topology());
    let mut observed = RouteSummary::default();

    for sequence in 0..32 {
        let route = router.select::<AsyncTask>(Priority::Normal, sequence);
        match route {
            SchedulerRoute::Process(route) => {
                assert!(route.async_lane.is_some());
                observed.process_routes += 1;
                observed.async_lane_routes += 1;
            }
            SchedulerRoute::Server(route) => {
                assert!(route.async_lane.is_some());
                observed.server_routes += 1;
                observed.async_lane_routes += 1;
            }
            SchedulerRoute::Thread(_) => {
                panic!("async hybrid route must leave thread-only path")
            }
            SchedulerRoute::Accelerator(_) => {
                panic!("hybrid policy must not produce accelerator routes")
            }
        }
    }

    assert_eq!(observed.total_routes(), 32);
    assert_eq!(observed.async_lane_routes, 32);
}

#[test]
fn sync_work_routes_across_thread_process_and_server_targets() {
    let router = HybridRouter::<HybridRoutePolicy>::new(topology());
    let summary = router.summarize::<SyncTask>(Priority::Normal, 128);

    assert!(summary.thread_routes > 0);
    assert!(summary.process_routes > 0);
    assert!(summary.server_routes > 0);
    assert_eq!(summary.async_lane_routes, 0);
    assert_eq!(summary.total_routes(), 128);
    assert_ne!(summary.checksum, 0);
}

#[test]
fn server_policy_increases_server_route_cadence() {
    let hybrid = HybridRouter::<HybridRoutePolicy>::new(topology());
    let server = HybridRouter::<ServerRoutePolicy>::new(topology());

    let hybrid_summary = hybrid.summarize::<BlockingTask>(Priority::High, 512);
    let server_summary = server.summarize::<BlockingTask>(Priority::High, 512);

    assert!(server_summary.server_routes > hybrid_summary.server_routes);
    assert_eq!(hybrid_summary.async_lane_routes, 0);
    assert_eq!(server_summary.async_lane_routes, 0);
}

#[test]
fn accelerator_policy_routes_metadata_across_device_kinds() {
    let topology = topology().with_accelerators(AcceleratorCounts::new(1, 2, 1, 1));
    let router = HybridRouter::<AcceleratorRoutePolicy>::new(topology);
    let summary = router.summarize::<SyncTask>(Priority::Normal, 128);

    assert_eq!(summary.total_routes(), 128);
    assert_eq!(summary.accelerator_routes, 18);
    assert_eq!(summary.cpu_routes, 3);
    assert_eq!(summary.gpu_routes, 8);
    assert_eq!(summary.tpu_routes, 4);
    assert_eq!(summary.npu_routes, 3);
    assert_eq!(summary.async_lane_routes, 0);
    assert_ne!(summary.checksum, 0);
}

#[test]
fn async_accelerator_metadata_retains_async_lane() {
    let topology = topology().with_accelerators(AcceleratorCounts::new(1, 1, 1, 1));
    let router = HybridRouter::<AcceleratorRoutePolicy>::new(topology);
    let summary = router.summarize::<AsyncTask>(Priority::Critical, 64);

    assert_eq!(summary.total_routes(), 64);
    assert_eq!(summary.accelerator_routes, 9);
    assert_eq!(summary.thread_routes, 0);
    assert_eq!(summary.async_lane_routes, 64);
    assert_eq!(
        summary.cpu_routes + summary.gpu_routes + summary.tpu_routes + summary.npu_routes,
        summary.accelerator_routes
    );
}

#[test]
fn thread_policy_stays_thread_local() {
    let router = HybridRouter::<ThreadRoutePolicy>::new(topology());
    let summary = router.summarize::<AsyncTask>(Priority::Critical, 64);

    assert_eq!(summary.thread_routes, 64);
    assert_eq!(summary.process_routes, 0);
    assert_eq!(summary.server_routes, 0);
    assert_eq!(summary.async_lane_routes, 0);
}