Cloacina Computation Graph
Core types for Cloacina computation graph plugins.
This crate contains the types that packaged computation graph cdylibs need
at compile time. It is the computation-graph equivalent of cloacina-workflow
— a thin crate that avoids pulling in the full engine.
The #[computation_graph] macro expands into code that references types from
this crate. Embedded-mode users get these types re-exported from cloacina.
pub
Derives: Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize
Identifies an accumulator source by name.
| Name | Type | Description |
|---|---|---|
0 |
String |
fn new (name : impl Into < String >) -> Self
Source
pub fn new(name: impl Into<String>) -> Self {
Self(name.into())
}
fn as_str (& self) -> & str
Source
pub fn as_str(&self) -> &str {
&self.0
}
pub
Derives: Debug, Clone
The input cache holds the last-seen serialized boundary per source.
The reactor’s receiver task updates this cache continuously. The executor takes a snapshot before calling the compiled graph function. Serialization format: bincode (compact binary). The FFI packaging bridge converts bincode→JSON at the boundary for plugin compatibility.
| Name | Type | Description |
|---|---|---|
entries |
HashMap < SourceName , Vec < u8 > > |
fn new () -> Self
Source
pub fn new() -> Self {
Self {
entries: HashMap::new(),
}
}
fn update (& mut self , source : SourceName , bytes : Vec < u8 >)
Update the cached value for a source.
Source
pub fn update(&mut self, source: SourceName, bytes: Vec<u8>) {
self.entries.insert(source, bytes);
}
fn get < T : DeserializeOwned > (& self , name : & str) -> Option < Result < T , GraphError > >
Get and deserialize a cached value by source name.
Source
pub fn get<T: DeserializeOwned>(&self, name: &str) -> Option<Result<T, GraphError>> {
let bytes = self.entries.get(&SourceName::new(name))?;
Some(deserialize::<T>(bytes))
}
fn has (& self , name : & str) -> bool
Check if a source has an entry in the cache.
Source
pub fn has(&self, name: &str) -> bool {
self.entries.contains_key(&SourceName::new(name))
}
fn get_raw (& self , name : & str) -> Option < & [u8] >
Get the raw bytes for a source.
Source
pub fn get_raw(&self, name: &str) -> Option<&[u8]> {
self.entries
.get(&SourceName::new(name))
.map(|v| v.as_slice())
}
fn snapshot (& self) -> InputCache
Create a snapshot (clone) of the cache.
Source
pub fn snapshot(&self) -> InputCache {
self.clone()
}
fn len (& self) -> usize
Number of sources in the cache.
Source
pub fn len(&self) -> usize {
self.entries.len()
}
fn is_empty (& self) -> bool
Whether the cache is empty.
Source
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
fn replace_all (& mut self , other : InputCache)
Replace all entries.
Source
pub fn replace_all(&mut self, other: InputCache) {
self.entries = other.entries;
}
fn sources (& self) -> Vec < & SourceName >
List all source names in the cache.
Source
pub fn sources(&self) -> Vec<&SourceName> {
self.entries.keys().collect()
}
fn entries_raw (& self) -> & HashMap < SourceName , Vec < u8 > >
Get a reference to the raw entries map.
Source
pub fn entries_raw(&self) -> &HashMap<SourceName, Vec<u8>> {
&self.entries
}
fn entries_as_json (& self) -> HashMap < String , String >
Return entries as a JSON-friendly map.
Source
pub fn entries_as_json(&self) -> HashMap<String, String> {
self.entries
.iter()
.map(|(name, bytes)| {
let value = if cfg!(debug_assertions) {
serde_json::from_slice::<serde_json::Value>(bytes)
.map(|v| v.to_string())
.unwrap_or_else(|_| hex_encode(bytes))
} else {
hex_encode(bytes)
};
(name.as_str().to_string(), value)
})
.collect()
}
pub
Metadata about a registered computation graph.
accumulator_names and reaction_mode are the canonical fields consumed
by the packaging FFI and the reconciler. Bundled-form graphs populate
these from the local declaration; split-form graphs mirror the
referenced reactor’s declaration. Trigger-less graphs carry empty
accumulator_names and trigger_reactor = None.
| Name | Type | Description |
|---|---|---|
graph_fn |
CompiledGraphFn |
The compiled graph function. |
trigger_reactor |
Option < String > |
Name of the reactor this graph is bound to, if any. None for |
| trigger-less graphs (T-02/T-03 invoke these directly from workflow | ||
| tasks or Python tasks). | ||
accumulator_names |
Vec < String > |
Accumulator names. For split-form graphs this mirrors the reactor’s |
| accumulators; for trigger-less graphs it is empty. | ||
reaction_mode |
String |
Reaction mode: "when_any", "when_all", or "none" for |
| trigger-less graphs. |
pub
Derives: Debug, Clone
Runtime-side description of a reactor.
Populated by the #[reactor] macro’s emitted inventory entry.
| Name | Type | Description |
|---|---|---|
name |
String |
|
accumulator_names |
Vec < String > |
|
reaction_mode |
ReactionMode |
Result of executing a compiled computation graph.
Completed- Graph executed to completion. Contains terminal node outputs.Error- Graph execution failed.
Errors that can occur during graph execution.
SerializationDeserializationMissingInputNodeExecutionExecution
How a reactor decides when to fire.
WhenAny- Fire as soon as any one accumulator has new input.WhenAll- Fire only when every accumulator has new input since the last firing.
pub
fn serialize < T : Serialize > (value : & T) -> Result < Vec < u8 > , GraphError >
Serialize a value to bincode bytes.
Bincode is used for all internal wire formats (boundary channels, checkpoint persistence, accumulator-to-reactor messaging).
Source
pub fn serialize<T: Serialize>(value: &T) -> Result<Vec<u8>, GraphError> {
bincode::serialize(value).map_err(|e| GraphError::Serialization(e.to_string()))
}
pub
fn deserialize < T : DeserializeOwned > (bytes : & [u8]) -> Result < T , GraphError >
Deserialize bincode bytes to a value.
Source
pub fn deserialize<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, GraphError> {
bincode::deserialize(bytes).map_err(|e| GraphError::Deserialization(e.to_string()))
}
private
fn hex_encode (bytes : & [u8]) -> String
Source
fn hex_encode(bytes: &[u8]) -> String {
bytes.iter().map(|b| format!("{:02x}", b)).collect()
}