Johanness
d093d64ad1
The initiator couldn't start because start_requested_ was never set when target_host was provided without target_info. Now on_add_reader signals startup when either target_info_str_ or target_host_ is available.
449 lines
17 KiB
C++
449 lines
17 KiB
C++
#include "fabric_bridge_node.hpp"
|
|
|
|
#include <mxl/mxl.h>
|
|
|
|
#include <spdlog/spdlog.h>
|
|
|
|
#include <sys/socket.h>
|
|
#include <netinet/in.h>
|
|
#include <arpa/inet.h>
|
|
#include <unistd.h>
|
|
#include <netdb.h>
|
|
|
|
namespace dmf_node {
|
|
|
|
FabricBridgeNode::~FabricBridgeNode() {
|
|
running_ = false;
|
|
if (worker_thread_.has_value() && worker_thread_->joinable()) {
|
|
worker_thread_->join();
|
|
}
|
|
}
|
|
|
|
std::vector<PortDef> FabricBridgeNode::input_ports() const {
|
|
if (mode_ == FabricMode::Target) {
|
|
return {};
|
|
}
|
|
return {{"video_in", PortDirection::Input, MediaType::VideoV210}};
|
|
}
|
|
|
|
std::vector<PortDef> FabricBridgeNode::output_ports() const {
|
|
if (mode_ == FabricMode::Initiator) {
|
|
return {};
|
|
}
|
|
return {{"video_out", PortDirection::Output, MediaType::VideoV210}};
|
|
}
|
|
|
|
void FabricBridgeNode::configure(const nlohmann::json& params) {
|
|
if (params.contains("mode")) {
|
|
auto mode_str = params["mode"].get<std::string>();
|
|
if (mode_str == "initiator") mode_ = FabricMode::Initiator;
|
|
else if (mode_str == "target") mode_ = FabricMode::Target;
|
|
else spdlog::warn("Fabric-bridge: unknown mode '{}', defaulting to initiator", mode_str);
|
|
}
|
|
|
|
if (params.contains("bind_node")) bind_node_ = params["bind_node"].get<std::string>();
|
|
if (params.contains("bind_service")) bind_service_ = params["bind_service"].get<std::string>();
|
|
if (params.contains("provider")) {
|
|
auto prov_str = params["provider"].get<std::string>();
|
|
mxlFabricsProviderFromString(prov_str.c_str(), &provider_);
|
|
}
|
|
|
|
if (params.contains("target_info")) {
|
|
target_info_str_ = params["target_info"].get<std::string>();
|
|
if (mode_ == FabricMode::Initiator && reader_.has_value() && !target_info_str_.empty()) {
|
|
start_requested_.store(true);
|
|
spdlog::info("Fabric-bridge: target_info set, signaling startup");
|
|
}
|
|
}
|
|
|
|
if (params.contains("target_host")) {
|
|
target_host_ = params["target_host"].get<std::string>();
|
|
target_port_ = params.value("target_port", 9100);
|
|
spdlog::info("Fabric-bridge: will fetch target_info from {}:{}", target_host_, target_port_);
|
|
}
|
|
}
|
|
|
|
void FabricBridgeNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
|
|
if (port_id == "video_out") {
|
|
writer_ = writer;
|
|
start_requested_.store(true);
|
|
spdlog::info("Fabric-bridge: writer added (target mode), signaling startup");
|
|
}
|
|
}
|
|
|
|
void FabricBridgeNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
|
|
if (port_id == "video_in") {
|
|
reader_ = reader;
|
|
if (mode_ == FabricMode::Initiator) {
|
|
if (!target_info_str_.empty() || !target_host_.empty()) {
|
|
start_requested_.store(true);
|
|
spdlog::info("Fabric-bridge: reader added (initiator mode), signaling startup");
|
|
} else {
|
|
spdlog::info("Fabric-bridge: reader added (initiator mode), waiting for target_info");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void FabricBridgeNode::on_remove_writer(const std::string& port_id) {
|
|
if (port_id == "video_out") {
|
|
writer_.reset();
|
|
spdlog::info("Fabric-bridge: writer removed");
|
|
}
|
|
}
|
|
|
|
void FabricBridgeNode::on_remove_reader(const std::string& port_id) {
|
|
if (port_id == "video_in") {
|
|
reader_.reset();
|
|
spdlog::info("Fabric-bridge: reader removed");
|
|
}
|
|
}
|
|
|
|
void FabricBridgeNode::fetch_target_info() {
|
|
if (target_host_.empty()) return;
|
|
|
|
spdlog::info("Fabric-bridge: fetching target_info from {}:{}", target_host_, target_port_);
|
|
|
|
std::string request = "POST /cmd HTTP/1.1\r\n"
|
|
"Host: " + target_host_ + "\r\n"
|
|
"Content-Type: application/json\r\n"
|
|
"Connection: close\r\n"
|
|
"Content-Length: 20\r\n"
|
|
"\r\n"
|
|
"{\"cmd\":\"status\"}";
|
|
|
|
for (int attempt = 0; attempt < 30 && running_ && target_info_str_.empty(); ++attempt) {
|
|
int sock = socket(AF_INET, SOCK_STREAM, 0);
|
|
if (sock < 0) {
|
|
spdlog::warn("Fabric-bridge: socket() failed, retry {}...", attempt + 1);
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
|
continue;
|
|
}
|
|
|
|
struct hostent* he = gethostbyname(target_host_.c_str());
|
|
if (!he) {
|
|
spdlog::warn("Fabric-bridge: cannot resolve {}, retry {}...", target_host_, attempt + 1);
|
|
close(sock);
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
|
continue;
|
|
}
|
|
|
|
struct sockaddr_in addr{};
|
|
addr.sin_family = AF_INET;
|
|
addr.sin_port = htons(target_port_);
|
|
std::memcpy(&addr.sin_addr, he->h_addr_list[0], he->h_length);
|
|
|
|
if (connect(sock, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) < 0) {
|
|
spdlog::warn("Fabric-bridge: connect to {}:{} failed, retry {}...", target_host_, target_port_, attempt + 1);
|
|
close(sock);
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
|
continue;
|
|
}
|
|
|
|
if (send(sock, request.c_str(), request.size(), 0) < 0) {
|
|
spdlog::warn("Fabric-bridge: send failed, retry {}...", attempt + 1);
|
|
close(sock);
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
|
continue;
|
|
}
|
|
|
|
std::string response;
|
|
char buf[4096];
|
|
ssize_t n;
|
|
while ((n = recv(sock, buf, sizeof(buf), 0)) > 0) {
|
|
response.append(buf, n);
|
|
}
|
|
close(sock);
|
|
|
|
auto body_pos = response.find("\r\n\r\n");
|
|
if (body_pos == std::string::npos) {
|
|
spdlog::warn("Fabric-bridge: no HTTP body in response, retry {}...", attempt + 1);
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
|
continue;
|
|
}
|
|
|
|
try {
|
|
auto json = nlohmann::json::parse(response.substr(body_pos + 4));
|
|
if (json.contains("data") && json["data"].contains("target_info")) {
|
|
auto ti = json["data"]["target_info"].get<std::string>();
|
|
if (!ti.empty()) {
|
|
target_info_str_ = ti;
|
|
start_requested_.store(true);
|
|
spdlog::info("Fabric-bridge: fetched target_info ({} bytes)", target_info_str_.size());
|
|
return;
|
|
}
|
|
}
|
|
} catch (const nlohmann::json::exception& e) {
|
|
spdlog::warn("Fabric-bridge: JSON parse error: {}", e.what());
|
|
}
|
|
|
|
spdlog::info("Fabric-bridge: target_info not ready yet, retry {}...", attempt + 1);
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
|
}
|
|
|
|
if (target_info_str_.empty()) {
|
|
spdlog::error("Fabric-bridge: failed to fetch target_info after 30 attempts");
|
|
}
|
|
}
|
|
|
|
void FabricBridgeNode::process() {
|
|
if (!worker_thread_.has_value() && start_requested_.load()) {
|
|
if (mode_ == FabricMode::Initiator && !target_host_.empty() && target_info_str_.empty()) {
|
|
fetch_target_info();
|
|
}
|
|
if (start_requested_.load()) {
|
|
running_ = true;
|
|
spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target");
|
|
if (mode_ == FabricMode::Initiator) {
|
|
worker_thread_ = std::thread(&FabricBridgeNode::run_initiator, this);
|
|
} else {
|
|
worker_thread_ = std::thread(&FabricBridgeNode::run_target, this);
|
|
}
|
|
}
|
|
}
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(100));
|
|
}
|
|
|
|
void FabricBridgeNode::run_initiator() {
|
|
spdlog::info("Fabric-bridge: starting initiator mode");
|
|
|
|
auto instance = get_mxl_instance();
|
|
if (!instance) {
|
|
spdlog::error("Fabric-bridge: no MXL instance");
|
|
running_ = false;
|
|
return;
|
|
}
|
|
|
|
spdlog::info("Fabric-bridge: creating fabrics instance...");
|
|
mxlFabricsInstance fab_inst = nullptr;
|
|
auto status = mxlFabricsCreateInstance(instance, nullptr, &fab_inst);
|
|
spdlog::info("Fabric-bridge: mxlFabricsCreateInstance returned {}", static_cast<int>(status));
|
|
if (status != MXL_STATUS_OK || !fab_inst) {
|
|
spdlog::error("Fabric-bridge: failed to create fabrics instance: {}", static_cast<int>(status));
|
|
running_ = false;
|
|
return;
|
|
}
|
|
fabrics_instance_ = fab_inst;
|
|
|
|
spdlog::info("Fabric-bridge: creating initiator...");
|
|
mxlFabricsInitiator initiator = nullptr;
|
|
status = mxlFabricsCreateInitiator(fab_inst, &initiator);
|
|
spdlog::info("Fabric-bridge: mxlFabricsCreateInitiator returned {}", static_cast<int>(status));
|
|
if (status != MXL_STATUS_OK || !initiator) {
|
|
spdlog::error("Fabric-bridge: failed to create initiator: {}", static_cast<int>(status));
|
|
mxlFabricsDestroyInstance(fab_inst);
|
|
running_ = false;
|
|
return;
|
|
}
|
|
initiator_ = initiator;
|
|
|
|
mxlFlowConfigInfo config{};
|
|
mxlFlowReaderGetConfigInfo(*reader_, &config);
|
|
grain_rate_ = config.common.grainRate;
|
|
|
|
mxlFabricsInitiatorConfig init_cfg{};
|
|
init_cfg.version = MXL_FABRICS_API_VERSION;
|
|
init_cfg.interface.version = MXL_FABRICS_API_VERSION;
|
|
init_cfg.interface.provider = provider_;
|
|
init_cfg.interface.address.node = bind_node_.c_str();
|
|
init_cfg.interface.address.service = bind_service_.c_str();
|
|
init_cfg.reader = *reader_;
|
|
|
|
status = mxlFabricsInitiatorSetup(initiator, &init_cfg, nullptr);
|
|
if (status != MXL_STATUS_OK) {
|
|
spdlog::error("Fabric-bridge: initiator setup failed: {}", static_cast<int>(status));
|
|
mxlFabricsDestroyInitiator(fab_inst, initiator);
|
|
mxlFabricsDestroyInstance(fab_inst);
|
|
running_ = false;
|
|
return;
|
|
}
|
|
|
|
mxlFabricsTargetInfo target_info = nullptr;
|
|
status = mxlFabricsTargetInfoFromString(target_info_str_.c_str(), &target_info);
|
|
if (status != MXL_STATUS_OK || !target_info) {
|
|
spdlog::error("Fabric-bridge: failed to parse target_info: {}", static_cast<int>(status));
|
|
mxlFabricsDestroyInitiator(fab_inst, initiator);
|
|
mxlFabricsDestroyInstance(fab_inst);
|
|
running_ = false;
|
|
return;
|
|
}
|
|
|
|
status = mxlFabricsInitiatorAddTarget(initiator, target_info);
|
|
if (status != MXL_STATUS_OK) {
|
|
spdlog::error("Fabric-bridge: failed to add target: {}", static_cast<int>(status));
|
|
mxlFabricsFreeTargetInfo(target_info);
|
|
mxlFabricsDestroyInitiator(fab_inst, initiator);
|
|
mxlFabricsDestroyInstance(fab_inst);
|
|
running_ = false;
|
|
return;
|
|
}
|
|
|
|
spdlog::info("Fabric-bridge: waiting for connection...");
|
|
do {
|
|
status = mxlFabricsInitiatorMakeProgressBlocking(initiator, 250);
|
|
} while (status == MXL_ERR_NOT_READY && running_);
|
|
spdlog::info("Fabric-bridge: connected");
|
|
|
|
auto now = mxlGetTime();
|
|
read_index_ = mxlTimestampToIndex(&grain_rate_, now) - 2;
|
|
|
|
spdlog::info("Fabric-bridge: initiator started, grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator);
|
|
|
|
while (running_) {
|
|
auto deadline = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
|
|
mxlSleepUntil(deadline);
|
|
|
|
mxlGrainInfo grain_info{};
|
|
uint8_t* payload = nullptr;
|
|
auto grain_status = mxlFlowReaderGetGrain(*reader_, read_index_, 5000000ULL, &grain_info, &payload);
|
|
if (grain_status != MXL_STATUS_OK) {
|
|
if (grain_status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || grain_status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
|
|
auto ts = mxlGetTime();
|
|
read_index_ = mxlTimestampToIndex(&grain_rate_, ts) - 2;
|
|
continue;
|
|
}
|
|
read_index_++;
|
|
continue;
|
|
}
|
|
|
|
auto transfer_status = mxlFabricsInitiatorTransferGrain(initiator, read_index_, 0, grain_info.validSlices);
|
|
if (transfer_status != MXL_STATUS_OK) {
|
|
spdlog::warn("Fabric-bridge: transfer grain {} failed: {}", read_index_, static_cast<int>(transfer_status));
|
|
read_index_++;
|
|
continue;
|
|
}
|
|
|
|
do {
|
|
status = mxlFabricsInitiatorMakeProgressBlocking(initiator, 10);
|
|
} while (status == MXL_ERR_NOT_READY && running_);
|
|
|
|
read_index_++;
|
|
grains_transferred_++;
|
|
|
|
if (grains_transferred_ == 1) {
|
|
spdlog::info("Fabric-bridge: first grain transferred, index={}", read_index_ - 1);
|
|
}
|
|
}
|
|
|
|
mxlFabricsInitiatorRemoveTarget(initiator, target_info);
|
|
mxlFabricsFreeTargetInfo(target_info);
|
|
mxlFabricsDestroyInitiator(fab_inst, initiator);
|
|
mxlFabricsDestroyInstance(fab_inst);
|
|
spdlog::info("Fabric-bridge: initiator stopped");
|
|
}
|
|
|
|
void FabricBridgeNode::run_target() {
|
|
spdlog::info("Fabric-bridge: starting target mode");
|
|
|
|
auto instance = get_mxl_instance();
|
|
if (!instance) {
|
|
spdlog::error("Fabric-bridge: no MXL instance");
|
|
running_ = false;
|
|
return;
|
|
}
|
|
|
|
spdlog::info("Fabric-bridge: creating fabrics instance...");
|
|
mxlFabricsInstance fab_inst = nullptr;
|
|
auto status = mxlFabricsCreateInstance(instance, nullptr, &fab_inst);
|
|
spdlog::info("Fabric-bridge: mxlFabricsCreateInstance returned {}", static_cast<int>(status));
|
|
if (status != MXL_STATUS_OK || !fab_inst) {
|
|
spdlog::error("Fabric-bridge: failed to create fabrics instance: {}", static_cast<int>(status));
|
|
running_ = false;
|
|
return;
|
|
}
|
|
fabrics_instance_ = fab_inst;
|
|
|
|
spdlog::info("Fabric-bridge: creating target...");
|
|
mxlFabricsTarget tgt = nullptr;
|
|
status = mxlFabricsCreateTarget(fab_inst, &tgt);
|
|
spdlog::info("Fabric-bridge: mxlFabricsCreateTarget returned {}", static_cast<int>(status));
|
|
if (status != MXL_STATUS_OK || !tgt) {
|
|
spdlog::error("Fabric-bridge: failed to create target: {}", static_cast<int>(status));
|
|
mxlFabricsDestroyInstance(fab_inst);
|
|
running_ = false;
|
|
return;
|
|
}
|
|
target_ = tgt;
|
|
|
|
auto* config = get_writer_config("video_out");
|
|
if (config) {
|
|
grain_rate_ = config->common.grainRate;
|
|
spdlog::info("Fabric-bridge: target grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator);
|
|
} else {
|
|
spdlog::warn("Fabric-bridge: no writer config, using default grain_rate");
|
|
}
|
|
|
|
mxlFabricsTargetConfig target_cfg{};
|
|
target_cfg.version = MXL_FABRICS_API_VERSION;
|
|
target_cfg.interface.version = MXL_FABRICS_API_VERSION;
|
|
target_cfg.interface.provider = provider_;
|
|
target_cfg.interface.address.node = bind_node_.c_str();
|
|
target_cfg.interface.address.service = bind_service_.c_str();
|
|
target_cfg.writer = *writer_;
|
|
|
|
spdlog::info("Fabric-bridge: calling mxlFabricsTargetSetup on {}:{}", bind_node_, bind_service_);
|
|
mxlFabricsTargetInfo target_info = nullptr;
|
|
status = mxlFabricsTargetSetup(tgt, &target_cfg, nullptr, &target_info);
|
|
if (status != MXL_STATUS_OK || !target_info) {
|
|
spdlog::error("Fabric-bridge: target setup failed: {}", static_cast<int>(status));
|
|
mxlFabricsDestroyTarget(fab_inst, tgt);
|
|
mxlFabricsDestroyInstance(fab_inst);
|
|
running_ = false;
|
|
return;
|
|
}
|
|
|
|
size_t info_size = 0;
|
|
mxlFabricsTargetInfoToString(target_info, nullptr, &info_size);
|
|
std::vector<char> info_buf(info_size);
|
|
mxlFabricsTargetInfoToString(target_info, info_buf.data(), &info_size);
|
|
target_info_str_ = std::string(info_buf.data(), info_size);
|
|
|
|
spdlog::info("Fabric-bridge: target ready, target_info length={}", target_info_str_.size());
|
|
spdlog::info("Fabric-bridge: TARGET_INFO={}", target_info_str_);
|
|
|
|
while (running_) {
|
|
uint64_t grain_index = 0;
|
|
status = mxlFabricsTargetReadGrain(tgt, 200, &grain_index);
|
|
if (status == MXL_ERR_TIMEOUT || status == MXL_ERR_NOT_READY) {
|
|
continue;
|
|
}
|
|
if (status != MXL_STATUS_OK) {
|
|
spdlog::warn("Fabric-bridge: target read grain failed: {}", static_cast<int>(status));
|
|
continue;
|
|
}
|
|
|
|
mxlGrainInfo grain_info{};
|
|
uint8_t* dummy_payload = nullptr;
|
|
auto open_status = mxlFlowWriterOpenGrain(*writer_, grain_index, &grain_info, &dummy_payload);
|
|
if (open_status == MXL_STATUS_OK) {
|
|
grain_info.validSlices = grain_info.totalSlices;
|
|
mxlFlowWriterCommitGrain(*writer_, &grain_info);
|
|
grains_transferred_++;
|
|
if (grains_transferred_ == 1) {
|
|
spdlog::info("Fabric-bridge: first grain committed, index={}", grain_index);
|
|
}
|
|
} else {
|
|
spdlog::warn("Fabric-bridge: open grain {} failed: {}", grain_index, static_cast<int>(open_status));
|
|
}
|
|
}
|
|
|
|
mxlFabricsFreeTargetInfo(target_info);
|
|
mxlFabricsDestroyTarget(fab_inst, tgt);
|
|
mxlFabricsDestroyInstance(fab_inst);
|
|
spdlog::info("Fabric-bridge: target stopped");
|
|
}
|
|
|
|
nlohmann::json FabricBridgeNode::status() const {
|
|
return {
|
|
{"type", "fabric-bridge"},
|
|
{"mode", mode_ == FabricMode::Initiator ? "initiator" : "target"},
|
|
{"grains_transferred", grains_transferred_},
|
|
{"running", running_.load()},
|
|
{"has_reader", reader_.has_value()},
|
|
{"has_writer", writer_.has_value()},
|
|
{"target_info_length", target_info_str_.size()},
|
|
{"target_info", target_info_str_},
|
|
};
|
|
}
|
|
|
|
} // namespace dmf_node
|