acfunlive-remake/progress.go

// 转换任务（合并 / 烧录弹幕）的实时进度与暂停、停止控制。
// 进度只保留一份全局状态：同一时间只允许一个合并任务在跑，串行靠下面的 mergeSlot 保证
// （mergeRun.busy 只拦得住手动合并之间，拦不住自动合并，也拦不住手动与自动互相穿插）。
package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"net/http"
	"os/exec"
	"sync"
	"time"
)

// 任务类型。阶段划分按它来取（见 jobPhasePlanFor）
const (
	jobKindManualMerge = "手动合并"
	jobKindAutoMerge   = "自动合并"
	jobKindTranscode   = "转码"
)

// 任务阶段，前端按这个顺序画步骤条
var jobPhaseNames = []string{"准备", "分段拼接", "获取头像", "画面渲染", "拼接与落盘"}

// 阶段序号，和 jobPhaseNames 一一对应
const (
	jobPhasePrepare = iota // 准备（读弹幕、探测视频、生成字幕）
	jobPhaseJoin           // 分段拼接（断流分成多段时才有事做）
	jobPhaseAvatar         // 获取头像
	jobPhaseRender         // 画面渲染（最耗时）
	jobPhasePlace          // 拼接分块与落盘
)

// 各阶段在总进度里占的权重（加起来是 1）。渲染最耗时，权重给得最多；
// 其余阶段按经验估：拼接分段与头像下载是分钟级，准备和落盘是秒级。
var jobPhaseWeights = []float64{0.02, 0.08, 0.10, 0.75, 0.05}

// 一类任务的阶段划分。转码没有弹幕和头像，步骤比合并少，硬套合并那套步骤条
// 会冒出两个永远走不到的空阶段
type jobPhasePlan struct {
	names   []string
	weights []float64
	render  int // 哪个阶段是真正耗时的主阶段，ETA 按它算
}

// 合并类任务的阶段（默认），转码类任务的阶段
var (
	mergePhasePlan = jobPhasePlan{names: jobPhaseNames, weights: jobPhaseWeights, render: jobPhaseRender}
	// 转码：探测一下参数就开始编，编完落到成片目录
	transcodePhasePlan = jobPhasePlan{
		names:   []string{"准备", "转码", "落盘"},
		weights: []float64{0.03, 0.92, 0.05},
		render:  1,
	}
)

// 转码任务阶段里「转码」这一步的序号
const jobPhaseTranscode = 1

// 按任务类型取阶段划分，没见过的类型按合并处理
func jobPhasePlanFor(kind string) jobPhasePlan {
	if kind == jobKindTranscode {
		return transcodePhasePlan
	}
	return mergePhasePlan
}

// 渲染进度的采样间隔与折线最多保留的点数（1 秒 1 个，够画两分钟）
const (
	jobSampleInterval = time.Second
	jobMaxSamples     = 120
)

// 任务被手动停止时各层返回这个错误：调用方据此清理临时文件，不再回退其它编码器
var errJobStopped = errors.New("任务已被手动停止")

// 起任务时任务位被别人占着：这次没有真的开始，调用方据此跳过 jobFinish，
// 不然会把正在跑的那个任务的进度卡片清掉
var errJobNotStarted = errors.New("已经有别的任务在跑，请先等它结束")

// 速度折线上的一个点
type jobSample struct {
	T     float64 `json:"t"`     // 任务开始后的秒数
	Speed float64 `json:"speed"` // 当时的渲染倍速
}

