Johanness
4b50f49172
- 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)
113 lines
3.4 KiB
C++
113 lines
3.4 KiB
C++
#include "passthrough_node.hpp"
|
|
|
|
#include <mxl/flow.h>
|
|
#include <mxl/mxl.h>
|
|
#include <mxl/time.h>
|
|
|
|
#include <spdlog/spdlog.h>
|
|
|
|
#include <cstring>
|
|
|
|
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;
|
|
|
|
mxlFlowConfigInfo config{};
|
|
mxlFlowReaderGetConfigInfo(*reader_, &config);
|
|
grain_rate_ = config.common.grainRate;
|
|
|
|
auto now = mxlGetTime();
|
|
auto current_index = mxlTimestampToIndex(&grain_rate_, now);
|
|
read_index_ = current_index - READ_DELAY_GRAINS;
|
|
aligned_ = true;
|
|
|
|
spdlog::info("Passthrough: reader added, grain_rate={}/{}, read_index={}, delay={} grains",
|
|
grain_rate_.numerator, grain_rate_.denominator, read_index_, READ_DELAY_GRAINS);
|
|
}
|
|
}
|
|
|
|
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_ || !aligned_) {
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
|
return;
|
|
}
|
|
|
|
auto deadline = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
|
|
mxlSleepUntil(deadline);
|
|
|
|
mxlGrainInfo grain_info{};
|
|
uint8_t* payload = nullptr;
|
|
|
|
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 now = mxlGetTime();
|
|
auto current_index = mxlTimestampToIndex(&grain_rate_, now);
|
|
read_index_ = current_index - READ_DELAY_GRAINS;
|
|
if (grains_processed_ == 0) {
|
|
spdlog::warn("Passthrough: realigned to index {}", read_index_);
|
|
}
|
|
}
|
|
return;
|
|
}
|
|
|
|
mxlGrainInfo out_grain{};
|
|
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<int>(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_++;
|
|
grains_processed_++;
|
|
|
|
if (grains_processed_ == 1) {
|
|
spdlog::info("Passthrough: first grain processed, index={}", grain_info.index);
|
|
}
|
|
}
|
|
|
|
nlohmann::json PassthroughNode::status() const {
|
|
return {
|
|
{"type", "passthrough"},
|
|
{"grains_processed", grains_processed_},
|
|
{"read_index", read_index_},
|
|
{"has_reader", reader_.has_value()},
|
|
{"has_writer", writer_.has_value()},
|
|
};
|
|
}
|
|
|
|
} // namespace dmf_node
|