use std::future::Future;
use std::sync::Arc;
use crate::runtime::BlockDescription;
use crate::runtime::BlockId;
use crate::runtime::BlockMessage;
use crate::runtime::BlockStatus;
use crate::runtime::Edge;
use crate::runtime::Error;
use crate::runtime::FlowgraphId;
use crate::runtime::FlowgraphMessage;
use crate::runtime::Result;
use crate::runtime::channel::mpsc::Receiver;
use crate::runtime::channel::mpsc::Sender;
use crate::runtime::channel::oneshot;
use crate::runtime::flowgraph_handle::RunningBlockEntry;
use crate::runtime::flowgraph_handle::RunningFlowgraphRegistry;
use crate::runtime::scheduler::LocalRunningDomain;
use crate::runtime::scheduler::NormalBlocks;
use crate::runtime::scheduler::PreparedLocalDomain;
use crate::runtime::scheduler::Scheduler;
use crate::runtime::scheduler::dev::DomainTopology;
use crate::runtime::scheduler::dev::NormalDomainSpec;
use crate::runtime::scheduler::dev::NormalRunningDomain;
use super::BlockSlot;
use super::Flowgraph;
use super::connector::FlowgraphConnector;
use super::connector::ResolvedEdge;
use super::domains::FlowgraphDomains;
use super::domains::RunningFlowgraphDomains;
use super::terminated::TerminatedFlowgraph;
use super::types::BlockLocation;
use super::types::BlockPlacement;
fn domain_topology(
block_ids: &[BlockId],
stream_edges: &[Edge],
message_edges: &[Edge],
) -> DomainTopology {
let relevant =
|edge: &Edge| block_ids.contains(&edge.src_block) || block_ids.contains(&edge.dst_block);
DomainTopology::new(
block_ids.to_vec(),
stream_edges
.iter()
.filter(|edge| relevant(edge))
.cloned()
.collect(),
message_edges
.iter()
.filter(|edge| relevant(edge))
.cloned()
.collect(),
)
}
pub(super) struct PreparedFlowgraph {
id: FlowgraphId,
placements: Vec<BlockPlacement>,
graph_domains: FlowgraphDomains,
registry: Arc<RunningFlowgraphRegistry>,
normal_topology: DomainTopology,
local_domains: Vec<PreparedLocalDomain>,
main_channel: Sender<FlowgraphMessage>,
}
impl PreparedFlowgraph {
pub(super) fn start_initialized<'a, S: Scheduler>(
self,
scheduler: &S,
main_rx: &'a Receiver<FlowgraphMessage>,
startup: oneshot::Sender<Result<Arc<RunningFlowgraphRegistry>, Error>>,
) -> impl Future<Output = Result<RunningFlowgraph, Error>> + 'a {
let Self {
id,
placements,
graph_domains,
registry,
normal_topology,
local_domains,
main_channel,
} = self;
let (running_domains, normal_blocks) = graph_domains.into_running();
let normal_spec = NormalDomainSpec::new(normal_blocks, normal_topology, main_channel);
let normal_domain = scheduler.start_normal_domain(normal_spec);
async move {
let normal_domain = match normal_domain {
Ok(domain) => domain,
Err(e) => {
let _ = startup.send(Err(e.clone()));
return Err(e);
}
};
let mut running = RunningFlowgraph {
id,
placements,
graph_domains: running_domains,
registry,
normal_domain,
local_domains: Vec::with_capacity(local_domains.len()),
active_blocks: 0,
};
for spec in local_domains {
match spec.start() {
Ok(domain) => running.local_domains.push(domain),
Err(e) => {
running.cleanup().await;
let _ = startup.send(Err(e.clone()));
return Err(e);
}
}
}
running.active_blocks = match running.initialize_blocks(main_rx).await {
Ok(active_blocks) => active_blocks,
Err(e) => {
running.cleanup().await;
let _ = startup.send(Err(e.clone()));
return Err(e);
}
};
if startup.send(Ok(running.registry.clone())).is_err() {
running.cleanup().await;
return Err(Error::RuntimeError(
"main thread dropped flowgraph startup receiver".to_string(),
));
}
Ok(running)
}
}
}
pub(super) struct RunningFlowgraph {
id: FlowgraphId,
placements: Vec<BlockPlacement>,
graph_domains: RunningFlowgraphDomains,
registry: Arc<RunningFlowgraphRegistry>,
normal_domain: NormalRunningDomain,
local_domains: Vec<LocalRunningDomain>,
active_blocks: u32,
}
impl RunningFlowgraph {
fn mark_block_terminated(&self, block_id: BlockId) {
self.registry.mark_terminated(block_id);
}
async fn initialize_blocks(
&mut self,
main_rx: &Receiver<FlowgraphMessage>,
) -> Result<u32, Error> {
debug!("init blocks");
let mut active_blocks = 0u32;
for inbox in self.registry.endpoints() {
inbox.send(BlockMessage::Initialize).await?;
active_blocks += 1;
}
debug!("wait for blocks init");
let mut initializing = active_blocks;
let mut block_error = None;
while initializing > 0 {
let message = main_rx.recv().await.ok_or_else(|| {
Error::RuntimeError("no reply from blocks during init phase".to_string())
})?;
match message {
FlowgraphMessage::Initialized => initializing -= 1,
FlowgraphMessage::BlockError { block_id, error } => {
self.mark_block_terminated(block_id);
initializing -= 1;
active_blocks -= 1;
error!("flowgraph init: block {:?} reported an error", block_id);
if block_error.is_none() {
block_error = Some(error);
}
}
FlowgraphMessage::BlockDone { block_id } => {
self.mark_block_terminated(block_id);
initializing -= 1;
active_blocks -= 1;
debug!("block {:?} terminated during initialization", block_id);
}
FlowgraphMessage::Terminate => {
return Err(Error::FlowgraphTerminated);
}
}
}
if let Some(error) = block_error {
return Err(error);
}
debug!("running blocks");
for inbox in self.registry.endpoints() {
if inbox.send(BlockMessage::Start).await.is_err() {
debug!("runtime wanted to start block that already terminated");
}
}
Ok(active_blocks)
}
async fn stop_domains(&mut self) {
self.normal_domain.stop().await;
for domain in &mut self.local_domains {
if let Err(e) = domain.stop().await {
debug!("runtime tried to stop local domain that was already terminated: {e}");
}
}
}
async fn join_domains(
normal_domain: NormalRunningDomain,
local_domains: Vec<LocalRunningDomain>,
) -> Result<NormalBlocks, Error> {
let normal_blocks = normal_domain.join().await;
let mut join_error = None;
for domain in local_domains {
if let Err(e) = domain.join().await
&& join_error.is_none()
{
join_error = Some(e);
}
}
if let Some(e) = join_error {
Err(e)
} else {
Ok(normal_blocks)
}
}
pub(super) async fn cleanup(mut self) {
self.stop_domains().await;
if let Err(e) = Self::join_domains(self.normal_domain, self.local_domains).await {
warn!("error while cleaning up started domains: {e}");
}
}
pub(super) async fn wait(
mut self,
main_rx: &Receiver<FlowgraphMessage>,
) -> Result<TerminatedFlowgraph, Error> {
let run_result = self.drive_runtime_loop(main_rx).await;
if let Err(e) = run_result {
self.cleanup().await;
return Err(e);
}
let Self {
id,
placements,
graph_domains,
registry: _,
normal_domain,
local_domains,
active_blocks: _,
} = self;
let normal_blocks = Self::join_domains(normal_domain, local_domains).await?;
let domains = graph_domains.restore_stopped_domains(normal_blocks)?;
Ok(TerminatedFlowgraph::new(id, placements, domains))
}
async fn drive_runtime_loop(
&mut self,
main_rx: &Receiver<FlowgraphMessage>,
) -> Result<(), Error> {
let mut terminated = false;
let mut block_error = None;
let mut active_blocks = self.active_blocks;
while active_blocks > 0 {
let message = main_rx.recv().await.ok_or_else(|| {
Error::RuntimeError("all senders to flowgraph inbox dropped".to_string())
})?;
match message {
FlowgraphMessage::BlockDone { block_id } => {
self.mark_block_terminated(block_id);
active_blocks -= 1;
}
FlowgraphMessage::BlockError { block_id, error } => {
self.mark_block_terminated(block_id);
if block_error.is_none() {
block_error = Some(error);
}
active_blocks -= 1;
if !terminated {
self.stop_domains().await;
terminated = true;
}
}
FlowgraphMessage::Terminate => {
if !terminated {
self.stop_domains().await;
terminated = true;
}
}
FlowgraphMessage::Initialized => {
warn!("flowgraph lifecycle loop received late initialization message");
}
}
}
if let Some(error) = block_error {
Err(error)
} else {
Ok(())
}
}
}
pub(super) async fn prepare_flowgraph(
mut flowgraph: Flowgraph,
main_channel: Sender<FlowgraphMessage>,
) -> Result<PreparedFlowgraph, Error> {
validate_stream_graph(&flowgraph)?;
let raw_stream_edges = flowgraph
.stream_edges
.iter()
.map(|edge| edge.edge())
.collect::<Vec<_>>();
let stream_edges_public = raw_stream_edges
.iter()
.map(|edge| flowgraph.named_stream_edge(edge))
.collect::<Result<Vec<_>, _>>()?;
let stream_edges = stream_edges_public
.iter()
.map(|edge| flowgraph.indexed_stream_edge(edge))
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.map(ResolvedEdge::from_indexed)
.collect::<Vec<_>>();
let message_edges_public = flowgraph.message_edges.clone();
let message_edges = message_edges_public
.iter()
.map(|edge| flowgraph.indexed_message_edge(edge))
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.map(ResolvedEdge::from_indexed)
.collect::<Vec<_>>();
let block_locations = flowgraph.block_locations()?;
let normal_block_ids = block_locations
.iter()
.filter_map(|location| location.is_normal().then_some(location.block_id))
.collect::<Vec<_>>();
let normal_topology = domain_topology(
&normal_block_ids,
&stream_edges_public,
&message_edges_public,
);
let local_domains = local_domain_specs(
&flowgraph,
&block_locations,
&stream_edges_public,
&message_edges_public,
&main_channel,
);
let mut connector = FlowgraphConnector::new(&mut flowgraph);
connector.apply_stream_edges(&stream_edges).await?;
connector.apply_message_edges(&message_edges).await?;
let Flowgraph {
id,
blocks,
domains: mut graph_domains,
stream_edges: _,
message_edges: _,
} = flowgraph;
let (placements, registry) = running_registry(
&mut graph_domains,
blocks,
stream_edges_public,
message_edges_public,
)
.await?;
Ok(PreparedFlowgraph {
id,
placements,
graph_domains,
registry,
normal_topology,
local_domains,
main_channel,
})
}
fn local_domain_specs(
flowgraph: &Flowgraph,
block_locations: &[BlockLocation],
stream_edges: &[Edge],
message_edges: &[Edge],
main_channel: &Sender<FlowgraphMessage>,
) -> Vec<PreparedLocalDomain> {
let mut local_slots_by_domain = vec![Vec::new(); flowgraph.domains.domain_len()];
for location in block_locations {
if location.is_local() {
local_slots_by_domain[location.domain_id]
.push((location.block_id, location.domain_slot));
}
}
flowgraph
.domains
.local_domain_ids()
.filter_map(|domain_id| {
let slots = std::mem::take(&mut local_slots_by_domain[domain_id]);
if slots.is_empty() {
return None;
}
let block_ids = slots
.iter()
.map(|(block_id, _)| *block_id)
.collect::<Vec<_>>();
Some(PreparedLocalDomain::new(
domain_id,
flowgraph
.domains
.local(domain_id)
.expect("planned local domain disappeared")
.inbox(),
slots,
domain_topology(&block_ids, stream_edges, message_edges),
main_channel.clone(),
))
})
.collect()
}
fn validate_stream_graph(flowgraph: &Flowgraph) -> Result<(), Error> {
let mut adjacency = vec![Vec::new(); flowgraph.blocks.len()];
let mut connected_inputs = Vec::with_capacity(flowgraph.stream_edges.len());
for edge in &flowgraph.stream_edges {
let (src, dst) = edge.endpoints();
if src == dst {
return Err(Error::ValidationError(format!(
"stream self-connections are not supported ({src:?})"
)));
}
if src.0 >= flowgraph.blocks.len() {
return Err(Error::InvalidBlock(src));
}
if dst.0 >= flowgraph.blocks.len() {
return Err(Error::InvalidBlock(dst));
}
let indexed_edge = flowgraph.indexed_stream_edge(&edge.edge)?;
if connected_inputs
.iter()
.any(|(block, port)| *block == dst && port == &indexed_edge.dst_port.index_value())
{
let dst_port = flowgraph.stream_input_name(dst, &edge.edge.dst_port)?;
return Err(Error::ValidationError(format!(
"stream input {:?}.{} has more than one connection",
dst,
dst_port.name()
)));
}
connected_inputs.push((dst, indexed_edge.dst_port.index_value()));
if edge.local_only {
let src_location = flowgraph.location(src)?;
let dst_location = flowgraph.location(dst)?;
Flowgraph::same_local_stream_locations(src_location, dst_location, false)?;
}
adjacency[src.0].push(dst.0);
}
fn visit(node: usize, adjacency: &[Vec<usize>], marks: &mut [u8]) -> bool {
match marks[node] {
1 => return false,
2 => return true,
_ => {}
}
marks[node] = 1;
for &next in &adjacency[node] {
if !visit(next, adjacency, marks) {
return false;
}
}
marks[node] = 2;
true
}
let mut marks = vec![0; flowgraph.blocks.len()];
for node in 0..flowgraph.blocks.len() {
if !visit(node, &adjacency, &mut marks) {
return Err(Error::ValidationError(
"stream connections must form a directed acyclic graph".to_string(),
));
}
}
Ok(())
}
async fn running_registry(
domains: &mut FlowgraphDomains,
block_slots: Vec<BlockSlot>,
stream_edges: Vec<Edge>,
message_edges: Vec<Edge>,
) -> Result<(Vec<BlockPlacement>, Arc<RunningFlowgraphRegistry>), Error> {
let mut placements = Vec::with_capacity(block_slots.len());
let mut blocks = Vec::with_capacity(block_slots.len());
for (id, entry) in block_slots.into_iter().enumerate() {
let block_id = BlockId(id);
let location = entry.location(block_id);
let (type_name, instance_name, blocking) = domains
.with_block_mut(location, |block| {
let type_name = block.type_name().to_string();
let instance_name = block
.instance_name()
.map(str::to_string)
.unwrap_or_else(|| type_name.clone());
Ok((type_name, instance_name, block.is_blocking()))
})
.await?;
let BlockSlot {
placement,
endpoint,
stream_inputs,
stream_outputs,
message_inputs,
message_outputs,
} = entry;
placements.push(placement);
blocks.push(RunningBlockEntry::new(
endpoint,
BlockDescription {
id: block_id,
status: BlockStatus::Running,
type_name,
instance_name,
stream_inputs,
stream_outputs,
message_inputs: message_inputs.iter().map(|name| name.to_string()).collect(),
message_outputs: message_outputs
.iter()
.map(|name| name.to_string())
.collect(),
blocking,
},
));
}
Ok((
placements,
Arc::new(RunningFlowgraphRegistry::new(
blocks,
stream_edges,
message_edges,
)),
))
}
#[cfg(all(test, not(target_arch = "wasm32")))]
mod tests {
use crate::blocks::NullSink;
use crate::blocks::NullSource;
use crate::runtime::Flowgraph;
use crate::runtime::FlowgraphMessage;
use crate::runtime::PortId;
use crate::runtime::channel::mpsc::channel;
use crate::runtime::wrapped_kernel::LocalWrappedKernel;
use super::*;
#[test]
fn preparation_produces_startup_state() -> Result<(), Error> {
let mut fg = Flowgraph::new();
let src = fg.add(NullSource::<f32>::new())?;
let snk = fg.add(NullSink::<f32>::new())?;
fg.stream_dyn(src, "output", snk, "input")?;
let (main_channel, _main_rx) = channel::<FlowgraphMessage>(8);
let prepared = crate::runtime::block_on(prepare_flowgraph(fg, main_channel))?;
assert_eq!(prepared.placements.len(), 2);
let description = prepared.registry.describe();
assert_eq!(
description
.blocks
.iter()
.map(|block| block.id)
.collect::<Vec<_>>(),
vec![src.id(), snk.id()]
);
assert_eq!(description.blocks.len(), 2);
assert!(
description
.blocks
.iter()
.all(|description| description.status == BlockStatus::Running)
);
assert_eq!(
description.stream_edges,
vec![Edge::new(
src.id(),
PortId::from("output"),
snk.id(),
PortId::from("input")
)]
);
assert!(prepared.local_domains.is_empty());
assert_eq!(prepared.normal_topology.blocks(), &[src.id(), snk.id()]);
assert_eq!(
prepared.normal_topology.stream_edges(),
&[Edge::new(
src.id(),
PortId::from("output"),
snk.id(),
PortId::from("input")
)]
);
Ok(())
}
#[test]
fn preparation_reads_local_block_metadata() -> Result<(), Error> {
let mut fg = Flowgraph::new();
let local = fg.local_domain()?;
let src = fg.with_local_domain(local, |ctx| Ok(ctx.add(NullSource::<f32>::new())))?;
let snk = fg.add(NullSink::<f32>::new())?;
fg.stream_dyn(src, "output", snk, "input")?;
let location = fg.location(src.id())?;
crate::runtime::block_on(fg.with_block_mut(location, move |block| {
let block = (block as &mut dyn std::any::Any)
.downcast_mut::<LocalWrappedKernel<NullSource<f32>>>()
.ok_or(Error::InvalidBlock(location.block_id))?;
block.meta.set_instance_name("local-source");
Ok(())
}))?;
let (main_channel, _main_rx) = channel::<FlowgraphMessage>(8);
let prepared = crate::runtime::block_on(prepare_flowgraph(fg, main_channel))?;
let description = prepared.registry.describe();
let src = description
.blocks
.iter()
.find(|block| block.id == src.id())
.unwrap();
assert_eq!(src.instance_name, "local-source");
Ok(())
}
}