arty_executor 0.6.0

Single-threaded async task executor for the Arty runtime.
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

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

use data_privacy::classified;
use observed::{Sink, emit, event};

use crate::TaskPlacement;

static SYSTEM_METADATA: data_privacy::DataClass = data_privacy::DataClass::new("arty", "SystemMetadata");

#[classified(SYSTEM_METADATA)]
#[derive(Clone, Copy)]
struct PlacementLabel(&'static str);

pub(crate) fn report_created(sink: &Sink, placement: TaskPlacement) {
    let placement = match placement {
        TaskPlacement::Any => "any",
        TaskPlacement::SameThread => "same_thread",
        TaskPlacement::SpecificThread => "specific_thread",
    };
    emit!(
        sink,
        TaskCreated {
            placement: PlacementLabel(placement)
        }
    );
}

pub(crate) fn report_succeeded(sink: &Sink, polls: u64) {
    emit!(sink, TaskSucceeded { polls });
}

pub(crate) fn report_panicked(sink: &Sink, polls: u64) {
    emit!(sink, TaskPanicked { polls });
}

#[event("arty.rt.task.spawned")]
#[counter(name = "arty.rt.task.spawned")]
struct TaskCreated {
    #[dimension(metric = "placement")]
    placement: PlacementLabel,
}

#[event("arty.rt.task.succeeded")]
#[counter(name = "arty.rt.task.succeeded")]
#[histogram(polls, name = "arty.rt.task.succeeded.polls", unit = "{poll}")]
struct TaskSucceeded {
    #[unredacted]
    polls: u64,
}

#[event("arty.rt.task.panicked")]
#[error("task panicked")]
#[counter(name = "arty.rt.task.panicked")]
#[histogram(polls, name = "arty.rt.task.panicked.polls", unit = "{poll}")]
struct TaskPanicked {
    #[unredacted]
    polls: u64,
}

#[cfg(test)]
mod tests {
    use observed_testing::{TEST_ID, test_emitter};

    use super::*;

    #[test]
    fn lifecycle_events_include_placement_and_poll_count() {
        let (sink, processor) = test_emitter(TEST_ID);

        report_created(&sink, TaskPlacement::SameThread);
        report_succeeded(&sink, 3);

        let events = processor.events();
        assert_eq!(events[0].name(), "arty.rt.task.spawned");
        assert_eq!(events[0].dimensions()[0], ("placement".to_owned(), "same_thread".into()));
        assert_eq!(events[1].name(), "arty.rt.task.succeeded");
        assert_eq!(events[1].field_metrics()[0].field_key(), "polls");
        assert_eq!(events[1].field_metrics()[0].metric().unit(), "{poll}");
        assert!((events[1].field_metrics()[0].value() - 3.0).abs() < f64::EPSILON);
    }

    #[test]
    fn specific_thread_placement_has_a_distinct_label() {
        let (sink, processor) = test_emitter(TEST_ID);

        report_created(&sink, TaskPlacement::SpecificThread);

        assert_eq!(
            processor.events()[0].dimensions()[0],
            ("placement".to_owned(), "specific_thread".into())
        );
    }
}