[fix]gop增加单个gop包数量的限制，防止异常情况内存飞涨

conf/lalserver.conf.json

@@ -9,6 +9,7 @@
     "rtmps_cert_file": "./conf/cert.pem",
     "rtmps_key_file": "./conf/key.pem",
     "gop_num": 0,
+    "single_gop_max_num": 0,
     "merge_write_size": 0
   },
   "in_session": {
@@ -25,7 +26,8 @@
     "enable": true,
     "enable_https": true,
     "url_pattern": "/",
-    "gop_num": 0
+    "gop_num": 0,
+    "single_gop_max_num": 0
   },
   "hls": {
     "enable": true,
@@ -42,7 +44,8 @@
     "enable": true,
     "enable_https": true,
     "url_pattern": "/",
-    "gop_num": 0
+    "gop_num": 0,
+    "single_gop_max_num": 0
   },
   "rtsp": {
     "enable": true,

conf/lalserver.conf.json.tmpl

@@ -9,6 +9,7 @@
     "rtmps_cert_file": "./conf/cert.pem",
     "rtmps_key_file": "./conf/key.pem",
     "gop_num": 0,
+    "single_gop_max_num": 0,
     "merge_write_size": 0
   },
   "in_session": {
@@ -25,7 +26,8 @@
     "enable": true,
     "enable_https": true,
     "url_pattern": "/",
-    "gop_num": 0
+    "gop_num": 0,
+    "single_gop_max_num": 0,
   },
   "hls": {
     "enable": true,
@@ -42,7 +44,8 @@
     "enable": true,
     "enable_https": true,
     "url_pattern": "/",
-    "gop_num": 0
+    "gop_num": 0,
+    "single_gop_max_num": 0,
   },
   "rtsp": {
     "enable": true,

pkg/logic/config.go

@@ -58,6 +58,7 @@ type RtmpConfig struct {
 	RtmpsCertFile  string `json:"rtmps_cert_file"`
 	RtmpsKeyFile   string `json:"rtmps_key_file"`
 	GopNum         int    `json:"gop_num"` // TODO(chef): refactor 更名为gop_cache_num
+	SingleGopMaxNum         int    `json:"single_gop_max_num"`
 	MergeWriteSize int    `json:"merge_write_size"`
 }
 
@@ -74,12 +75,14 @@ type HttpflvConfig struct {
 	CommonHttpServerConfig
 
 	GopNum int `json:"gop_num"`
+	SingleGopMaxNum         int    `json:"single_gop_max_num"`
 }
 
 type HttptsConfig struct {
 	CommonHttpServerConfig
 
 	GopNum int `json:"gop_num"`
+	SingleGopMaxNum         int    `json:"single_gop_max_num"`
 }
 
 type HlsConfig struct {

pkg/logic/group__.go

@@ -147,9 +147,9 @@ func NewGroup(appName string, streamName string, config *Config, observer IGroup
 		httptsSubSessionSet:        make(map[*httpts.SubSession]struct{}),
 		rtspSubSessionSet:          make(map[*rtsp.SubSession]struct{}),
 		waitRtspSubSessionSet:      make(map[*rtsp.SubSession]struct{}),
-		rtmpGopCache:               remux.NewGopCache("rtmp", uk, config.RtmpConfig.GopNum),
-		httpflvGopCache:            remux.NewGopCache("httpflv", uk, config.HttpflvConfig.GopNum),
-		httptsGopCache:             remux.NewGopCacheMpegts(uk, config.HttptsConfig.GopNum),
+		rtmpGopCache:               remux.NewGopCache("rtmp", uk, config.RtmpConfig.GopNum, config.RtmpConfig.SingleGopMaxNum),
+		httpflvGopCache:            remux.NewGopCache("httpflv", uk, config.HttpflvConfig.GopNum, config.HttpflvConfig.SingleGopMaxNum),
+		httptsGopCache:             remux.NewGopCacheMpegts(uk, config.HttptsConfig.GopNum, config.HttptsConfig.SingleGopMaxNum),
 		psPubPrevInactiveCheckTick: -1,
 	}
 

pkg/remux/gop_cache.go

@@ -63,6 +63,7 @@ type GopCache struct {
 	gopRingFirst int
 	gopRingLast  int
 	gopSize      int
+	singleGopMaxNum int
 }
 
 // NewGopCache
@@ -70,7 +71,7 @@ type GopCache struct {
 // @param gopNum: gop缓存大小。
 //   - 如果为0，则不缓存音频数据，也即GOP缓存功能不生效。
 //   - 如果>0，则缓存[0, gopNum]个GOP，最多缓存 gopNum 个GOP。注意，最后一个GOP可能是不完整的。
-func NewGopCache(t string, uniqueKey string, gopNum int) *GopCache {
+func NewGopCache(t string, uniqueKey string, gopNum int,singleGopMaxNum int) *GopCache {
 	return &GopCache{
 		t:            t,
 		uniqueKey:    uniqueKey,
@@ -78,6 +79,7 @@ func NewGopCache(t string, uniqueKey string, gopNum int) *GopCache {
 		gopRing:      make([]Gop, gopNum+1, gopNum+1),
 		gopRingFirst: 0,
 		gopRingLast:  0,
+		singleGopMaxNum:singleGopMaxNum,
 	}
 }
 
@@ -153,7 +155,9 @@ func (gc *GopCache) Clear() {
 // 注意，如果GopCache为空，则不缓存msg
 func (gc *GopCache) feedLastGop(msg base.RtmpMsg, b []byte) {
 	if !gc.isGopRingEmpty() {
-		gc.gopRing[(gc.gopRingLast-1+gc.gopSize)%gc.gopSize].Feed(msg, b)
+		if gc.gopRing[(gc.gopRingLast-1+gc.gopSize)%gc.gopSize].len()<=gc.singleGopMaxNum ||gc.singleGopMaxNum==0{
+			gc.gopRing[(gc.gopRingLast-1+gc.gopSize)%gc.gopSize].Feed(msg, b)
+		}
 	}
 }
 
@@ -193,3 +197,6 @@ func (g *Gop) Feed(msg base.RtmpMsg, b []byte) {
 func (g *Gop) Clear() {
 	g.data = g.data[:0]
 }
+func (g *Gop) len()int {
+	return len(g.data)
+}

pkg/remux/gop_cache_mpegts.go

@@ -21,9 +21,10 @@ type GopCacheMpegts struct {
 	gopRingFirst int
 	gopRingLast  int
 	gopSize      int
+	singleGopMaxNum int
 }
 
-func NewGopCacheMpegts(uniqueKey string, gopNum int) *GopCacheMpegts {
+func NewGopCacheMpegts(uniqueKey string, gopNum int,singleGopMaxNum int) *GopCacheMpegts {
 	return &GopCacheMpegts{
 		uniqueKey:    uniqueKey,
 		gopNum:       gopNum,
@@ -31,6 +32,7 @@ func NewGopCacheMpegts(uniqueKey string, gopNum int) *GopCacheMpegts {
 		gopRing:      make([]GopMpegts, gopNum+1, gopNum+1),
 		gopRingFirst: 0,
 		gopRingLast:  0,
+		singleGopMaxNum: singleGopMaxNum,
 	}
 }
 
@@ -72,7 +74,10 @@ func (gc *GopCacheMpegts) Clear() {
 // 注意，如果GopCache为空，则不缓存msg
 func (gc *GopCacheMpegts) feedLastGop(b []byte) {
 	if !gc.isGopRingEmpty() {
-		gc.gopRing[(gc.gopRingLast-1+gc.gopSize)%gc.gopSize].Feed(b)
+		if gc.gopRing[(gc.gopRingLast-1+gc.gopSize)%gc.gopSize].len()<=gc.singleGopMaxNum ||gc.singleGopMaxNum==0{
+			gc.gopRing[(gc.gopRingLast-1+gc.gopSize)%gc.gopSize].Feed(b)
+		}
+
 	}
 }
 
@@ -115,3 +120,6 @@ func (g *GopMpegts) Feed(b []byte) {
 func (g *GopMpegts) Clear() {
 	g.data = g.data[:0]
 }
+func (g *GopMpegts) len()int {
+	return len(g.data)
+}