// 给前端的任务快照
type jobInfo struct {
	Active       bool        `json:"active"`
	Kind         string      `json:"kind"`     // 手动合并 / 自动合并
	OutName      string      `json:"outName"`  // 成片文件名
	PhaseIdx     int         `json:"phaseIdx"` // 当前阶段序号，-1 表示已全部完成
	Phases       []string    `json:"phases"`   // 阶段名列表
	PhasePercent float64     `json:"phasePercent"`
	Percent      float64     `json:"percent"` // 总进度 0~100
	Paused       bool        `json:"paused"`
	Stopping     bool        `json:"stopping"`
	Speed        float64     `json:"speed"` // 渲染倍速，例如 1.37
	FPS          float64     `json:"fps"`
	DoneSec      float64     `json:"doneSec"`  // 已渲染时长
	TotalSec     float64     `json:"totalSec"` // 视频总时长
	ChunkDone    int         `json:"chunkDone"`
	ChunkTotal   int         `json:"chunkTotal"`
	AvatarDone   int         `json:"avatarDone"`
	AvatarTotal  int         `json:"avatarTotal"`
	QueueDone    int         `json:"queueDone"`   // 批量任务：已经做完第几个（从 1 开始），单任务为 0
	QueueTotal   int         `json:"queueTotal"`  // 批量任务：一共几个
	InputBytes   int64       `json:"inputBytes"`  // 源视频总大小
	EstOutBytes  int64       `json:"estOutBytes"` // 预计成片大小
	EstSource    string      `json:"estSource"`   // 预计值的来源：bitrate / measured
	OutBytes     int64       `json:"outBytes"`    // 已经写出多少字节
	ETASec       float64     `json:"etaSec"`      // 预计剩余秒数，0 表示算不出来
	ElapsedSec   float64     `json:"elapsedSec"`  // 已用时间
	Samples      []jobSample `json:"samples"`     // 速度折线
	Message      string      `json:"message"`     // 结束后的提示（完成 / 失败原因）
	Err          string      `json:"err"`         // 失败原因，成功为空
	DoneAt       int64       `json:"doneAt"`      // 结束时刻（Unix 秒）
	Pausable     bool        `json:"pausable"`    // 现在是否能暂停
}

// 全局唯一的任务状态
type jobState struct {
	sync.RWMutex
	active   bool
	kind     string
	outName  string
	phaseIdx int
	phasePct float64
	started  time.Time
	paused   bool
	stopped  bool
	proc     *exec.Cmd // 当前在跑的 ffmpeg，暂停/停止都作用在它身上

	// 当前任务的阶段划分，随任务类型在 jobStart 时定下来
	phases    []string
	weights   []float64
	renderIdx int
	// 批量任务（手动合并队列、转码队列）的进度，单任务时都是 0
	queueDone  int
	queueTotal int

	totalSec float64
	doneSec  float64
	speed    float64
	fps      float64

	chunkDone   int
	chunkTotal  int
	avatarDone  int
	avatarTotal int

	inputBytes int64
	// 按目标码率算出来的预计成片大小
	planOutBytes int64
	// 按「已写出字节 ÷ 已渲染比例」外推出来的预计成片大小，渲染到 5% 后才有值
	liveOutBytes int64
	outBytes     int64

	samples    []jobSample
	lastSample time.Time

	errText string
	message string
	doneAt  int64
}

var job jobState

// 合并任务的串行锁。全局 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
	plan := jobPhasePlanFor(kind)
	job.phases = plan.names
	job.weights = plan.weights
	job.renderIdx = plan.render
	job.phaseIdx = 0
	job.phasePct = 0
	job.started = time.Now()
	job.paused = false
	job.stopped = false
	job.proc = nil
	job.totalSec = totalSec
	job.doneSec = 0
	job.speed = 0
	job.fps = 0
	job.chunkDone, job.chunkTotal = 0, 0
	job.avatarDone, job.avatarTotal = 0, 0
	job.queueDone, job.queueTotal = 0, 0
	job.inputBytes = inputBytes
	job.planOutBytes = 0
	job.liveOutBytes = 0
	job.outBytes = 0
	job.samples = nil
	job.lastSample = time.Time{}
	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() 表示被手动停止
