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
use aligned_vec::{AVec, ConstAlign};
use crate::{
DataflowId,
id::{DataId, NodeId},
metadata::Metadata,
};
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
#[allow(clippy::large_enum_variant)]
#[non_exhaustive]
pub enum InterDaemonEvent {
Output {
dataflow_id: DataflowId,
node_id: NodeId,
output_id: DataId,
metadata: Metadata,
#[serde(with = "crate::bulk_bytes::option")]
data: Option<AVec<u8, ConstAlign<128>>>,
},
OutputClosed {
dataflow_id: DataflowId,
node_id: NodeId,
output_id: DataId,
},
/// An opaque message between the daemons of one dataflow, sent on
/// behalf of an out-of-tree extension. dora never interprets
/// `namespace` or `payload` — it moves the bytes and nothing else.
///
/// This is the inter-daemon half of the extension seam (the node-facing
/// halves are [`crate::node_to_daemon::DaemonRequest::ExtensionRequest`]
/// and the extension table). Naming the variants after any one
/// transport would freeze that transport's architecture into the 1.0
/// protocol, which is exactly what `docs/extensions.md` exists to
/// avoid; a second extension needs no change here at all.
ExtensionMessage {
dataflow_id: DataflowId,
/// Extension that owns the payload. A daemon with no extension
/// registered under this namespace drops the message.
namespace: String,
/// When set, only the daemon whose machine id matches acts on the
/// message; the others drop it. `None` addresses every daemon in
/// the dataflow.
target_machine: Option<String>,
#[serde(with = "crate::bulk_bytes::vec")]
payload: Vec<u8>,
},
}
impl InterDaemonEvent {
/// Bulk bytes this event will contribute to its encoding, for
/// [`crate::encode_presized`].
///
/// Not to be confused with [`crate::metadata::debug_frame_wire_size`], which
/// answers "how big was this on the wire" and deliberately prefers the
/// daemon-stamped `WIRE_SIZE` parameter over the buffer length (#2584). This
/// one must be the actual buffer length, since it sizes an allocation.
pub fn encode_size_hint(&self) -> usize {
match self {
Self::Output { data, .. } => data.as_ref().map_or(0, |d| d.len()),
Self::OutputClosed { .. } => 0,
// Opaque to dora, but it can carry a full tensor frame.
Self::ExtensionMessage { payload, .. } => payload.len(),
}
}
}