moirai-executor 0.8.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

//! Route sequence summaries.

use super::{AcceleratorKind, AsyncLaneId, SchedulerRoute};

/// Value summary of a route sequence.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RouteSummary {
    /// Number of local thread routes.
    pub thread_routes: usize,
    /// Number of process routes.
    pub process_routes: usize,
    /// Number of server routes.
    pub server_routes: usize,
    /// Number of accelerator metadata routes.
    pub accelerator_routes: usize,
    /// Number of CPU accelerator metadata routes.
    pub cpu_routes: usize,
    /// Number of GPU accelerator metadata routes.
    pub gpu_routes: usize,
    /// Number of TPU accelerator metadata routes.
    pub tpu_routes: usize,
    /// Number of NPU accelerator metadata routes.
    pub npu_routes: usize,
    /// Number of routes with an async lane.
    pub async_lane_routes: usize,
    /// Deterministic checksum over route placements.
    pub checksum: usize,
}

impl RouteSummary {
    /// Return the total number of summarized route decisions.
    #[inline]
    pub const fn total_routes(self) -> usize {
        self.thread_routes + self.process_routes + self.server_routes + self.accelerator_routes
    }

    #[inline]
    pub(crate) fn record(&mut self, route: SchedulerRoute) {
        match route {
            SchedulerRoute::Thread(route) => {
                self.thread_routes += 1;
                self.checksum = mix_route_checksum(
                    self.checksum,
                    1,
                    0,
                    route.process.get(),
                    route.thread.get(),
                    None,
                );
            }
            SchedulerRoute::Process(route) => {
                self.process_routes += 1;
                if route.async_lane.is_some() {
                    self.async_lane_routes += 1;
                }
                self.checksum = mix_route_checksum(
                    self.checksum,
                    2,
                    0,
                    route.process.get(),
                    route.thread.get(),
                    route.async_lane,
                );
            }
            SchedulerRoute::Server(route) => {
                self.server_routes += 1;
                if route.async_lane.is_some() {
                    self.async_lane_routes += 1;
                }
                self.checksum = mix_route_checksum(
                    self.checksum,
                    3,
                    route.server.get(),
                    route.process.get(),
                    route.thread.get(),
                    route.async_lane,
                );
            }
            SchedulerRoute::Accelerator(route) => {
                self.accelerator_routes += 1;
                if route.async_lane.is_some() {
                    self.async_lane_routes += 1;
                }
                match route.kind {
                    AcceleratorKind::Cpu => self.cpu_routes += 1,
                    AcceleratorKind::Gpu => self.gpu_routes += 1,
                    AcceleratorKind::Tpu => self.tpu_routes += 1,
                    AcceleratorKind::Npu => self.npu_routes += 1,
                }
                self.checksum = mix_route_checksum(
                    self.checksum,
                    4,
                    route
                        .kind
                        .checksum_tag()
                        .wrapping_mul(1_009)
                        .wrapping_add(route.accelerator.get()),
                    route.process.get(),
                    route.thread.get(),
                    route.async_lane,
                );
            }
        }
    }
}

#[inline]
fn mix_route_checksum(
    current: usize,
    route_kind: usize,
    server: usize,
    process: usize,
    thread: usize,
    async_lane: Option<AsyncLaneId>,
) -> usize {
    current
        .wrapping_mul(1_000_003)
        .wrapping_add(route_kind.wrapping_mul(97))
        .wrapping_add(server.wrapping_mul(31))
        .wrapping_add(process.wrapping_mul(17))
        .wrapping_add(thread.wrapping_mul(13))
        .wrapping_add(async_lane.map_or(0, |lane| lane.get().wrapping_add(1)))
}