func jobFinish(errText string) {
	job.Lock()
	defer job.Unlock()
	job.active = false
	job.paused = false
	job.stopped = false
	job.proc = nil
	job.speed = 0
	job.errText = errText
	job.doneAt = time.Now().Unix()
	switch errText {
	case "":
		job.phaseIdx = len(job.phases)
		job.phasePct = 1
		job.message = "已完成"
	case errJobStopped.Error():
		job.message = "已停止"
	default:
		job.message = "失败"
	}
}

// 是否已经被要求停止
func jobStopRequested() bool {
	job.Lock()
	defer job.Unlock()
	return job.stopped
}

// 各阶段在进入下一步前调用这个「闸门」：暂停时在这里等，被停止时立刻返回错误。
// 放在循环体里，这样下载头像、逐块渲染这些阶段都能被暂停和停止。
func jobGate() error {
	for {
		job.Lock()
		stopped, paused := job.stopped, job.paused
		job.Unlock()
		if stopped {
			return errJobStopped
		}
		if !paused {
			return nil
		}
		time.Sleep(120 * time.Millisecond)
	}
}

// 切换阶段。序号只往前走，重复设置同一阶段不会把进度打回 0
func jobSetPhase(idx int) {
	job.Lock()
	defer job.Unlock()
	if !job.active {
		return
	}
	if idx > job.phaseIdx {
		job.phaseIdx = idx
		job.phasePct = 0
	}
}

// 直接设置当前阶段的完成比例（0~1），只取更大值，避免折返
func jobSetPhasePercent(p float64) {
	job.Lock()
	defer job.Unlock()
	if !job.active {
		return
	}
	if p > job.phasePct {
		job.phasePct = clamp01(p)
	}
}

// 渲染进度：doneSec 是累计已渲染的秒数（分块时已折算成整片时间轴），
// cumBytes 是累计已写出的字节数（分块时是「已完成块 + 当前块」）
func jobRender(doneSec, speed, fps float64, cumBytes int64) {
	job.Lock()
	defer job.Unlock()
	if !job.active {
		return
	}
	if doneSec > job.doneSec {
		job.doneSec = doneSec
	}
	if speed > 0 {
		job.speed = speed
	}
	if fps > 0 {
		job.fps = fps
	}
	if cumBytes > job.outBytes {
		job.outBytes = cumBytes
	}
	ratio := 0.0
	if job.totalSec > 0 {
		ratio = clamp01(job.doneSec / job.totalSec)
	}
	if ratio > job.phasePct {
		job.phasePct = ratio
	}
	// 渲染到 5% 之后，用「已写出 ÷ 已完成比例」外推整片体积，比按码率估的准
	if ratio >= 0.05 && job.outBytes > 0 {
		if est := int64(float64(job.outBytes) / ratio); est > 0 {
			job.liveOutBytes = est
		}
	}
	// 速度折线：最多 1 秒一个点
	now := time.Now()
	if job.lastSample.IsZero() || now.Sub(job.lastSample) >= jobSampleInterval {
		job.lastSample = now
		job.samples = append(job.samples, jobSample{T: now.Sub(job.started).Seconds(), Speed: speed})
		if len(job.samples) > jobMaxSamples {
			job.samples = append([]jobSample(nil), job.samples[len(job.samples)-jobMaxSamples:]...)
		}
	}
}

// 已分块渲染的块数，给界面显示「第 3/12 块」
func jobChunks(done, total int) {
	job.Lock()
	defer job.Unlock()
	if !job.active {
		return
	}
	job.chunkDone, job.chunkTotal = done, total
}

// 头像下载进度（按用户数）
func jobAvatars(done, total int) {
	job.Lock()
	defer job.Unlock()
	if !job.active {
		return
	}
	job.avatarDone, job.avatarTotal = done, total
	if total > 0 {
		p := float64(done) / float64(total)
		if p > job.phasePct {
			job.phasePct = clamp01(p)
		}
	}
}

