sim-lib-stream-fabric 0.1.0

Content-addressed distributed evaluation for remote stream realization.
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

use sim_kernel::{CapabilityName, Expr};
use sim_lib_stream_core::{
    LatencyClass, PlacedFragment, StreamEdge, StreamEnvelope, TransportProfile,
    stream_remote_network_capability, stream_remote_preview_capability,
};

use crate::placement::{
    ServerPlacementRequest, server_placement_request_symbol, server_placement_result_symbol,
    server_site_symbol,
};
use crate::placement_security::{
    placement_expr_uses_host_device, placement_host_device_capability,
    placement_remote_render_capability, placement_run_node_on_server_capability,
    redact_placement_expr, redact_placement_symbol,
};

pub(crate) fn request_report_expr(
    fragment: &PlacedFragment,
    request: &ServerPlacementRequest,
) -> Expr {
    Expr::Map(vec![
        field("operation", Expr::Symbol(server_placement_request_symbol())),
        field("site", Expr::Symbol(server_site_symbol())),
        field(
            "fragment-id",
            Expr::Symbol(redact_placement_symbol(fragment.id())),
        ),
        field("node", redact_placement_expr(fragment.node())),
        field("realtime-pinned", Expr::Bool(request.realtime_pinned())),
        field(
            "placement-report",
            redact_placement_expr(request.placement_report()),
        ),
        field("declared-profile", request.output_profile().to_expr()),
        field(
            "declared-latency",
            Expr::Symbol(request.output_profile().latency_class().symbol()),
        ),
        field("resource-limits", request.resource_limits().to_expr()),
        field(
            "required-capabilities",
            required_capabilities_expr(fragment, request),
        ),
        field("input-edges", edge_list_expr(fragment.input_edges())),
        field("output-edges", edge_list_expr(fragment.output_edges())),
    ])
}

pub(crate) fn result_report_expr(
    fragment: &PlacedFragment,
    payload: Expr,
    output_profile: &TransportProfile,
) -> Expr {
    Expr::Map(vec![
        field("operation", Expr::Symbol(server_placement_result_symbol())),
        field("site", Expr::Symbol(server_site_symbol())),
        field(
            "fragment-id",
            Expr::Symbol(redact_placement_symbol(fragment.id())),
        ),
        field("node-result", redact_placement_expr(&payload)),
        field(
            "output-envelope-count",
            Expr::String(fragment.output_envelopes().len().to_string()),
        ),
        field("declared-profile", output_profile.to_expr()),
        field(
            "declared-latency",
            Expr::Symbol(output_profile.latency_class().symbol()),
        ),
    ])
}

pub(crate) fn placement_uses_host_device(
    fragment: &PlacedFragment,
    request: &ServerPlacementRequest,
) -> bool {
    placement_expr_uses_host_device(fragment.node())
        || placement_expr_uses_host_device(request.placement_report())
        || fragment.input_edges().iter().any(edge_uses_host_device)
        || fragment.output_edges().iter().any(edge_uses_host_device)
}

fn edge_list_expr(edges: &[StreamEdge]) -> Expr {
    Expr::List(edges.iter().map(edge_expr).collect())
}

fn edge_expr(edge: &StreamEdge) -> Expr {
    let rate = edge.rate_contract();
    Expr::Map(vec![
        field("port", Expr::Symbol(redact_placement_symbol(edge.port()))),
        field("clock-domain", Expr::Symbol(rate.clock_domain().symbol())),
        field("latency-class", Expr::Symbol(rate.latency_class().symbol())),
        field(
            "nominal-rate-hz",
            rate.nominal_rate_hz()
                .map(|rate| Expr::String(rate.to_string()))
                .unwrap_or(Expr::Nil),
        ),
        field(
            "metadata",
            redact_placement_expr(&edge.metadata().table_expr()),
        ),
        field(
            "envelopes",
            Expr::List(
                edge.envelopes()
                    .iter()
                    .map(StreamEnvelope::to_expr)
                    .map(|expr| redact_placement_expr(&expr))
                    .collect(),
            ),
        ),
    ])
}

fn edge_uses_host_device(edge: &StreamEdge) -> bool {
    placement_expr_uses_host_device(&Expr::Symbol(edge.port().clone()))
        || placement_expr_uses_host_device(&edge.metadata().table_expr())
        || edge
            .envelopes()
            .iter()
            .any(|envelope| placement_expr_uses_host_device(&envelope.to_expr()))
}

fn required_capabilities_expr(fragment: &PlacedFragment, request: &ServerPlacementRequest) -> Expr {
    let mut capabilities = vec![
        placement_run_node_on_server_capability(),
        stream_remote_network_capability(),
    ];
    capabilities.extend(profile_capabilities(request.output_profile()));
    if placement_uses_host_device(fragment, request) {
        capabilities.push(placement_host_device_capability());
    }
    capabilities.sort_by(|left, right| left.as_str().cmp(right.as_str()));
    capabilities.dedup();
    Expr::List(
        capabilities
            .into_iter()
            .map(|capability| Expr::String(capability.as_str().to_owned()))
            .collect(),
    )
}

fn profile_capabilities(profile: &TransportProfile) -> Vec<CapabilityName> {
    match profile.latency_class() {
        LatencyClass::BufferedPreview => vec![stream_remote_preview_capability()],
        LatencyClass::OfflineRender => vec![placement_remote_render_capability()],
        LatencyClass::Interactive => Vec::new(),
        _ => Vec::new(),
    }
}

use sim_value::build::entry as field;