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]") {