// 批量任务（手动合并队列、转码队列）排到第几个。done 从 1 开始，total 是总数
func jobQueue(done, total int) {
	job.Lock()
	defer job.Unlock()
	if !job.active {
		return
	}
	job.queueDone, job.queueTotal = done, total
}

// 设置任务总时长（视频秒数），ETA 和渲染百分比都靠它
func jobSetTotal(sec float64) {
	job.Lock()
	defer job.Unlock()
	if !job.active {
		return
	}
	job.totalSec = sec
}

// 按目标码率算出来的预计成片大小（byte）
func jobPlanSize(bytes int64) {
	job.Lock()
	defer job.Unlock()
	if !job.active {
		return
	}
	job.planOutBytes = bytes
}

// 把当前 ffmpeg 进程注册进来，暂停/停止才作用得到。
// 如果注册时任务已经处于暂停状态，直接把它挂起（在启动前点了暂停的情况）；
// 如果任务已经被要求停止，就地杀掉 —— jobGate 已经过去了，不补这一刀的话
// 用户点了停止还得等整个块渲完（大块的渲染可能要几分钟）
func jobAttach(cmd *exec.Cmd) {
	job.Lock()
	job.proc = cmd
	paused, stopped := job.paused, job.stopped
	job.Unlock()
	if cmd == nil || cmd.Process == nil {
		return
	}
	if stopped {
		_ = cmd.Process.Kill()
		return
	}
	if paused {
		_ = suspendProcess(cmd.Process)
	}
}

func jobDetach(cmd *exec.Cmd) {
	job.Lock()
	if job.proc == cmd {
		job.proc = nil
	}
	job.Unlock()
}

// 暂停：挂起当前 ffmpeg 进程；不在 ffmpeg 阶段时靠 jobGate 在下一步生效
func jobPause() error {
	job.Lock()
	if !job.active {
		job.Unlock()
		return errors.New("当前没有正在运行的合并任务")
	}
	if job.paused {
		job.Unlock()
		return nil
	}
	job.paused = true
	cmd := job.proc
	job.Unlock()
	if cmd != nil && cmd.Process != nil {
		if err := suspendProcess(cmd.Process); err != nil {
			// 没挂住就别留着「已暂停」：界面会显示暂停了、ffmpeg 其实还在跑，
			// 用户以为停住了却在烧 CPU。只在进程还是同一个时回滚，免得踩到已经换过的进程
			job.Lock()
			if job.proc == cmd {
				job.paused = false
			}
			job.Unlock()
			return err
		}
	}
	return nil
}

// 恢复
func jobResume() error {
	job.Lock()
	if !job.active {
		job.Unlock()
		return errors.New("当前没有正在运行的合并任务")
	}
	if !job.paused {
		job.Unlock()
		return nil
	}
	job.paused = false
	cmd := job.proc
	job.Unlock()
	if cmd != nil && cmd.Process != nil {
		if err := resumeProcess(cmd.Process); err != nil {
			// 没恢复成就把状态放回「已暂停」：否则界面显示在跑、ffmpeg 其实还停着
			job.Lock()
			if job.proc == cmd {
				job.paused = true
			}
			job.Unlock()
			return err
		}
	}
	return nil
}

// 停止：杀掉当前 ffmpeg，并置位让各层在下一个闸门返回 errJobStopped
func jobStop() error {
	job.Lock()
	if !job.active {
		job.Unlock()
		return errors.New("当前没有正在运行的合并任务")
	}
	job.stopped = true
	job.paused = false
	cmd := job.proc
	job.Unlock()
	if cmd != nil && cmd.Process != nil {
		_ = cmd.Process.Kill()
	}
	return nil
}

func clamp01(v float64) float64 {
	if v < 0 {
		return 0
	}
	if v > 1 {
		return 1
	}
	return v
}

