6 Commits

Author SHA1 Message Date
Johanness 12896182c9 fix: correct HTTP request in fetch_target_info, add socket timeouts and debug logging
- Content-Length was wrong (20 vs actual 18 bytes)
- Removed running_ check from loop (was false during startup)
- Added SO_RCVTIMEO/SO_SNDTIMEO to prevent infinite recv hangs
- Added response size logging for debugging
2026-06-16 11:38:48 +03:00
Johanness d093d64ad1 fix: trigger startup on add_reader when target_host is set
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.
2026-06-16 11:29:23 +03:00
Johanness e6e439b16e feat: add node register endpoint and fabric create-target/connect APIs
- POST /api/graph/nodes/{id}/register: register manually-started node with control port
- POST /api/fabric/create-target: add_writer + poll for target_info
- POST /api/fabric/connect: auto-propagate target_info to initiator
- fabric-bridge: auto-fetch target_info from remote target via HTTP
2026-06-16 11:11:13 +03:00
Johanness 0944f574ad fabric-bridge: remove unnecessary win32 includes 2026-06-16 11:09:10 +03:00
Johanness cca0cd810a feat: fabric-bridge auto-fetch target_info from remote target node
Initiator can now auto-discover target_info by polling the target's
HTTP control server. Just set target_host and target_port in config:
  --config '{"mode":"initiator",...,"target_host":"10.10.100.1","target_port":9100}'

