fix: LWS HTTP connection leak, grain index alignment, and reader head tracking
- control_server: add Connection: close header + always return -1 after HTTP response to force-close connection. Without this, lws_service() blocks forever after the first POST /cmd, freezing the node process loop. - passthrough: initialize read_index_ to runtime.headIndex when reader is added (prevents TOO_EARLY errors on first read) - passthrough: handle MXL_ERR_OUT_OF_RANGE_TOO_EARLY in addition to TOO_LATE (both jump to head index) - passthrough: use grain_info.index as writer index instead of separate write_index_ counter (MXL writers must use TAI-based grain indices) - passthrough: reduce grain read timeout from 100ms to 20ms for tighter loop with LWS poll - node_runner: add 'status' command handler that sends node status back via control server
This commit is contained in:
@@ -47,12 +47,14 @@ static int send_http_json(struct lws* wsi, const std::string& status, const std:
|
|||||||
auto hdr = "HTTP/1.1 " + status + "\r\n"
|
auto hdr = "HTTP/1.1 " + status + "\r\n"
|
||||||
"Content-Type: application/json\r\n"
|
"Content-Type: application/json\r\n"
|
||||||
"Content-Length: " + std::to_string(body.size()) + "\r\n"
|
"Content-Length: " + std::to_string(body.size()) + "\r\n"
|
||||||
|
"Connection: close\r\n"
|
||||||
"\r\n";
|
"\r\n";
|
||||||
std::vector<uint8_t> buf(LWS_PRE + hdr.size() + body.size());
|
std::vector<uint8_t> buf(LWS_PRE + hdr.size() + body.size());
|
||||||
std::memcpy(buf.data() + LWS_PRE, hdr.data(), hdr.size());
|
std::memcpy(buf.data() + LWS_PRE, hdr.data(), hdr.size());
|
||||||
std::memcpy(buf.data() + LWS_PRE + hdr.size(), body.data(), body.size());
|
std::memcpy(buf.data() + LWS_PRE + hdr.size(), body.data(), body.size());
|
||||||
lws_write(wsi, buf.data() + LWS_PRE, hdr.size() + body.size(), LWS_WRITE_HTTP);
|
lws_write(wsi, buf.data() + LWS_PRE, hdr.size() + body.size(), LWS_WRITE_HTTP);
|
||||||
return lws_http_transaction_completed(wsi) ? -1 : 0;
|
lws_http_transaction_completed(wsi);
|
||||||
|
return -1;
|
||||||
}
|
}
|
||||||
|
|
||||||
struct PerSession {
|
struct PerSession {
|
||||||
|
|||||||
@@ -144,6 +144,14 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
control_server->register_command("status", [&](const nlohmann::json& /*msg*/) {
|
||||||
|
nlohmann::json resp;
|
||||||
|
resp["event"] = "status";
|
||||||
|
resp["node_id"] = node_id_;
|
||||||
|
resp["data"] = node->status();
|
||||||
|
control_server->send_event(resp);
|
||||||
|
});
|
||||||
|
|
||||||
control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) {
|
control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) {
|
||||||
spdlog::info("Shutdown command received");
|
spdlog::info("Shutdown command received");
|
||||||
g_running = false;
|
g_running = false;
|
||||||
|
|||||||
@@ -19,7 +19,10 @@ void PassthroughNode::on_add_writer(const std::string& port_id, mxlFlowWriter wr
|
|||||||
void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
|
void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
|
||||||
if (port_id == "video_in") {
|
if (port_id == "video_in") {
|
||||||
reader_ = reader;
|
reader_ = reader;
|
||||||
spdlog::info("Passthrough: reader added on video_in");
|
mxlFlowRuntimeInfo runtime{};
|
||||||
|
mxlFlowReaderGetRuntimeInfo(*reader_, &runtime);
|
||||||
|
read_index_ = runtime.headIndex;
|
||||||
|
spdlog::info("Passthrough: reader added on video_in, starting at index {}", read_index_);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -45,26 +48,23 @@ void PassthroughNode::process() {
|
|||||||
mxlGrainInfo grain_info{};
|
mxlGrainInfo grain_info{};
|
||||||
uint8_t* payload = nullptr;
|
uint8_t* payload = nullptr;
|
||||||
|
|
||||||
auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 100000000ULL, &grain_info, &payload);
|
auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload);
|
||||||
if (status != MXL_STATUS_OK) {
|
if (status != MXL_STATUS_OK) {
|
||||||
if (status == MXL_ERR_TIMEOUT) {
|
if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
|
||||||
return;
|
|
||||||
}
|
|
||||||
if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
|
|
||||||
mxlFlowRuntimeInfo runtime{};
|
mxlFlowRuntimeInfo runtime{};
|
||||||
mxlFlowReaderGetRuntimeInfo(*reader_, &runtime);
|
mxlFlowReaderGetRuntimeInfo(*reader_, &runtime);
|
||||||
|
spdlog::warn("Passthrough: index {} out of range, jumping to head {}", read_index_, runtime.headIndex);
|
||||||
read_index_ = runtime.headIndex;
|
read_index_ = runtime.headIndex;
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
mxlGrainInfo out_grain{};
|
mxlGrainInfo out_grain{};
|
||||||
uint8_t* out_payload = nullptr;
|
uint8_t* out_payload = nullptr;
|
||||||
status = mxlFlowWriterOpenGrain(*writer_, write_index_, &out_grain, &out_payload);
|
status = mxlFlowWriterOpenGrain(*writer_, grain_info.index, &out_grain, &out_payload);
|
||||||
if (status != MXL_STATUS_OK) {
|
if (status != MXL_STATUS_OK) {
|
||||||
spdlog::warn("Passthrough: failed to open output grain at index {}: {}", write_index_, static_cast<int>(status));
|
spdlog::warn("Passthrough: failed to open output grain at index {}: {}", grain_info.index, static_cast<int>(status));
|
||||||
read_index_++;
|
read_index_ = grain_info.index + 1;
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -75,9 +75,12 @@ void PassthroughNode::process() {
|
|||||||
out_grain.flags = grain_info.flags;
|
out_grain.flags = grain_info.flags;
|
||||||
mxlFlowWriterCommitGrain(*writer_, &out_grain);
|
mxlFlowWriterCommitGrain(*writer_, &out_grain);
|
||||||
|
|
||||||
read_index_++;
|
read_index_ = grain_info.index + 1;
|
||||||
write_index_++;
|
|
||||||
grains_processed_++;
|
grains_processed_++;
|
||||||
|
|
||||||
|
if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) {
|
||||||
|
spdlog::info("Passthrough: grain #{}, index={}, size={}", grains_processed_, grain_info.index, grain_info.grainSize);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
nlohmann::json PassthroughNode::status() const {
|
nlohmann::json PassthroughNode::status() const {
|
||||||
@@ -85,7 +88,6 @@ nlohmann::json PassthroughNode::status() const {
|
|||||||
{"type", "passthrough"},
|
{"type", "passthrough"},
|
||||||
{"grains_processed", grains_processed_},
|
{"grains_processed", grains_processed_},
|
||||||
{"read_index", read_index_},
|
{"read_index", read_index_},
|
||||||
{"write_index", write_index_},
|
|
||||||
{"has_reader", reader_.has_value()},
|
{"has_reader", reader_.has_value()},
|
||||||
{"has_writer", writer_.has_value()},
|
{"has_writer", writer_.has_value()},
|
||||||
};
|
};
|
||||||
|
|||||||
Reference in New Issue
Block a user