// 总进度：已走完的阶段按权重全算，当前阶段按阶段内比例折算
func (j *jobState) percentLocked() float64 {
	weights := j.weights
	if len(weights) == 0 {
		weights = jobPhaseWeights
	}
	p := 0.0
	for i, w := range weights {
		switch {
		case i < j.phaseIdx:
			p += w
		case i == j.phaseIdx:
			p += w * clamp01(j.phasePct)
		}
	}
	return math.Min(p*100, 100)
}

// 预计剩余时间：渲染阶段用「剩余时长 ÷ 倍速」，其它阶段按已用时间线性外推
func (j *jobState) etaLocked(now time.Time) float64 {
	if !j.active {
		return 0
	}
	if j.phaseIdx == j.renderIdx && j.speed > 0 && j.totalSec > j.doneSec {
		return (j.totalSec - j.doneSec) / j.speed
	}
	pct := j.percentLocked()
	if pct <= 0.5 {
		return 0
	}
	return now.Sub(j.started).Seconds() * (100 - pct) / pct
}

// 给页面的快照
func jobSnapshot() jobInfo {
	job.Lock()
	defer job.Unlock()

	info := jobInfo{
		Active:      job.active,
		Kind:        job.kind,
		OutName:     job.outName,
		PhaseIdx:    job.phaseIdx,
		Phases:      job.phases,
		Paused:      job.paused,
		Stopping:    job.stopped,
		Speed:       job.speed,
		FPS:         job.fps,
		DoneSec:     job.doneSec,
		TotalSec:    job.totalSec,
		ChunkDone:   job.chunkDone,
		ChunkTotal:  job.chunkTotal,
		AvatarDone:  job.avatarDone,
		AvatarTotal: job.avatarTotal,
		QueueDone:   job.queueDone,
		QueueTotal:  job.queueTotal,
		InputBytes:  job.inputBytes,
		OutBytes:    job.outBytes,
		Message:     job.message,
		Err:         job.errText,
		DoneAt:      job.doneAt,
	}
	info.PhasePercent = clamp01(job.phasePct) * 100
	info.Percent = job.percentLocked()
	// 预计成片大小：有实测外推就用实测，否则用按码率估的
	info.EstOutBytes, info.EstSource = job.planOutBytes, "bitrate"
	if job.liveOutBytes > 0 {
		info.EstOutBytes, info.EstSource = job.liveOutBytes, "measured"
	}
	now := time.Now()
	info.ETASec = job.etaLocked(now)
	if job.active {
		info.ElapsedSec = now.Sub(job.started).Seconds()
	} else if job.doneAt > 0 && !job.started.IsZero() {
		info.ElapsedSec = time.Unix(job.doneAt, 0).Sub(job.started).Seconds()
	}
	if len(job.samples) > 0 {
		info.Samples = append([]jobSample(nil), job.samples...)
	}
	// 暂停靠 jobGate 生效：ffmpeg 在跑时直接挂起进程，不在跑时在下一个闸门处停住
	info.Pausable = job.active
	return info
}

// GET /api/job：当前转换任务的快照，页面轮询它来画进度
func jobHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(jobSnapshot()); err != nil {
		lPrintErrf("输出任务进度失败：%v", err)
	}
}

// POST /api/job/control：暂停 / 恢复 / 停止当前任务
func jobControlHandler(w http.ResponseWriter, r *http.Request) {
	var req struct {
		Action string `json:"action"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "请求不是合法的 json："+err.Error(), http.StatusBadRequest)
		return
	}
	var err error
	switch req.Action {
	case "pause":
		err = jobPause()
	case "resume":
		err = jobResume()
	case "stop":
		err = jobStop()
	default:
		http.Error(w, "不认识的操作："+req.Action, http.StatusBadRequest)
		return
	}
	if err != nil {
		http.Error(w, err.Error(), http.StatusConflict)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_, _ = w.Write([]byte(`{"ok":true}`))
}