diff --git a/automerge.go b/automerge.go index f6496db..8ebfb6e 100644 --- a/automerge.go +++ b/automerge.go @@ -242,9 +242,15 @@ func remapJsonlTimes(srcFile, dstFile string, starts []int64, durs []float64, da return nil } -// 自动合并入口:把结果写进任务进度;日志措辞仍由内部负责 +// 自动合并入口:排队跑完一场,把结果写进任务进度;日志措辞仍由内部负责 func autoMerge(uid int, segs []recordSegment, jsonlFile string, danmuStartNs int64) { - err := autoMergeJob(uid, segs, jsonlFile, danmuStartNs) + // 多个主播可能同时下播,而全局只有一份任务状态:先排队,跑到了再登记进度 + if jobActive() { + lPrintf("已有合并任务在跑(%s),本场先排队等待", jobBusyName()) + } + err := runMergeExclusive(func() error { + return autoMergeJob(uid, segs, jsonlFile, danmuStartNs) + }) if err == errJobStopped { // 停止不是故障,内部不打印,这里补一句 lPrintWarn("自动合并已被手动停止" + longID(uid)) @@ -300,7 +306,10 @@ func autoMergeJob(uid int, segs []recordSegment, jsonlFile string, danmuStartNs inputBytes += fileSizeOrZero(p) } } - jobStart("自动合并", filepath.Base(outFile), inputBytes, 0) + if !jobStart("自动合并", filepath.Base(outFile), inputBytes, 0) { + // 走到这里说明没先拿 mergeSlot,与其把别人的进度冲掉不如直接放弃这一场 + return errors.New("开始任务失败:还有别的合并任务没结束,本场合并已跳过") + } video, jsonl, offset, cleanup, err := prepareForMerge(ffmpegFile, segs, jsonlFile, danmuStartNs) defer cleanup() diff --git a/manualmerge.go b/manualmerge.go index 64f07a6..05ed739 100644 --- a/manualmerge.go +++ b/manualmerge.go @@ -471,7 +471,9 @@ func runManualMerge(plan mergePlan) error { for _, v := range plan.info.Videos { inputBytes += v.Size } - jobStart("手动合并", plan.info.OutName, inputBytes, plan.info.Duration) + if !jobStart("手动合并", plan.info.OutName, inputBytes, plan.info.Duration) { + return errors.New("已经有别的合并任务在跑,请先等它结束") + } src, err := prepareDanmuSource(plan.danmu, true) if err != nil { @@ -596,12 +598,23 @@ func mergeHandler(w http.ResponseWriter, r *http.Request) { http.Error(w, "已经有一个手动合并在进行("+busyOut+"),等它结束再说", http.StatusConflict) return } + // 任务卡片只有一份:自动合并也可能占着它,那时手动的也得等 + if !tryTakeMergeSlot() { + mergeRun.Unlock() + busy := jobBusyName() + if busy == "" { + busy = "另一个合并任务" + } + http.Error(w, "已经有一个合并在进行("+busy+"),等它结束再说", http.StatusConflict) + return + } mergeRun.busy = true mergeRun.outName = plan.info.OutName mergeRun.lastErr = "" mergeRun.Unlock() safeGo("手动合并", func() { + defer releaseMergeSlot() defer func() { mergeRun.Lock() mergeRun.busy = false diff --git a/manualmerge_test.go b/manualmerge_test.go index 137c52c..86bfa96 100644 --- a/manualmerge_test.go +++ b/manualmerge_test.go @@ -63,6 +63,16 @@ Dialogue: 0,0:00:03.50,0:00:13.50,Danmu,丙(303),20,20,2,,{\move(400,24,-40,24)} // 测试用的 ffmpeg,找不到就跳过 // 依次找 PATH 和源码目录(以及上一级目录)里的 ffmpeg,不写死某台机器上的路径 +// 按真实链路跑一次手动合并:独占任务位,跑完一定要收尾。 +// 不能直接调 runManualMerge —— 上一个用例如果没收尾,jobStart 会拒掉这一场 +// (正是它的职责),测试就会莫名其妙失败 +func execManualMerge(plan mergePlan) error { + jobFinish("") + err := runMergeExclusive(func() error { return runManualMerge(plan) }) + jobFinish(jobErrText(err)) + return err +} + func testFFmpeg(t *testing.T) string { t.Helper() cands := []string{"ffmpeg"} @@ -641,7 +651,7 @@ Dialogue: 0,0:00:05.00,0:00:15.00,Danmu,甲(101),20,20,2,,{\move(400,72,-40,72)} } // 真正合并 - if err := runManualMerge(plan); err != nil { + if err := execManualMerge(plan); err != nil { t.Fatalf("手动合并失败:%v", err) } if _, err := os.Stat(plan.out); err != nil { @@ -708,7 +718,7 @@ Dialogue: 0,0:00:05.00,0:00:15.00,Danmu,乙(202),0,0,0,,{\move(400,48,-40,48)} if filepath.Base(plan.videos[0]) != filepath.Base(v1) { t.Fatalf("分段顺序应纠正为第一段在前,实际 %q", filepath.Base(plan.videos[0])) } - if err := runManualMerge(plan); err != nil { + if err := execManualMerge(plan); err != nil { t.Fatalf("多段手动合并失败:%v", err) } d := probeDuration(ffmpeg, plan.out) diff --git a/progress.go b/progress.go index bf6f6c3..bd86862 100644 --- a/progress.go +++ b/progress.go @@ -1,10 +1,12 @@ // 转换任务(合并 / 烧录弹幕)的实时进度与暂停、停止控制。 -// 进度只保留一份全局状态:同一时间只允许一个合并任务在跑(mergeRun 已经拦住了)。 +// 进度只保留一份全局状态:同一时间只允许一个合并任务在跑,串行靠下面的 mergeSlot 保证 +// (mergeRun.busy 只拦得住手动合并之间,拦不住自动合并,也拦不住手动与自动互相穿插)。 package main import ( "encoding/json" "errors" + "fmt" "math" "net/http" "os/exec" @@ -77,7 +79,7 @@ type jobInfo struct { // 全局唯一的任务状态 type jobState struct { - sync.Mutex + sync.RWMutex active bool kind string outName string @@ -115,10 +117,49 @@ type jobState struct { var job jobState -// 开始一个新任务。kind 是任务类型,totalSec 是视频总时长(拿不到填 0) -func jobStart(kind, outName string, inputBytes int64, totalSec float64) { +// 合并任务的串行锁。全局 job 状态只有一份,两个任务同时跑的话后启动的会把前一个的 +// 进度和输出名清掉,「停止」也会杀掉别家任务的 ffmpeg,所以必须排队。 +// 手动合并用 TryLock(抢不到立刻告诉用户),自动合并用 Lock(后台任务,排队等就行)。 +var mergeSlot sync.Mutex + +// 尝试占用合并任务位,被占着返回 false。给手动合并用:不能让 HTTP 请求干等着 +func tryTakeMergeSlot() bool { + return mergeSlot.TryLock() +} + +// 占用合并任务位,被占着就等到前一个跑完。给自动合并用 +func takeMergeSlot() { + mergeSlot.Lock() +} + +func releaseMergeSlot() { + mergeSlot.Unlock() +} + +// 排队跑完一整个合并任务:先独占任务位,跑完后一定释放,被调用方 panic 也不漏。 +// 内部出错会被翻成 error 而不是让上层去 recover —— 任务卡片只有走到 jobFinish 才会收尾, +// 漏掉的话进度会永远停在界面上(之前 panic 后卡片一直转就是这么来的) +func runMergeExclusive(f func() error) (err error) { + takeMergeSlot() + defer releaseMergeSlot() + defer func() { + if p := recover(); p != nil { + err = fmt.Errorf("合并任务内部出错:%v", p) + lPrintErrf("合并任务内部出错,已兜住:%v", p) + } + }() + return f() +} + +// 开始一个新任务。kind 是任务类型,totalSec 是视频总时长(拿不到填 0)。 +// 已经有任务在跑时返回 false 并且什么都不动:调用方应该先拿到 mergeSlot, +// 这里再做一次判断是为了让误用(忘记排队)只会失败而不会互相覆盖 +func jobStart(kind, outName string, inputBytes int64, totalSec float64) bool { job.Lock() defer job.Unlock() + if job.active { + return false + } job.active = true job.kind = kind job.outName = outName @@ -143,6 +184,24 @@ func jobStart(kind, outName string, inputBytes int64, totalSec float64) { job.errText = "" job.message = "" job.doneAt = 0 + return true +} + +// 当前是否有任务在跑。给界面提示和「现在能不能接新任务」的判断用 +func jobActive() bool { + job.RLock() + defer job.RUnlock() + return job.active +} + +// 当前任务的名字(没任务时为空),拒绝新任务时用它告诉用户在忙什么 +func jobBusyName() string { + job.RLock() + defer job.RUnlock() + if !job.active { + return "" + } + return job.outName } // 任务结束。errText 为空表示成功,errJobStopped.Error() 表示被手动停止 @@ -168,13 +227,6 @@ func jobFinish(errText string) { } } -// 任务是否正在跑 -func jobActive() bool { - job.Lock() - defer job.Unlock() - return job.active -} - // 是否已经被要求停止 func jobStopRequested() bool { job.Lock() diff --git a/progress_test.go b/progress_test.go index fe85c52..71e0d18 100644 --- a/progress_test.go +++ b/progress_test.go @@ -3,10 +3,22 @@ package main import ( "errors" "strings" + "sync" "testing" "time" ) +// 测试用的「开一个新任务」。全局任务只有一份,用例之间必须清干净: +// 上一个没收尾的话 jobStart 会拒绝覆盖(这是刻意的防串台),断言就会读到别人的数据 +func startJobForTest(t *testing.T, kind, outName string, inputBytes int64, totalSec float64) { + t.Helper() + jobFinish("") + if !jobStart(kind, outName, inputBytes, totalSec) { + t.Fatalf("jobStart 失败:还残留着未结束的任务 %q", jobBusyName()) + } + t.Cleanup(func() { jobFinish("") }) +} + // 阶段权重加起来必须正好是 1,否则总进度永远到不了 100% func TestJobPhaseWeightsSumToOne(t *testing.T) { if len(jobPhaseNames) != len(jobPhaseWeights) { @@ -23,8 +35,7 @@ func TestJobPhaseWeightsSumToOne(t *testing.T) { // 总进度:走过的阶段按权重满算,当前阶段按比例折算 func TestJobPercentByPhase(t *testing.T) { - jobStart("测试", "x.mp4", 0, 100) - t.Cleanup(func() { jobFinish("") }) + startJobForTest(t, "测试", "x.mp4", 0, 100) if got := jobSnapshot().Percent; got != 0 { t.Errorf("刚开始时总进度应为 0,实际 %v", got) @@ -50,7 +61,7 @@ func TestJobPercentByPhase(t *testing.T) { // 阶段序号只往前走,重复设置同一阶段不会把进度打回 0 func TestJobPhaseOnlyMovesForward(t *testing.T) { - jobStart("测试", "x.mp4", 0, 100) + startJobForTest(t, "测试", "x.mp4", 0, 100) defer jobFinish("") jobSetPhase(jobPhaseRender) jobRender(60, 1, 25, 0) @@ -66,7 +77,7 @@ func TestJobPhaseOnlyMovesForward(t *testing.T) { // 渲染阶段的 ETA = 剩余时长 ÷ 倍速 func TestJobEtaFromSpeed(t *testing.T) { - jobStart("测试", "x.mp4", 0, 200) + startJobForTest(t, "测试", "x.mp4", 0, 200) defer jobFinish("") jobSetPhase(jobPhaseRender) jobRender(50, 2.0, 25, 0) @@ -78,7 +89,7 @@ func TestJobEtaFromSpeed(t *testing.T) { // 预计输出大小:先给按码率估的值,渲染过 5% 之后换成按已写出字节外推 func TestJobEstOutSize(t *testing.T) { - jobStart("测试", "x.mp4", 0, 100) + startJobForTest(t, "测试", "x.mp4", 0, 100) defer jobFinish("") jobSetPhase(jobPhaseRender) @@ -99,7 +110,7 @@ func TestJobEstOutSize(t *testing.T) { } // 比例太小的时候不急着外推,免得开头抖动 - jobStart("测试", "x.mp4", 0, 100) + startJobForTest(t, "测试", "x.mp4", 0, 100) jobSetPhase(jobPhaseRender) jobPlanSize(5000) jobRender(1, 1, 25, 100) @@ -110,7 +121,7 @@ func TestJobEstOutSize(t *testing.T) { // 速度折线:最多保留固定点数,且按秒采样 func TestJobSamples(t *testing.T) { - jobStart("测试", "x.mp4", 0, 100) + startJobForTest(t, "测试", "x.mp4", 0, 100) defer jobFinish("") jobSetPhase(jobPhaseRender) // 同一秒内连着报多次只留一个点 @@ -136,7 +147,7 @@ func TestJobSamples(t *testing.T) { // 暂停 / 恢复 / 停止的状态流转;停止后闸门必须让上层拿到 errJobStopped func TestJobPauseStopGate(t *testing.T) { - jobStart("测试", "x.mp4", 0, 100) + startJobForTest(t, "测试", "x.mp4", 0, 100) defer jobFinish("") if err := jobPause(); err != nil { @@ -281,7 +292,7 @@ func TestScanFFmpegProgressTolerant(t *testing.T) { // 截图式地看一眼任务快照的字段都填上了 func TestJobSnapshotFields(t *testing.T) { - jobStart("手动合并", "成片.mp4", 123456, 3600) + startJobForTest(t, "手动合并", "成片.mp4", 123456, 3600) defer jobFinish("") jobSetPhase(jobPhaseAvatar) jobAvatars(3, 10) @@ -306,3 +317,99 @@ func TestJobSnapshotFields(t *testing.T) { t.Error("任务在跑的时候应该可暂停") } } + +// ==================== 任务串行化 ==================== +// 全局任务状态只有一份:两个合并同时跑的话,后启动的会把前一个的进度和输出名清掉, +// 点「停止」还会杀掉别家任务的 ffmpeg。下面几条例外情况锁住改串行之后的语义。 + +// jobStart 在已经有任务在跑时必须拒绝,并且一个字段都不能动 +func TestJobStartRejectsWhenBusy(t *testing.T) { + startJobForTest(t, "自动合并", "甲_弹幕合并.mp4", 1000, 100) + jobSetPhase(jobPhaseRender) + jobRender(30, 1.2, 25, 5000) + + if ok := jobStart("手动合并", "乙_弹幕合并.mp4", 2000, 200); ok { + t.Fatal("已经有任务在跑,jobStart 应当返回 false") + } + info := jobSnapshot() + if info.Kind != "自动合并" || info.OutName != "甲_弹幕合并.mp4" { + t.Errorf("被拒之后不该动别人的任务信息:%+v", info) + } + if info.InputBytes != 1000 || info.TotalSec != 100 { + t.Errorf("被拒之后输入大小和总时长也不该变:%+v", info) + } + if info.Percent < 25 || info.Percent > 45 { + t.Errorf("被拒之后进度必须还是原任务自己的(25~45),实际 %v", info.Percent) + } + if !info.Active { + t.Error("拒绝新任务不代表把老任务结束掉") + } +} + +// 任务位被占着时手动合并要立刻知道(不能让页面干等),抢不到只能有一个赢 +func TestMergeSlotTryLockExclusive(t *testing.T) { + if !tryTakeMergeSlot() { + t.Fatal("空闲时手动合并应该抢得到任务位") + } + if tryTakeMergeSlot() { + t.Fatal("任务位已经被占着,第二次 TryLock 不该成功") + } + releaseMergeSlot() + if !tryTakeMergeSlot() { + t.Fatal("释放之后应该又能抢到") + } + releaseMergeSlot() +} + +// 自动合并(十几个主播同时下播很常见)必须排队而不是并发 +func TestMergeSlotSerializes(t *testing.T) { + if !tryTakeMergeSlot() { + t.Fatal("前置条件:任务位本该是空的") + } + const n = 4 + order := make([]int, 0, n) + var mu sync.Mutex + done := make(chan struct{}, n) + for i := 0; i < n; i++ { + idx := i + go func() { + defer func() { done <- struct{}{} }() + // 和真实链路一致:自动合并会一直等到轮到自己 + _ = runMergeExclusive(func() error { + mu.Lock() + order = append(order, idx) + mu.Unlock() + time.Sleep(20 * time.Millisecond) + return nil + }) + }() + } + time.Sleep(60 * time.Millisecond) // 让它们先堆在门外 + releaseMergeSlot() + for i := 0; i < n; i++ { + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatalf("第 %d 个任务没跑完,队列卡住了", i) + } + } + if len(order) != n { + t.Fatalf("应当 %d 个任务都跑完,实际 %d 个", n, len(order)) + } +} + +// 任务内部 panic 时也必须释放任务位、并且能翻成错误 +// —— 否则卡片会永远停在「进行中」,队列也被锁死 +func TestMergeExclusiveReleasesOnPanic(t *testing.T) { + err := runMergeExclusive(func() error { panic("模拟内部出错") }) + if err == nil { + t.Fatal("内部 panic 应当被翻成错误返回") + } + if !strings.Contains(err.Error(), "模拟内部出错") { + t.Errorf("错误信息里应带上原因,实际:%v", err) + } + if !tryTakeMergeSlot() { + t.Fatal("panic 之后任务位必须已经释放") + } + releaseMergeSlot() +}