weirflow 0.1.0

GPU-first dataflow analysis primitives for Vyre and Santh compiler pipelines.
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251

use crate::dense_domain::{
    plan_sparse_dense_domain, SparseDenseDomainMode, SparseDenseDomainObservation,
    SparseDenseDomainPolicy,
};

/// Execution family suggested by observed fixed-point frontier density.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub enum FrontierExecutionMode {
    /// No active frontier bits were observed.
    Empty,
    /// Low active and low delta density; a sparse frontier kernel should win.
    Sparse,
    /// Mixed density; a sparse/dense hybrid or adaptive kernel should win.
    Hybrid,
    /// High active density; a dense bitset kernel should win.
    Dense,
}

/// Frontier density counters captured by a fixed-point scratch session.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct FrontierDensityTelemetry {
    /// Number of bits in the analysis domain for this run.
    pub domain_bits: u64,
    /// Number of decoded frontier samples, including the seed frontier.
    pub samples: u64,
    /// Number of GPU fixed-point iterations observed.
    pub iterations: u64,
    /// Sum of active frontier bits across all samples.
    pub active_bits_total: u64,
    /// Sum of changed frontier bits across decoded GPU transitions.
    pub delta_bits_total: u64,
    /// Active bits in the most recently decoded frontier sample.
    pub last_active_bits: u64,
    /// Changed bits between the previous frontier and the most recent sample.
    pub last_delta_bits: u64,
    /// Maximum active bits observed in one frontier sample.
    pub peak_active_bits: u64,
    /// Maximum changed bits observed in one decoded GPU transition.
    pub peak_delta_bits: u64,
    /// Number of frontier telemetry samples where decoded words were shorter than the domain.
    pub truncated_frontier_samples: u64,
    /// Number of times the telemetry domain could not fit the host index width.
    pub domain_overflow_events: u64,
    /// Number of telemetry counter additions that pinned at `u64::MAX`.
    pub arithmetic_overflow_events: u64,
}

impl FrontierDensityTelemetry {
    pub(crate) fn begin(domain_bits: u32) -> Self {
        Self {
            domain_bits: u64::from(domain_bits),
            samples: 0,
            iterations: 0,
            active_bits_total: 0,
            delta_bits_total: 0,
            last_active_bits: 0,
            last_delta_bits: 0,
            peak_active_bits: 0,
            peak_delta_bits: 0,
            truncated_frontier_samples: 0,
            domain_overflow_events: 0,
            arithmetic_overflow_events: 0,
        }
    }

    pub(crate) fn record_sample(&mut self, active: u64) {
        self.samples = bounded_add(self.samples, 1);
        self.active_bits_total = bounded_add(self.active_bits_total, active);
        self.last_active_bits = active;
        self.peak_active_bits = self.peak_active_bits.max(active);
    }

    pub(crate) fn record_transition(&mut self, active: u64, delta: u64) {
        self.iterations = bounded_add(self.iterations, 1);
        self.record_sample(active);
        self.record_delta(delta);
    }

    pub(crate) fn record_window(&mut self, iterations: u32, active: u64, delta: u64) {
        self.iterations = bounded_add(self.iterations, u64::from(iterations));
        self.record_sample(active);
        self.record_delta(delta);
    }

    fn record_delta(&mut self, delta: u64) {
        self.delta_bits_total = bounded_add(self.delta_bits_total, delta);
        self.last_delta_bits = delta;
        self.peak_delta_bits = self.peak_delta_bits.max(delta);
    }

    /// Malformed frontier decode length (cold error telemetry).
    #[cold]
    pub(crate) fn record_truncated_frontier_sample(&mut self) {
        self.truncated_frontier_samples = bounded_add(self.truncated_frontier_samples, 1);
    }

    /// Domain bit width overflow (cold error telemetry).
    #[cold]
    pub(crate) fn record_domain_overflow(&mut self) {
        self.domain_overflow_events = bounded_add(self.domain_overflow_events, 1);
    }

    /// Telemetry counter saturation (cold error telemetry).
    #[cold]
    pub(crate) fn record_arithmetic_overflow(&mut self) {
        self.arithmetic_overflow_events = bounded_add(self.arithmetic_overflow_events, 1);
    }

    /// Mean active frontier density across all samples in parts per million.
    #[must_use]
    pub fn active_density_ppm(self) -> u64 {
        let denominator = u128::from(self.domain_bits) * u128::from(self.samples);
        if denominator == 0 {
            0
        } else {
            ppm(self.active_bits_total, denominator)
        }
    }

