Live: Limit cached max frames by gop_cache_max_frames (#3236)

* add gop_cache_max_frames

* Live: Limit cached max frames by gop_cache_max_frames. v5.0.93

Co-authored-by: wanglei <wanglei@unicloud.com>
Co-authored-by: winlin <winlin@vip.126.com>

trunk/conf/full.conf

@@ -1096,6 +1096,12 @@ vhost play.srs.com {
         # Overwrite by env SRS_VHOST_PLAY_GOP_CACHE for all vhosts.
         # default: on
         gop_cache off;
+
+        # Limit the max frames in gop cache. It might cause OOM if video stream has no IDR frame, so we limit to N
+        # frames by default.
+        # default: 250
+        gop_cache_max_frames 250;
+
         # the max live queue length in seconds.
         # if the messages in the queue exceed the max length,
         # drop the old whole gop.

trunk/conf/srs.conf

@@ -36,4 +36,8 @@ vhost __defaultVhost__ {
         # @see https://ossrs.net/lts/zh-cn/docs/v4/doc/webrtc#rtc-to-rtmp
         rtc_to_rtmp off;
     }
+
+    play{
+        gop_cache_max_frames 250;
+    }
 }

trunk/doc/CHANGELOG.md

@@ -8,6 +8,7 @@ The changelog for SRS.
 
 ## SRS 5.0 Changelog
 
+* v5.0, 2022-11-22, Merge [#3236](https://github.com/ossrs/srs/pull/3236): Live: Limit cached max frames by gop_cache_max_frames. v5.0.93
 * v5.0, 2022-11-22, Asan: Check libasan and show tips. v5.0.92
 * v5.0, 2022-11-21, Merge [#3264](https://github.com/ossrs/srs/pull/3264): Asan: Try to fix st_memory_leak for asan check. (#3264). v5.0.91
 * v5.0, 2022-11-21, Asan: Fix global ip address leak check. v5.0.90

trunk/src/app/srs_app_config.cpp

@@ -517,7 +517,8 @@ srs_error_t srs_config_transform_vhost(SrsConfDirective* root)
             //  SRS3+:
             //      vhost { play { shadow; } }
             if (n == "time_jitter" || n == "mix_correct" || n == "atc" || n == "atc_auto"
-                || n == "mw_latency" || n == "gop_cache" || n == "queue_length" || n == "send_min_interval"
+                || n == "mw_latency" || n == "gop_cache" || n == "gop_cache_max_frames" 
+                || n == "queue_length" || n == "send_min_interval"
                 || n == "reduce_sequence_header") {
                 it = dir->directives.erase(it);
 
@@ -2548,7 +2549,7 @@ srs_error_t SrsConfig::check_normal_config()
                 for (int j = 0; j < (int)conf->directives.size(); j++) {
                     string m = conf->at(j)->name;
                     if (m != "time_jitter" && m != "mix_correct" && m != "atc" && m != "atc_auto" && m != "mw_latency"
-                        && m != "gop_cache" && m != "queue_length" && m != "send_min_interval" && m != "reduce_sequence_header"
+                        && m != "gop_cache" && m != "gop_cache_max_frames" && m != "queue_length" && m != "send_min_interval" && m != "reduce_sequence_header"
                         && m != "mw_msgs") {
                         return srs_error_new(ERROR_SYSTEM_CONFIG_INVALID, "illegal vhost.play.%s of %s", m.c_str(), vhost->arg0().c_str());
                     }
@@ -4657,6 +4658,32 @@ bool SrsConfig::get_gop_cache(string vhost)
     return SRS_CONF_PERFER_TRUE(conf->arg0());
 }
 
+
+int SrsConfig::get_gop_cache_max_frames(string vhost)
+{
+    SRS_OVERWRITE_BY_ENV_INT("srs.vhost.play.gop_cache_max_frames");
+
+    static int DEFAULT = 250;
+
+    SrsConfDirective* conf = get_vhost(vhost);
+    if (!conf) {
+        return DEFAULT;
+    }
+
+    conf = conf->get("play");
+    if (!conf) {
+        return DEFAULT;
+    }
+
+    conf = conf->get("gop_cache_max_frames");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+
+    return ::atoi(conf->arg0().c_str());
+}
+
+
 bool SrsConfig::get_debug_srs_upnode(string vhost)
 {
     static bool DEFAULT = true;

trunk/src/app/srs_app_config.hpp

@@ -553,6 +553,8 @@ class SrsConfig
     // @return true when gop_cache is ok; otherwise, false.
     // @remark, default true.
     virtual bool get_gop_cache(std::string vhost);
+    // Get the limit max frames for gop cache.
+    virtual int get_gop_cache_max_frames(std::string vhost);
     // Whether debug_srs_upnode is enabled of vhost.
     // debug_srs_upnode is very important feature for tracable log,
     // but some server, for instance, flussonic donot support it.

trunk/src/app/srs_app_rtmp_conn.cpp

@@ -557,9 +557,12 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
     srs_assert(source != NULL);
 
     bool enabled_cache = _srs_config->get_gop_cache(req->vhost);
-    srs_trace("source url=%s, ip=%s, cache=%d, is_edge=%d, source_id=%s/%s",
-        req->get_stream_url().c_str(), ip.c_str(), enabled_cache, info->edge, source->source_id().c_str(), source->pre_source_id().c_str());
+    int gcmf = _srs_config->get_gop_cache_max_frames(req->vhost);
+    srs_trace("source url=%s, ip=%s, cache=%d/%d, is_edge=%d, source_id=%s/%s",
+        req->get_stream_url().c_str(), ip.c_str(), enabled_cache, gcmf, info->edge, source->source_id().c_str(),
+        source->pre_source_id().c_str());
     source->set_cache(enabled_cache);
+    source->set_gop_cache_max_frames(gcmf);
 
     switch (info->type) {
         case SrsRtmpConnPlay: {

trunk/src/app/srs_app_source.cpp

@@ -575,6 +575,7 @@ SrsGopCache::SrsGopCache()
     cached_video_count = 0;
     enable_gop_cache = true;
     audio_after_last_video_count = 0;
+    gop_cache_max_frames_ = 0;
 }
 
 SrsGopCache::~SrsGopCache()
@@ -597,6 +598,11 @@ void SrsGopCache::set(bool v)
     }
 }
 
+void SrsGopCache::set_gop_cache_max_frames(int v)
+{
+    gop_cache_max_frames_ = v;
+}
+
 bool SrsGopCache::enabled()
 {
     return enable_gop_cache;
@@ -648,10 +654,17 @@ srs_error_t SrsGopCache::cache(SrsSharedPtrMessage* shared_msg)
         // curent msg is video frame, so we set to 1.
         cached_video_count = 1;
     }
-    
+
     // cache the frame.
     gop_cache.push_back(msg->copy());
-
+
+    // Clear gop cache if exceed the max frames.
+    if (gop_cache.size() > (size_t)gop_cache_max_frames_) {
+        srs_warn("Gop cache exceed max frames=%d, total=%d, videos=%d, aalvc=%d",
+            gop_cache_max_frames_, (int)gop_cache.size(), cached_video_count, audio_after_last_video_count);
+        clear();
+    }
+
     return err;
 }
 
@@ -2086,6 +2099,7 @@ srs_error_t SrsLiveSource::on_reload_vhost_play(string vhost)
             string url = req->get_stream_url();
             srs_trace("vhost %s gop_cache changed to %d, source url=%s", vhost.c_str(), v, url.c_str());
             gop_cache->set(v);
+            gop_cache->set_gop_cache_max_frames(_srs_config->get_gop_cache_max_frames(vhost));
         }
     }
 
@@ -2725,6 +2739,11 @@ void SrsLiveSource::set_cache(bool enabled)
     gop_cache->set(enabled);
 }
 
