lmflow 0.3.1

Dataflow-graph compute engine: a pure-Rust scheduler behind a stable C ABI
Documentation
pub(crate) const FONT: &str = "sans";
pub(crate) const CLUSTER_COLOR: &str = "#888888";
pub(crate) const PORT_FILL: &str = "#e8e8e8";
pub(crate) const PORT_COLOR: &str = "#777777";
pub(crate) const BACK_EDGE_COLOR: &str = "#7b61a8";

pub(crate) fn escape(value: &str) -> String {
    value.replace('\\', "\\\\").replace('"', "\\\"")
}

pub(crate) fn begin(name: &str, label: Option<&str>) -> String {
    let mut dot = format!(
        "digraph {name} {{\n  rankdir=LR;\n  newrank=true;\n  graph [fontname=\"{FONT}\", nodesep=0.35, ranksep=0.65"
    );
    if let Some(label) = label {
        dot.push_str(&format!(", labelloc=t, label=\"{}\"", escape(label)));
    }
    dot.push_str("];\n");
    dot.push_str(&format!(
        "  node [fontname=\"{FONT}\", shape=box, style=\"rounded,filled\", fillcolor=white, ordering=out];\n"
    ));
    dot.push_str(&format!("  edge [fontname=\"{FONT}\", fontsize=10];\n"));
    dot
}

pub(crate) fn executor_label(name: &str, kind: &str, threads: usize) -> String {
    if threads == 0 {
        format!("{name} · {kind}")
    } else {
        format!("{name} · {kind} · {threads}t")
    }
}

pub(crate) fn render_plan(plan: &crate::config::GraphPlan) -> String {
    let mut dot = begin("lmflow_plan", Some("lmflow configuration plan"));
    dot.push_str(&format!(
        "  graph_input [shape=cds, style=filled, fillcolor=\"{PORT_FILL}\", color=\"{PORT_COLOR}\", label=\"graph input\"];\n"
    ));
    dot.push_str(&format!(
        "  graph_output [shape=cds, style=filled, fillcolor=\"{PORT_FILL}\", color=\"{PORT_COLOR}\", label=\"graph output\"];\n"
    ));
    let mut executors = vec![(
        "default".to_string(),
        "ThreadPoolExecutor".to_string(),
        0,
        Vec::new(),
        0,
    )];
    executors.extend(plan.config.executors.iter().map(|executor| {
        (
            executor.name.clone(),
            if executor.r#type.is_empty() {
                "ThreadPoolExecutor".to_string()
            } else {
                executor.r#type.clone()
            },
            executor.num_threads,
            executor.affinity.clone(),
            executor.priority,
        )
    }));
    for (index, (name, kind, threads, affinity, priority)) in executors.iter().enumerate() {
        let mut label = executor_label(name, kind, *threads);
        if !affinity.is_empty() {
            label.push_str(&format!(
                " · cores[{}]",
                affinity
                    .iter()
                    .map(usize::to_string)
                    .collect::<Vec<_>>()
                    .join(",")
            ));
        }
        if *priority != 0 {
            label.push_str(&format!(" · priority {priority}"));
        }
        dot.push_str(&format!(
            "  subgraph cluster_executor_{index} {{\n    label=\"{}\";\n    color=\"{CLUSTER_COLOR}\";\n",
            escape(&label),
        ));
        for node in plan.nodes.iter().filter(|node| node.executor == *name) {
            let kind = node.route.as_ref().map_or_else(
                || node.kernel.clone(),
                |route| format!("Route · {:?}", route.mode).to_lowercase(),
            );
            let route_lines = node.route.as_ref().map_or_else(String::new, |route| {
                route
                    .routes
                    .iter()
                    .enumerate()
                    .map(|(index, rule)| {
                        if rule.default {
                            format!("\\n{}: default → {}", index + 1, rule.to)
                        } else {
                            format!(
                                "\\n{}: {} → {}",
                                index + 1,
                                rule.when.as_ref().map_or_else(
                                    || "condition".to_string(),
                                    |predicate| predicate.summary()
                                ),
                                rule.to
                            )
                        }
                    })
                    .collect::<String>()
            });
            let label = format!(
                "{}\\n{}{}\\nexecutor: {}\\npolicy: {}\\nmax_in_flight: {}\\nrate: {} Hz\\ninputs: {}\\noutputs: {}",
                node.name,
                kind,
                route_lines,
                node.executor,
                plan.config.nodes[node.index].input_policy.r#type,
                plan.config.nodes[node.index].max_in_flight.max(1),
                plan.config.nodes[node.index].rate,
                node.inputs.join(", "),
                node.outputs.join(", ")
            );
            dot.push_str(&format!(
                "    node_{} [label=\"{}\"];\n",
                node.index,
                escape(&label)
            ));
        }
        dot.push_str("  }\n");
    }
    let diagnostics = plan.diagnostics();
    if !diagnostics.is_empty() {
        dot.push_str("  subgraph cluster_diagnostics { label=\"diagnostics\"; style=dashed; color=\"#d98c00\";\n");
        for (index, diagnostic) in diagnostics.iter().enumerate() {
            dot.push_str(&format!(
                "    diagnostic_{index} [shape=note, style=filled, fillcolor=\"#fff0c2\", color=\"#d98c00\", label=\"{}\\n{}\"];\n",
                escape(&diagnostic.code),
                escape(&diagnostic.message)
            ));
        }
        dot.push_str("  }\n");
    }
    for (edge_index, edge) in plan.edges.iter().enumerate() {
        let edge_id = format!("edge_{edge_index}");
        dot.push_str(&format!(
            "  {edge_id} [shape=point, width=0.08, label=\"\"];\n"
        ));
        if edge.graph_input {
            dot.push_str(&format!(
                "  graph_input -> {edge_id} [label=\"{}\"];\n",
                escape(&edge.name)
            ));
        }
        if let Some(producer) = edge.producer {
            dot.push_str(&format!("  node_{producer} -> {edge_id};\n"));
        }
        for consumer in &edge.consumers {
            let node = &plan.config.nodes[*consumer];
            let capacity = node
                .input_queues
                .ports
                .get(&edge.name)
                .and_then(|limits| limits.packets)
                .unwrap_or(node.input_queues.packets);
            let mut attributes = vec![format!("label=\"{}\"", escape(&edge.name))];
            if capacity != 0 {
                attributes.push(format!("xlabel=\"queue {capacity} packets\""));
            }
            if node.back_edges.contains(&edge.name) {
                attributes.extend([
                    "style=dashed".to_string(),
                    format!("color=\"{BACK_EDGE_COLOR}\""),
                    format!("fontcolor=\"{BACK_EDGE_COLOR}\""),
                    "constraint=false".to_string(),
                    "tooltip=\"back-edge latest-value register\"".to_string(),
                ]);
            }
            dot.push_str(&format!(
                "  {edge_id} -> node_{consumer} [{}];\n",
                attributes.join(", ")
            ));
        }
        if edge.graph_output {
            dot.push_str(&format!(
                "  {edge_id} -> graph_output [label=\"{}\"];\n",
                escape(&edge.name)
            ));
        }
    }
    dot.push_str("}\n");
    dot
}