dmf-studio-rnd/studio-manager/main.cpp

// Studio Manager — launches and monitors node processes for a single-host pipeline.
// Graph is defined in build_graph(). Later: load from graph.json, WebSocket API.
// Node binaries are looked up next to this binary (same directory).

#include <cstdio>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <map>
#include <sstream>
#include <string>
#include <utility>
#include <vector>
#include <fcntl.h>
#include <sys/wait.h>
#include <unistd.h>
#include <mxl/mxl.h>
#include <mxl/time.h>
#include "Signal.hpp"
#include "FlowGraph.hpp"

namespace fs = std::filesystem;

// --- UUID generation ---------------------------------------------------------

static std::string gen_uuid() {
    uint8_t b[16];
    int fd = open("/dev/urandom", O_RDONLY);
    if (fd < 0) { perror("open /dev/urandom"); exit(1); }
    if (read(fd, b, sizeof(b)) != sizeof(b)) { perror("read"); exit(1); }
    close(fd);
    b[6] = (b[6] & 0x0Fu) | 0x40u;  // version 4
    b[8] = (b[8] & 0x3Fu) | 0x80u;  // variant 1
    char s[37];
    snprintf(s, sizeof(s),
        "%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x",
        b[0],b[1],b[2],b[3],  b[4],b[5],  b[6],b[7],
        b[8],b[9],  b[10],b[11],b[12],b[13],b[14],b[15]);
    return s;
}

// --- Pipeline graph ----------------------------------------------------------

static dmf::FlowGraph load_graph(const std::string& path) {
    std::ifstream f(path);
    if (!f) {
        fprintf(stderr, "[studio-manager] cannot open graph file: %s\n", path.c_str());
        exit(1);
    }
    std::ostringstream ss;
    ss << f.rdbuf();

    auto j = nlohmann::json::parse(ss.str(), nullptr, /*allow_exceptions=*/false);
    if (j.is_discarded()) {
        fprintf(stderr, "[studio-manager] invalid JSON in %s\n", path.c_str());
        exit(1);
    }

    dmf::FlowGraph g;
    for (const auto& n : j.at("nodes")) {
        g.nodes.push_back({
            n.at("id").get<std::string>(),
            n.at("type").get<std::string>(),
            n.value("params", nlohmann::json::object())
        });
    }
    // Edges sharing the same from_node+from_port reuse the same flow UUID,
    // so one writer can be read by multiple consumers (MXL supports N readers per flow).
    std::map<std::pair<std::string,std::string>, std::string> flow_ids;
    for (const auto& e : j.at("edges")) {
        auto from_node = e.at("from").get<std::string>();
        auto from_port = e.at("from_port").get<std::string>();
        auto key = std::make_pair(from_node, from_port);
        if (!flow_ids.count(key)) flow_ids[key] = gen_uuid();
        g.edges.push_back({
            flow_ids.at(key),
            from_node,
            from_port,
            e.value("to",      std::string{}),
            e.value("to_port", std::string{}),
            e.at("format")
        });
    }

    fprintf(stderr, "[studio-manager] loaded graph from %s (%zu nodes, %zu edges)\n",
        path.c_str(), g.nodes.size(), g.edges.size());
    return g;
}

static dmf::FlowGraph build_graph() {
    dmf::FlowGraph g;
    const std::string tp_video_flow = gen_uuid();
    const std::string tp_audio_flow = gen_uuid();
    const std::string ndi_video_flow = gen_uuid();
    const std::string ndi_audio_flow = gen_uuid();
    g.nodes = {
        { "testpattern", "testpattern", {{"pattern", "bars"}} },
        // { "ndiin",       "ndiin",       {} },
        // { "fakesink",    "fakesink",    {} },
        { "ndiout",      "ndiout",      {} },
    };
    const nlohmann::json video_fmt = {
        {"kind","video"}, {"width",1920}, {"height",1080}, {"fps_num",25}, {"fps_den",1}
    };
    const nlohmann::json audio_fmt = {
        {"kind","audio"}, {"sample_rate",48000}, {"channels",2}, {"bit_depth",32}
    };
    g.edges = {
        // { tp_video_flow,  "testpattern", "video_flow_id",       "fakesink", "video_flow_id", video_fmt },
        // { tp_audio_flow,  "testpattern", "audio_flow_id", "",         "",        audio_fmt },
        // { ndi_video_flow, "ndiin",       "video_flow_id", "ndiout",   "video_flow_id", video_fmt },
        // { ndi_audio_flow, "ndiin",       "audio_flow_id", "",         "",        audio_fmt },
        { tp_video_flow,  "testpattern", "video_flow_id",       "ndiout", "video_flow_id", video_fmt },
        { tp_audio_flow,  "testpattern", "audio_flow_id", "ndiout", "audio_flow_id", audio_fmt },
    };
    return g;
}

