beamr 0.4.2

A Rust runtime with the BEAM's execution model, targeting Gleam
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

//! OpenTelemetry metric helpers for VM health and per-process scheduler state.

use std::sync::OnceLock;
use std::time::Duration;

use opentelemetry::KeyValue;
use opentelemetry::global;
use opentelemetry::metrics::{Counter, Gauge, Histogram};

const METER_NAME: &str = "beamr";

struct Instruments {
    processes_alive: Gauge<u64>,
    scheduler_utilization: Gauge<f64>,
    gc_collections: Counter<u64>,
    gc_duration: Histogram<f64>,
    messages_sent: Counter<u64>,
    memory_heap_words: Gauge<u64>,
    process_message_queue_len: Gauge<u64>,
    process_reductions: Counter<u64>,
}

impl Instruments {
    fn new() -> Self {
        let meter = global::meter(METER_NAME);
        Self {
            processes_alive: meter
                .u64_gauge("beamr.processes.alive")
                .with_description("Current number of live Beamr processes")
                .with_unit("{process}")
                .build(),
            scheduler_utilization: meter
                .f64_gauge("beamr.scheduler.utilization")
                .with_description(
                    "Fraction of scheduler time spent executing processes rather than idle",
                )
                .with_unit("1")
                .build(),
            gc_collections: meter
                .u64_counter("beamr.gc.collections")
                .with_description("Total number of Beamr garbage collections")
                .with_unit("{collection}")
                .build(),
            gc_duration: meter
                .f64_histogram("beamr.gc.duration")
                .with_description("Beamr garbage collection duration")
                .with_unit("s")
                .build(),
            messages_sent: meter
                .u64_counter("beamr.messages.sent")
                .with_description("Total number of Beamr messages sent")
                .with_unit("{message}")
                .build(),
            memory_heap_words: meter
                .u64_gauge("beamr.memory.heap_words")
                .with_description("Total process heap words allocated")
                .with_unit("{word}")
                .build(),
            process_message_queue_len: meter
                .u64_gauge("beamr.process.message_queue_len")
                .with_description(
                    "Current process mailbox depth sampled at scheduler slice boundaries",
                )
                .with_unit("{message}")
                .build(),
            process_reductions: meter
                .u64_counter("beamr.process.reductions")
                .with_description("Total scheduler reductions consumed by process")
                .with_unit("{reduction}")
                .build(),
        }
    }
}

fn instruments() -> &'static Instruments {
    static INSTRUMENTS: OnceLock<Instruments> = OnceLock::new();
    INSTRUMENTS.get_or_init(Instruments::new)
}

/// Record a VM health snapshot.
pub(crate) fn record_vm_health(
    processes_alive: usize,
    heap_words: usize,
    scheduler_utilization: f64,
) {
    let instruments = instruments();
    instruments
        .processes_alive
        .record(usize_to_u64(processes_alive), &[]);
    instruments
        .memory_heap_words
        .record(usize_to_u64(heap_words), &[]);
    instruments
        .scheduler_utilization
        .record(scheduler_utilization.clamp(0.0, 1.0), &[]);
}

/// Record one successfully completed GC collection.
pub(crate) fn record_gc_collection(kind: &'static str, duration: Duration) {
    let attributes = [KeyValue::new("gc.kind", kind)];
    let instruments = instruments();
    instruments.gc_collections.add(1, &attributes);
    instruments
        .gc_duration
        .record(duration.as_secs_f64(), &attributes);
}

/// Record one successfully sent message.
pub(crate) fn record_message_sent() {
    instruments().messages_sent.add(1, &[]);
}

/// Record sampled process state at a scheduler slice boundary.
pub(crate) fn record_process_slice(pid: u64, reductions: u32, message_queue_len: usize) {
    let pid_value = match i64::try_from(pid) {
        Ok(value) => value,
        Err(_) => i64::MAX,
    };
    let attributes = [KeyValue::new("pid", pid_value)];
    let instruments = instruments();
    instruments
        .process_reductions
        .add(u64::from(reductions), &attributes);
    instruments
        .process_message_queue_len
        .record(usize_to_u64(message_queue_len), &attributes);
}

fn usize_to_u64(value: usize) -> u64 {
    match u64::try_from(value) {
        Ok(value) => value,
        Err(_) => u64::MAX,
    }
}