+void SrsLiveSource::set_gop_cache_max_frames(int v)
+{
+    gop_cache->set_gop_cache_max_frames(v);
+}
+
 SrsRtmpJitterAlgorithm SrsLiveSource::jitter()
 {
     return jitter_algorithm;

trunk/src/app/srs_app_source.hpp

@@ -228,6 +228,10 @@ class SrsGopCache
     // The client will wait for the next keyframe for h264,
     // and will be black-screen.
     bool enable_gop_cache;
+    // to limit the max gop cache frames
+    // without this limit, if ingest stream always has no IDR frame
+    // it will cause srs run out of memory
+    int gop_cache_max_frames_;
     // The video frame count, avoid cache for pure audio stream.
     int cached_video_count;
     // when user disabled video when publishing, and gop cache enalbed,
@@ -251,6 +255,7 @@ class SrsGopCache
     virtual void dispose();
     // To enable or disable the gop cache.
     virtual void set(bool v);
+    virtual void set_gop_cache_max_frames(int v);
     virtual bool enabled();
     // only for h264 codec
     // 1. cache the gop when got h264 video packet.
@@ -589,6 +594,7 @@ class SrsLiveSource : public ISrsReloadHandler
     virtual srs_error_t consumer_dumps(SrsLiveConsumer* consumer, bool ds = true, bool dm = true, bool dg = true);
     virtual void on_consumer_destroy(SrsLiveConsumer* consumer);
     virtual void set_cache(bool enabled);
+    virtual void set_gop_cache_max_frames(int v);
     virtual SrsRtmpJitterAlgorithm jitter();
 public:
     // For edge, when publish edge stream, check the state

trunk/src/core/srs_core_version5.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR       5
 #define VERSION_MINOR       0
-#define VERSION_REVISION    92
+#define VERSION_REVISION    93
 
 #endif