    /// Mean changed-frontier density across decoded GPU transitions in parts per million.
    #[must_use]
    pub fn delta_density_ppm(self) -> u64 {
        let denominator = u128::from(self.domain_bits) * u128::from(self.iterations);
        if denominator == 0 {
            0
        } else {
            ppm(self.delta_bits_total, denominator)
        }
    }

    /// Density of the latest decoded frontier sample in parts per million.
    #[must_use]
    pub fn last_active_density_ppm(self) -> u64 {
        if self.domain_bits == 0 {
            0
        } else {
            ppm(self.last_active_bits, u128::from(self.domain_bits))
        }
    }

    /// Density of the latest decoded frontier delta in parts per million.
    #[must_use]
    pub fn last_delta_density_ppm(self) -> u64 {
        if self.domain_bits == 0 {
            0
        } else {
            ppm(self.last_delta_bits, u128::from(self.domain_bits))
        }
    }

    /// Choose a sparse, dense, or hybrid execution family from measured density.
    #[must_use]
    pub fn recommended_execution_mode(self) -> FrontierExecutionMode {
        let plan = plan_sparse_dense_domain(
            SparseDenseDomainPolicy::frontier_density(),
            SparseDenseDomainObservation {
                domain_bits: self.domain_bits,
                samples: self.samples,
                iterations: self.iterations,
                active_bits_total: self.active_bits_total,
                delta_bits_total: self.delta_bits_total,
                last_active_bits: self.last_active_bits,
                peak_active_bits: self.peak_active_bits,
                average_degree_ppm: 0,
            },
        );
        match plan.mode {
            SparseDenseDomainMode::Empty => FrontierExecutionMode::Empty,
            SparseDenseDomainMode::Sparse => FrontierExecutionMode::Sparse,
            SparseDenseDomainMode::Hybrid => FrontierExecutionMode::Hybrid,
            SparseDenseDomainMode::Dense => FrontierExecutionMode::Dense,
        }
    }
}

fn bounded_add(left: u64, right: u64) -> u64 {
    #[allow(clippy::manual_saturating_arithmetic)]
    left.checked_add(right).unwrap_or(u64::MAX)
}

fn ppm(numerator: u64, denominator: u128) -> u64 {
    let scaled = u128::from(numerator) * 1_000_000;
    let value = scaled / denominator;
    if value > u128::from(u64::MAX) {
        u64::MAX
    } else {
        value as u64
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn frontier_density_telemetry_does_not_panic_on_counter_overflow() {
        let mut telemetry = FrontierDensityTelemetry {
            domain_bits: u64::from(u32::MAX),
            samples: u64::MAX,
            iterations: u64::MAX,
            active_bits_total: u64::MAX,
            delta_bits_total: u64::MAX,
            last_active_bits: 0,
            last_delta_bits: 0,
            peak_active_bits: 0,
            peak_delta_bits: 0,
            truncated_frontier_samples: 0,
            domain_overflow_events: 0,
            arithmetic_overflow_events: 0,
        };

        telemetry.record_transition(u64::MAX, u64::MAX);
        telemetry.record_window(u32::MAX, u64::MAX, u64::MAX);

        assert_eq!(telemetry.samples, u64::MAX);
        assert_eq!(telemetry.iterations, u64::MAX);
        assert_eq!(telemetry.active_bits_total, u64::MAX);
        assert_eq!(telemetry.delta_bits_total, u64::MAX);
        telemetry.record_truncated_frontier_sample();
        telemetry.record_domain_overflow();
        telemetry.record_arithmetic_overflow();
        assert_eq!(telemetry.truncated_frontier_samples, 1);
        assert_eq!(telemetry.domain_overflow_events, 1);
        assert_eq!(telemetry.arithmetic_overflow_events, 1);
        let _ = telemetry.active_density_ppm();
        let _ = telemetry.delta_density_ppm();
        assert!(telemetry.last_active_density_ppm() > 1_000_000);
        assert!(telemetry.last_delta_density_ppm() > 1_000_000);
    }

    #[test]
    fn frontier_density_telemetry_source_has_no_panic_arithmetic() {
        let source = include_str!("telemetry.rs");
        let production = source
            .split("#[cfg(test)]")
            .next()
            .expect("telemetry production source must precede tests");
        assert!(
            !production.contains(concat!("panic", "!("))
                && !production.contains(".unwrap_or_else(")
                && !production.contains(concat!(".", "saturating_add")),
            "Fix: fixed-point frontier telemetry must not abort or hide overflow with saturating arithmetic."
        );
        assert!(
            production.contains("u128::from(self.domain_bits) * u128::from(self.samples)")
                && production.contains("fn bounded_add(")
                && production.contains("record_truncated_frontier_sample"),
            "Fix: fixed-point frontier telemetry must use widened density math and bounded explicit counter updates."
        );
    }
}