No more manual copy-paste of target_info JSON between hosts.
2026-06-16 11:08:06 +03:00
Johanness ac3a506a5c feat: add engine API for fabric-bridge target_info exchange
- POST /api/fabric/create-target: add_writer to target node, poll for target_info
- POST /api/fabric/connect: get target_info from target, configure initiator node
2026-06-16 10:50:04 +03:00
3 changed files with 311 additions and 12 deletions
+167
View File
@@ -12,6 +12,7 @@
#include <cstring> #include <cstring>
#include <string> #include <string>
#include <thread>
namespace dmf_engine { namespace dmf_engine {
@@ -320,6 +321,172 @@ static void handle_request(const std::string& method, const std::string& path,
} else { } else {
error_resp(500, "Failed to send command to node"); error_resp(500, "Failed to send command to node");
} }
} else if (path.find("/api/graph/nodes/") == 0 && path.find("/register") != std::string::npos && method == "POST") {
auto prefix = std::string("/api/graph/nodes/");
auto suffix_start = path.find("/register");
auto node_id = path.substr(prefix.length(), suffix_start - prefix.length());
if (!req_body.contains("port")) {
error_resp(400, "Missing port");
return;
}
auto control_port = static_cast<uint16_t>(req_body["port"].get<int>());
cc.register_node(node_id, control_port);
auto* node = graph.get_node_mut(node_id);
if (node) {
node->control_port = control_port;
node->state = NodeState::Running;
}
ok({{"node_id", node_id}, {"control_port", control_port}});
} else if (path == "/api/fabric/create-target" && method == "POST") {
if (!req_body.contains("node_id")) {
error_resp(400, "Missing node_id");
return;
}
auto node_id = req_body["node_id"].get<std::string>();
auto port_id = req_body.value("port_id", "video_out");
auto flow_id = req_body.value("flow_id", fm.create_flow_id());
auto port_num = cc.get_port(node_id);
if (port_num == 0) {
error_resp(404, "Node not running: " + node_id);
return;
}
int fps_num = 25, fps_den = 1;
int width = 1920, height = 1080;
nlohmann::json status_cmd;
status_cmd["cmd"] = "status";
auto status_resp = cc.send_command_with_response(port_num, status_cmd.dump());
if (!status_resp.empty()) {
try {
auto sr = nlohmann::json::parse(status_resp);
if (sr.contains("data")) {
auto& d = sr["data"];
if (d.contains("grain_rate")) {
fps_num = d["grain_rate"].value("numerator", fps_num);
fps_den = d["grain_rate"].value("denominator", fps_den);
}
width = d.value("width", width);
height = d.value("height", height);
}
} catch (...) {}
}
auto flow_def = fm.create_v210_flow_def(flow_id, width, height, fps_num, fps_den);
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())) {
error_resp(500, "Failed to add writer to target node");
return;
}
std::string target_info;
constexpr int max_poll_attempts = 30;
for (int i = 0; i < max_poll_attempts; ++i) {
auto resp = cc.send_command_with_response(port_num, status_cmd.dump());
if (!resp.empty()) {
try {
auto sr = nlohmann::json::parse(resp);
if (sr.contains("data") && sr["data"].contains("target_info") && !sr["data"]["target_info"].get<std::string>().empty()) {
target_info = sr["data"]["target_info"].get<std::string>();
break;
}
} catch (...) {}
}
std::this_thread::sleep_for(std::chrono::milliseconds(500));
}
if (target_info.empty()) {
error_resp(504, "Target node did not produce target_info in time");
return;
}
ok({
{"node_id", node_id},
{"flow_id", flow_id},
{"target_info", target_info}
});
} else if (path == "/api/fabric/connect" && method == "POST") {
if (!req_body.contains("target_node_id") || !req_body.contains("initiator_node_id")) {
error_resp(400, "Missing target_node_id/initiator_node_id");
return;
}
auto target_node_id = req_body["target_node_id"].get<std::string>();
auto initiator_node_id = req_body["initiator_node_id"].get<std::string>();
auto target_port = cc.get_port(target_node_id);
auto initiator_port = cc.get_port(initiator_node_id);
if (target_port == 0) {
error_resp(404, "Target node not running: " + target_node_id);
return;
}
if (initiator_port == 0) {
error_resp(404, "Initiator node not running: " + initiator_node_id);
return;
}
nlohmann::json status_cmd;
status_cmd["cmd"] = "status";
std::string target_info;
constexpr int max_poll_attempts = 30;
for (int i = 0; i < max_poll_attempts; ++i) {
auto resp = cc.send_command_with_response(target_port, status_cmd.dump());
if (!resp.empty()) {
try {
auto sr = nlohmann::json::parse(resp);
if (sr.contains("data") && sr["data"].contains("target_info") && !sr["data"]["target_info"].get<std::string>().empty()) {
target_info = sr["data"]["target_info"].get<std::string>();
break;
}
} catch (...) {}
}
std::this_thread::sleep_for(std::chrono::milliseconds(500));
}
if (target_info.empty()) {
error_resp(504, "Target node did not produce target_info in time");
return;
}
nlohmann::json configure_cmd;
configure_cmd["cmd"] = "configure";
configure_cmd["params"]["target_info"] = target_info;
if (!cc.send_command(initiator_port, configure_cmd.dump())) {
error_resp(500, "Failed to send target_info to initiator node");
return;
}
std::this_thread::sleep_for(std::chrono::milliseconds(200));
bool initiator_running = false;
auto init_resp = cc.send_command_with_response(initiator_port, status_cmd.dump());
if (!init_resp.empty()) {
try {
auto ir = nlohmann::json::parse(init_resp);
if (ir.contains("data") && ir["data"].contains("running")) {
initiator_running = ir["data"]["running"].get<bool>();
}
} catch (...) {}
}
ok({
{"target_node_id", target_node_id},
{"initiator_node_id", initiator_node_id},
{"target_info_length", target_info.size()},
{"initiator_running", initiator_running}
});
} else if (path.find("/api/graph/nodes/") == 0 && path.find("/command") != std::string::npos && method == "POST") { } else if (path.find("/api/graph/nodes/") == 0 && path.find("/command") != std::string::npos && method == "POST") {
auto prefix = std::string("/api/graph/nodes/"); auto prefix = std::string("/api/graph/nodes/");
auto suffix_start = path.find("/command"); auto suffix_start = path.find("/command");
+129 -2
View File
@@ -4,6 +4,13 @@
#include <spdlog/spdlog.h> #include <spdlog/spdlog.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <netdb.h>
#include <cerrno>
namespace dmf_node { namespace dmf_node {
FabricBridgeNode::~FabricBridgeNode() { FabricBridgeNode::~FabricBridgeNode() {
@@ -49,6 +56,12 @@ void FabricBridgeNode::configure(const nlohmann::json& params) {
spdlog::info("Fabric-bridge: target_info set, signaling startup"); 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", static_cast<uint16_t>(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) { void FabricBridgeNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
@@ -62,13 +75,15 @@ void FabricBridgeNode::on_add_writer(const std::string& port_id, mxlFlowWriter w
void FabricBridgeNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) { void FabricBridgeNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
if (port_id == "video_in") { if (port_id == "video_in") {
reader_ = reader; reader_ = reader;
if (mode_ == FabricMode::Initiator && !target_info_str_.empty()) { if (mode_ == FabricMode::Initiator) {
if (!target_info_str_.empty() || !target_host_.empty()) {
start_requested_.store(true); start_requested_.store(true);
spdlog::info("Fabric-bridge: reader added (initiator mode), signaling startup"); spdlog::info("Fabric-bridge: reader added (initiator mode), signaling startup");
} else { } else {
spdlog::info("Fabric-bridge: reader added (initiator mode), waiting for target_info"); spdlog::info("Fabric-bridge: reader added (initiator mode), waiting for target_info");
} }
} }
}
} }
void FabricBridgeNode::on_remove_writer(const std::string& port_id) { void FabricBridgeNode::on_remove_writer(const std::string& port_id) {
@@ -85,8 +100,110 @@ void FabricBridgeNode::on_remove_reader(const std::string& port_id) {
} }
} }
void FabricBridgeNode::fetch_target_info() {
if (target_host_.empty()) return;
spdlog::info("Fabric-bridge: fetching target_info from {}:{}", target_host_, target_port_);
std::string body = R"({"cmd":"status"})";
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: " + std::to_string(body.size()) + "\r\n"
"\r\n" +
body;
for (int attempt = 0; attempt < 30 && 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::seconds(1));
continue;
}
struct timeval tv{};
tv.tv_sec = 2;
tv.tv_usec = 0;
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
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::seconds(1));
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 (errno={}), retry {}...", target_host_, target_port_, errno, attempt + 1);
close(sock);
std::this_thread::sleep_for(std::chrono::seconds(1));
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(500));
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);
spdlog::info("Fabric-bridge: received {} bytes from target", response.size());
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;
}
std::string response_body = response.substr(body_pos + 4);
try {
auto json = nlohmann::json::parse(response_body);
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;
spdlog::info("Fabric-bridge: fetched target_info ({} bytes)", target_info_str_.size());
return;
}
}
spdlog::debug("Fabric-bridge: response data: {}", json.contains("data") ? json["data"].dump() : "(no data)");
} 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() { void FabricBridgeNode::process() {
if (!worker_thread_.has_value() && start_requested_.load()) { if (!worker_thread_.has_value() && start_requested_.load()) {
if (mode_ == FabricMode::Initiator && reader_.has_value() && !target_host_.empty() && target_info_str_.empty()) {
fetch_target_info();
}
if (start_requested_.load()) {
running_ = true; running_ = true;
spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target"); spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target");
if (mode_ == FabricMode::Initiator) { if (mode_ == FabricMode::Initiator) {
@@ -95,6 +212,7 @@ void FabricBridgeNode::process() {
worker_thread_ = std::thread(&FabricBridgeNode::run_target, this); worker_thread_ = std::thread(&FabricBridgeNode::run_target, this);
} }
} }
}
std::this_thread::sleep_for(std::chrono::milliseconds(100)); std::this_thread::sleep_for(std::chrono::milliseconds(100));
} }
@@ -143,6 +261,7 @@ void FabricBridgeNode::run_initiator() {
init_cfg.interface.address.service = bind_service_.c_str(); init_cfg.interface.address.service = bind_service_.c_str();
init_cfg.reader = *reader_; init_cfg.reader = *reader_;
spdlog::info("Fabric-bridge: calling mxlFabricsInitiatorSetup on {}:{}", bind_node_, bind_service_);
status = mxlFabricsInitiatorSetup(initiator, &init_cfg, nullptr); status = mxlFabricsInitiatorSetup(initiator, &init_cfg, nullptr);
if (status != MXL_STATUS_OK) { if (status != MXL_STATUS_OK) {
spdlog::error("Fabric-bridge: initiator setup failed: {}", static_cast<int>(status)); spdlog::error("Fabric-bridge: initiator setup failed: {}", static_cast<int>(status));
@@ -152,7 +271,16 @@ void FabricBridgeNode::run_initiator() {
return; return;
} }
if (target_info_str_.empty()) {
spdlog::error("Fabric-bridge: no target_info available, cannot add target");
mxlFabricsDestroyInitiator(fab_inst, initiator);
mxlFabricsDestroyInstance(fab_inst);
running_ = false;
return;
}
mxlFabricsTargetInfo target_info = nullptr; mxlFabricsTargetInfo target_info = nullptr;
spdlog::info("Fabric-bridge: parsing target_info ({} bytes)", target_info_str_.size());
status = mxlFabricsTargetInfoFromString(target_info_str_.c_str(), &target_info); status = mxlFabricsTargetInfoFromString(target_info_str_.c_str(), &target_info);
if (status != MXL_STATUS_OK || !target_info) { if (status != MXL_STATUS_OK || !target_info) {
spdlog::error("Fabric-bridge: failed to parse target_info: {}", static_cast<int>(status)); spdlog::error("Fabric-bridge: failed to parse target_info: {}", static_cast<int>(status));
@@ -176,7 +304,6 @@ void FabricBridgeNode::run_initiator() {
do { do {
status = mxlFabricsInitiatorMakeProgressBlocking(initiator, 250); status = mxlFabricsInitiatorMakeProgressBlocking(initiator, 250);
} while (status == MXL_ERR_NOT_READY && running_); } while (status == MXL_ERR_NOT_READY && running_);
spdlog::info("Fabric-bridge: connected");
auto now = mxlGetTime(); auto now = mxlGetTime();
read_index_ = mxlTimestampToIndex(&grain_rate_, now) - 2; read_index_ = mxlTimestampToIndex(&grain_rate_, now) - 2;
@@ -51,6 +51,11 @@ private:
std::string target_info_str_; std::string target_info_str_;
std::string target_host_;
uint16_t target_port_ = 9100;
void fetch_target_info();
mxlFabricsInstance fabrics_instance_ = nullptr; mxlFabricsInstance fabrics_instance_ = nullptr;
mxlFabricsInitiator initiator_ = nullptr; mxlFabricsInitiator initiator_ = nullptr;
mxlFabricsTarget target_ = nullptr; mxlFabricsTarget target_ = nullptr;