cxh/yituo_video
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加推流期间无帧的保护机制

Browse Source
This commit is contained in:
cxh 2025-12-19 09:01:23 +08:00
parent 99228c832c
commit 55bc6eda31
2 changed files with 78 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/rtmp_manager.hpp
View File

@ -51,6 +51,9 @@ class RTMPManager
StreamStatus status; StreamStatus status;
std::mutex status_mutex; std::mutex status_mutex;
// 新增：记录最后收到帧的时间（毫秒）
std::atomic<int64_t> last_frame_ms{0};
}; };
static void stream_loop(Camera cam, StreamContext* ctx); static void stream_loop(Camera cam, StreamContext* ctx);

147
src/rtmp_manager.cpp
View File

@ -116,9 +116,12 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
{ {
const std::string key = make_key(cam.name); const std::string key = make_key(cam.name);
constexpr int64_t START_TIMEOUT_MS = 5000; // 启动阶段无帧
constexpr int64_t NO_FRAME_TIMEOUT_MS = 10000; // 运行阶段突然无帧 ⇒ pipeline 卡死重启
while (ctx->thread_running) while (ctx->thread_running)
{ {
// 1. 检查设备节点是否存在 // 1) 检查设备节点
struct stat st{}; struct stat st{};
if (stat(cam.device.c_str(), &st) != 0) if (stat(cam.device.c_str(), &st) != 0)
{ {
@ -130,7 +133,7 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
continue; continue;
} }
// 2. 创建 GStreamer 管线 // 2) 创建 pipeline
GstElement* pipeline = create_pipeline(cam); GstElement* pipeline = create_pipeline(cam);
if (!pipeline) if (!pipeline)
{ {
@ -143,8 +146,8 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
gst_element_set_name(pipeline, key.c_str()); gst_element_set_name(pipeline, key.c_str());
GstBus* bus = gst_element_get_bus(pipeline); GstBus* bus = gst_element_get_bus(pipeline);
// 3. 在 v4l2src 的 src pad 上挂探测 probe // 3) 帧探测：记录 last_frame_ms
bool got_frame = false; ctx->last_frame_ms.store(0, std::memory_order_relaxed);
{ {
GstElement* src = gst_bin_get_by_name(GST_BIN(pipeline), "src"); GstElement* src = gst_bin_get_by_name(GST_BIN(pipeline), "src");
if (src) if (src)
@ -154,76 +157,69 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
{ {
gst_pad_add_probe( gst_pad_add_probe(
pad, GST_PAD_PROBE_TYPE_BUFFER, pad, GST_PAD_PROBE_TYPE_BUFFER,
[](GstPad*, GstPadProbeInfo*, gpointer user_data) -> GstPadProbeReturn [](GstPad*, GstPadProbeInfo*, gpointer data) -> GstPadProbeReturn
{ {
*static_cast<bool*>(user_data) = true; auto ts = static_cast<std::atomic<int64_t>*>(data);
auto now = std::chrono::steady_clock::now().time_since_epoch();
auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(now).count();
ts->store(ms, std::memory_order_relaxed);
return GST_PAD_PROBE_OK; return GST_PAD_PROBE_OK;
}, },
&got_frame, nullptr); &(ctx->last_frame_ms), nullptr);
gst_object_unref(pad); gst_object_unref(pad);
} }
else
{
LOG_WARN("[RTMP] " + key + " - src has no 'src' pad?");
}
gst_object_unref(src); gst_object_unref(src);
} }
else
{
LOG_WARN("[RTMP] " + key + " - cannot find element 'src' for pad-probe");
}
} }
// 4. 启动播放 // 4) 启动 pipeline
LOG_INFO("[RTMP] Starting stream: " + key); LOG_INFO("[RTMP] Starting stream: " + key);
gst_element_set_state(pipeline, GST_STATE_PLAYING); gst_element_set_state(pipeline, GST_STATE_PLAYING);
// 等待进入 PLAYING 状态（最长 5s）
GstState state = GST_STATE_NULL, pending = GST_STATE_NULL; GstState state = GST_STATE_NULL, pending = GST_STATE_NULL;
bool confirmed_running = false; if (gst_element_get_state(pipeline, &state, &pending, 5 * GST_SECOND) != GST_STATE_CHANGE_SUCCESS ||
if (gst_element_get_state(pipeline, &state, &pending, 5 * GST_SECOND) == GST_STATE_CHANGE_SUCCESS && state != GST_STATE_PLAYING)
state == GST_STATE_PLAYING)
{
std::lock_guard<std::mutex> lk(ctx->status_mutex);
confirmed_running = true;
ctx->status.running = true;
ctx->status.last_error.clear();
LOG_INFO("[RTMP] " + key + " confirmed PLAYING");
}
else
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false; ctx->status.running = false;
ctx->status.last_error = "Pipeline failed to confirm PLAYING"; ctx->status.last_error = "Failed to enter PLAYING";
LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error); LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
}
if (!confirmed_running)
{
gst_element_set_state(pipeline, GST_STATE_NULL); gst_element_set_state(pipeline, GST_STATE_NULL);
if (bus) gst_object_unref(bus); if (bus) gst_object_unref(bus);
gst_object_unref(pipeline); gst_object_unref(pipeline);
std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS)); std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_BASE_DELAY_MS));
continue; continue;
} }
// 5. 运行阶段：监测帧和错误 // pipeline 成功启动
const auto start_t = std::chrono::steady_clock::now(); {
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = true;
ctx->status.last_error.clear();
}
LOG_INFO("[RTMP] " + key + " confirmed PLAYING");
auto launch_tp = std::chrono::steady_clock::now();
bool need_restart = false; bool need_restart = false;
// 5) 主循环：检测停帧或 error
while (ctx->thread_running) while (ctx->thread_running)
{ {
// 检查是否收到帧（前 5s 内） auto now = std::chrono::steady_clock::now();
if (!got_frame) int64_t now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
int64_t last_ms = ctx->last_frame_ms.load(std::memory_order_relaxed);
if (last_ms == 0)
{ {
auto elapsed = auto startup_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now - launch_tp).count();
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start_t) if (startup_ms > START_TIMEOUT_MS)
.count();
if (elapsed > 5)
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false; ctx->status.running = false;
ctx->status.last_error = "No frames detected (no video signal)"; ctx->status.last_error = "No frames detected at startup";
LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error); LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
need_restart = true; need_restart = true;
break; break;
@ -231,53 +227,59 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
} }
else else
{ {
std::lock_guard<std::mutex> lk(ctx->status_mutex); int64_t idle_ms = now_ms - last_ms;
ctx->status.running = true; if (idle_ms > NO_FRAME_TIMEOUT_MS)
ctx->status.last_error.clear(); {
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false;
ctx->status.last_error = "Frame stalled for " + std::to_string(idle_ms) + " ms";
LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
need_restart = true;
break;
}
else
{
std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = true;
ctx->status.last_error.clear();
}
} }
// 等待错误或 EOS 消息 // 错误消息检测
GstMessage* msg = gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND, GstMessage* msg = gst_bus_timed_pop_filtered(bus, 200 * GST_MSECOND,
(GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS)); (GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
if (!msg) continue; if (!msg) continue;
switch (GST_MESSAGE_TYPE(msg)) if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_ERROR)
{ {
case GST_MESSAGE_ERROR: GError* err = nullptr;
{ gst_message_parse_error(msg, &err, nullptr);
GError* err = nullptr; std::lock_guard<std::mutex> lk(ctx->status_mutex);
gst_message_parse_error(msg, &err, nullptr); ctx->status.running = false;
std::lock_guard<std::mutex> lk(ctx->status_mutex); ctx->status.last_error = err ? err->message : "GStreamer error";
ctx->status.running = false; LOG_ERROR("[RTMP] " + key + " - " + ctx->status.last_error);
ctx->status.last_error = err ? err->message : "GStreamer error"; if (err) g_error_free(err);
LOG_ERROR("[RTMP] " + key + " stream error: " + ctx->status.last_error); need_restart = true;
if (err) g_error_free(err); }
need_restart = true; else if (GST_MESSAGE_TYPE(msg) == GST_MESSAGE_EOS)
break; {
} std::lock_guard<std::mutex> lk(ctx->status_mutex);
case GST_MESSAGE_EOS: ctx->status.running = false;
{ ctx->status.last_error = "EOS (End of stream)";
std::lock_guard<std::mutex> lk(ctx->status_mutex); LOG_WARN("[RTMP] " + key + " - " + ctx->status.last_error);
ctx->status.running = false; need_restart = true;
ctx->status.last_error = "End of stream (EOS)";
LOG_WARN("[RTMP] " + key + " reached EOS");
need_restart = true;
break;
}
default: break;
} }
gst_message_unref(msg);
gst_message_unref(msg);
if (need_restart) break; if (need_restart) break;
} }
// 6. 收尾清理 // 清理 & 重建
gst_element_set_state(pipeline, GST_STATE_NULL); gst_element_set_state(pipeline, GST_STATE_NULL);
if (bus) gst_object_unref(bus); if (bus) gst_object_unref(bus);
gst_object_unref(pipeline); gst_object_unref(pipeline);
// 7. 若仍在运行状态，准备重启
if (ctx->thread_running) if (ctx->thread_running)
{ {
LOG_WARN("[RTMP] Restarting " + key + " in 3s..."); LOG_WARN("[RTMP] Restarting " + key + " in 3s...");
@ -289,6 +291,7 @@ void RTMPManager::stream_loop(Camera cam, StreamContext* ctx)
std::lock_guard<std::mutex> lk(ctx->status_mutex); std::lock_guard<std::mutex> lk(ctx->status_mutex);
ctx->status.running = false; ctx->status.running = false;
} }
LOG_INFO("[RTMP] Stream thread exited for " + key); LOG_INFO("[RTMP] Stream thread exited for " + key);
} }
@ -301,7 +304,7 @@ void RTMPManager::start_all()
int delay_ms = 0; int delay_ms = 0;
for (const auto& cam : g_app_config.cameras) for (const auto& cam : g_app_config.cameras)
{ {
// 🔴 核心修正：跳过未启用摄像头 // 跳过未启用摄像头
if (!cam.enabled) if (!cam.enabled)
{ {
LOG_INFO("[RTMP] Skip disabled camera: " + cam.name); LOG_INFO("[RTMP] Skip disabled camera: " + cam.name);