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 <string>
#include <vector>
#include <fcntl.h>
#include <sys/wait.h>
#include <unistd.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 build_graph() {
    dmf::FlowGraph g;
    g.nodes = {
        { "testpattern", "testpattern", {} },
        { "fakesink",    "fakesink",    {} },
    };
    g.edges = {
        { gen_uuid(), "testpattern", "flow_id", "fakesink", "flow_id",
          { {"kind","video"}, {"width",1920}, {"height",1080}, {"fps_num",25}, {"fps_den",1} } },
    };
    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());

    // --- Build and launch the pipeline graph ---
    const dmf::FlowGraph graph = 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;
}