somatize_core/
transport.rs

1//! Who carries a slice of plan elsewhere and brings back what it produced.
2//!
3//! The core does not know what a wire is: a [`Host`](crate::Host) is a name and
4//! a `Transport` is someone who knows how to reach it. It neither serializes nor
5//! spawns processes nor knows about sockets — a wire format would require
6//! `serde`, and the core has no dependencies.
7//!
8//! Declared versus injected: a [`Node`](crate::Node) is put there by whoever
9//! declares the graph, a `Transport` and a [`Watcher`] by whoever **executes**.
10//! That is why a host is a name and not an address.
11
12use crate::{Keys, Memory, NodeId, Placement, Plan, Value, Watcher};
13use std::fmt;
14
15/// Knows how to execute a plan elsewhere.
16pub trait Transport: Send + Sync {
17    /// Executes `plan` over there, with what it needs in order to do so, and
18    /// tells `seen` whatever the far side says while it is at it.
19    ///
20    /// A [`Value::Opaque`] has to fail here: it carries something that only
21    /// exists in this process.
22    ///
23    /// # Why `seen` is an argument and not a field of either
24    ///
25    /// It is not in [`Cargo`] because everything in a `Cargo` **travels**, and a
26    /// watcher is injected and stays. And it is not given to the transport when
27    /// the transport is built because a worker is opened once and used for many
28    /// runs, while a watcher belongs to **one** run. A watcher is of the call,
29    /// so it goes in the call.
30    ///
31    /// Whatever comes back is passed on as it was emitted: saying *where* it
32    /// happened is the engine's, which is the only one that knows the host by
33    /// the name the graph gave it.
34    fn dispatch(
35        &self,
36        plan: &Plan,
37        cargo: &Cargo<'_>,
38        seen: Option<&dyn Watcher>,
39    ) -> Result<Outcome, TransportError>;
40}
41
42/// What a plan needs beyond itself in order to run elsewhere.
43pub struct Cargo<'a> {
44    /// The graph's input, for the steps over there that read from nobody.
45    pub input: &'a Value,
46    /// What was already produced **here** that the plan over there reads and
47    /// does not produce. Only that: the wire is the expensive part.
48    pub known: &'a [(NodeId, Value)],
49    /// What each of those is called. Without them the slice over there can name
50    /// nothing it produces, and a cache that stops at the process boundary is a
51    /// cache nobody can rely on.
52    pub keys: &'a [(NodeId, Keys)],
53    /// Where each node runs. It travels because a placement is data and the
54    /// catalog is not.
55    pub placement: &'a Placement,
56    /// What is remembered about each node, for the same reason and by the same
57    /// rule: it is data. Without it the other side does not know what is frozen,
58    /// what is worth keeping, or what any of it is called.
59    pub memory: &'a Memory,
60}
61
62/// What came back from executing a plan elsewhere.
63#[derive(Debug, Clone, PartialEq)]
64#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
65pub struct Outcome {
66    /// What the last step returned, exactly as it would have returned it here.
67    /// Not "the last one in the map": a wave has no single output.
68    pub last: Value,
69    /// What each node produced, to be merged with what is here.
70    pub produced: Vec<(NodeId, Value)>,
71    /// And what each of those is called, so the chain of keys carries on below
72    /// the slice that went away.
73    pub keys: Vec<(NodeId, Keys)>,
74}
75
76impl Outcome {
77    /// The same outcome with whatever cannot leave this process left out of
78    /// `produced`.
79    ///
80    /// An intermediate value of a slice is read by the steps of that slice,
81    /// which ran where it did: sending it back was never the point, and
82    /// refusing the whole answer over one is refusing the case this exists for —
83    /// two steps on one host, with something live in between them. What is
84    /// **not** filtered is `last`, which is the value of the slice itself: that
85    /// one has a reader here by definition.
86    ///
87    /// Whoever does read one of the dropped values gets [`RunError::Lost`](crate::RunError::Lost),
88    /// naming both ends.
89    pub fn travelling(mut self) -> Self {
90        self.produced.retain(|(_, value)| value.travels());
91        self
92    }
93}
94
95/// What a transport can answer when it cannot carry something.
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub struct TransportError(String);
98
99impl TransportError {
100    /// A failure described by a message.
101    pub fn new(message: impl Into<String>) -> Self {
102        Self(message.into())
103    }
104
105    /// The message.
106    pub fn message(&self) -> &str {
107        &self.0
108    }
109}
110
111impl fmt::Display for TransportError {
112    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
113        f.write_str(&self.0)
114    }
115}
116
117impl std::error::Error for TransportError {}
somatize_core/transport.rs

somatize_core/
transport.rs
