diff --git a/CMakeLists.txt b/CMakeLists.txt index cfa16f2..7cadc7c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -115,5 +115,21 @@ endif() add_subdirectory(nodes/videoin) +# ── Asio standalone (needed by Crow; no Boost) ─────────────────────────────── +FetchContent_Declare(asio_fc + GIT_REPOSITORY https://github.com/chriskohlhoff/asio.git + GIT_TAG asio-1-30-2) +FetchContent_MakeAvailable(asio_fc) +set(ASIO_INCLUDE_DIR "${asio_fc_SOURCE_DIR}/asio/include" CACHE PATH "" FORCE) + +# ── Crow (HTTP + WebSocket) ─────────────────────────────────────────────────── +set(CROW_USE_BOOST OFF CACHE BOOL "" FORCE) +set(CROW_BUILD_EXAMPLES OFF CACHE BOOL "" FORCE) +set(CROW_BUILD_TESTS OFF CACHE BOOL "" FORCE) +FetchContent_Declare(crow_fc + GIT_REPOSITORY https://github.com/CrowCpp/Crow.git + GIT_TAG v1.2.0) +FetchContent_MakeAvailable(crow_fc) + # ── Core server ────────────────────────────────────────────────────────────── add_subdirectory(studio-manager) diff --git a/studio-manager/CMakeLists.txt b/studio-manager/CMakeLists.txt index cd1d4b6..8ad4599 100644 --- a/studio-manager/CMakeLists.txt +++ b/studio-manager/CMakeLists.txt @@ -1,4 +1,4 @@ add_executable(dmf-studio-manager main.cpp) target_compile_features(dmf-studio-manager PRIVATE cxx_std_20) -target_link_libraries(dmf-studio-manager PRIVATE dmf-shared) +target_link_libraries(dmf-studio-manager PRIVATE dmf-shared Crow::Crow) install(TARGETS dmf-studio-manager RUNTIME DESTINATION bin) diff --git a/studio-manager/StudioManager.hpp b/studio-manager/StudioManager.hpp new file mode 100644 index 0000000..8017f09 --- /dev/null +++ b/studio-manager/StudioManager.hpp @@ -0,0 +1,283 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "FlowGraph.hpp" +#include "Signal.hpp" + +namespace fs = std::filesystem; + +namespace dmf { + +enum class NodeState { Running, Stopped, Crashed }; + +struct NodeProcess { + std::string id; + std::string type; + pid_t pid{-1}; + NodeState state{NodeState::Stopped}; +}; + +class StudioManager { +public: + StudioManager(fs::path bin_dir, std::string domain) + : bin_dir_(std::move(bin_dir)), domain_(std::move(domain)) {} + + ~StudioManager() { shutdown(); } + + // Replace the running pipeline with a new graph. + // Stops all current nodes, then starts the new ones. + nlohmann::json load_graph(const nlohmann::json& graph_json) { + nlohmann::json s; + std::string err; + { + std::lock_guard lk(mutex_); + stop_all_locked(); + garbage_collect_locked(); + try { + graph_ = parse_graph(graph_json); + start_all_locked(); + } catch (const std::exception& e) { + err = e.what(); + } + s = status_locked(); + } + notify(s); + if (!err.empty()) + return {{"type", "error"}, {"message", err}}; + return s; + } + + nlohmann::json load_graph_file(const std::string& path) { + std::ifstream f(path); + if (!f) return {{"type", "error"}, {"message", "cannot open: " + path}}; + std::ostringstream ss; + ss << f.rdbuf(); + auto j = nlohmann::json::parse(ss.str(), nullptr, false); + if (j.is_discarded()) + return {{"type", "error"}, {"message", "invalid JSON in: " + path}}; + return load_graph(j); + } + + nlohmann::json stop_node(const std::string& id) { + nlohmann::json s; + { + std::lock_guard lk(mutex_); + for (auto& p : processes_) { + if (p.id == id && p.pid > 0) { + kill(p.pid, SIGTERM); + waitpid(p.pid, nullptr, 0); + p.pid = -1; + p.state = NodeState::Stopped; + fprintf(stderr, "[studio-manager] stopped node %s\n", id.c_str()); + break; + } + } + s = status_locked(); + } + notify(s); + return s; + } + + nlohmann::json start_node(const std::string& id) { + nlohmann::json s; + { + std::lock_guard lk(mutex_); + for (auto& p : processes_) { + if (p.id == id && p.pid <= 0) { + launch_locked(p); + break; + } + } + s = status_locked(); + } + notify(s); + return s; + } + + nlohmann::json get_status() { + std::lock_guard lk(mutex_); + return status_locked(); + } + + // Register a callback invoked (without mutex held) whenever node state changes. + // Safe to call send_text() from inside the callback. + void on_status_change(std::function cb) { + std::lock_guard lk(mutex_); + status_cb_ = std::move(cb); + } + + void shutdown() { + std::lock_guard lk(mutex_); + stop_all_locked(); + } + + // Blocks until g_running is false. Call from a dedicated thread. + void run_monitor() { + while (dmf::g_running.load(std::memory_order_relaxed)) { + mxlSleepForNs(500'000'000); + check_children(); + } + } + +private: + fs::path bin_dir_; + std::string domain_; + FlowGraph graph_; + std::vector processes_; + std::mutex mutex_; + std::function status_cb_; + + // ── locked helpers — call only while holding mutex_ ────────────────────── + + void stop_all_locked() { + for (auto& p : processes_) + if (p.pid > 0) kill(p.pid, SIGTERM); + for (auto& p : processes_) { + if (p.pid > 0) { + waitpid(p.pid, nullptr, 0); + fprintf(stderr, "[studio-manager] stopped node %s\n", p.id.c_str()); + p.pid = -1; + p.state = NodeState::Stopped; + } + } + processes_.clear(); + } + + void garbage_collect_locked() { + mxlInstance gc = mxlCreateInstance(domain_.c_str(), ""); + if (gc) { + mxlGarbageCollectFlows(gc); + mxlDestroyInstance(gc); + } + } + + void start_all_locked() { + processes_.clear(); + for (const auto& node : graph_.nodes) { + processes_.push_back({node.id, node.type, -1, NodeState::Stopped}); + launch_locked(processes_.back()); + } + } + + void launch_locked(NodeProcess& p) { + const std::string binary = (bin_dir_ / ("dmf-node-" + p.type)).string(); + const nlohmann::json cfg = graph_.node_config(p.id); + + pid_t pid = fork(); + if (pid < 0) { perror("fork"); return; } + if (pid == 0) { + setenv("MXL_DOMAIN", domain_.c_str(), 1); + setenv("NODE_CONFIG", cfg.dump().c_str(), 1); + execl(binary.c_str(), binary.c_str(), nullptr); + perror(("execl " + binary).c_str()); + _exit(1); + } + p.pid = pid; + p.state = NodeState::Running; + fprintf(stderr, "[studio-manager] launched %s pid=%d\n", p.id.c_str(), pid); + } + + nlohmann::json status_locked() const { + auto nodes = nlohmann::json::array(); + for (const auto& p : processes_) { + const char* state = + p.state == NodeState::Running ? "running" : + p.state == NodeState::Crashed ? "crashed" : "stopped"; + nodes.push_back({{"id", p.id}, {"type", p.type}, + {"pid", p.pid}, {"state", state}}); + } + return {{"type", "status"}, {"nodes", nodes}}; + } + + // ── unlocked helpers ───────────────────────────────────────────────────── + + void notify(const nlohmann::json& s) { + // Called WITHOUT mutex_ held so the callback can safely call back into us. + if (status_cb_) status_cb_(s); + } + + void check_children() { + bool changed = false; + nlohmann::json s; + { + std::lock_guard lk(mutex_); + for (auto& p : processes_) { + if (p.pid <= 0) continue; + int ws = 0; + if (waitpid(p.pid, &ws, WNOHANG) == p.pid) { + fprintf(stderr, "[studio-manager] node %s (pid=%d) exited (status=%d)\n", + p.id.c_str(), p.pid, WEXITSTATUS(ws)); + p.pid = -1; + p.state = NodeState::Crashed; + changed = true; + } + } + if (changed) s = status_locked(); + } + if (changed) notify(s); + } + + // ── static helpers ─────────────────────────────────────────────────────── + + static FlowGraph parse_graph(const nlohmann::json& j) { + FlowGraph g; + for (const auto& n : j.at("nodes")) + g.nodes.push_back({ + n.at("id").get(), + n.at("type").get(), + n.value("params", nlohmann::json::object()) + }); + + std::map, std::string> flow_ids; + for (const auto& e : j.at("edges")) { + auto from = e.at("from").get(); + auto port = e.at("from_port").get(); + auto key = std::make_pair(from, port); + if (!flow_ids.count(key)) flow_ids[key] = gen_uuid(); + g.edges.push_back({ + flow_ids.at(key), + from, port, + e.value("to", std::string{}), + e.value("to_port", std::string{}), + e.at("format") + }); + } + return g; + } + + 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)) != (ssize_t)sizeof(b)) { perror("read"); exit(1); } + close(fd); + b[6] = (b[6] & 0x0Fu) | 0x40u; + b[8] = (b[8] & 0x3Fu) | 0x80u; + 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; + } +}; + +} // namespace dmf diff --git a/studio-manager/main.cpp b/studio-manager/main.cpp index a7591b9..df7acca 100644 --- a/studio-manager/main.cpp +++ b/studio-manager/main.cpp @@ -1,190 +1,16 @@ -// 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 -#include #include -#include -#include -#include +#include #include -#include -#include -#include -#include -#include -#include -#include -#include "Signal.hpp" -#include "FlowGraph.hpp" +#include +#include +#include +#include "StudioManager.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(), - n.at("type").get(), - 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::string> flow_ids; - for (const auto& e : j.at("edges")) { - auto from_node = e.at("from").get(); - auto from_port = e.at("from_port").get(); - 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& 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& 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__ @@ -193,47 +19,87 @@ int main(int argc, char* argv[]) { 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()); + fprintf(stderr, "[studio-manager] cannot create domain dir: %s\n", 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"); + dmf::StudioManager manager(bin_dir, domain); + + if (argc > 1) { + auto r = manager.load_graph_file(argv[1]); + if (r.value("type", "") == "error") { + fprintf(stderr, "[studio-manager] %s\n", + r.value("message", "load failed").c_str()); + return 1; } } - // --- Build and launch the pipeline graph --- - const dmf::FlowGraph graph = (argc > 1) ? load_graph(argv[1]) : build_graph(); + // ── WebSocket server ────────────────────────────────────────────────────── - 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()); + crow::SimpleApp app; + app.signal_clear(); // let our signal handler manage SIGTERM/SIGINT - std::vector 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))); - } + std::mutex ws_mutex; + std::unordered_set clients; - // --- 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 - } + // Push status to every connected client (called without manager mutex held). + manager.on_status_change([&](nlohmann::json status) { + std::lock_guard lk(ws_mutex); + const std::string msg = status.dump(); + for (auto* c : clients) c->send_text(msg); + }); - shutdown_children(nodes); + CROW_WEBSOCKET_ROUTE(app, "/ws") + .onopen([&](crow::websocket::connection& conn) { + { + std::lock_guard lk(ws_mutex); + clients.insert(&conn); + } + // get_status() acquires manager mutex — must NOT hold ws_mutex here + conn.send_text(manager.get_status().dump()); + fprintf(stderr, "[studio-manager] client connected\n"); + }) + .onclose([&](crow::websocket::connection& conn, const std::string&) { + std::lock_guard lk(ws_mutex); + clients.erase(&conn); + fprintf(stderr, "[studio-manager] client disconnected\n"); + }) + .onmessage([&](crow::websocket::connection& conn, const std::string& data, bool) { + auto j = nlohmann::json::parse(data, nullptr, false); + if (j.is_discarded()) { + conn.send_text(R"({"type":"error","message":"invalid JSON"})"); + return; + } + + nlohmann::json response; + const std::string type = j.value("type", ""); + try { + if (type == "load_graph") response = manager.load_graph(j.at("graph")); + else if (type == "stop_node") response = manager.stop_node(j.at("id")); + else if (type == "start_node") response = manager.start_node(j.at("id")); + else if (type == "get_status") response = manager.get_status(); + else response = {{"type","error"},{"message","unknown command: " + type}}; + } catch (const std::exception& e) { + response = {{"type","error"},{"message", e.what()}}; + } + conn.send_text(response.dump()); + }); + + // ── Monitor thread — detects crashes, stops Crow on shutdown ───────────── + + std::thread monitor([&] { + manager.run_monitor(); + app.stop(); + }); + + fprintf(stderr, "[studio-manager] WebSocket API at ws://0.0.0.0:7070/ws\n"); + app.port(7070).multithreaded().run(); + + manager.shutdown(); + monitor.join(); fprintf(stderr, "[studio-manager] done\n"); return 0; }