diff --git a/nodes/fabric-bridge/src/fabric_bridge_node.cpp b/nodes/fabric-bridge/src/fabric_bridge_node.cpp index e7ce366..fa1bb13 100644 --- a/nodes/fabric-bridge/src/fabric_bridge_node.cpp +++ b/nodes/fabric-bridge/src/fabric_bridge_node.cpp @@ -9,6 +9,7 @@ #include #include #include +#include namespace dmf_node { @@ -58,7 +59,7 @@ void FabricBridgeNode::configure(const nlohmann::json& params) { if (params.contains("target_host")) { target_host_ = params["target_host"].get(); - target_port_ = params.value("target_port", 9100); + target_port_ = params.value("target_port", static_cast(9100)); spdlog::info("Fabric-bridge: will fetch target_info from {}:{}", target_host_, target_port_); } } @@ -104,27 +105,34 @@ void FabricBridgeNode::fetch_target_info() { 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: 20\r\n" - "\r\n" - "{\"cmd\":\"status\"}"; + "Content-Length: " + std::to_string(body.size()) + "\r\n" + "\r\n" + + body; - for (int attempt = 0; attempt < 30 && running_ && target_info_str_.empty(); ++attempt) { + 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::milliseconds(1000)); + 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::milliseconds(1000)); + std::this_thread::sleep_for(std::chrono::seconds(1)); continue; } @@ -134,16 +142,16 @@ void FabricBridgeNode::fetch_target_info() { std::memcpy(&addr.sin_addr, he->h_addr_list[0], he->h_length); if (connect(sock, reinterpret_cast(&addr), sizeof(addr)) < 0) { - spdlog::warn("Fabric-bridge: connect to {}:{} failed, retry {}...", target_host_, target_port_, attempt + 1); + 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::milliseconds(1000)); + 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(1000)); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); continue; } @@ -155,6 +163,8 @@ void FabricBridgeNode::fetch_target_info() { } 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); @@ -162,17 +172,19 @@ void FabricBridgeNode::fetch_target_info() { continue; } + std::string response_body = response.substr(body_pos + 4); + try { - auto json = nlohmann::json::parse(response.substr(body_pos + 4)); + auto json = nlohmann::json::parse(response_body); if (json.contains("data") && json["data"].contains("target_info")) { auto ti = json["data"]["target_info"].get(); if (!ti.empty()) { target_info_str_ = ti; - start_requested_.store(true); 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()); } @@ -188,7 +200,7 @@ void FabricBridgeNode::fetch_target_info() { void FabricBridgeNode::process() { if (!worker_thread_.has_value() && start_requested_.load()) { - if (mode_ == FabricMode::Initiator && !target_host_.empty() && target_info_str_.empty()) { + if (mode_ == FabricMode::Initiator && reader_.has_value() && !target_host_.empty() && target_info_str_.empty()) { fetch_target_info(); } if (start_requested_.load()) { @@ -249,6 +261,7 @@ void FabricBridgeNode::run_initiator() { init_cfg.interface.address.service = bind_service_.c_str(); init_cfg.reader = *reader_; + spdlog::info("Fabric-bridge: calling mxlFabricsInitiatorSetup on {}:{}", bind_node_, bind_service_); status = mxlFabricsInitiatorSetup(initiator, &init_cfg, nullptr); if (status != MXL_STATUS_OK) { spdlog::error("Fabric-bridge: initiator setup failed: {}", static_cast(status)); @@ -258,7 +271,16 @@ void FabricBridgeNode::run_initiator() { 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; + spdlog::info("Fabric-bridge: parsing target_info ({} bytes)", target_info_str_.size()); 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(status)); @@ -282,7 +304,6 @@ void FabricBridgeNode::run_initiator() { 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;