// --- Process management ------------------------------------------------------

struct NodeProcess {
    std::string name;
    pid_t       pid{-1};
};

// Fork the node binary, passing domain and config via environment variables.
static NodeProcess launch_node(
    const std::string&    binary,
    const std::string&    domain,
    const nlohmann::json& config)
{
    NodeProcess proc;
    proc.name = config.value("node_id", binary);

    pid_t pid = fork();
    if (pid < 0) { perror("fork"); return proc; }

    if (pid == 0) {
        // Child: set env and exec
        setenv("MXL_DOMAIN",  domain.c_str(),       1);
        setenv("NODE_CONFIG", config.dump().c_str(), 1);
        execl(binary.c_str(), binary.c_str(), nullptr);
        // execl only returns on error
        perror(("execl " + binary).c_str());
        _exit(1);
    }

    proc.pid = pid;
    fprintf(stderr, "[studio-manager] launched %s  pid=%d\n", proc.name.c_str(), pid);
    return proc;
}

// Poll children with WNOHANG; log and clear pid if one has exited.
static void check_children(std::vector<NodeProcess>& nodes) {
    for (auto& node : nodes) {
        if (node.pid <= 0) continue;
        int wstatus = 0;
        if (waitpid(node.pid, &wstatus, WNOHANG) == node.pid) {
            fprintf(stderr, "[studio-manager] node %s (pid=%d) exited (status=%d)\n",
                node.name.c_str(), node.pid, WEXITSTATUS(wstatus));
            node.pid = -1;
        }
    }
}

// Send SIGTERM to all live children, then wait for them.
static void shutdown_children(std::vector<NodeProcess>& nodes) {
    fprintf(stderr, "[studio-manager] sending SIGTERM to all nodes\n");
    for (auto& node : nodes) {
        if (node.pid > 0) {
            kill(node.pid, SIGTERM);
        }
    }
    for (auto& node : nodes) {
        if (node.pid > 0) {
            waitpid(node.pid, nullptr, 0);
            fprintf(stderr, "[studio-manager] node %s stopped\n", node.name.c_str());
        }
    }
}

// -----------------------------------------------------------------------------

int main(int argc, char* argv[]) {
    dmf::install_signal_handlers();

    // Resolve node binary paths relative to this binary
    const fs::path bin_dir = fs::path(argv[0]).parent_path();

#ifdef __APPLE__
    const std::string domain = "/tmp/dmf-studio";
#else
    const std::string domain = "/dev/shm/dmf-studio";
#endif

    // Ensure the domain directory exists
    std::error_code ec;
    fs::create_directories(domain, ec);
    if (ec) {
        fprintf(stderr, "[studio-manager] cannot create domain dir %s: %s\n",
            domain.c_str(), ec.message().c_str());
        return 1;
    }
    fprintf(stderr, "[studio-manager] domain: %s\n", domain.c_str());

    // Clean up stale flow directories left by any previous crashed run
    {
        mxlInstance gc = mxlCreateInstance(domain.c_str(), "");
        if (gc) {
            mxlGarbageCollectFlows(gc);
            mxlDestroyInstance(gc);
            fprintf(stderr, "[studio-manager] garbage collected stale flows\n");
        }
    }

    // --- Build and launch the pipeline graph ---
    const dmf::FlowGraph graph = (argc > 1) ? load_graph(argv[1]) : build_graph();

    for (const auto& e : graph.edges)
        fprintf(stderr, "[studio-manager] flow %s → %s  id=%s\n",
            e.from_node.c_str(), e.to_node.c_str(), e.id.c_str());

    std::vector<NodeProcess> nodes;
    for (const auto& node : graph.nodes) {
        const std::string binary = (bin_dir / ("dmf-node-" + node.type)).string();
        nodes.push_back(launch_node(binary, domain, graph.node_config(node.id)));
    }

    // --- Run until Ctrl+C or SIGTERM ---
    fprintf(stderr, "[studio-manager] running — Ctrl+C to stop\n");
    while (dmf::g_running.load(std::memory_order_relaxed)) {
        check_children(nodes);
        mxlSleepForNs(500'000'000); // check every 500 ms
    }

    shutdown_children(nodes);
    fprintf(stderr, "[studio-manager] done\n");
    return 0;
}