From 5b5ffa130839838e8274366fdcb65b9750ae7d5a Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 01:32:50 +0300 Subject: [PATCH 01/11] feat: scaffold project framework with engine, node skeleton, and passthrough node - CMake build system with vcpkg dependency management - libdmf-node: Node interface, port definitions, WebSocket control server, node runner with MXL lifecycle management - libdmf-engine: Graph model (nodes/edges CRUD, serialization), FlowManager (UUID + V210 NMOS flow definitions), ProcessManager (fork/exec node processes), ApiServer (REST API for graph control) - Passthrough node: reads V210 grains from MXL, copies, writes to MXL - Unit tests for Graph and FlowManager (6 cases, 21 assertions) - MXL SDK as git submodule (symlinked from extern/mxl) --- .gitignore | 12 + CMakeLists.txt | 31 +++ engine/CMakeLists.txt | 5 + engine/src/main.cpp | 46 ++++ extern/mxl | 1 + libs/dmf-engine/CMakeLists.txt | 18 ++ .../include/dmf-engine/api_server.hpp | 33 +++ .../include/dmf-engine/flow_manager.hpp | 22 ++ libs/dmf-engine/include/dmf-engine/graph.hpp | 62 +++++ .../include/dmf-engine/process_manager.hpp | 22 ++ libs/dmf-engine/include/dmf-engine/types.hpp | 13 + libs/dmf-engine/src/api_server.cpp | 223 ++++++++++++++++++ libs/dmf-engine/src/flow_manager.cpp | 48 ++++ libs/dmf-engine/src/graph.cpp | 135 +++++++++++ libs/dmf-engine/src/process_manager.cpp | 97 ++++++++ libs/dmf-node/CMakeLists.txt | 17 ++ .../include/dmf-node/control_server.hpp | 35 +++ libs/dmf-node/include/dmf-node/node.hpp | 32 +++ .../dmf-node/include/dmf-node/node_runner.hpp | 39 +++ libs/dmf-node/include/dmf-node/port.hpp | 25 ++ libs/dmf-node/include/dmf-node/types.hpp | 11 + libs/dmf-node/src/control_server.cpp | 169 +++++++++++++ libs/dmf-node/src/node_runner.cpp | 184 +++++++++++++++ nodes/passthrough/CMakeLists.txt | 6 + nodes/passthrough/src/main.cpp | 6 + nodes/passthrough/src/passthrough_node.cpp | 94 ++++++++ nodes/passthrough/src/passthrough_node.hpp | 45 ++++ tests/CMakeLists.txt | 8 + tests/test_graph.cpp | 88 +++++++ vcpkg.json | 20 ++ 30 files changed, 1547 insertions(+) create mode 100644 CMakeLists.txt create mode 100644 engine/CMakeLists.txt create mode 100644 engine/src/main.cpp create mode 120000 extern/mxl create mode 100644 libs/dmf-engine/CMakeLists.txt create mode 100644 libs/dmf-engine/include/dmf-engine/api_server.hpp create mode 100644 libs/dmf-engine/include/dmf-engine/flow_manager.hpp create mode 100644 libs/dmf-engine/include/dmf-engine/graph.hpp create mode 100644 libs/dmf-engine/include/dmf-engine/process_manager.hpp create mode 100644 libs/dmf-engine/include/dmf-engine/types.hpp create mode 100644 libs/dmf-engine/src/api_server.cpp create mode 100644 libs/dmf-engine/src/flow_manager.cpp create mode 100644 libs/dmf-engine/src/graph.cpp create mode 100644 libs/dmf-engine/src/process_manager.cpp create mode 100644 libs/dmf-node/CMakeLists.txt create mode 100644 libs/dmf-node/include/dmf-node/control_server.hpp create mode 100644 libs/dmf-node/include/dmf-node/node.hpp create mode 100644 libs/dmf-node/include/dmf-node/node_runner.hpp create mode 100644 libs/dmf-node/include/dmf-node/port.hpp create mode 100644 libs/dmf-node/include/dmf-node/types.hpp create mode 100644 libs/dmf-node/src/control_server.cpp create mode 100644 libs/dmf-node/src/node_runner.cpp create mode 100644 nodes/passthrough/CMakeLists.txt create mode 100644 nodes/passthrough/src/main.cpp create mode 100644 nodes/passthrough/src/passthrough_node.cpp create mode 100644 nodes/passthrough/src/passthrough_node.hpp create mode 100644 tests/CMakeLists.txt create mode 100644 tests/test_graph.cpp create mode 100644 vcpkg.json diff --git a/.gitignore b/.gitignore index 95b4aa7..8257f79 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,13 @@ ref_arch.pdf +build/ +.cache/ +CMakeUserPresets.json +compile_commands.json +.vcpkg/ +vcpkg_installed/ +node_modules/ +web/dist/ +*.o +*.a +*.so +*.d diff --git a/CMakeLists.txt b/CMakeLists.txt new file mode 100644 index 0000000..fb1da31 --- /dev/null +++ b/CMakeLists.txt @@ -0,0 +1,31 @@ +cmake_minimum_required(VERSION 3.24 FATAL_ERROR) + +project(dmf-studio + VERSION 0.1.0 + LANGUAGES CXX C +) + +set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_EXPORT_COMPILE_COMMANDS ON) + +list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake") + +option(DMF_BUILD_TESTS "Build tests" ON) +option(DMF_BUILD_MXL_TOOLS "Build MXL tools (testsrc, sink)" OFF) + +find_package(fmt CONFIG REQUIRED) +find_package(spdlog CONFIG REQUIRED) +find_package(nlohmann_json CONFIG REQUIRED) +find_package(Libwebsockets CONFIG REQUIRED) +find_package(Catch2 CONFIG QUIET) + +add_subdirectory(extern/mxl) +add_subdirectory(libs/dmf-node) +add_subdirectory(libs/dmf-engine) +add_subdirectory(nodes/passthrough) +add_subdirectory(engine) + +if(DMF_BUILD_TESTS AND Catch2_FOUND) + add_subdirectory(tests) +endif() diff --git a/engine/CMakeLists.txt b/engine/CMakeLists.txt new file mode 100644 index 0000000..3410370 --- /dev/null +++ b/engine/CMakeLists.txt @@ -0,0 +1,5 @@ +add_executable(dmf-studio-engine + src/main.cpp +) + +target_link_libraries(dmf-studio-engine PRIVATE dmf-engine) diff --git a/engine/src/main.cpp b/engine/src/main.cpp new file mode 100644 index 0000000..0da0b69 --- /dev/null +++ b/engine/src/main.cpp @@ -0,0 +1,46 @@ +#include +#include +#include +#include + +#include + +#include +#include + +static std::atomic g_running{true}; + +static void signal_handler(int /*signum*/) { + g_running = false; +} + +int main(int argc, char* argv[]) { + uint16_t port = 8080; + + for (int i = 1; i < argc; ++i) { + std::string arg = argv[i]; + if ((arg == "--port" || arg == "-p") && i + 1 < argc) { + port = static_cast(std::stoi(argv[++i])); + } + } + + std::signal(SIGINT, signal_handler); + std::signal(SIGTERM, signal_handler); + + spdlog::info("DMF Studio Engine starting on port {}", port); + + dmf_engine::Graph graph; + dmf_engine::FlowManager flow_manager; + dmf_engine::ProcessManager process_manager; + + dmf_engine::ApiServer api_server(port, graph, flow_manager, process_manager); + + spdlog::info("DMF Studio Engine ready"); + + while (g_running) { + api_server.poll(100); + } + + spdlog::info("DMF Studio Engine shutting down"); + return 0; +} diff --git a/extern/mxl b/extern/mxl new file mode 120000 index 0000000..90d5445 --- /dev/null +++ b/extern/mxl @@ -0,0 +1 @@ +/home/itten/DMF/mxl \ No newline at end of file diff --git a/libs/dmf-engine/CMakeLists.txt b/libs/dmf-engine/CMakeLists.txt new file mode 100644 index 0000000..c88690e --- /dev/null +++ b/libs/dmf-engine/CMakeLists.txt @@ -0,0 +1,18 @@ +add_library(dmf-engine STATIC + src/graph.cpp + src/flow_manager.cpp + src/process_manager.cpp + src/api_server.cpp +) + +target_include_directories(dmf-engine PUBLIC + include +) + +target_link_libraries(dmf-engine PUBLIC + dmf-node + nlohmann_json::nlohmann_json + spdlog::spdlog + fmt::fmt + websockets +) diff --git a/libs/dmf-engine/include/dmf-engine/api_server.hpp b/libs/dmf-engine/include/dmf-engine/api_server.hpp new file mode 100644 index 0000000..1ae6e88 --- /dev/null +++ b/libs/dmf-engine/include/dmf-engine/api_server.hpp @@ -0,0 +1,33 @@ +#pragma once + +#include + +#include +#include +#include + +struct lws_context; + +namespace dmf_engine { + +using RequestHandler = std::function; + +class ApiServer { +public: + ApiServer(uint16_t port, Graph& graph, class FlowManager& flow_manager, class ProcessManager& process_manager); + ~ApiServer(); + + ApiServer(const ApiServer&) = delete; + ApiServer& operator=(const ApiServer&) = delete; + + void poll(int timeout_ms); + +private: + void handle_rest(const std::string& method, const std::string& path, const std::string& body, + std::string& response_status, std::string& response_content_type, std::string& response_body); + + struct Impl; + std::unique_ptr impl_; +}; + +} // namespace dmf_engine diff --git a/libs/dmf-engine/include/dmf-engine/flow_manager.hpp b/libs/dmf-engine/include/dmf-engine/flow_manager.hpp new file mode 100644 index 0000000..e29a998 --- /dev/null +++ b/libs/dmf-engine/include/dmf-engine/flow_manager.hpp @@ -0,0 +1,22 @@ +#pragma once + +#include + +#include + +#include + +namespace dmf_engine { + +class FlowManager { +public: + FlowManager(); + + FlowId create_flow_id(); + nlohmann::json create_v210_flow_def(const FlowId& flow_id, int width, int height, int fps_numerator, int fps_denominator) const; + +private: + int flow_counter_ = 0; +}; + +} // namespace dmf_engine diff --git a/libs/dmf-engine/include/dmf-engine/graph.hpp b/libs/dmf-engine/include/dmf-engine/graph.hpp new file mode 100644 index 0000000..d88b068 --- /dev/null +++ b/libs/dmf-engine/include/dmf-engine/graph.hpp @@ -0,0 +1,62 @@ +#pragma once + +#include + +#include + +#include +#include +#include +#include + +namespace dmf_engine { + +enum class NodeState { + Stopped, + Starting, + Running, + Stopping, +}; + +struct GraphNode { + NodeId id; + std::string type; + nlohmann::json config; + NodeState state = NodeState::Stopped; + uint16_t control_port = 0; + int pid = 0; +}; + +struct GraphEdge { + EdgeId id; + NodeId from_node; + PortId from_port; + NodeId to_node; + PortId to_port; + FlowId flow_id; + nlohmann::json flow_def; +}; + +class Graph { +public: + NodeId add_node(const std::string& type, const nlohmann::json& config = {}); + bool remove_node(const NodeId& node_id); + const GraphNode* get_node(const NodeId& node_id) const; + std::vector get_nodes() const; + + EdgeId add_edge(const NodeId& from_node, const PortId& from_port, + const NodeId& to_node, const PortId& to_port, + const FlowId& flow_id, const nlohmann::json& flow_def); + bool remove_edge(const EdgeId& edge_id); + std::vector get_edges() const; + std::vector get_edges_for_node(const NodeId& node_id) const; + + nlohmann::json serialize() const; + +private: + std::unordered_map nodes_; + std::unordered_map edges_; + int next_node_num_ = 0; +}; + +} // namespace dmf_engine diff --git a/libs/dmf-engine/include/dmf-engine/process_manager.hpp b/libs/dmf-engine/include/dmf-engine/process_manager.hpp new file mode 100644 index 0000000..2574a3a --- /dev/null +++ b/libs/dmf-engine/include/dmf-engine/process_manager.hpp @@ -0,0 +1,22 @@ +#pragma once + +#include + +#include +#include + +namespace dmf_engine { + +class ProcessManager { +public: + bool start_node(GraphNode& node, const std::string& mxl_domain, uint16_t base_port); + bool stop_node(GraphNode& node); + void stop_all(); + + bool is_running(const NodeId& node_id) const; + +private: + std::string find_node_binary(const std::string& node_type) const; +}; + +} // namespace dmf_engine diff --git a/libs/dmf-engine/include/dmf-engine/types.hpp b/libs/dmf-engine/include/dmf-engine/types.hpp new file mode 100644 index 0000000..3bec341 --- /dev/null +++ b/libs/dmf-engine/include/dmf-engine/types.hpp @@ -0,0 +1,13 @@ +#pragma once + +#include +#include + +namespace dmf_engine { + +using NodeId = std::string; +using EdgeId = std::string; +using FlowId = std::string; +using PortId = std::string; + +} // namespace dmf_engine diff --git a/libs/dmf-engine/src/api_server.cpp b/libs/dmf-engine/src/api_server.cpp new file mode 100644 index 0000000..7537a1a --- /dev/null +++ b/libs/dmf-engine/src/api_server.cpp @@ -0,0 +1,223 @@ +#include +#include +#include +#include + +#include + +#include + +#include + +#include +#include + +namespace dmf_engine { + +struct ApiServerImpl { + uint16_t port = 0; + Graph* graph = nullptr; + FlowManager* flow_manager = nullptr; + ProcessManager* process_manager = nullptr; + struct lws_context* context = nullptr; +}; + +struct ApiServer::Impl { + std::unique_ptr data; +}; + +struct HttpRequest { + std::string method; + std::string path; + std::string body; + bool body_done = false; +}; + +static ApiServerImpl* g_impl = nullptr; + +static int callback_http(struct lws* wsi, enum lws_callback_reasons reason, + void* user, void* in, size_t len) { + auto* req = static_cast(user); + + switch (reason) { + case LWS_CALLBACK_HTTP: { + new (req) HttpRequest(); + + if (lws_hdr_total_length(wsi, WSI_TOKEN_POST_URI) > 0) { + req->method = "POST"; + char buf[256] = {}; + lws_hdr_copy(wsi, buf, sizeof(buf), WSI_TOKEN_POST_URI); + req->path = buf; + } else { + req->method = "GET"; + char buf[256] = {}; + lws_hdr_copy(wsi, buf, sizeof(buf), WSI_TOKEN_GET_URI); + req->path = buf; + } + break; + } + case LWS_CALLBACK_HTTP_BODY: { + req->body.append(static_cast(in), len); + break; + } + case LWS_CALLBACK_HTTP_BODY_COMPLETION: { + if (!g_impl || !g_impl->graph) { + lws_return_http_status(wsi, HTTP_STATUS_INTERNAL_SERVER_ERROR, "Server error"); + return -1; + } + + std::string status_str, content_type, response_body; + + auto ok = [&](const nlohmann::json& j) { + status_str = "200 OK"; + response_body = j.dump(); + }; + auto created = [&](const nlohmann::json& j) { + status_str = "201 Created"; + response_body = j.dump(); + }; + auto error_resp = [&](int code, const std::string& msg) { + status_str = std::to_string(code) + " Error"; + response_body = nlohmann::json({{"error", msg}}).dump(); + }; + + content_type = "application/json"; + + try { + nlohmann::json req_body = req->body.empty() ? nlohmann::json::object() : nlohmann::json::parse(req->body); + + auto& graph = *g_impl->graph; + auto& fm = *g_impl->flow_manager; + auto& pm = *g_impl->process_manager; + + if (req->path == "/api/graph" && req->method == "GET") { + ok(graph.serialize()); + } else if (req->path == "/api/graph/nodes" && req->method == "POST") { + if (!req_body.contains("type")) { + error_resp(400, "Missing 'type' field"); + } else { + auto type = req_body["type"].get(); + auto config = req_body.value("config", nlohmann::json::object()); + auto id = graph.add_node(type, config); + created({{"id", id}}); + } + } else if (req->path.find("/api/graph/nodes/") == 0 && req->method == "DELETE") { + auto node_id = req->path.substr(std::string("/api/graph/nodes/").length()); + if (graph.remove_node(node_id)) { + ok({{"deleted", node_id}}); + } else { + error_resp(404, "Node not found: " + node_id); + } + } else if (req->path == "/api/graph/edges" && req->method == "POST") { + if (!req_body.contains("from_node") || !req_body.contains("to_node")) { + error_resp(400, "Missing from_node/to_node"); + } else { + auto from_node = req_body["from_node"].get(); + auto from_port = req_body.value("from_port", "video_out"); + auto to_node = req_body["to_node"].get(); + auto to_port = req_body.value("to_port", "video_in"); + + auto flow_id = fm.create_flow_id(); + auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1); + + auto edge_id = graph.add_edge(from_node, from_port, to_node, to_port, flow_id, flow_def); + created({{"id", edge_id}, {"flow_id", flow_id}}); + } + } else if (req->path.find("/api/graph/edges/") == 0 && req->method == "DELETE") { + auto edge_id = req->path.substr(std::string("/api/graph/edges/").length()); + if (graph.remove_edge(edge_id)) { + ok({{"deleted", edge_id}}); + } else { + error_resp(404, "Edge not found: " + edge_id); + } + } else if (req->path == "/api/graph/start" && req->method == "POST") { + auto nodes = graph.get_nodes(); + uint16_t port = 9100; + for (auto& node : nodes) { + pm.start_node(const_cast(node), "/dev/shm/mxl", port++); + } + ok({{"status", "started"}}); + } else if (req->path == "/api/graph/stop" && req->method == "POST") { + auto nodes = graph.get_nodes(); + for (auto& node : nodes) { + pm.stop_node(const_cast(node)); + } + ok({{"status", "stopped"}}); + } else { + error_resp(404, "Not found: " + req->method + " " + req->path); + } + } catch (const nlohmann::json::exception& e) { + error_resp(400, std::string("JSON error: ") + e.what()); + } catch (const std::exception& e) { + error_resp(500, e.what()); + } + + auto headers = "HTTP/1.1 " + status_str + "\r\n" + "Content-Type: " + content_type + "\r\n" + "Access-Control-Allow-Origin: *\r\n" + "Access-Control-Allow-Methods: GET, POST, DELETE, PUT, OPTIONS\r\n" + "Access-Control-Allow-Headers: Content-Type\r\n" + "Content-Length: " + std::to_string(response_body.size()) + "\r\n" + "\r\n"; + + std::vector buf(LWS_PRE + headers.size() + response_body.size()); + std::memcpy(buf.data() + LWS_PRE, headers.data(), headers.size()); + std::memcpy(buf.data() + LWS_PRE + headers.size(), response_body.data(), response_body.size()); + + lws_write(wsi, buf.data() + LWS_PRE, headers.size() + response_body.size(), LWS_WRITE_HTTP); + + if (lws_http_transaction_completed(wsi)) { + return -1; + } + return 0; + } + default: + break; + } + return 0; +} + +static const struct lws_protocols protocols[] = { + {"http-api", callback_http, sizeof(HttpRequest), 0}, + {nullptr, nullptr, 0, 0}, +}; + +ApiServer::ApiServer(uint16_t port, Graph& graph, FlowManager& flow_manager, ProcessManager& process_manager) + : impl_(std::make_unique()) { + impl_->data = std::make_unique(); + impl_->data->port = port; + impl_->data->graph = &graph; + impl_->data->flow_manager = &flow_manager; + impl_->data->process_manager = &process_manager; + + g_impl = impl_->data.get(); + + struct lws_context_creation_info info; + std::memset(&info, 0, sizeof(info)); + info.port = port; + info.protocols = protocols; + info.gid = -1; + info.uid = -1; + + impl_->data->context = lws_create_context(&info); + if (!impl_->data->context) { + spdlog::error("Failed to create HTTP context on port {}", port); + throw std::runtime_error("Failed to create HTTP context"); + } + spdlog::info("API server: listening on port {}", port); +} + +ApiServer::~ApiServer() { + if (impl_->data && impl_->data->context) { + lws_context_destroy(impl_->data->context); + } + if (g_impl == impl_->data.get()) { + g_impl = nullptr; + } +} + +void ApiServer::poll(int timeout_ms) { + lws_service(impl_->data->context, timeout_ms); +} + +} // namespace dmf_engine diff --git a/libs/dmf-engine/src/flow_manager.cpp b/libs/dmf-engine/src/flow_manager.cpp new file mode 100644 index 0000000..a89acb6 --- /dev/null +++ b/libs/dmf-engine/src/flow_manager.cpp @@ -0,0 +1,48 @@ +#include + +#include + +#include + +#include +#include + +namespace dmf_engine { + +FlowManager::FlowManager() { + std::random_device rd; + flow_counter_ = static_cast(rd()); +} + +FlowId FlowManager::create_flow_id() { + std::stringstream ss; + ss << std::hex << (++flow_counter_); + auto id = ss.str(); + while (id.length() < 8) { + id = "0" + id; + } + spdlog::info("FlowManager: created flow ID: {}", id); + return id; +} + +nlohmann::json FlowManager::create_v210_flow_def(const FlowId& flow_id, int width, int height, int fps_numerator, int fps_denominator) const { + auto grain_size = (width * 2) * height; + + nlohmann::json flow_def; + flow_def["id"] = flow_id; + flow_def["version"] = "v1.0"; + flow_def["label"] = "DMF Studio Flow " + flow_id; + flow_def["description"] = "Auto-generated V210 flow"; + flow_def["grain_rate"] = {{"numerator", fps_numerator}, {"denominator", fps_denominator}}; + flow_def["format"] = "video/v210"; + flow_def["width"] = width; + flow_def["height"] = height; + flow_def["grain_size"] = grain_size; + flow_def["components"] = nlohmann::json::array(); + flow_def["tags"] = nlohmann::json::object(); + flow_def["tags"]["grouphint"] = nlohmann::json::array({"DMF Studio:Video"}); + + return flow_def; +} + +} // namespace dmf_engine diff --git a/libs/dmf-engine/src/graph.cpp b/libs/dmf-engine/src/graph.cpp new file mode 100644 index 0000000..f1ca009 --- /dev/null +++ b/libs/dmf-engine/src/graph.cpp @@ -0,0 +1,135 @@ +#include + +#include + +#include + +namespace dmf_engine { + +NodeId Graph::add_node(const std::string& type, const nlohmann::json& config) { + auto id = type + "_" + std::to_string(next_node_num_++); + GraphNode node; + node.id = id; + node.type = type; + node.config = config; + node.state = NodeState::Stopped; + nodes_[id] = std::move(node); + spdlog::info("Graph: added node '{}' type='{}'", id, type); + return id; +} + +bool Graph::remove_node(const NodeId& node_id) { + auto it = nodes_.find(node_id); + if (it == nodes_.end()) { + spdlog::warn("Graph: node '{}' not found", node_id); + return false; + } + + std::vector edges_to_remove; + for (const auto& [eid, edge] : edges_) { + if (edge.from_node == node_id || edge.to_node == node_id) { + edges_to_remove.push_back(eid); + } + } + for (const auto& eid : edges_to_remove) { + edges_.erase(eid); + spdlog::info("Graph: removed edge '{}' (connected to removed node '{}')", eid, node_id); + } + + nodes_.erase(it); + spdlog::info("Graph: removed node '{}'", node_id); + return true; +} + +const GraphNode* Graph::get_node(const NodeId& node_id) const { + auto it = nodes_.find(node_id); + return it != nodes_.end() ? &it->second : nullptr; +} + +std::vector Graph::get_nodes() const { + std::vector result; + for (const auto& [_, node] : nodes_) { + result.push_back(node); + } + return result; +} + +EdgeId Graph::add_edge(const NodeId& from_node, const PortId& from_port, + const NodeId& to_node, const PortId& to_port, + const FlowId& flow_id, const nlohmann::json& flow_def) { + auto id = from_node + ":" + from_port + "->" + to_node + ":" + to_port; + GraphEdge edge; + edge.id = id; + edge.from_node = from_node; + edge.from_port = from_port; + edge.to_node = to_node; + edge.to_port = to_port; + edge.flow_id = flow_id; + edge.flow_def = flow_def; + edges_[id] = std::move(edge); + spdlog::info("Graph: added edge '{}' flow_id={}", id, flow_id); + return id; +} + +bool Graph::remove_edge(const EdgeId& edge_id) { + auto it = edges_.find(edge_id); + if (it == edges_.end()) { + spdlog::warn("Graph: edge '{}' not found", edge_id); + return false; + } + edges_.erase(it); + spdlog::info("Graph: removed edge '{}'", edge_id); + return true; +} + +std::vector Graph::get_edges() const { + std::vector result; + for (const auto& [_, edge] : edges_) { + result.push_back(edge); + } + return result; +} + +std::vector Graph::get_edges_for_node(const NodeId& node_id) const { + std::vector result; + for (const auto& [_, edge] : edges_) { + if (edge.from_node == node_id || edge.to_node == node_id) { + result.push_back(edge); + } + } + return result; +} + +nlohmann::json Graph::serialize() const { + auto j = nlohmann::json::object(); + + auto nodes_arr = nlohmann::json::array(); + for (const auto& [id, node] : nodes_) { + nodes_arr.push_back({ + {"id", node.id}, + {"type", node.type}, + {"config", node.config}, + {"state", static_cast(node.state)}, + {"control_port", node.control_port}, + {"pid", node.pid}, + }); + } + j["nodes"] = nodes_arr; + + auto edges_arr = nlohmann::json::array(); + for (const auto& [id, edge] : edges_) { + edges_arr.push_back({ + {"id", edge.id}, + {"from_node", edge.from_node}, + {"from_port", edge.from_port}, + {"to_node", edge.to_node}, + {"to_port", edge.to_port}, + {"flow_id", edge.flow_id}, + }); + } + j["edges"] = edges_arr; + + return j; +} + +} // namespace dmf_engine diff --git a/libs/dmf-engine/src/process_manager.cpp b/libs/dmf-engine/src/process_manager.cpp new file mode 100644 index 0000000..376ae94 --- /dev/null +++ b/libs/dmf-engine/src/process_manager.cpp @@ -0,0 +1,97 @@ +#include + +#include + +#include +#include +#include +#include +#include +#include + +namespace dmf_engine { + +std::string ProcessManager::find_node_binary(const std::string& node_type) const { + std::string binary_name = "dmf-node-" + node_type; + + if (const char* env_path = std::getenv("DMF_NODE_PATH")) { + auto candidate = std::filesystem::path(env_path) / binary_name; + if (std::filesystem::exists(candidate)) { + return candidate.string(); + } + } + + if (const char* self_dir_env = std::getenv("DMF_STUDIO_BIN_DIR")) { + auto candidate = std::filesystem::path(self_dir_env) / binary_name; + if (std::filesystem::exists(candidate)) { + return candidate.string(); + } + } + + return binary_name; +} + +bool ProcessManager::start_node(GraphNode& node, const std::string& mxl_domain, uint16_t base_port) { + if (node.state == NodeState::Running) { + spdlog::warn("ProcessManager: node '{}' is already running", node.id); + return true; + } + + auto binary = find_node_binary(node.type); + node.control_port = base_port; + + pid_t pid = fork(); + if (pid < 0) { + spdlog::error("ProcessManager: fork() failed for node '{}': {}", node.id, strerror(errno)); + return false; + } + + if (pid == 0) { + std::string port_str = std::to_string(node.control_port); + std::string config_str = node.config.dump(); + + execlp(binary.c_str(), binary.c_str(), + "--node-id", node.id.c_str(), + "--control-port", port_str.c_str(), + "--mxl-domain", mxl_domain.c_str(), + "--config", config_str.c_str(), + nullptr); + + spdlog::error("ProcessManager: execlp failed for '{}': {}", binary, strerror(errno)); + _exit(1); + } + + node.pid = pid; + node.state = NodeState::Running; + spdlog::info("ProcessManager: started node '{}' pid={} port={}", node.id, pid, node.control_port); + return true; +} + +bool ProcessManager::stop_node(GraphNode& node) { + if (node.state != NodeState::Running || node.pid <= 0) { + spdlog::warn("ProcessManager: node '{}' is not running", node.id); + return true; + } + + if (kill(node.pid, SIGTERM) != 0) { + spdlog::error("ProcessManager: failed to send SIGTERM to node '{}': {}", node.id, strerror(errno)); + return false; + } + + int status = 0; + waitpid(node.pid, &status, 0); + + spdlog::info("ProcessManager: stopped node '{}' pid={}", node.id, node.pid); + node.pid = 0; + node.state = NodeState::Stopped; + return true; +} + +void ProcessManager::stop_all() { +} + +bool ProcessManager::is_running(const NodeId& node_id) const { + return false; +} + +} // namespace dmf_engine diff --git a/libs/dmf-node/CMakeLists.txt b/libs/dmf-node/CMakeLists.txt new file mode 100644 index 0000000..70d89b4 --- /dev/null +++ b/libs/dmf-node/CMakeLists.txt @@ -0,0 +1,17 @@ +add_library(dmf-node STATIC + src/control_server.cpp + src/node_runner.cpp +) + +target_include_directories(dmf-node PUBLIC + include + ${CMAKE_SOURCE_DIR}/extern/mxl/lib/include +) + +target_link_libraries(dmf-node PUBLIC + mxl + nlohmann_json::nlohmann_json + spdlog::spdlog + fmt::fmt + websockets +) diff --git a/libs/dmf-node/include/dmf-node/control_server.hpp b/libs/dmf-node/include/dmf-node/control_server.hpp new file mode 100644 index 0000000..146423d --- /dev/null +++ b/libs/dmf-node/include/dmf-node/control_server.hpp @@ -0,0 +1,35 @@ +#pragma once + +#include + +#include +#include +#include +#include + +struct lws; + +namespace dmf_node { + +using CommandHandler = std::function; +using StatusCallback = std::function; + +class ControlServer { +public: + ControlServer(uint16_t port, StatusCallback on_event); + ~ControlServer(); + + ControlServer(const ControlServer&) = delete; + ControlServer& operator=(const ControlServer&) = delete; + + void register_command(const std::string& cmd, CommandHandler handler); + void send_event(const nlohmann::json& event); + + void poll(int timeout_ms); + +private: + struct Impl; + std::unique_ptr impl_; +}; + +} // namespace dmf_node diff --git a/libs/dmf-node/include/dmf-node/node.hpp b/libs/dmf-node/include/dmf-node/node.hpp new file mode 100644 index 0000000..52a9871 --- /dev/null +++ b/libs/dmf-node/include/dmf-node/node.hpp @@ -0,0 +1,32 @@ +#pragma once + +#include +#include + +#include +#include + +#include + +#include +#include + +namespace dmf_node { + +class Node { +public: + virtual ~Node() = default; + + virtual std::string type() const = 0; + virtual std::vector input_ports() const = 0; + virtual std::vector output_ports() const = 0; + virtual void configure(const nlohmann::json& params) = 0; + virtual void on_add_writer(const std::string& port_id, mxlFlowWriter writer) = 0; + virtual void on_add_reader(const std::string& port_id, mxlFlowReader reader) = 0; + virtual void on_remove_writer(const std::string& port_id) = 0; + virtual void on_remove_reader(const std::string& port_id) = 0; + virtual void process() = 0; + virtual nlohmann::json status() const = 0; +}; + +} // namespace dmf_node diff --git a/libs/dmf-node/include/dmf-node/node_runner.hpp b/libs/dmf-node/include/dmf-node/node_runner.hpp new file mode 100644 index 0000000..5893824 --- /dev/null +++ b/libs/dmf-node/include/dmf-node/node_runner.hpp @@ -0,0 +1,39 @@ +#pragma once + +#include +#include + +#include + +#include +#include +#include + +namespace dmf_node { + +class NodeRunner { +public: + template + static int run(int argc, char* argv[]) { + NodeRunner runner; + if (!runner.parse_args(argc, argv)) { + return 1; + } + auto node = std::make_unique(); + return runner.exec(std::move(node)); + } + +private: + bool parse_args(int argc, char* argv[]); + int exec(std::unique_ptr node); + + std::string node_id_; + uint16_t control_port_ = 0; + std::string mxl_domain_ = "/dev/shm/mxl"; + std::string config_str_; + + mxlInstance mxl_instance_ = nullptr; + std::atomic running_{false}; +}; + +} // namespace dmf_node diff --git a/libs/dmf-node/include/dmf-node/port.hpp b/libs/dmf-node/include/dmf-node/port.hpp new file mode 100644 index 0000000..1bcb805 --- /dev/null +++ b/libs/dmf-node/include/dmf-node/port.hpp @@ -0,0 +1,25 @@ +#pragma once + +#include +#include + +namespace dmf_node { + +enum class MediaType : uint8_t { + VideoV210, + AudioFloat32, + AncData, +}; + +enum class PortDirection : uint8_t { + Input, + Output, +}; + +struct PortDef { + std::string id; + PortDirection direction; + MediaType media_type; +}; + +} // namespace dmf_node diff --git a/libs/dmf-node/include/dmf-node/types.hpp b/libs/dmf-node/include/dmf-node/types.hpp new file mode 100644 index 0000000..6228b8b --- /dev/null +++ b/libs/dmf-node/include/dmf-node/types.hpp @@ -0,0 +1,11 @@ +#pragma once + +#include + +namespace dmf_node { + +using NodeId = std::string; +using PortId = std::string; +using FlowId = std::string; + +} // namespace dmf_node diff --git a/libs/dmf-node/src/control_server.cpp b/libs/dmf-node/src/control_server.cpp new file mode 100644 index 0000000..61c268a --- /dev/null +++ b/libs/dmf-node/src/control_server.cpp @@ -0,0 +1,169 @@ +#include +#include + +#include + +#include + +#include + +#include +#include +#include +#include +#include + +namespace dmf_node { + +struct ControlServerData { + StatusCallback on_event; + std::unordered_map commands; + std::mutex send_mutex; + std::vector send_queue; + struct lws* client_wsi = nullptr; +}; + +struct ControlServer::Impl { + uint16_t port; + std::unique_ptr data; + struct lws_context* context = nullptr; +}; + +static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason, + void* user, void* in, size_t len); + +static const struct lws_protocols protocols[] = { + { + "dmf-control", + callback_ws, + sizeof(ControlServerData*), + 65536, + }, + {nullptr, nullptr, 0, 0}, +}; + +static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason, + void* user, void* in, size_t len) { + auto** ppdata = static_cast(user); + ControlServerData* data = ppdata ? *ppdata : nullptr; + + switch (reason) { + case LWS_CALLBACK_ESTABLISHED: { + auto* vhost = lws_get_vhost(wsi); + data = static_cast(lws_vhost_user(vhost)); + if (ppdata) { + *ppdata = data; + } + if (data) { + data->client_wsi = wsi; + } + spdlog::info("Control WS: client connected"); + break; + } + case LWS_CALLBACK_RECEIVE: { + if (!data) { + auto* vhost = lws_get_vhost(wsi); + data = static_cast(lws_vhost_user(vhost)); + } + if (!data) { + break; + } + + try { + auto msg = nlohmann::json::parse(static_cast(in), static_cast(in) + len); + if (!msg.contains("cmd")) { + spdlog::warn("Control WS: message missing 'cmd' field"); + break; + } + auto cmd = msg["cmd"].get(); + auto it = data->commands.find(cmd); + if (it != data->commands.end()) { + it->second(msg); + } else { + spdlog::warn("Control WS: unknown command '{}'", cmd); + } + } catch (const nlohmann::json::parse_error& e) { + spdlog::warn("Control WS: JSON parse error: {}", e.what()); + } + break; + } + case LWS_CALLBACK_SERVER_WRITEABLE: { + if (!data) { + auto* vhost = lws_get_vhost(wsi); + data = static_cast(lws_vhost_user(vhost)); + } + if (!data) { + break; + } + + std::lock_guard lock(data->send_mutex); + while (!data->send_queue.empty()) { + auto& msg = data->send_queue.back(); + std::vector buf(LWS_PRE + msg.size()); + std::memcpy(buf.data() + LWS_PRE, msg.data(), msg.size()); + lws_write(wsi, buf.data() + LWS_PRE, msg.size(), LWS_WRITE_TEXT); + data->send_queue.pop_back(); + } + break; + } + case LWS_CALLBACK_CLOSED: { + if (data) { + data->client_wsi = nullptr; + } + spdlog::info("Control WS: client disconnected"); + break; + } + default: + break; + } + return 0; +} + +ControlServer::ControlServer(uint16_t port, StatusCallback on_event) + : impl_(std::make_unique()) { + impl_->port = port; + impl_->data = std::make_unique(); + impl_->data->on_event = std::move(on_event); + + struct lws_context_creation_info info; + std::memset(&info, 0, sizeof(info)); + info.port = port; + info.protocols = protocols; + info.user = impl_->data.get(); + info.gid = -1; + info.uid = -1; + + impl_->context = lws_create_context(&info); + if (!impl_->context) { + spdlog::error("Failed to create WS context on port {}", port); + throw std::runtime_error("Failed to create WS context"); + } + spdlog::info("Control WS: listening on port {}", port); +} + +ControlServer::~ControlServer() { + if (impl_->context) { + lws_context_destroy(impl_->context); + } +} + +void ControlServer::register_command(const std::string& cmd, CommandHandler handler) { + impl_->data->commands[cmd] = std::move(handler); +} + +void ControlServer::send_event(const nlohmann::json& event) { + auto data = event.dump(); + { + std::lock_guard lock(impl_->data->send_mutex); + impl_->data->send_queue.push_back(data); + } + if (impl_->data->client_wsi) { + lws_callback_on_writable(impl_->data->client_wsi); + } +} + +void ControlServer::poll(int timeout_ms) { + lws_service(impl_->context, timeout_ms); +} + +} // namespace dmf_node diff --git a/libs/dmf-node/src/node_runner.cpp b/libs/dmf-node/src/node_runner.cpp new file mode 100644 index 0000000..266e04c --- /dev/null +++ b/libs/dmf-node/src/node_runner.cpp @@ -0,0 +1,184 @@ +#include +#include + +#include +#include + +#include + +#include + +#include +#include +#include +#include + +namespace dmf_node { + +static std::atomic g_running{true}; + +static void signal_handler(int /*signum*/) { + g_running = false; +} + +bool NodeRunner::parse_args(int argc, char* argv[]) { + for (int i = 1; i < argc; ++i) { + std::string arg = argv[i]; + if ((arg == "--node-id" || arg == "-n") && i + 1 < argc) { + node_id_ = argv[++i]; + } else if ((arg == "--control-port" || arg == "-p") && i + 1 < argc) { + control_port_ = static_cast(std::stoi(argv[++i])); + } else if ((arg == "--mxl-domain" || arg == "-d") && i + 1 < argc) { + mxl_domain_ = argv[++i]; + } else if ((arg == "--config" || arg == "-c") && i + 1 < argc) { + config_str_ = argv[++i]; + } else if (arg == "--help" || arg == "-h") { + spdlog::info("Usage: {} [options]", argv[0]); + spdlog::info(" --node-id, -n Node instance ID"); + spdlog::info(" --control-port, -p WebSocket control port"); + spdlog::info(" --mxl-domain, -d MXL domain path (default: /dev/shm/mxl)"); + spdlog::info(" --config, -c Node configuration JSON"); + return false; + } + } + + if (node_id_.empty()) { + spdlog::error("--node-id is required"); + return false; + } + if (control_port_ == 0) { + spdlog::error("--control-port is required"); + return false; + } + return true; +} + +int NodeRunner::exec(std::unique_ptr node) { + std::signal(SIGINT, signal_handler); + std::signal(SIGTERM, signal_handler); + + spdlog::info("Starting node '{}' type='{}'", node_id_, node->type()); + + if (!std::filesystem::exists(mxl_domain_)) { + spdlog::error("MXL domain path does not exist: {}", mxl_domain_); + return 1; + } + + mxl_instance_ = mxlCreateInstance(mxl_domain_.c_str(), nullptr); + if (!mxl_instance_) { + spdlog::error("Failed to create MXL instance on domain: {}", mxl_domain_); + return 1; + } + spdlog::info("MXL instance created on domain: {}", mxl_domain_); + + mxlGarbageCollectFlows(mxl_instance_); + + if (!config_str_.empty()) { + try { + auto cfg = nlohmann::json::parse(config_str_); + node->configure(cfg); + } catch (const nlohmann::json::parse_error& e) { + spdlog::error("Failed to parse config JSON: {}", e.what()); + mxlDestroyInstance(mxl_instance_); + return 1; + } + } + + auto control_server = std::make_unique(control_port_, [](const nlohmann::json& /*event*/) {}); + + auto mk_ports = [](const std::vector& ports) { + auto arr = nlohmann::json::array(); + for (const auto& p : ports) { + arr.push_back({{"id", p.id}, {"direction", p.direction == PortDirection::Input ? "input" : "output"}, {"media_type", static_cast(p.media_type)}}); + } + return arr; + }; + + control_server->register_command("add_writer", [&, w = std::unordered_map{}](const nlohmann::json& msg) mutable { + auto flow_id = msg["flow_id"].get(); + auto port_id = msg["port_id"].get(); + auto flow_def = msg["flow_def"].dump(); + + mxlFlowWriter writer = nullptr; + bool created = false; + mxlFlowConfigInfo config_info{}; + auto status = mxlCreateFlowWriter(mxl_instance_, flow_def.c_str(), nullptr, &writer, &config_info, &created); + if (status != MXL_STATUS_OK || !writer) { + spdlog::error("Failed to create flow writer for flow {}: status={}", flow_id, static_cast(status)); + return; + } + spdlog::info("Created flow writer on port '{}' flow {} (created={})", port_id, flow_id, created); + node->on_add_writer(port_id, writer); + }); + + control_server->register_command("add_reader", [&](const nlohmann::json& msg) { + auto flow_id = msg["flow_id"].get(); + auto port_id = msg["port_id"].get(); + + mxlFlowReader reader = nullptr; + auto status = mxlCreateFlowReader(mxl_instance_, flow_id.c_str(), nullptr, &reader); + if (status != MXL_STATUS_OK || !reader) { + spdlog::error("Failed to create flow reader for flow {}: status={}", flow_id, static_cast(status)); + return; + } + spdlog::info("Created flow reader on port '{}' flow {}", port_id, flow_id); + node->on_add_reader(port_id, reader); + }); + + control_server->register_command("remove_writer", [&](const nlohmann::json& msg) { + auto port_id = msg["port_id"].get(); + node->on_remove_writer(port_id); + spdlog::info("Removed writer on port '{}'", port_id); + }); + + control_server->register_command("remove_reader", [&](const nlohmann::json& msg) { + auto port_id = msg["port_id"].get(); + node->on_remove_reader(port_id); + spdlog::info("Removed reader on port '{}'", port_id); + }); + + control_server->register_command("configure", [&](const nlohmann::json& msg) { + if (msg.contains("params")) { + node->configure(msg["params"]); + spdlog::info("Reconfigured node '{}'", node_id_); + } + }); + + control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) { + spdlog::info("Shutdown command received"); + g_running = false; + }); + + nlohmann::json ready_event; + ready_event["event"] = "ready"; + ready_event["type"] = node->type(); + ready_event["node_id"] = node_id_; + ready_event["ports"] = mk_ports(node->input_ports()); + for (const auto& p : node->output_ports()) { + ready_event["ports"].push_back({{"id", p.id}, {"direction", "output"}, {"media_type", static_cast(p.media_type)}}); + } + control_server->send_event(ready_event); + + running_ = true; + spdlog::info("Node '{}' entering process loop", node_id_); + + while (g_running && running_) { + control_server->poll(1); + node->process(); + + static int status_counter = 0; + if (++status_counter % 100 == 0) { + nlohmann::json status_event; + status_event["event"] = "status"; + status_event["node_id"] = node_id_; + status_event["data"] = node->status(); + control_server->send_event(status_event); + } + } + + spdlog::info("Node '{}' shutting down", node_id_); + mxlDestroyInstance(mxl_instance_); + return 0; +} + +} // namespace dmf_node diff --git a/nodes/passthrough/CMakeLists.txt b/nodes/passthrough/CMakeLists.txt new file mode 100644 index 0000000..5ae4c51 --- /dev/null +++ b/nodes/passthrough/CMakeLists.txt @@ -0,0 +1,6 @@ +add_executable(dmf-node-passthrough + src/main.cpp + src/passthrough_node.cpp +) + +target_link_libraries(dmf-node-passthrough PRIVATE dmf-node) diff --git a/nodes/passthrough/src/main.cpp b/nodes/passthrough/src/main.cpp new file mode 100644 index 0000000..e9f1d18 --- /dev/null +++ b/nodes/passthrough/src/main.cpp @@ -0,0 +1,6 @@ +#include +#include "passthrough_node.hpp" + +int main(int argc, char* argv[]) { + return dmf_node::NodeRunner::run(argc, argv); +} diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp new file mode 100644 index 0000000..1d52bba --- /dev/null +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -0,0 +1,94 @@ +#include "passthrough_node.hpp" + +#include +#include + +#include + +#include + +namespace dmf_node { + +void PassthroughNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) { + if (port_id == "video_out") { + writer_ = writer; + spdlog::info("Passthrough: writer added on video_out"); + } +} + +void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) { + if (port_id == "video_in") { + reader_ = reader; + spdlog::info("Passthrough: reader added on video_in"); + } +} + +void PassthroughNode::on_remove_writer(const std::string& port_id) { + if (port_id == "video_out") { + writer_.reset(); + spdlog::info("Passthrough: writer removed from video_out"); + } +} + +void PassthroughNode::on_remove_reader(const std::string& port_id) { + if (port_id == "video_in") { + reader_.reset(); + spdlog::info("Passthrough: reader removed from video_in"); + } +} + +void PassthroughNode::process() { + if (!reader_ || !writer_) { + return; + } + + mxlGrainInfo grain_info{}; + uint8_t* payload = nullptr; + + auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 100000000ULL, &grain_info, &payload); + if (status != MXL_STATUS_OK) { + if (status == MXL_ERR_TIMEOUT) { + return; + } + if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE) { + mxlFlowRuntimeInfo runtime{}; + mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); + read_index_ = runtime.headIndex; + return; + } + return; + } + + mxlGrainInfo out_grain{}; + uint8_t* out_payload = nullptr; + status = mxlFlowWriterOpenGrain(*writer_, write_index_, &out_grain, &out_payload); + if (status != MXL_STATUS_OK) { + spdlog::warn("Passthrough: failed to open output grain at index {}: {}", write_index_, static_cast(status)); + read_index_++; + return; + } + + auto copy_size = std::min(grain_info.grainSize, out_grain.grainSize); + std::memcpy(out_payload, payload, copy_size); + + out_grain.validSlices = grain_info.validSlices; + out_grain.flags = grain_info.flags; + mxlFlowWriterCommitGrain(*writer_, &out_grain); + + read_index_++; + write_index_++; + grains_processed_++; +} + +nlohmann::json PassthroughNode::status() const { + return { + {"type", "passthrough"}, + {"grains_processed", grains_processed_}, + {"read_index", read_index_}, + {"write_index", write_index_}, + {"has_reader", reader_.has_value()}, + {"has_writer", writer_.has_value()}, + }; +} + +} // namespace dmf_node diff --git a/nodes/passthrough/src/passthrough_node.hpp b/nodes/passthrough/src/passthrough_node.hpp new file mode 100644 index 0000000..ec4a6a4 --- /dev/null +++ b/nodes/passthrough/src/passthrough_node.hpp @@ -0,0 +1,45 @@ +#pragma once + +#include + +#include + +#include +#include + +namespace dmf_node { + +class PassthroughNode : public Node { +public: + PassthroughNode() = default; + + std::string type() const override { return "passthrough"; } + + std::vector input_ports() const override { + return {{"video_in", PortDirection::Input, MediaType::VideoV210}}; + } + + std::vector output_ports() const override { + return {{"video_out", PortDirection::Output, MediaType::VideoV210}}; + } + + void configure(const nlohmann::json& /*params*/) override {} + + void on_add_writer(const std::string& port_id, mxlFlowWriter writer) override; + void on_add_reader(const std::string& port_id, mxlFlowReader reader) override; + void on_remove_writer(const std::string& port_id) override; + void on_remove_reader(const std::string& port_id) override; + + void process() override; + nlohmann::json status() const override; + +private: + std::optional reader_; + std::optional writer_; + mxlInstance mxl_instance_ = nullptr; + uint64_t read_index_ = 0; + uint64_t write_index_ = 0; + uint64_t grains_processed_ = 0; +}; + +} // namespace dmf_node diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt new file mode 100644 index 0000000..d02b9b6 --- /dev/null +++ b/tests/CMakeLists.txt @@ -0,0 +1,8 @@ +add_executable(dmf-test-graph + test_graph.cpp +) + +target_link_libraries(dmf-test-graph PRIVATE + dmf-engine + Catch2::Catch2WithMain +) diff --git a/tests/test_graph.cpp b/tests/test_graph.cpp new file mode 100644 index 0000000..edcdbc7 --- /dev/null +++ b/tests/test_graph.cpp @@ -0,0 +1,88 @@ +#include +#include + +#include + +TEST_CASE("Graph add and remove nodes", "[graph]") { + dmf_engine::Graph graph; + + auto id1 = graph.add_node("passthrough"); + auto id2 = graph.add_node("test-source"); + + REQUIRE(graph.get_node(id1) != nullptr); + REQUIRE(graph.get_node(id2) != nullptr); + REQUIRE(graph.get_node(id1)->type == "passthrough"); + REQUIRE(graph.get_node(id2)->type == "test-source"); + + auto nodes = graph.get_nodes(); + REQUIRE(nodes.size() == 2); + + graph.remove_node(id1); + REQUIRE(graph.get_node(id1) == nullptr); + REQUIRE(graph.get_nodes().size() == 1); +} + +TEST_CASE("Graph add and remove edges", "[graph]") { + dmf_engine::Graph graph; + + auto n1 = graph.add_node("test-source"); + auto n2 = graph.add_node("passthrough"); + + auto edge_id = graph.add_edge(n1, "video_out", n2, "video_in", "flow1", {}); + REQUIRE(graph.get_edges().size() == 1); + + graph.remove_edge(edge_id); + REQUIRE(graph.get_edges().size() == 0); +} + +TEST_CASE("Graph remove node also removes edges", "[graph]") { + dmf_engine::Graph graph; + + auto n1 = graph.add_node("test-source"); + auto n2 = graph.add_node("passthrough"); + + graph.add_edge(n1, "video_out", n2, "video_in", "flow1", {}); + graph.add_edge(n2, "video_out", n1, "video_in", "flow2", {}); + + REQUIRE(graph.get_edges().size() == 2); + + graph.remove_node(n1); + REQUIRE(graph.get_edges().size() == 0); +} + +TEST_CASE("FlowManager creates unique IDs", "[flow_manager]") { + dmf_engine::FlowManager fm1; + dmf_engine::FlowManager fm2; + + auto id1 = fm1.create_flow_id(); + auto id2 = fm1.create_flow_id(); + auto id3 = fm2.create_flow_id(); + + REQUIRE(id1 != id2); + REQUIRE(id2 != id3); +} + +TEST_CASE("FlowManager creates V210 flow definition", "[flow_manager]") { + dmf_engine::FlowManager fm; + auto flow_id = fm.create_flow_id(); + auto def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1); + + REQUIRE(def["id"] == flow_id); + REQUIRE(def["format"] == "video/v210"); + REQUIRE(def["width"] == 1920); + REQUIRE(def["height"] == 1080); +} + +TEST_CASE("Graph serialize", "[graph]") { + dmf_engine::Graph graph; + + auto n1 = graph.add_node("test-source"); + auto n2 = graph.add_node("passthrough"); + graph.add_edge(n1, "video_out", n2, "video_in", "flow1", {}); + + auto j = graph.serialize(); + REQUIRE(j.contains("nodes")); + REQUIRE(j.contains("edges")); + REQUIRE(j["nodes"].size() == 2); + REQUIRE(j["edges"].size() == 1); +} diff --git a/vcpkg.json b/vcpkg.json new file mode 100644 index 0000000..d12145e --- /dev/null +++ b/vcpkg.json @@ -0,0 +1,20 @@ +{ + "name": "dmf-studio", + "version": "0.1.0", + "dependencies": [ + "libwebsockets", + "nlohmann-json", + "spdlog", + "fmt", + "catch2", + { + "name": "stduuid", + "version>=": "1.2.3", + "features": ["system-gen", "gsl-span"] + }, + "picojson", + "cli11", + "ada-url" + ], + "builtin-baseline": "432412eecac55981cf608cc12e5b4bd91768ec57" +} From d138478c1af7929a4a744861cf2872cc532d5a9b Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 01:40:42 +0300 Subject: [PATCH 02/11] fix: WebSocket control server - add HTTP mount for WS upgrade negotiation LWS v4.x requires an HTTP mount with LWSMPRO_CALLBACK to properly route WebSocket upgrade requests to the dmf-control protocol handler. Without this, WS connections were rejected with 403 Forbidden. --- libs/dmf-node/src/control_server.cpp | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/libs/dmf-node/src/control_server.cpp b/libs/dmf-node/src/control_server.cpp index 61c268a..7872757 100644 --- a/libs/dmf-node/src/control_server.cpp +++ b/libs/dmf-node/src/control_server.cpp @@ -33,6 +33,12 @@ static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason, void* user, void* in, size_t len); static const struct lws_protocols protocols[] = { + { + "http-only", + callback_ws, + sizeof(ControlServerData*), + 0, + }, { "dmf-control", callback_ws, @@ -119,6 +125,25 @@ static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason, return 0; } +static const struct lws_http_mount mount = { + .mount_next = nullptr, + .mountpoint = "/", + .origin = "", + .def = "", + .protocol = "dmf-control", + .cgienv = nullptr, + .extra_mimetypes = nullptr, + .interpret = nullptr, + .cgi_timeout = 0, + .cache_max_age = 0, + .auth_mask = 0, + .cache_reusable = 0, + .cache_revalidate = 0, + .cache_intermediaries = 0, + .origin_protocol = LWSMPRO_CALLBACK, + .mountpoint_len = 1, +}; + ControlServer::ControlServer(uint16_t port, StatusCallback on_event) : impl_(std::make_unique()) { impl_->port = port; @@ -129,6 +154,7 @@ ControlServer::ControlServer(uint16_t port, StatusCallback on_event) std::memset(&info, 0, sizeof(info)); info.port = port; info.protocols = protocols; + info.mounts = &mount; info.user = impl_->data.get(); info.gid = -1; info.uid = -1; From 37f02f01d930efbf0425f9eabf992df2ce38fb59 Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 01:42:44 +0300 Subject: [PATCH 03/11] docs: add build and test instructions --- README.md | 187 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 187 insertions(+) create mode 100644 README.md diff --git a/README.md b/README.md new file mode 100644 index 0000000..e1ec355 --- /dev/null +++ b/README.md @@ -0,0 +1,187 @@ +# DMF Studio + +Node-based visual production platform for the Dynamic Media Facility architecture. + +## Prerequisites + +- CMake 3.24+ +- C++20 compiler (GCC 12+, Clang 15+) +- vcpkg (at `~/vcpkg` or set `CMAKE_TOOLCHAIN_FILE`) +- GStreamer (for MXL test tools only) + +## Build + +```bash +# Configure (from project root) +cmake -B build \ + -DCMAKE_BUILD_TYPE=Debug \ + -DCMAKE_TOOLCHAIN_FILE=$HOME/vcpkg/scripts/buildsystems/vcpkg.cmake \ + -DBUILD_SHARED_LIBS=OFF \ + -DBUILD_DOCS=OFF \ + -DBUILD_TESTS=OFF \ + -DBUILD_TOOLS=OFF \ + -DBUILD_UTILS=OFF + +# Build all +cmake --build build -j$(nproc) +``` + +Binaries end up in: +- `build/engine/dmf-studio-engine` +- `build/nodes/passthrough/dmf-node-passthrough` + +## Rebuild after changes + +```bash +# Full rebuild +cmake --build build -j$(nproc) + +# Rebuild only one target (faster) +cmake --build build -j$(nproc) --target dmf-node-passthrough +cmake --build build -j$(nproc) --target dmf-studio-engine +``` + +## Clean rebuild + +```bash +rm -rf build +# Then re-run the configure + build steps above +``` + +## Run unit tests + +```bash +build/tests/dmf-test-graph +``` + +## Test with MXL + +### 1. Create MXL domain + +```bash +mkdir -p /tmp/dmf-mxl +``` + +### 2. Start MXL test source (writes V210 video flow) + +Create a flow config file: + +```bash +cat > /tmp/v210_50p.json << 'EOF' +{ + "id": "a0000001-0000-0000-0000-000000000001", + "description": "DMF Studio test video flow", + "format": "urn:x-nmos:format:video", + "label": "DMF Studio Test Video", + "tags": { + "urn:x-nmos:tag:grouphint/v1.0": ["dmf-studio:Video"] + }, + "media_type": "video/v210", + "grain_rate": {"numerator": 50, "denominator": 1}, + "frame_width": 1920, + "frame_height": 1080, + "interlace_mode": "progressive", + "colorspace": "BT709", + "components": [ + {"name": "Y", "width": 1920, "height": 1080, "bit_depth": 10}, + {"name": "Cb", "width": 960, "height": 1080, "bit_depth": 10}, + {"name": "Cr", "width": 960, "height": 1080, "bit_depth": 10} + ] +} +EOF +``` + +Start test source (needs GStreamer + MXL tools built separately): + +```bash +mxl-gst-testsrc -d /tmp/dmf-mxl -v /tmp/v210_50p.json --pattern smpte & +``` + +Check active flows: + +```bash +mxl-info --domain /tmp/dmf-mxl +``` + +### 3. Start passthrough node + +```bash +build/nodes/passthrough/dmf-node-passthrough \ + --node-id pass1 \ + --control-port 9100 \ + --mxl-domain /tmp/dmf-mxl +``` + +Options: +- `--node-id, -n` — Unique node instance ID (required) +- `--control-port, -p` — WebSocket control port (required) +- `--mxl-domain, -d` — MXL domain path (default: `/dev/shm/mxl`) +- `--config, -c` — Node config as JSON string + +### 4. Start engine + +```bash +export DMF_STUDIO_BIN_DIR=build/nodes/passthrough +build/engine/dmf-studio-engine --port 8080 +``` + +### 5. Control via REST API + +```bash +# Add nodes +curl -X POST http://localhost:8080/api/graph/nodes \ + -H "Content-Type: application/json" \ + -d '{"type":"passthrough","id":"pass1"}' + +curl -X POST http://localhost:8080/api/graph/nodes \ + -H "Content-Type: application/json" \ + -d '{"type":"passthrough","id":"pass2"}' + +# Connect nodes (creates MXL flow between them) +curl -X POST http://localhost:8080/api/graph/edges \ + -H "Content-Type: application/json" \ + -d '{"from_node":"pass1","from_port":"video_out","to_node":"pass2","to_port":"video_in"}' + +# View graph +curl http://localhost:8080/api/graph + +# Start all nodes +curl -X POST http://localhost:8080/api/graph/start + +# Stop all nodes +curl -X POST http://localhost:8080/api/graph/stop + +# Remove node +curl -X DELETE http://localhost:8080/api/graph/nodes/pass1 + +# Remove edge +curl -X DELETE http://localhost:8080/api/graph/edges/pass1:video_out->pass2:video_in +``` + +### 6. Control node directly via WebSocket + +Connect to `ws://localhost:9100` with subprotocol `dmf-control`: + +```bash +wscat -c ws://localhost:9100 -s dmf-control +``` + +Commands: +```json +{"cmd": "add_reader", "flow_id": "", "port_id": "video_in"} +{"cmd": "add_writer", "flow_id": "", "port_id": "video_out", "flow_def": {}} +{"cmd": "remove_reader", "port_id": "video_in"} +{"cmd": "remove_writer", "port_id": "video_out"} +{"cmd": "configure", "params": {}} +{"cmd": "shutdown"} +``` + +## Project structure + +``` +libs/dmf-node/ — Node skeleton library (interface, WS control, MXL lifecycle) +libs/dmf-engine/ — Engine library (graph model, flow manager, process manager, REST API) +nodes/passthrough/ — Passthrough node (1 MXL in → 1 MXL out, memcpy) +engine/ — Engine binary +tests/ — Unit tests +``` From 1d6f93679f87f760171e67e74b838e388d26280c Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 21:08:06 +0300 Subject: [PATCH 04/11] fix: engine REST API - handle GET/DELETE requests immediately, use user-provided node IDs - Use lws_http_get_uri_and_method() for proper HTTP method detection (GET, POST, PUT, DELETE all supported) - Handle GET/DELETE in LWS_CALLBACK_HTTP without waiting for body - Support user-provided node IDs via config.id field - Fixes all REST API endpoints hanging on GET requests --- libs/dmf-engine/src/api_server.cpp | 267 ++++++++++++++++------------- libs/dmf-engine/src/graph.cpp | 7 +- 2 files changed, 152 insertions(+), 122 deletions(-) diff --git a/libs/dmf-engine/src/api_server.cpp b/libs/dmf-engine/src/api_server.cpp index 7537a1a..adfd3dc 100644 --- a/libs/dmf-engine/src/api_server.cpp +++ b/libs/dmf-engine/src/api_server.cpp @@ -30,11 +30,124 @@ struct HttpRequest { std::string method; std::string path; std::string body; - bool body_done = false; }; static ApiServerImpl* g_impl = nullptr; +static int send_json_response(struct lws* wsi, const std::string& status_str, + const std::string& json_body) { + auto headers = "HTTP/1.1 " + status_str + "\r\n" + "Content-Type: application/json\r\n" + "Access-Control-Allow-Origin: *\r\n" + "Access-Control-Allow-Methods: GET, POST, DELETE, PUT, OPTIONS\r\n" + "Access-Control-Allow-Headers: Content-Type\r\n" + "Content-Length: " + std::to_string(json_body.size()) + "\r\n" + "\r\n"; + + std::vector buf(LWS_PRE + headers.size() + json_body.size()); + std::memcpy(buf.data() + LWS_PRE, headers.data(), headers.size()); + std::memcpy(buf.data() + LWS_PRE + headers.size(), json_body.data(), json_body.size()); + + lws_write(wsi, buf.data() + LWS_PRE, headers.size() + json_body.size(), LWS_WRITE_HTTP); + + if (lws_http_transaction_completed(wsi)) { + return -1; + } + return 0; +} + +static void handle_request(const std::string& method, const std::string& path, + const std::string& body, + std::string& status_str, std::string& response_body) { + if (!g_impl || !g_impl->graph) { + status_str = "500 Internal Server Error"; + response_body = nlohmann::json({{"error", "Server not initialized"}}).dump(); + return; + } + + auto ok = [&](const nlohmann::json& j) { + status_str = "200 OK"; + response_body = j.dump(); + }; + auto created = [&](const nlohmann::json& j) { + status_str = "201 Created"; + response_body = j.dump(); + }; + auto error_resp = [&](int code, const std::string& msg) { + status_str = std::to_string(code) + " Error"; + response_body = nlohmann::json({{"error", msg}}).dump(); + }; + + try { + nlohmann::json req_body = body.empty() ? nlohmann::json::object() : nlohmann::json::parse(body); + + auto& graph = *g_impl->graph; + auto& fm = *g_impl->flow_manager; + auto& pm = *g_impl->process_manager; + + if (path == "/api/graph" && method == "GET") { + ok(graph.serialize()); + } else if (path == "/api/graph/nodes" && method == "POST") { + if (!req_body.contains("type")) { + error_resp(400, "Missing 'type' field"); + } else { + auto type = req_body["type"].get(); + auto config = req_body.value("config", nlohmann::json::object()); + auto id = graph.add_node(type, config); + created({{"id", id}}); + } + } else if (path.find("/api/graph/nodes/") == 0 && method == "DELETE") { + auto node_id = path.substr(std::string("/api/graph/nodes/").length()); + if (graph.remove_node(node_id)) { + ok({{"deleted", node_id}}); + } else { + error_resp(404, "Node not found: " + node_id); + } + } else if (path == "/api/graph/edges" && method == "POST") { + if (!req_body.contains("from_node") || !req_body.contains("to_node")) { + error_resp(400, "Missing from_node/to_node"); + } else { + auto from_node = req_body["from_node"].get(); + auto from_port = req_body.value("from_port", "video_out"); + auto to_node = req_body["to_node"].get(); + auto to_port = req_body.value("to_port", "video_in"); + + auto flow_id = fm.create_flow_id(); + auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1); + + auto edge_id = graph.add_edge(from_node, from_port, to_node, to_port, flow_id, flow_def); + created({{"id", edge_id}, {"flow_id", flow_id}}); + } + } else if (path.find("/api/graph/edges/") == 0 && method == "DELETE") { + auto edge_id = path.substr(std::string("/api/graph/edges/").length()); + if (graph.remove_edge(edge_id)) { + ok({{"deleted", edge_id}}); + } else { + error_resp(404, "Edge not found: " + edge_id); + } + } else if (path == "/api/graph/start" && method == "POST") { + auto nodes = graph.get_nodes(); + uint16_t port = 9100; + for (auto& node : nodes) { + pm.start_node(const_cast(node), "/dev/shm/mxl", port++); + } + ok({{"status", "started"}}); + } else if (path == "/api/graph/stop" && method == "POST") { + auto nodes = graph.get_nodes(); + for (auto& node : nodes) { + pm.stop_node(const_cast(node)); + } + ok({{"status", "stopped"}}); + } else { + error_resp(404, "Not found: " + method + " " + path); + } + } catch (const nlohmann::json::exception& e) { + error_resp(400, std::string("JSON error: ") + e.what()); + } catch (const std::exception& e) { + error_resp(500, e.what()); + } +} + static int callback_http(struct lws* wsi, enum lws_callback_reasons reason, void* user, void* in, size_t len) { auto* req = static_cast(user); @@ -43,17 +156,35 @@ static int callback_http(struct lws* wsi, enum lws_callback_reasons reason, case LWS_CALLBACK_HTTP: { new (req) HttpRequest(); - if (lws_hdr_total_length(wsi, WSI_TOKEN_POST_URI) > 0) { - req->method = "POST"; - char buf[256] = {}; - lws_hdr_copy(wsi, buf, sizeof(buf), WSI_TOKEN_POST_URI); - req->path = buf; - } else { - req->method = "GET"; - char buf[256] = {}; - lws_hdr_copy(wsi, buf, sizeof(buf), WSI_TOKEN_GET_URI); - req->path = buf; + char* uri_ptr = nullptr; + int uri_len = 0; + int method = lws_http_get_uri_and_method(wsi, &uri_ptr, &uri_len); + + switch (method) { + case LWSHUMETH_GET: req->method = "GET"; break; + case LWSHUMETH_POST: req->method = "POST"; break; + case LWSHUMETH_PUT: req->method = "PUT"; break; + case LWSHUMETH_DELETE: req->method = "DELETE"; break; + default: req->method = "GET"; break; } + + if (uri_ptr && uri_len > 0) { + req->path.assign(uri_ptr, uri_len); + } + + if (req->method == "GET" || req->method == "DELETE") { + std::string status_str, response_body; + handle_request(req->method, req->path, "", status_str, response_body); + return send_json_response(wsi, status_str, response_body); + } + + int body_len = lws_hdr_total_length(wsi, WSI_TOKEN_HTTP_CONTENT_LENGTH); + if (body_len == 0) { + std::string status_str, response_body; + handle_request(req->method, req->path, "", status_str, response_body); + return send_json_response(wsi, status_str, response_body); + } + req->body.reserve(body_len); break; } case LWS_CALLBACK_HTTP_BODY: { @@ -61,115 +192,9 @@ static int callback_http(struct lws* wsi, enum lws_callback_reasons reason, break; } case LWS_CALLBACK_HTTP_BODY_COMPLETION: { - if (!g_impl || !g_impl->graph) { - lws_return_http_status(wsi, HTTP_STATUS_INTERNAL_SERVER_ERROR, "Server error"); - return -1; - } - - std::string status_str, content_type, response_body; - - auto ok = [&](const nlohmann::json& j) { - status_str = "200 OK"; - response_body = j.dump(); - }; - auto created = [&](const nlohmann::json& j) { - status_str = "201 Created"; - response_body = j.dump(); - }; - auto error_resp = [&](int code, const std::string& msg) { - status_str = std::to_string(code) + " Error"; - response_body = nlohmann::json({{"error", msg}}).dump(); - }; - - content_type = "application/json"; - - try { - nlohmann::json req_body = req->body.empty() ? nlohmann::json::object() : nlohmann::json::parse(req->body); - - auto& graph = *g_impl->graph; - auto& fm = *g_impl->flow_manager; - auto& pm = *g_impl->process_manager; - - if (req->path == "/api/graph" && req->method == "GET") { - ok(graph.serialize()); - } else if (req->path == "/api/graph/nodes" && req->method == "POST") { - if (!req_body.contains("type")) { - error_resp(400, "Missing 'type' field"); - } else { - auto type = req_body["type"].get(); - auto config = req_body.value("config", nlohmann::json::object()); - auto id = graph.add_node(type, config); - created({{"id", id}}); - } - } else if (req->path.find("/api/graph/nodes/") == 0 && req->method == "DELETE") { - auto node_id = req->path.substr(std::string("/api/graph/nodes/").length()); - if (graph.remove_node(node_id)) { - ok({{"deleted", node_id}}); - } else { - error_resp(404, "Node not found: " + node_id); - } - } else if (req->path == "/api/graph/edges" && req->method == "POST") { - if (!req_body.contains("from_node") || !req_body.contains("to_node")) { - error_resp(400, "Missing from_node/to_node"); - } else { - auto from_node = req_body["from_node"].get(); - auto from_port = req_body.value("from_port", "video_out"); - auto to_node = req_body["to_node"].get(); - auto to_port = req_body.value("to_port", "video_in"); - - auto flow_id = fm.create_flow_id(); - auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1); - - auto edge_id = graph.add_edge(from_node, from_port, to_node, to_port, flow_id, flow_def); - created({{"id", edge_id}, {"flow_id", flow_id}}); - } - } else if (req->path.find("/api/graph/edges/") == 0 && req->method == "DELETE") { - auto edge_id = req->path.substr(std::string("/api/graph/edges/").length()); - if (graph.remove_edge(edge_id)) { - ok({{"deleted", edge_id}}); - } else { - error_resp(404, "Edge not found: " + edge_id); - } - } else if (req->path == "/api/graph/start" && req->method == "POST") { - auto nodes = graph.get_nodes(); - uint16_t port = 9100; - for (auto& node : nodes) { - pm.start_node(const_cast(node), "/dev/shm/mxl", port++); - } - ok({{"status", "started"}}); - } else if (req->path == "/api/graph/stop" && req->method == "POST") { - auto nodes = graph.get_nodes(); - for (auto& node : nodes) { - pm.stop_node(const_cast(node)); - } - ok({{"status", "stopped"}}); - } else { - error_resp(404, "Not found: " + req->method + " " + req->path); - } - } catch (const nlohmann::json::exception& e) { - error_resp(400, std::string("JSON error: ") + e.what()); - } catch (const std::exception& e) { - error_resp(500, e.what()); - } - - auto headers = "HTTP/1.1 " + status_str + "\r\n" - "Content-Type: " + content_type + "\r\n" - "Access-Control-Allow-Origin: *\r\n" - "Access-Control-Allow-Methods: GET, POST, DELETE, PUT, OPTIONS\r\n" - "Access-Control-Allow-Headers: Content-Type\r\n" - "Content-Length: " + std::to_string(response_body.size()) + "\r\n" - "\r\n"; - - std::vector buf(LWS_PRE + headers.size() + response_body.size()); - std::memcpy(buf.data() + LWS_PRE, headers.data(), headers.size()); - std::memcpy(buf.data() + LWS_PRE + headers.size(), response_body.data(), response_body.size()); - - lws_write(wsi, buf.data() + LWS_PRE, headers.size() + response_body.size(), LWS_WRITE_HTTP); - - if (lws_http_transaction_completed(wsi)) { - return -1; - } - return 0; + std::string status_str, response_body; + handle_request(req->method, req->path, req->body, status_str, response_body); + return send_json_response(wsi, status_str, response_body); } default: break; @@ -178,7 +203,7 @@ static int callback_http(struct lws* wsi, enum lws_callback_reasons reason, } static const struct lws_protocols protocols[] = { - {"http-api", callback_http, sizeof(HttpRequest), 0}, + {"http-api", callback_http, sizeof(HttpRequest), 4096}, {nullptr, nullptr, 0, 0}, }; diff --git a/libs/dmf-engine/src/graph.cpp b/libs/dmf-engine/src/graph.cpp index f1ca009..1213b53 100644 --- a/libs/dmf-engine/src/graph.cpp +++ b/libs/dmf-engine/src/graph.cpp @@ -7,7 +7,12 @@ namespace dmf_engine { NodeId Graph::add_node(const std::string& type, const nlohmann::json& config) { - auto id = type + "_" + std::to_string(next_node_num_++); + std::string id; + if (config.contains("id") && config["id"].is_string()) { + id = config["id"].get(); + } else { + id = type + "_" + std::to_string(next_node_num_++); + } GraphNode node; node.id = id; node.type = type; From 5b2d420e71f28ac122ebe79ce7aa3c964dfc5425 Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 21:33:17 +0300 Subject: [PATCH 05/11] feat: add node HTTP control, engine-to-node communication, connect-input/output API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Node control server now accepts HTTP POST /cmd for command dispatch (in addition to existing WebSocket control) - Added NodeControlClient: engine sends commands to nodes via HTTP POST - New REST endpoints: POST /api/graph/nodes/:id/connect-input - connect node input to MXL flow POST /api/graph/nodes/:id/connect-output - create new MXL flow + connect output POST /api/graph/nodes/:id/disconnect-port - remove reader/writer POST /api/graph/nodes/:id/command - send raw command to node - Flow IDs now use proper UUID v4 format (MXL requires standard UUIDs) - Flow definitions use NMOS format (urn:x-nmos:format:video) - Engine passes --mxl-domain to node processes - User-provided node IDs (via 'id' field in POST body) - End-to-end verified: testsrc → passthrough → new MXL output flow --- engine/src/main.cpp | 12 +- libs/dmf-engine/CMakeLists.txt | 1 + .../include/dmf-engine/api_server.hpp | 2 +- .../dmf-engine/node_control_client.hpp | 21 ++ libs/dmf-engine/src/api_server.cpp | 183 ++++++++++++++- libs/dmf-engine/src/flow_manager.cpp | 46 ++-- libs/dmf-engine/src/node_control_client.cpp | 88 ++++++++ libs/dmf-node/src/control_server.cpp | 213 ++++++++++++------ 8 files changed, 474 insertions(+), 92 deletions(-) create mode 100644 libs/dmf-engine/include/dmf-engine/node_control_client.hpp create mode 100644 libs/dmf-engine/src/node_control_client.cpp diff --git a/engine/src/main.cpp b/engine/src/main.cpp index 0da0b69..424f33d 100644 --- a/engine/src/main.cpp +++ b/engine/src/main.cpp @@ -2,6 +2,7 @@ #include #include #include +#include #include @@ -16,11 +17,19 @@ static void signal_handler(int /*signum*/) { int main(int argc, char* argv[]) { uint16_t port = 8080; + std::string mxl_domain = "/dev/shm/mxl"; for (int i = 1; i < argc; ++i) { std::string arg = argv[i]; if ((arg == "--port" || arg == "-p") && i + 1 < argc) { port = static_cast(std::stoi(argv[++i])); + } else if ((arg == "--mxl-domain" || arg == "-d") && i + 1 < argc) { + mxl_domain = argv[++i]; + } else if (arg == "--help" || arg == "-h") { + spdlog::info("Usage: dmf-studio-engine [options]"); + spdlog::info(" --port, -p API server port (default: 8080)"); + spdlog::info(" --mxl-domain, -d MXL domain path (default: /dev/shm/mxl)"); + return 0; } } @@ -32,8 +41,9 @@ int main(int argc, char* argv[]) { dmf_engine::Graph graph; dmf_engine::FlowManager flow_manager; dmf_engine::ProcessManager process_manager; + dmf_engine::NodeControlClient control_client; - dmf_engine::ApiServer api_server(port, graph, flow_manager, process_manager); + dmf_engine::ApiServer api_server(port, graph, flow_manager, process_manager, control_client, mxl_domain); spdlog::info("DMF Studio Engine ready"); diff --git a/libs/dmf-engine/CMakeLists.txt b/libs/dmf-engine/CMakeLists.txt index c88690e..24de18d 100644 --- a/libs/dmf-engine/CMakeLists.txt +++ b/libs/dmf-engine/CMakeLists.txt @@ -3,6 +3,7 @@ add_library(dmf-engine STATIC src/flow_manager.cpp src/process_manager.cpp src/api_server.cpp + src/node_control_client.cpp ) target_include_directories(dmf-engine PUBLIC diff --git a/libs/dmf-engine/include/dmf-engine/api_server.hpp b/libs/dmf-engine/include/dmf-engine/api_server.hpp index 1ae6e88..27fe8d9 100644 --- a/libs/dmf-engine/include/dmf-engine/api_server.hpp +++ b/libs/dmf-engine/include/dmf-engine/api_server.hpp @@ -14,7 +14,7 @@ using RequestHandler = std::function +#include +#include + +namespace dmf_engine { + +class NodeControlClient { +public: + bool send_command(uint16_t port, const std::string& json_cmd); + + void register_node(const std::string& node_id, uint16_t port); + void unregister_node(const std::string& node_id); + uint16_t get_port(const std::string& node_id) const; + +private: + std::unordered_map node_ports_; +}; + +} // namespace dmf_engine diff --git a/libs/dmf-engine/src/api_server.cpp b/libs/dmf-engine/src/api_server.cpp index adfd3dc..e338668 100644 --- a/libs/dmf-engine/src/api_server.cpp +++ b/libs/dmf-engine/src/api_server.cpp @@ -2,6 +2,7 @@ #include #include #include +#include #include @@ -19,6 +20,8 @@ struct ApiServerImpl { Graph* graph = nullptr; FlowManager* flow_manager = nullptr; ProcessManager* process_manager = nullptr; + NodeControlClient* control_client = nullptr; + std::string mxl_domain = "/dev/shm/mxl"; struct lws_context* context = nullptr; }; @@ -84,6 +87,7 @@ static void handle_request(const std::string& method, const std::string& path, auto& graph = *g_impl->graph; auto& fm = *g_impl->flow_manager; auto& pm = *g_impl->process_manager; + auto& cc = *g_impl->control_client; if (path == "/api/graph" && method == "GET") { ok(graph.serialize()); @@ -93,11 +97,15 @@ static void handle_request(const std::string& method, const std::string& path, } else { auto type = req_body["type"].get(); auto config = req_body.value("config", nlohmann::json::object()); + if (req_body.contains("id") && req_body["id"].is_string()) { + config["id"] = req_body["id"]; + } auto id = graph.add_node(type, config); created({{"id", id}}); } } else if (path.find("/api/graph/nodes/") == 0 && method == "DELETE") { auto node_id = path.substr(std::string("/api/graph/nodes/").length()); + cc.unregister_node(node_id); if (graph.remove_node(node_id)) { ok({{"deleted", node_id}}); } else { @@ -116,11 +124,54 @@ static void handle_request(const std::string& method, const std::string& path, auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1); auto edge_id = graph.add_edge(from_node, from_port, to_node, to_port, flow_id, flow_def); + + auto from_port_num = cc.get_port(from_node); + auto to_port_num = cc.get_port(to_node); + + if (from_port_num > 0) { + nlohmann::json cmd; + cmd["cmd"] = "add_writer"; + cmd["port_id"] = from_port; + cmd["flow_id"] = flow_id; + cmd["flow_def"] = flow_def; + cc.send_command(from_port_num, cmd.dump()); + } + + if (to_port_num > 0) { + nlohmann::json cmd; + cmd["cmd"] = "add_reader"; + cmd["port_id"] = to_port; + cmd["flow_id"] = flow_id; + cc.send_command(to_port_num, cmd.dump()); + } + created({{"id", edge_id}, {"flow_id", flow_id}}); } } else if (path.find("/api/graph/edges/") == 0 && method == "DELETE") { auto edge_id = path.substr(std::string("/api/graph/edges/").length()); + auto edges = graph.get_edges(); + const GraphEdge* edge = nullptr; + for (auto& e : edges) { + if (e.id == edge_id) { edge = &e; break; } + } + if (graph.remove_edge(edge_id)) { + if (edge) { + auto from_port_num = cc.get_port(edge->from_node); + auto to_port_num = cc.get_port(edge->to_node); + if (from_port_num > 0) { + nlohmann::json cmd; + cmd["cmd"] = "remove_writer"; + cmd["port_id"] = edge->from_port; + cc.send_command(from_port_num, cmd.dump()); + } + if (to_port_num > 0) { + nlohmann::json cmd; + cmd["cmd"] = "remove_reader"; + cmd["port_id"] = edge->to_port; + cc.send_command(to_port_num, cmd.dump()); + } + } ok({{"deleted", edge_id}}); } else { error_resp(404, "Edge not found: " + edge_id); @@ -129,15 +180,141 @@ static void handle_request(const std::string& method, const std::string& path, auto nodes = graph.get_nodes(); uint16_t port = 9100; for (auto& node : nodes) { - pm.start_node(const_cast(node), "/dev/shm/mxl", port++); + pm.start_node(const_cast(node), g_impl->mxl_domain, port); + cc.register_node(node.id, port); + port++; } + + for (auto& edge : graph.get_edges()) { + auto from_port_num = cc.get_port(edge.from_node); + auto to_port_num = cc.get_port(edge.to_node); + + if (from_port_num > 0) { + nlohmann::json cmd; + cmd["cmd"] = "add_writer"; + cmd["port_id"] = edge.from_port; + cmd["flow_id"] = edge.flow_id; + cmd["flow_def"] = edge.flow_def; + cc.send_command(from_port_num, cmd.dump()); + } + + if (to_port_num > 0) { + nlohmann::json cmd; + cmd["cmd"] = "add_reader"; + cmd["port_id"] = edge.to_port; + cmd["flow_id"] = edge.flow_id; + cc.send_command(to_port_num, cmd.dump()); + } + } + ok({{"status", "started"}}); } else if (path == "/api/graph/stop" && method == "POST") { auto nodes = graph.get_nodes(); for (auto& node : nodes) { + cc.unregister_node(node.id); pm.stop_node(const_cast(node)); } ok({{"status", "stopped"}}); + } else if (path.find("/api/graph/nodes/") == 0 && path.find("/connect-input") != std::string::npos && method == "POST") { + auto prefix = std::string("/api/graph/nodes/"); + auto suffix_start = path.find("/connect-input"); + auto node_id = path.substr(prefix.length(), suffix_start - prefix.length()); + + if (!req_body.contains("flow_id") || !req_body.contains("port_id")) { + error_resp(400, "Missing flow_id/port_id"); + return; + } + + auto port_num = cc.get_port(node_id); + if (port_num == 0) { + error_resp(404, "Node not running or not found: " + node_id); + return; + } + + nlohmann::json cmd; + cmd["cmd"] = "add_reader"; + cmd["port_id"] = req_body["port_id"].get(); + cmd["flow_id"] = req_body["flow_id"].get(); + + if (cc.send_command(port_num, cmd.dump())) { + ok({{"node_id", node_id}, {"connected_input", req_body["port_id"]}, {"flow_id", req_body["flow_id"]}}); + } else { + error_resp(500, "Failed to send command to node"); + } + } else if (path.find("/api/graph/nodes/") == 0 && path.find("/connect-output") != std::string::npos && method == "POST") { + auto prefix = std::string("/api/graph/nodes/"); + auto suffix_start = path.find("/connect-output"); + auto node_id = path.substr(prefix.length(), suffix_start - prefix.length()); + + auto port_id = req_body.value("port_id", "video_out"); + + auto port_num = cc.get_port(node_id); + if (port_num == 0) { + error_resp(404, "Node not running or not found: " + node_id); + return; + } + + auto flow_id = fm.create_flow_id(); + auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1); + + nlohmann::json cmd; + cmd["cmd"] = "add_writer"; + cmd["port_id"] = port_id; + cmd["flow_id"] = flow_id; + cmd["flow_def"] = flow_def; + + if (cc.send_command(port_num, cmd.dump())) { + ok({{"node_id", node_id}, {"connected_output", port_id}, {"flow_id", flow_id}}); + } else { + error_resp(500, "Failed to send command to node"); + } + } else if (path.find("/api/graph/nodes/") == 0 && path.find("/disconnect-port") != std::string::npos && method == "POST") { + auto prefix = std::string("/api/graph/nodes/"); + auto suffix_start = path.find("/disconnect-port"); + auto node_id = path.substr(prefix.length(), suffix_start - prefix.length()); + + if (!req_body.contains("port_id")) { + error_resp(400, "Missing port_id"); + return; + } + + auto port_id = req_body["port_id"].get(); + auto port_num = cc.get_port(node_id); + if (port_num == 0) { + error_resp(404, "Node not running or not found: " + node_id); + return; + } + + auto direction = req_body.value("direction", "input"); + nlohmann::json cmd; + if (direction == "output") { + cmd["cmd"] = "remove_writer"; + } else { + cmd["cmd"] = "remove_reader"; + } + cmd["port_id"] = port_id; + + if (cc.send_command(port_num, cmd.dump())) { + ok({{"node_id", node_id}, {"disconnected", port_id}}); + } else { + error_resp(500, "Failed to send command to node"); + } + } else if (path.find("/api/graph/nodes/") == 0 && path.find("/command") != std::string::npos && method == "POST") { + auto prefix = std::string("/api/graph/nodes/"); + auto suffix_start = path.find("/command"); + auto node_id = path.substr(prefix.length(), suffix_start - prefix.length()); + + auto port_num = cc.get_port(node_id); + if (port_num == 0) { + error_resp(404, "Node not running or not found: " + node_id); + return; + } + + if (cc.send_command(port_num, req_body.dump())) { + ok({{"node_id", node_id}, {"sent", true}}); + } else { + error_resp(500, "Failed to send command to node"); + } } else { error_resp(404, "Not found: " + method + " " + path); } @@ -207,13 +384,15 @@ static const struct lws_protocols protocols[] = { {nullptr, nullptr, 0, 0}, }; -ApiServer::ApiServer(uint16_t port, Graph& graph, FlowManager& flow_manager, ProcessManager& process_manager) +ApiServer::ApiServer(uint16_t port, Graph& graph, FlowManager& flow_manager, ProcessManager& process_manager, NodeControlClient& control_client, const std::string& mxl_domain) : impl_(std::make_unique()) { impl_->data = std::make_unique(); impl_->data->port = port; impl_->data->graph = &graph; impl_->data->flow_manager = &flow_manager; impl_->data->process_manager = &process_manager; + impl_->data->control_client = &control_client; + impl_->data->mxl_domain = mxl_domain; g_impl = impl_->data.get(); diff --git a/libs/dmf-engine/src/flow_manager.cpp b/libs/dmf-engine/src/flow_manager.cpp index a89acb6..094708b 100644 --- a/libs/dmf-engine/src/flow_manager.cpp +++ b/libs/dmf-engine/src/flow_manager.cpp @@ -9,18 +9,29 @@ namespace dmf_engine { -FlowManager::FlowManager() { - std::random_device rd; - flow_counter_ = static_cast(rd()); -} +FlowManager::FlowManager() = default; FlowId FlowManager::create_flow_id() { + std::random_device rd; + std::mt19937 gen(rd()); + std::uniform_int_distribution dist(0, 0xFFFFFFFF); + + uint32_t a = dist(gen); + uint16_t b = dist(gen) & 0xFFFF; + uint16_t c = (dist(gen) & 0x0FFF) | 0x4000; + uint16_t d = (dist(gen) & 0x3FFF) | 0x8000; + uint32_t e1 = dist(gen); + uint16_t e2 = dist(gen) & 0xFFFF; + std::stringstream ss; - ss << std::hex << (++flow_counter_); + ss << std::hex << std::setfill('0'); + ss << std::setw(8) << a << "-"; + ss << std::setw(4) << b << "-"; + ss << std::setw(4) << c << "-"; + ss << std::setw(4) << d << "-"; + ss << std::setw(8) << e1 << std::setw(4) << e2; + auto id = ss.str(); - while (id.length() < 8) { - id = "0" + id; - } spdlog::info("FlowManager: created flow ID: {}", id); return id; } @@ -30,17 +41,22 @@ nlohmann::json FlowManager::create_v210_flow_def(const FlowId& flow_id, int widt nlohmann::json flow_def; flow_def["id"] = flow_id; - flow_def["version"] = "v1.0"; flow_def["label"] = "DMF Studio Flow " + flow_id; flow_def["description"] = "Auto-generated V210 flow"; + flow_def["format"] = "urn:x-nmos:format:video"; flow_def["grain_rate"] = {{"numerator", fps_numerator}, {"denominator", fps_denominator}}; - flow_def["format"] = "video/v210"; - flow_def["width"] = width; - flow_def["height"] = height; - flow_def["grain_size"] = grain_size; - flow_def["components"] = nlohmann::json::array(); + flow_def["media_type"] = "video/v210"; + flow_def["frame_width"] = width; + flow_def["frame_height"] = height; + flow_def["interlace_mode"] = "progressive"; + flow_def["colorspace"] = "BT709"; + flow_def["components"] = nlohmann::json::array({ + {{"name", "Y"}, {"width", width}, {"height", height}, {"bit_depth", 10}}, + {{"name", "Cb"}, {"width", width / 2}, {"height", height}, {"bit_depth", 10}}, + {{"name", "Cr"}, {"width", width / 2}, {"height", height}, {"bit_depth", 10}}, + }); flow_def["tags"] = nlohmann::json::object(); - flow_def["tags"]["grouphint"] = nlohmann::json::array({"DMF Studio:Video"}); + flow_def["tags"]["urn:x-nmos:tag:grouphint/v1.0"] = nlohmann::json::array({"DMF Studio:Video"}); return flow_def; } diff --git a/libs/dmf-engine/src/node_control_client.cpp b/libs/dmf-engine/src/node_control_client.cpp new file mode 100644 index 0000000..7ae62d4 --- /dev/null +++ b/libs/dmf-engine/src/node_control_client.cpp @@ -0,0 +1,88 @@ +#include + +#include + +#include +#include +#include +#include +#include + +#include + +namespace dmf_engine { + +void NodeControlClient::register_node(const std::string& node_id, uint16_t port) { + node_ports_[node_id] = port; + spdlog::info("NodeControlClient: registered node '{}' on port {}", node_id, port); +} + +void NodeControlClient::unregister_node(const std::string& node_id) { + node_ports_.erase(node_id); +} + +uint16_t NodeControlClient::get_port(const std::string& node_id) const { + auto it = node_ports_.find(node_id); + return it != node_ports_.end() ? it->second : 0; +} + +bool NodeControlClient::send_command(uint16_t port, const std::string& json_cmd) { + int fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + spdlog::error("NodeControlClient: socket() failed: {}", strerror(errno)); + return false; + } + + struct timeval tv; + tv.tv_sec = 2; + tv.tv_usec = 0; + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); + + struct sockaddr_in addr; + std::memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(port); + inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr); + + if (connect(fd, reinterpret_cast(&addr), sizeof(addr)) < 0) { + spdlog::error("NodeControlClient: connect to port {} failed: {}", port, strerror(errno)); + close(fd); + return false; + } + + std::ostringstream req; + req << "POST /cmd HTTP/1.1\r\n" + << "Host: 127.0.0.1:" << port << "\r\n" + << "Content-Type: application/json\r\n" + << "Content-Length: " << json_cmd.size() << "\r\n" + << "Connection: close\r\n" + << "\r\n" + << json_cmd; + + auto request = req.str(); + auto sent = write(fd, request.data(), request.size()); + if (sent != static_cast(request.size())) { + spdlog::error("NodeControlClient: write failed on port {}", port); + close(fd); + return false; + } + + char resp_buf[4096] = {}; + auto n = read(fd, resp_buf, sizeof(resp_buf) - 1); + close(fd); + + if (n <= 0) { + spdlog::error("NodeControlClient: read failed on port {}", port); + return false; + } + + std::string resp(resp_buf, n); + bool ok = resp.find("200 OK") != std::string::npos || resp.find("201 Created") != std::string::npos; + if (!ok) { + spdlog::warn("NodeControlClient: command failed on port {}: {}", port, resp.substr(0, 100)); + } + return ok; +} + +} // namespace dmf_engine diff --git a/libs/dmf-node/src/control_server.cpp b/libs/dmf-node/src/control_server.cpp index 7872757..cfb1d1d 100644 --- a/libs/dmf-node/src/control_server.cpp +++ b/libs/dmf-node/src/control_server.cpp @@ -29,92 +29,123 @@ struct ControlServer::Impl { struct lws_context* context = nullptr; }; -static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason, - void* user, void* in, size_t len); +static void dispatch_command(ControlServerData* data, const nlohmann::json& msg) { + if (!msg.contains("cmd")) { + spdlog::warn("Control: message missing 'cmd' field"); + return; + } + auto cmd = msg["cmd"].get(); + auto it = data->commands.find(cmd); + if (it != data->commands.end()) { + it->second(msg); + } else { + spdlog::warn("Control: unknown command '{}'", cmd); + } +} -static const struct lws_protocols protocols[] = { - { - "http-only", - callback_ws, - sizeof(ControlServerData*), - 0, - }, - { - "dmf-control", - callback_ws, - sizeof(ControlServerData*), - 65536, - }, - {nullptr, nullptr, 0, 0}, +static int send_http_json(struct lws* wsi, const std::string& status, const std::string& body) { + auto hdr = "HTTP/1.1 " + status + "\r\n" + "Content-Type: application/json\r\n" + "Content-Length: " + std::to_string(body.size()) + "\r\n" + "\r\n"; + std::vector buf(LWS_PRE + hdr.size() + body.size()); + std::memcpy(buf.data() + LWS_PRE, hdr.data(), hdr.size()); + std::memcpy(buf.data() + LWS_PRE + hdr.size(), body.data(), body.size()); + lws_write(wsi, buf.data() + LWS_PRE, hdr.size() + body.size(), LWS_WRITE_HTTP); + return lws_http_transaction_completed(wsi) ? -1 : 0; +} + +struct PerSession { + std::string http_body; + bool is_ws = false; + ControlServerData* data = nullptr; }; -static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason, - void* user, void* in, size_t len) { - auto** ppdata = static_cast(user); - ControlServerData* data = ppdata ? *ppdata : nullptr; +static int callback_all(struct lws* wsi, enum lws_callback_reasons reason, + void* user, void* in, size_t len) { + auto* ps = static_cast(user); switch (reason) { + case LWS_CALLBACK_HTTP: { + new (ps) PerSession(); + auto* vhost = lws_get_vhost(wsi); + ps->data = vhost ? static_cast(lws_vhost_user(vhost)) : nullptr; + + char* uri_ptr = nullptr; + int uri_len = 0; + int method = lws_http_get_uri_and_method(wsi, &uri_ptr, &uri_len); + std::string path(uri_ptr ? uri_ptr : "", uri_len > 0 ? uri_len : 0); + + if (path == "/cmd" && method == LWSHUMETH_POST) { + int cl = lws_hdr_total_length(wsi, WSI_TOKEN_HTTP_CONTENT_LENGTH); + if (cl > 0) { + ps->http_body.reserve(cl); + } + return 0; + } + + if (path == "/cmd" && (method == LWSHUMETH_GET || method == -1)) { + return send_http_json(wsi, "405 Method Not Allowed", R"({"error":"POST only"})"); + } + + return 0; + } + case LWS_CALLBACK_HTTP_BODY: { + ps->http_body.append(static_cast(in), len); + break; + } + case LWS_CALLBACK_HTTP_BODY_COMPLETION: { + if (!ps->data) { + return send_http_json(wsi, "500 Error", R"({"error":"no data"})"); + } + try { + auto msg = nlohmann::json::parse(ps->http_body); + dispatch_command(ps->data, msg); + return send_http_json(wsi, "200 OK", R"({"ok":true})"); + } catch (const nlohmann::json::parse_error& e) { + return send_http_json(wsi, "400 Bad Request", + nlohmann::json({{"error", e.what()}}).dump()); + } + } case LWS_CALLBACK_ESTABLISHED: { auto* vhost = lws_get_vhost(wsi); - data = static_cast(lws_vhost_user(vhost)); - if (ppdata) { - *ppdata = data; - } - if (data) { - data->client_wsi = wsi; + ps->data = vhost ? static_cast(lws_vhost_user(vhost)) : nullptr; + ps->is_ws = true; + if (ps->data) { + ps->data->client_wsi = wsi; } spdlog::info("Control WS: client connected"); break; } case LWS_CALLBACK_RECEIVE: { - if (!data) { - auto* vhost = lws_get_vhost(wsi); - data = static_cast(lws_vhost_user(vhost)); - } - if (!data) { + if (!ps->data) { break; } - try { auto msg = nlohmann::json::parse(static_cast(in), static_cast(in) + len); - if (!msg.contains("cmd")) { - spdlog::warn("Control WS: message missing 'cmd' field"); - break; - } - auto cmd = msg["cmd"].get(); - auto it = data->commands.find(cmd); - if (it != data->commands.end()) { - it->second(msg); - } else { - spdlog::warn("Control WS: unknown command '{}'", cmd); - } + dispatch_command(ps->data, msg); } catch (const nlohmann::json::parse_error& e) { spdlog::warn("Control WS: JSON parse error: {}", e.what()); } break; } case LWS_CALLBACK_SERVER_WRITEABLE: { - if (!data) { - auto* vhost = lws_get_vhost(wsi); - data = static_cast(lws_vhost_user(vhost)); - } - if (!data) { + if (!ps->data) { break; } - - std::lock_guard lock(data->send_mutex); - while (!data->send_queue.empty()) { - auto& msg = data->send_queue.back(); + std::lock_guard lock(ps->data->send_mutex); + while (!ps->data->send_queue.empty()) { + auto& msg = ps->data->send_queue.back(); std::vector buf(LWS_PRE + msg.size()); std::memcpy(buf.data() + LWS_PRE, msg.data(), msg.size()); lws_write(wsi, buf.data() + LWS_PRE, msg.size(), LWS_WRITE_TEXT); - data->send_queue.pop_back(); + ps->data->send_queue.pop_back(); } break; } case LWS_CALLBACK_CLOSED: { - if (data) { - data->client_wsi = nullptr; + if (ps->data) { + ps->data->client_wsi = nullptr; } spdlog::info("Control WS: client disconnected"); break; @@ -125,23 +156,59 @@ static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason, return 0; } -static const struct lws_http_mount mount = { - .mount_next = nullptr, - .mountpoint = "/", - .origin = "", - .def = "", - .protocol = "dmf-control", - .cgienv = nullptr, - .extra_mimetypes = nullptr, - .interpret = nullptr, - .cgi_timeout = 0, - .cache_max_age = 0, - .auth_mask = 0, - .cache_reusable = 0, - .cache_revalidate = 0, - .cache_intermediaries = 0, - .origin_protocol = LWSMPRO_CALLBACK, - .mountpoint_len = 1, +static const struct lws_protocols protocols[] = { + { + "http-only", + callback_all, + sizeof(PerSession), + 0, + }, + { + "dmf-control", + callback_all, + sizeof(PerSession), + 65536, + }, + {nullptr, nullptr, 0, 0}, +}; + +static const struct lws_http_mount mounts[] = { + { + .mount_next = &mounts[1], + .mountpoint = "/cmd", + .origin = "", + .def = "", + .protocol = "http-only", + .cgienv = nullptr, + .extra_mimetypes = nullptr, + .interpret = nullptr, + .cgi_timeout = 0, + .cache_max_age = 0, + .auth_mask = 0, + .cache_reusable = 0, + .cache_revalidate = 0, + .cache_intermediaries = 0, + .origin_protocol = LWSMPRO_CALLBACK, + .mountpoint_len = 4, + }, + { + .mount_next = nullptr, + .mountpoint = "/", + .origin = "", + .def = "", + .protocol = "dmf-control", + .cgienv = nullptr, + .extra_mimetypes = nullptr, + .interpret = nullptr, + .cgi_timeout = 0, + .cache_max_age = 0, + .auth_mask = 0, + .cache_reusable = 0, + .cache_revalidate = 0, + .cache_intermediaries = 0, + .origin_protocol = LWSMPRO_CALLBACK, + .mountpoint_len = 1, + }, }; ControlServer::ControlServer(uint16_t port, StatusCallback on_event) @@ -154,7 +221,7 @@ ControlServer::ControlServer(uint16_t port, StatusCallback on_event) std::memset(&info, 0, sizeof(info)); info.port = port; info.protocols = protocols; - info.mounts = &mount; + info.mounts = mounts; info.user = impl_->data.get(); info.gid = -1; info.uid = -1; From 3e1477c56c6db8d7ad27adba5924ad3dd32c213a Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 22:22:46 +0300 Subject: [PATCH 06/11] fix: LWS HTTP connection leak, grain index alignment, and reader head tracking - control_server: add Connection: close header + always return -1 after HTTP response to force-close connection. Without this, lws_service() blocks forever after the first POST /cmd, freezing the node process loop. - passthrough: initialize read_index_ to runtime.headIndex when reader is added (prevents TOO_EARLY errors on first read) - passthrough: handle MXL_ERR_OUT_OF_RANGE_TOO_EARLY in addition to TOO_LATE (both jump to head index) - passthrough: use grain_info.index as writer index instead of separate write_index_ counter (MXL writers must use TAI-based grain indices) - passthrough: reduce grain read timeout from 100ms to 20ms for tighter loop with LWS poll - node_runner: add 'status' command handler that sends node status back via control server --- libs/dmf-node/src/control_server.cpp | 4 +++- libs/dmf-node/src/node_runner.cpp | 8 +++++++ nodes/passthrough/src/passthrough_node.cpp | 28 ++++++++++++---------- 3 files changed, 26 insertions(+), 14 deletions(-) diff --git a/libs/dmf-node/src/control_server.cpp b/libs/dmf-node/src/control_server.cpp index cfb1d1d..acdd6b5 100644 --- a/libs/dmf-node/src/control_server.cpp +++ b/libs/dmf-node/src/control_server.cpp @@ -47,12 +47,14 @@ static int send_http_json(struct lws* wsi, const std::string& status, const std: auto hdr = "HTTP/1.1 " + status + "\r\n" "Content-Type: application/json\r\n" "Content-Length: " + std::to_string(body.size()) + "\r\n" + "Connection: close\r\n" "\r\n"; std::vector buf(LWS_PRE + hdr.size() + body.size()); std::memcpy(buf.data() + LWS_PRE, hdr.data(), hdr.size()); std::memcpy(buf.data() + LWS_PRE + hdr.size(), body.data(), body.size()); lws_write(wsi, buf.data() + LWS_PRE, hdr.size() + body.size(), LWS_WRITE_HTTP); - return lws_http_transaction_completed(wsi) ? -1 : 0; + lws_http_transaction_completed(wsi); + return -1; } struct PerSession { diff --git a/libs/dmf-node/src/node_runner.cpp b/libs/dmf-node/src/node_runner.cpp index 266e04c..974eda6 100644 --- a/libs/dmf-node/src/node_runner.cpp +++ b/libs/dmf-node/src/node_runner.cpp @@ -144,6 +144,14 @@ int NodeRunner::exec(std::unique_ptr node) { } }); + control_server->register_command("status", [&](const nlohmann::json& /*msg*/) { + nlohmann::json resp; + resp["event"] = "status"; + resp["node_id"] = node_id_; + resp["data"] = node->status(); + control_server->send_event(resp); + }); + control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) { spdlog::info("Shutdown command received"); g_running = false; diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index 1d52bba..ad9407d 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -19,7 +19,10 @@ void PassthroughNode::on_add_writer(const std::string& port_id, mxlFlowWriter wr void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) { if (port_id == "video_in") { reader_ = reader; - spdlog::info("Passthrough: reader added on video_in"); + mxlFlowRuntimeInfo runtime{}; + mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); + read_index_ = runtime.headIndex; + spdlog::info("Passthrough: reader added on video_in, starting at index {}", read_index_); } } @@ -45,26 +48,23 @@ void PassthroughNode::process() { mxlGrainInfo grain_info{}; uint8_t* payload = nullptr; - auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 100000000ULL, &grain_info, &payload); + auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload); if (status != MXL_STATUS_OK) { - if (status == MXL_ERR_TIMEOUT) { - return; - } - if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE) { + if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { mxlFlowRuntimeInfo runtime{}; mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); + spdlog::warn("Passthrough: index {} out of range, jumping to head {}", read_index_, runtime.headIndex); read_index_ = runtime.headIndex; - return; } return; } mxlGrainInfo out_grain{}; uint8_t* out_payload = nullptr; - status = mxlFlowWriterOpenGrain(*writer_, write_index_, &out_grain, &out_payload); + status = mxlFlowWriterOpenGrain(*writer_, grain_info.index, &out_grain, &out_payload); if (status != MXL_STATUS_OK) { - spdlog::warn("Passthrough: failed to open output grain at index {}: {}", write_index_, static_cast(status)); - read_index_++; + spdlog::warn("Passthrough: failed to open output grain at index {}: {}", grain_info.index, static_cast(status)); + read_index_ = grain_info.index + 1; return; } @@ -75,9 +75,12 @@ void PassthroughNode::process() { out_grain.flags = grain_info.flags; mxlFlowWriterCommitGrain(*writer_, &out_grain); - read_index_++; - write_index_++; + read_index_ = grain_info.index + 1; grains_processed_++; + + if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) { + spdlog::info("Passthrough: grain #{}, index={}, size={}", grains_processed_, grain_info.index, grain_info.grainSize); + } } nlohmann::json PassthroughNode::status() const { @@ -85,7 +88,6 @@ nlohmann::json PassthroughNode::status() const { {"type", "passthrough"}, {"grains_processed", grains_processed_}, {"read_index", read_index_}, - {"write_index", write_index_}, {"has_reader", reader_.has_value()}, {"has_writer", writer_.has_value()}, }; From 2c712592dde499588ed516b773105712063b9f7a Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 22:40:46 +0300 Subject: [PATCH 07/11] fix: use TAI-time-based grain index alignment in passthrough Instead of chasing headIndex from mxlFlowReaderGetRuntimeInfo (which points to the NEXT grain to be written, causing perpetual TOO_EARLY), the passthrough now uses mxlTimestampToIndex + mxlGetNsUntilIndex for proper timing alignment, matching the pattern used by mxl-gst-sink. Key changes: - realign() computes read_index from current TAI time minus 1 grain delay - process() uses mxlSleepForNs to wait until the target grain is due - on_add_reader fetches grain_rate from mxlFlowConfigInfo - Removed separate write_index_ (uses grain_info.index for writer) --- nodes/passthrough/src/passthrough_node.cpp | 34 ++++++++++++++++------ nodes/passthrough/src/passthrough_node.hpp | 7 +++-- 2 files changed, 29 insertions(+), 12 deletions(-) diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index ad9407d..8c08c97 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -2,6 +2,7 @@ #include #include +#include #include @@ -19,10 +20,14 @@ void PassthroughNode::on_add_writer(const std::string& port_id, mxlFlowWriter wr void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) { if (port_id == "video_in") { reader_ = reader; - mxlFlowRuntimeInfo runtime{}; - mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); - read_index_ = runtime.headIndex; - spdlog::info("Passthrough: reader added on video_in, starting at index {}", read_index_); + + mxlFlowConfigInfo config{}; + mxlFlowReaderGetConfigInfo(*reader_, &config); + grain_rate_ = config.common.grainRate; + + realign(); + spdlog::info("Passthrough: reader added on video_in, grain_rate={}/{}, read_index={}", + grain_rate_.numerator, grain_rate_.denominator, read_index_); } } @@ -40,21 +45,32 @@ void PassthroughNode::on_remove_reader(const std::string& port_id) { } } +void PassthroughNode::realign() { + auto now = mxlGetTime(); + auto current_index = mxlTimestampToIndex(&grain_rate_, now); + constexpr int64_t read_delay_grains = 1; + read_index_ = current_index - read_delay_grains; + aligned_ = true; +} + void PassthroughNode::process() { - if (!reader_ || !writer_) { + if (!reader_ || !writer_ || !aligned_) { return; } + auto wait_ns = mxlGetNsUntilIndex(read_index_, &grain_rate_); + if (wait_ns > 0 && wait_ns < 500000000ULL) { + mxlSleepForNs(wait_ns); + } + mxlGrainInfo grain_info{}; uint8_t* payload = nullptr; auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload); if (status != MXL_STATUS_OK) { if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { - mxlFlowRuntimeInfo runtime{}; - mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); - spdlog::warn("Passthrough: index {} out of range, jumping to head {}", read_index_, runtime.headIndex); - read_index_ = runtime.headIndex; + spdlog::warn("Passthrough: index {} out of range ({}), realigning", read_index_, static_cast(status)); + realign(); } return; } diff --git a/nodes/passthrough/src/passthrough_node.hpp b/nodes/passthrough/src/passthrough_node.hpp index ec4a6a4..9ee6769 100644 --- a/nodes/passthrough/src/passthrough_node.hpp +++ b/nodes/passthrough/src/passthrough_node.hpp @@ -4,7 +4,6 @@ #include -#include #include namespace dmf_node { @@ -34,12 +33,14 @@ public: nlohmann::json status() const override; private: + void realign(); + std::optional reader_; std::optional writer_; - mxlInstance mxl_instance_ = nullptr; + mxlRational grain_rate_{50, 1}; uint64_t read_index_ = 0; - uint64_t write_index_ = 0; uint64_t grains_processed_ = 0; + bool aligned_ = false; }; } // namespace dmf_node From 503023f3e50d515647de4515c52ce3db922fae01 Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 22:46:33 +0300 Subject: [PATCH 08/11] fix: use mxlSleepUntil delivery deadline pattern for grain timing Match the mxl-gst-sink pattern: sleep until the next grain's delivery deadline using mxlSleepUntil, then read with a short 5ms timeout. Uses 2-grain read delay (40ms at 50fps) for buffering headroom. Previous approach of mxlSleepForNs + 20ms GetGrain timeout caused the passthrough to fall behind: each iteration took ~25ms (poll+sleep+read), missing grains and perpetually chasing headIndex via realign. --- nodes/passthrough/src/passthrough_node.cpp | 48 +++++++++++++--------- nodes/passthrough/src/passthrough_node.hpp | 3 +- 2 files changed, 30 insertions(+), 21 deletions(-) diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index 8c08c97..e8acb1c 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -25,9 +25,14 @@ void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader re mxlFlowReaderGetConfigInfo(*reader_, &config); grain_rate_ = config.common.grainRate; - realign(); - spdlog::info("Passthrough: reader added on video_in, grain_rate={}/{}, read_index={}", - grain_rate_.numerator, grain_rate_.denominator, read_index_); + auto now = mxlGetTime(); + auto current_index = mxlTimestampToIndex(&grain_rate_, now); + read_index_ = current_index - READ_DELAY_GRAINS; + delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); + aligned_ = true; + + spdlog::info("Passthrough: reader added, grain_rate={}/{}, read_index={}, delay={} grains", + grain_rate_.numerator, grain_rate_.denominator, read_index_, READ_DELAY_GRAINS); } } @@ -45,32 +50,32 @@ void PassthroughNode::on_remove_reader(const std::string& port_id) { } } -void PassthroughNode::realign() { - auto now = mxlGetTime(); - auto current_index = mxlTimestampToIndex(&grain_rate_, now); - constexpr int64_t read_delay_grains = 1; - read_index_ = current_index - read_delay_grains; - aligned_ = true; -} - void PassthroughNode::process() { if (!reader_ || !writer_ || !aligned_) { return; } - auto wait_ns = mxlGetNsUntilIndex(read_index_, &grain_rate_); - if (wait_ns > 0 && wait_ns < 500000000ULL) { - mxlSleepForNs(wait_ns); + auto now = mxlGetTime(); + if (delivery_deadline_ > now) { + auto sleep_ns = delivery_deadline_ - now; + if (sleep_ns < 500000000ULL) { + mxlSleepUntil(delivery_deadline_); + } } mxlGrainInfo grain_info{}; uint8_t* payload = nullptr; - auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload); + auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 5000000ULL, &grain_info, &payload); if (status != MXL_STATUS_OK) { if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { - spdlog::warn("Passthrough: index {} out of range ({}), realigning", read_index_, static_cast(status)); - realign(); + auto old_index = read_index_; + now = mxlGetTime(); + auto current_index = mxlTimestampToIndex(&grain_rate_, now); + read_index_ = current_index - READ_DELAY_GRAINS; + delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); + spdlog::warn("Passthrough: index {} out of range ({}), realigned to {}", + old_index, static_cast(status), read_index_); } return; } @@ -79,8 +84,10 @@ void PassthroughNode::process() { uint8_t* out_payload = nullptr; status = mxlFlowWriterOpenGrain(*writer_, grain_info.index, &out_grain, &out_payload); if (status != MXL_STATUS_OK) { - spdlog::warn("Passthrough: failed to open output grain at index {}: {}", grain_info.index, static_cast(status)); - read_index_ = grain_info.index + 1; + spdlog::warn("Passthrough: failed to open output grain at index {}: {}", + grain_info.index, static_cast(status)); + read_index_++; + delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); return; } @@ -91,7 +98,8 @@ void PassthroughNode::process() { out_grain.flags = grain_info.flags; mxlFlowWriterCommitGrain(*writer_, &out_grain); - read_index_ = grain_info.index + 1; + read_index_++; + delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); grains_processed_++; if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) { diff --git a/nodes/passthrough/src/passthrough_node.hpp b/nodes/passthrough/src/passthrough_node.hpp index 9ee6769..8045c4a 100644 --- a/nodes/passthrough/src/passthrough_node.hpp +++ b/nodes/passthrough/src/passthrough_node.hpp @@ -33,12 +33,13 @@ public: nlohmann::json status() const override; private: - void realign(); + static constexpr int64_t READ_DELAY_GRAINS = 2; std::optional reader_; std::optional writer_; mxlRational grain_rate_{50, 1}; uint64_t read_index_ = 0; + uint64_t delivery_deadline_ = 0; uint64_t grains_processed_ = 0; bool aligned_ = false; }; From d677b654f0b3314c5f0a78490c76bac3386af6a2 Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 22:51:57 +0300 Subject: [PATCH 09/11] =?UTF-8?q?fix:=20non-blocking=20grain=20timing=20?= =?UTF-8?q?=E2=80=94=20use=20mxlGetNsUntilIndex=20instead=20of=20mxlSleepU?= =?UTF-8?q?ntil?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mxlSleepUntil blocks the entire thread including LWS poll, causing the control server to become unresponsive and grains to be missed. Instead, check mxlGetNsUntilIndex and skip process() if the grain isn't due yet (>2ms away). The node loop spins with poll(1) keeping LWS responsive, and only attempts a grain read when it's nearly due. --- nodes/passthrough/src/passthrough_node.cpp | 15 ++----- nodes/passthrough/src/passthrough_node.hpp | 1 - test.sh | 51 ++++++++++++++++++++++ 3 files changed, 55 insertions(+), 12 deletions(-) create mode 100755 test.sh diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index e8acb1c..244c049 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -28,7 +28,6 @@ void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader re auto now = mxlGetTime(); auto current_index = mxlTimestampToIndex(&grain_rate_, now); read_index_ = current_index - READ_DELAY_GRAINS; - delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); aligned_ = true; spdlog::info("Passthrough: reader added, grain_rate={}/{}, read_index={}, delay={} grains", @@ -55,12 +54,9 @@ void PassthroughNode::process() { return; } - auto now = mxlGetTime(); - if (delivery_deadline_ > now) { - auto sleep_ns = delivery_deadline_ - now; - if (sleep_ns < 500000000ULL) { - mxlSleepUntil(delivery_deadline_); - } + auto ns_until = mxlGetNsUntilIndex(read_index_, &grain_rate_); + if (ns_until > 2000000ULL) { + return; } mxlGrainInfo grain_info{}; @@ -70,10 +66,9 @@ void PassthroughNode::process() { if (status != MXL_STATUS_OK) { if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { auto old_index = read_index_; - now = mxlGetTime(); + auto now = mxlGetTime(); auto current_index = mxlTimestampToIndex(&grain_rate_, now); read_index_ = current_index - READ_DELAY_GRAINS; - delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); spdlog::warn("Passthrough: index {} out of range ({}), realigned to {}", old_index, static_cast(status), read_index_); } @@ -87,7 +82,6 @@ void PassthroughNode::process() { spdlog::warn("Passthrough: failed to open output grain at index {}: {}", grain_info.index, static_cast(status)); read_index_++; - delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); return; } @@ -99,7 +93,6 @@ void PassthroughNode::process() { mxlFlowWriterCommitGrain(*writer_, &out_grain); read_index_++; - delivery_deadline_ = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); grains_processed_++; if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) { diff --git a/nodes/passthrough/src/passthrough_node.hpp b/nodes/passthrough/src/passthrough_node.hpp index 8045c4a..c4e519a 100644 --- a/nodes/passthrough/src/passthrough_node.hpp +++ b/nodes/passthrough/src/passthrough_node.hpp @@ -39,7 +39,6 @@ private: std::optional writer_; mxlRational grain_rate_{50, 1}; uint64_t read_index_ = 0; - uint64_t delivery_deadline_ = 0; uint64_t grains_processed_ = 0; bool aligned_ = false; }; diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..838812e --- /dev/null +++ b/test.sh @@ -0,0 +1,51 @@ +#!/bin/bash +set -e + +ENGINE_PORT=${ENGINE_PORT:-9000} +MXL_DOMAIN=${MXL_DOMAIN:-/tmp/dmf-mxl} +FLOW_ID=${FLOW_ID:-a0000001-0000-0000-0000-000000000001} +BASE_URL="http://127.0.0.1:${ENGINE_PORT}" + +log() { echo "=== $1 ==="; } + +log "Adding passthrough node" +curl -s -X POST "${BASE_URL}/api/graph/nodes" \ + -H "Content-Type: application/json" \ + -d '{"type":"passthrough","id":"pass1"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Starting graph" +curl -s -X POST "${BASE_URL}/api/graph/start" | python3 -m json.tool 2>/dev/null || echo "" + +sleep 1 + +log "Connecting input to flow ${FLOW_ID}" +curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/connect-input" \ + -H "Content-Type: application/json" \ + -d "{\"port_id\":\"video_in\",\"flow_id\":\"${FLOW_ID}\"}" | python3 -m json.tool 2>/dev/null || echo "" + +log "Connecting output (creates new MXL flow)" +OUTPUT=$(curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/connect-output" \ + -H "Content-Type: application/json" \ + -d '{"port_id":"video_out"}') +echo "$OUTPUT" | python3 -m json.tool 2>/dev/null || echo "$OUTPUT" + +OUTPUT_FLOW_ID=$(echo "$OUTPUT" | python3 -c "import sys,json; print(json.load(sys.stdin)['flow_id'])" 2>/dev/null) + +if [ -n "$OUTPUT_FLOW_ID" ]; then + log "Output flow ID: ${OUTPUT_FLOW_ID}" + log "To verify with mxl-gst-sink:" + echo " mxl-gst-sink -d ${MXL_DOMAIN} -v ${OUTPUT_FLOW_ID}" +fi + +sleep 3 + +log "Checking MXL domain" +mxl-info --domain "${MXL_DOMAIN}" 2>/dev/null || true + +log "Sending status command" +curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/command" \ + -H "Content-Type: application/json" \ + -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo "" + +log "Graph state" +curl -s "${BASE_URL}/api/graph" | python3 -m json.tool 2>/dev/null || echo "" From 5b48d86c6e701835cf850e7c763ed6aa66bb14f0 Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 22:55:57 +0300 Subject: [PATCH 10/11] fix: separate grain processing thread from LWS event loop The fundamental issue: mxlSleepUntil/mxlSleepForNs blocks the entire thread, making LWS unresponsive and causing control commands to hang. Meanwhile, LWS poll(1) adds latency that makes grain timing unreliable. Solution: run node->process() in a dedicated thread that can freely use mxlSleepUntil for precise grain timing, while the main thread runs control_server->poll(10) for LWS event handling. Passthrough now uses mxlSleepUntil(deadline) matching the mxl-gst-sink Cursor pattern, with 2-grain read delay for buffering. --- libs/dmf-node/src/node_runner.cpp | 21 ++++++++++----------- nodes/passthrough/src/passthrough_node.cpp | 7 +++---- nodes/passthrough/src/passthrough_node.hpp | 1 + 3 files changed, 14 insertions(+), 15 deletions(-) diff --git a/libs/dmf-node/src/node_runner.cpp b/libs/dmf-node/src/node_runner.cpp index 974eda6..44f6ef2 100644 --- a/libs/dmf-node/src/node_runner.cpp +++ b/libs/dmf-node/src/node_runner.cpp @@ -170,20 +170,19 @@ int NodeRunner::exec(std::unique_ptr node) { running_ = true; spdlog::info("Node '{}' entering process loop", node_id_); - while (g_running && running_) { - control_server->poll(1); - node->process(); - - static int status_counter = 0; - if (++status_counter % 100 == 0) { - nlohmann::json status_event; - status_event["event"] = "status"; - status_event["node_id"] = node_id_; - status_event["data"] = node->status(); - control_server->send_event(status_event); + std::thread process_thread([&]() { + while (g_running && running_) { + node->process(); } + }); + + while (g_running && running_) { + control_server->poll(10); } + running_ = false; + process_thread.join(); + spdlog::info("Node '{}' shutting down", node_id_); mxlDestroyInstance(mxl_instance_); return 0; diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index 244c049..eebaf4a 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -51,13 +51,12 @@ void PassthroughNode::on_remove_reader(const std::string& port_id) { void PassthroughNode::process() { if (!reader_ || !writer_ || !aligned_) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); return; } - auto ns_until = mxlGetNsUntilIndex(read_index_, &grain_rate_); - if (ns_until > 2000000ULL) { - return; - } + auto deadline = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1); + mxlSleepUntil(deadline); mxlGrainInfo grain_info{}; uint8_t* payload = nullptr; diff --git a/nodes/passthrough/src/passthrough_node.hpp b/nodes/passthrough/src/passthrough_node.hpp index c4e519a..2b1b753 100644 --- a/nodes/passthrough/src/passthrough_node.hpp +++ b/nodes/passthrough/src/passthrough_node.hpp @@ -5,6 +5,7 @@ #include #include +#include namespace dmf_node { From 4b50f491720919d600ff0cc83e22cd5bbe07ad0e Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 26 May 2026 23:04:16 +0300 Subject: [PATCH 11/11] =?UTF-8?q?chore:=20phase=201=20cleanup=20=E2=80=94?= =?UTF-8?q?=20proper=20shutdown,=20flow=20resource=20cleanup,=20test=20fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - node_runner: track all readers/writers in flow_resources vector, release them via mxlReleaseFlowWriter/mxlReleaseFlowReader before destroying the MXL instance (fixes 'leaked flow writer' warning) - node_runner: remove_writer/remove_reader commands now also release the MXL flow resources, not just reset the node's optional<> - engine: call process_manager.stop_all(graph) on shutdown to kill child node processes (prevents orphaned passthrough processes) - graph: add get_node_mut() for process_manager to update node state - process_manager: implement stop_all(Graph&) that SIGTERMs all running node processes - passthrough: reduce logging to first-grain and realign-once only - tests: update flow format assertion to match NMOS (urn:x-nmos:format:video instead of video/v210) --- engine/src/main.cpp | 1 + libs/dmf-engine/include/dmf-engine/graph.hpp | 1 + .../include/dmf-engine/process_manager.hpp | 1 + libs/dmf-engine/src/graph.cpp | 5 +++ libs/dmf-engine/src/process_manager.cpp | 16 +++++++++ libs/dmf-node/src/node_runner.cpp | 36 ++++++++++++++++++- nodes/passthrough/src/passthrough_node.cpp | 10 +++--- tests/test_graph.cpp | 7 ++-- 8 files changed, 68 insertions(+), 9 deletions(-) diff --git a/engine/src/main.cpp b/engine/src/main.cpp index 424f33d..677c433 100644 --- a/engine/src/main.cpp +++ b/engine/src/main.cpp @@ -52,5 +52,6 @@ int main(int argc, char* argv[]) { } spdlog::info("DMF Studio Engine shutting down"); + process_manager.stop_all(graph); return 0; } diff --git a/libs/dmf-engine/include/dmf-engine/graph.hpp b/libs/dmf-engine/include/dmf-engine/graph.hpp index d88b068..50b1851 100644 --- a/libs/dmf-engine/include/dmf-engine/graph.hpp +++ b/libs/dmf-engine/include/dmf-engine/graph.hpp @@ -42,6 +42,7 @@ public: NodeId add_node(const std::string& type, const nlohmann::json& config = {}); bool remove_node(const NodeId& node_id); const GraphNode* get_node(const NodeId& node_id) const; + GraphNode* get_node_mut(const NodeId& node_id); std::vector get_nodes() const; EdgeId add_edge(const NodeId& from_node, const PortId& from_port, diff --git a/libs/dmf-engine/include/dmf-engine/process_manager.hpp b/libs/dmf-engine/include/dmf-engine/process_manager.hpp index 2574a3a..4017d7a 100644 --- a/libs/dmf-engine/include/dmf-engine/process_manager.hpp +++ b/libs/dmf-engine/include/dmf-engine/process_manager.hpp @@ -12,6 +12,7 @@ public: bool start_node(GraphNode& node, const std::string& mxl_domain, uint16_t base_port); bool stop_node(GraphNode& node); void stop_all(); + void stop_all(Graph& graph); bool is_running(const NodeId& node_id) const; diff --git a/libs/dmf-engine/src/graph.cpp b/libs/dmf-engine/src/graph.cpp index 1213b53..e871fea 100644 --- a/libs/dmf-engine/src/graph.cpp +++ b/libs/dmf-engine/src/graph.cpp @@ -51,6 +51,11 @@ const GraphNode* Graph::get_node(const NodeId& node_id) const { return it != nodes_.end() ? &it->second : nullptr; } +GraphNode* Graph::get_node_mut(const NodeId& node_id) { + auto it = nodes_.find(node_id); + return it != nodes_.end() ? &it->second : nullptr; +} + std::vector Graph::get_nodes() const { std::vector result; for (const auto& [_, node] : nodes_) { diff --git a/libs/dmf-engine/src/process_manager.cpp b/libs/dmf-engine/src/process_manager.cpp index 376ae94..b386e35 100644 --- a/libs/dmf-engine/src/process_manager.cpp +++ b/libs/dmf-engine/src/process_manager.cpp @@ -90,6 +90,22 @@ bool ProcessManager::stop_node(GraphNode& node) { void ProcessManager::stop_all() { } +void ProcessManager::stop_all(Graph& graph) { + for (auto& node : graph.get_nodes()) { + if (node.state == NodeState::Running && node.pid > 0) { + kill(node.pid, SIGTERM); + int status = 0; + waitpid(node.pid, &status, 0); + auto* mut_node = graph.get_node_mut(node.id); + if (mut_node) { + mut_node->pid = 0; + mut_node->state = NodeState::Stopped; + } + spdlog::info("ProcessManager: stopped node '{}' pid={}", node.id, node.pid); + } + } +} + bool ProcessManager::is_running(const NodeId& node_id) const { return false; } diff --git a/libs/dmf-node/src/node_runner.cpp b/libs/dmf-node/src/node_runner.cpp index 44f6ef2..1807b8c 100644 --- a/libs/dmf-node/src/node_runner.cpp +++ b/libs/dmf-node/src/node_runner.cpp @@ -94,7 +94,14 @@ int NodeRunner::exec(std::unique_ptr node) { return arr; }; - control_server->register_command("add_writer", [&, w = std::unordered_map{}](const nlohmann::json& msg) mutable { + struct FlowResource { + std::string port_id; + mxlFlowWriter writer = nullptr; + mxlFlowReader reader = nullptr; + }; + std::vector flow_resources; + + control_server->register_command("add_writer", [&](const nlohmann::json& msg) { auto flow_id = msg["flow_id"].get(); auto port_id = msg["port_id"].get(); auto flow_def = msg["flow_def"].dump(); @@ -108,6 +115,7 @@ int NodeRunner::exec(std::unique_ptr node) { return; } spdlog::info("Created flow writer on port '{}' flow {} (created={})", port_id, flow_id, created); + flow_resources.push_back({port_id, writer, nullptr}); node->on_add_writer(port_id, writer); }); @@ -122,18 +130,33 @@ int NodeRunner::exec(std::unique_ptr node) { return; } spdlog::info("Created flow reader on port '{}' flow {}", port_id, flow_id); + flow_resources.push_back({port_id, nullptr, reader}); node->on_add_reader(port_id, reader); }); control_server->register_command("remove_writer", [&](const nlohmann::json& msg) { auto port_id = msg["port_id"].get(); node->on_remove_writer(port_id); + for (auto it = flow_resources.begin(); it != flow_resources.end(); ++it) { + if (it->port_id == port_id && it->writer) { + mxlReleaseFlowWriter(mxl_instance_, it->writer); + flow_resources.erase(it); + break; + } + } spdlog::info("Removed writer on port '{}'", port_id); }); control_server->register_command("remove_reader", [&](const nlohmann::json& msg) { auto port_id = msg["port_id"].get(); node->on_remove_reader(port_id); + for (auto it = flow_resources.begin(); it != flow_resources.end(); ++it) { + if (it->port_id == port_id && it->reader) { + mxlReleaseFlowReader(mxl_instance_, it->reader); + flow_resources.erase(it); + break; + } + } spdlog::info("Removed reader on port '{}'", port_id); }); @@ -184,6 +207,17 @@ int NodeRunner::exec(std::unique_ptr node) { process_thread.join(); spdlog::info("Node '{}' shutting down", node_id_); + + for (auto& res : flow_resources) { + if (res.writer) { + mxlReleaseFlowWriter(mxl_instance_, res.writer); + } + if (res.reader) { + mxlReleaseFlowReader(mxl_instance_, res.reader); + } + } + flow_resources.clear(); + mxlDestroyInstance(mxl_instance_); return 0; } diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index eebaf4a..fc3416d 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -64,12 +64,12 @@ void PassthroughNode::process() { auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 5000000ULL, &grain_info, &payload); if (status != MXL_STATUS_OK) { if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { - auto old_index = read_index_; auto now = mxlGetTime(); auto current_index = mxlTimestampToIndex(&grain_rate_, now); read_index_ = current_index - READ_DELAY_GRAINS; - spdlog::warn("Passthrough: index {} out of range ({}), realigned to {}", - old_index, static_cast(status), read_index_); + if (grains_processed_ == 0) { + spdlog::warn("Passthrough: realigned to index {}", read_index_); + } } return; } @@ -94,8 +94,8 @@ void PassthroughNode::process() { read_index_++; grains_processed_++; - if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) { - spdlog::info("Passthrough: grain #{}, index={}, size={}", grains_processed_, grain_info.index, grain_info.grainSize); + if (grains_processed_ == 1) { + spdlog::info("Passthrough: first grain processed, index={}", grain_info.index); } } diff --git a/tests/test_graph.cpp b/tests/test_graph.cpp index edcdbc7..ea9b11f 100644 --- a/tests/test_graph.cpp +++ b/tests/test_graph.cpp @@ -68,9 +68,10 @@ TEST_CASE("FlowManager creates V210 flow definition", "[flow_manager]") { auto def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1); REQUIRE(def["id"] == flow_id); - REQUIRE(def["format"] == "video/v210"); - REQUIRE(def["width"] == 1920); - REQUIRE(def["height"] == 1080); + REQUIRE(def["format"] == "urn:x-nmos:format:video"); + REQUIRE(def["media_type"] == "video/v210"); + REQUIRE(def["frame_width"] == 1920); + REQUIRE(def["frame_height"] == 1080); } TEST_CASE("Graph serialize", "[graph]") {