From fa51e360b9d61ebf8e931c92c654d3e202922f69 Mon Sep 17 00:00:00 2001 From: limjoe Date: Sat, 17 Aug 2024 22:38:10 +0800 Subject: [PATCH] Transcript: support custom fix and overlay queue --- DEVELOPER.md | 2 ++ platform/main.go | 8 ++++- platform/transcript.go | 67 ++++++++++++++++++++++++++++++++++++++++-- platform/utils.go | 8 +++++ 4 files changed, 81 insertions(+), 4 deletions(-) diff --git a/DEVELOPER.md b/DEVELOPER.md index 3870e133..5d04d826 100644 --- a/DEVELOPER.md +++ b/DEVELOPER.md @@ -1190,6 +1190,8 @@ For limit that you can control: * `SRS_FORWARD_LIMIT`: The limit for SRS forward. Default: `10`. * `SRS_VLIVE_LIMIT`: The limit for SRS virtual live. Default: `10`. +* `SRS_TRANSCRIPT_FIX_QUEUE_LIMIT`: The limit for SRS transcript manually fix queue. Default: `2`. +* `SRS_TRANSCRIPT_OVERLAY_QUEUE_LIMIT`: The limit for SRS transcript overlay queue. Default: `9`. For feature control: diff --git a/platform/main.go b/platform/main.go index 9bd97915..1f8dae93 100644 --- a/platform/main.go +++ b/platform/main.go @@ -132,13 +132,18 @@ func doMain(ctx context.Context) error { setEnvDefault("SRS_VLIVE_LIMIT", "10") setEnvDefault("SRS_CAMERA_LIMIT", "10") + // For transcript queue limit. + setEnvDefault("SRS_TRANSCRIPT_FIX_QUEUE_LIMIT", "2") + setEnvDefault("SRS_TRANSCRIPT_OVERLAY_QUEUE_LIMIT", "9") + logger.Tf(ctx, "load .env as MGMT_PASSWORD=%vB, GO_PPROF=%v, "+ "SRS_PLATFORM_SECRET=%vB, CLOUD=%v, REGION=%v, SOURCE=%v, SRT_PORT=%v, RTC_PORT=%v, "+ "NODE_ENV=%v, LOCAL_RELEASE=%v, REDIS_DATABASE=%v, REDIS_HOST=%v, REDIS_PASSWORD=%vB, REDIS_PORT=%v, RTMP_PORT=%v, "+ "PUBLIC_URL=%v, BUILD_PATH=%v, REACT_APP_LOCALE=%v, PLATFORM_LISTEN=%v, HTTP_PORT=%v, "+ "REGISTRY=%v, MGMT_LISTEN=%v, HTTPS_LISTEN=%v, AUTO_SELF_SIGNED_CERTIFICATE=%v, "+ "NAME_LOOKUP=%v, PLATFORM_DOCKER=%v, SRS_FORWARD_LIMIT=%v, SRS_VLIVE_LIMIT=%v, "+ - "SRS_CAMERA_LIMIT=%v, YTDL_PROXY=%v", + "SRS_CAMERA_LIMIT=%v, YTDL_PROXY=%v"+ + "SRS_TRANSCRIPT_FIX_QUEUE_LIMIT=%v, SRS_TRANSCRIPT_OVERLAY_QUEUE_LIMIT=%v", len(envMgmtPassword()), envGoPprof(), len(envApiSecret()), envCloud(), envRegion(), envSource(), envSrtListen(), envRtcListen(), envNodeEnv(), envLocalRelease(), @@ -149,6 +154,7 @@ func doMain(ctx context.Context) error { envSelfSignedCertificate(), envNameLookup(), envPlatformDocker(), envForwardLimit(), envVLiveLimit(), envCameraLimit(), envYtdlProxy(), + envTranscriptFixQueueLimit(), envTranscriptOverlayQueueLimit(), ) // Start the Go pprof if enabled. diff --git a/platform/transcript.go b/platform/transcript.go index 51340915..1cf376c7 100644 --- a/platform/transcript.go +++ b/platform/transcript.go @@ -27,8 +27,39 @@ import ( "github.com/sashabaranov/go-openai" ) -// The total segments in overlay HLS. -const maxOverlaySegments = 9 +// The default total segments in overlay HLS. +const defaultMaxOverlaySegments = 9 + +// Get the total segments in overlay HLS. +func GetMaxOverlaySegments() (int, error) { + var maxOverlaySegments int + if envTranscriptOverlayQueueLimit() != "" { + if iv, err := strconv.ParseInt(envTranscriptOverlayQueueLimit(), 10, 64); err != nil { + return defaultMaxOverlaySegments, errors.Wrapf(err, "parse env transcript overlay queue limit %v", envTranscriptOverlayQueueLimit()) + } else { + maxOverlaySegments = int(iv) + } + } + + return maxOverlaySegments, nil +} + +// The default max fix queue limits. +const defaultMaxFixQueueLimit = 2 + +// Get the manually fix queue limit. +func GetMaxFixQueueLimit() (int, error) { + var maxFixQueueLimit int + if envTranscriptFixQueueLimit() != "" { + if iv, err := strconv.ParseInt(envTranscriptFixQueueLimit(), 10, 64); err != nil { + return defaultMaxFixQueueLimit, errors.Wrapf(err, "parse env transcript manually fix queue limit %v", envTranscriptFixQueueLimit()) + } else { + maxFixQueueLimit = int(iv) + } + } + + return maxFixQueueLimit, nil +} var transcriptWorker *TranscriptWorker @@ -1717,6 +1748,12 @@ func (v *TranscriptTask) DriveLiveQueue(ctx context.Context) error { return nil } + // Get the maxOverlaySegments value + maxOverlaySegments, err := GetMaxOverlaySegments() + if err != nil { + logger.Wf(ctx, "transcript: ignore get maxOverlaySegments err %+v, use default value %v", err, defaultMaxOverlaySegments) + } + // Wait if ASR queue is full. if v.AsrQueue.count() >= maxOverlaySegments+1 { return nil @@ -1792,6 +1829,12 @@ func (v *TranscriptTask) DriveAsrQueue(ctx context.Context) error { return nil } + // Get the maxOverlaySegments value + maxOverlaySegments, err := GetMaxOverlaySegments() + if err != nil { + logger.Wf(ctx, "transcript: ignore get maxOverlaySegments err %+v, use default value %v", err, defaultMaxOverlaySegments) + } + // Wait if Fix queue is full. if v.FixQueue.count() >= maxOverlaySegments+1 { return nil @@ -1917,8 +1960,14 @@ func (v *TranscriptTask) DriveFixQueue(ctx context.Context) error { return nil } + // Get the maxFixQueueLimit value + maxFixQueueLimit, err := GetMaxFixQueueLimit() + if err != nil { + logger.Wf(ctx, "transcript: ignore get maxFixQueueLimit err %+v, use default value %v", err, defaultMaxFixQueueLimit) + } + // Ignore if not enough segments. - if v.FixQueue.count() <= 2 { + if v.FixQueue.count() <= maxFixQueueLimit { return nil } @@ -1937,6 +1986,12 @@ func (v *TranscriptTask) DriveFixQueue(ctx context.Context) error { return nil } + // Get the maxOverlaySegments value + maxOverlaySegments, err := GetMaxOverlaySegments() + if err != nil { + logger.Wf(ctx, "transcript: ignore get maxOverlaySegments err %+v, use default value %v", err, defaultMaxOverlaySegments) + } + // Wait if Overlay queue is full. if v.OverlayQueue.count() >= maxOverlaySegments+1 { return nil @@ -2028,6 +2083,12 @@ func (v *TranscriptTask) DriveOverlayQueue(ctx context.Context) error { return nil } + // Get the maxOverlaySegments value + maxOverlaySegments, err := GetMaxOverlaySegments() + if err != nil { + logger.Wf(ctx, "transcript: ignore get maxOverlaySegments err %+v, use default value %v", err, defaultMaxOverlaySegments) + } + // Ignore if not enough segments. if v.OverlayQueue.count() <= maxOverlaySegments { select { diff --git a/platform/utils.go b/platform/utils.go index 5a116582..805d207a 100644 --- a/platform/utils.go +++ b/platform/utils.go @@ -494,6 +494,14 @@ func envYtdlProxy() string { return os.Getenv("YTDL_PROXY") } +func envTranscriptFixQueueLimit() string { + return os.Getenv("SRS_TRANSCRIPT_FIX_QUEUE_LIMIT") +} + +func envTranscriptOverlayQueueLimit() string { + return os.Getenv("SRS_TRANSCRIPT_OVERLAY_QUEUE_LIMIT") +} + // rdb is a global redis client object. var rdb *redis.Client