add: queue total ETA from current encode speed
The queue card now shows an estimated time to clear: current job remaining + sum(pending source durations) / current speed. Durations come from a path+mtime-keyed cache that probes each file once in the background (ffprobe format=duration, header-only), so the 1s poll never re-probes; entries are pruned to the live queue. No speed (idle/held) -> no estimate; files still being probed mark it partial (shown as '~Xh+'). Closes #9
This commit is contained in:
+6
-1
@@ -113,7 +113,12 @@ func main() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if addr := *cfg.HTTPAddr; addr != "" {
|
if addr := *cfg.HTTPAddr; addr != "" {
|
||||||
srv := &http.Server{Addr: addr, Handler: server.New(tracker, log, store, controls, cfg.Paths.Input, cfg.Paths.Failed).Handler()}
|
probeDuration := func(p string) (float64, error) {
|
||||||
|
pctx, pcancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||||
|
defer pcancel()
|
||||||
|
return enc.GetDuration(pctx, p)
|
||||||
|
}
|
||||||
|
srv := &http.Server{Addr: addr, Handler: server.New(tracker, log, store, controls, probeDuration, cfg.Paths.Input, cfg.Paths.Failed).Handler()}
|
||||||
go func() {
|
go func() {
|
||||||
log.Info(fmt.Sprintf("Status server listening on %s", addr))
|
log.Info(fmt.Sprintf("Status server listening on %s", addr))
|
||||||
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||||
|
|||||||
@@ -97,6 +97,7 @@
|
|||||||
.qsum { display:flex; align-items:baseline; gap:.5rem; flex-wrap:wrap; margin-bottom:.8rem; }
|
.qsum { display:flex; align-items:baseline; gap:.5rem; flex-wrap:wrap; margin-bottom:.8rem; }
|
||||||
.qsum .qn { font-family:var(--mono); font-size:1.4rem; color:var(--amber); line-height:1; }
|
.qsum .qn { font-family:var(--mono); font-size:1.4rem; color:var(--amber); line-height:1; }
|
||||||
.qsum .ql { color:var(--muted); font-size:.85rem; }
|
.qsum .ql { color:var(--muted); font-size:.85rem; }
|
||||||
|
.qsum .qeta { font-family:var(--mono); font-size:.8rem; color:var(--amber-soft); margin-left:auto; }
|
||||||
ul.list { list-style:none; margin:0; padding:0; font-family:var(--mono); font-size:.82rem; }
|
ul.list { list-style:none; margin:0; padding:0; font-family:var(--mono); font-size:.82rem; }
|
||||||
ul.list li { padding:.35rem 0; border-bottom:1px solid var(--line); color:var(--muted); word-break:break-word; }
|
ul.list li { padding:.35rem 0; border-bottom:1px solid var(--line); color:var(--muted); word-break:break-word; }
|
||||||
ul.list li:last-child { border-bottom:0; }
|
ul.list li:last-child { border-bottom:0; }
|
||||||
@@ -325,7 +326,8 @@
|
|||||||
for (const s of arr) { while (!s.startsWith(p)) p = p.slice(0, -1); if (!p) break; }
|
for (const s of arr) { while (!s.startsWith(p)) p = p.slice(0, -1); if (!p) break; }
|
||||||
return p;
|
return p;
|
||||||
}
|
}
|
||||||
function renderQueue(queue, current) {
|
function renderQueue(d) {
|
||||||
|
const queue = d.queue, current = d.current;
|
||||||
const cur = base(current && current.file);
|
const cur = base(current && current.file);
|
||||||
const pending = (queue || []).filter(f => f !== cur);
|
const pending = (queue || []).filter(f => f !== cur);
|
||||||
const el = $("queue");
|
const el = $("queue");
|
||||||
@@ -334,7 +336,10 @@
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const pre = pending.length > 1 ? commonPrefix(pending) : "";
|
const pre = pending.length > 1 ? commonPrefix(pending) : "";
|
||||||
const head = `<div class="qsum"><span class="qn">${pending.length}</span><span class="ql">${pending.length === 1 ? "file" : "files"} waiting</span></div>`;
|
const eta = d.queue_eta_sec > 0
|
||||||
|
? `<span class="qeta">~${fmtDurLong(d.queue_eta_sec)}${d.queue_eta_partial ? "+" : ""} to clear</span>`
|
||||||
|
: "";
|
||||||
|
const head = `<div class="qsum"><span class="qn">${pending.length}</span><span class="ql">${pending.length === 1 ? "file" : "files"} waiting</span>${eta}</div>`;
|
||||||
const rows = pending.slice(0, 3).map(f => {
|
const rows = pending.slice(0, 3).map(f => {
|
||||||
const tail = pre ? f.slice(pre.length) : f;
|
const tail = pre ? f.slice(pre.length) : f;
|
||||||
return `<li>${pre ? `<span class="pre">${esc(pre)}</span>` : ""}<span class="key">${esc(tail)}</span></li>`;
|
return `<li>${pre ? `<span class="pre">${esc(pre)}</span>` : ""}<span class="key">${esc(tail)}</span></li>`;
|
||||||
@@ -429,7 +434,7 @@
|
|||||||
renderControls(d);
|
renderControls(d);
|
||||||
renderJob(d.current);
|
renderJob(d.current);
|
||||||
renderFailed(d.failed);
|
renderFailed(d.failed);
|
||||||
renderQueue(d.queue, d.current);
|
renderQueue(d);
|
||||||
renderEvents(d.recent);
|
renderEvents(d.recent);
|
||||||
const active = d.current && d.current.file && d.current.phase !== "idle";
|
const active = d.current && d.current.file && d.current.phase !== "idle";
|
||||||
setLive(d.current && d.current.paused ? "idle" : active ? "ok" : "idle");
|
setLive(d.current && d.current.paused ? "idle" : active ? "ok" : "idle");
|
||||||
|
|||||||
+122
-12
@@ -6,7 +6,10 @@ import (
|
|||||||
_ "embed"
|
_ "embed"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"av1dae/internal/logger"
|
"av1dae/internal/logger"
|
||||||
"av1dae/internal/settings"
|
"av1dae/internal/settings"
|
||||||
@@ -35,12 +38,91 @@ type Server struct {
|
|||||||
log *logger.Logger
|
log *logger.Logger
|
||||||
store *settings.Store
|
store *settings.Store
|
||||||
controls Controls
|
controls Controls
|
||||||
|
durCache *durationCache
|
||||||
inputDir string
|
inputDir string
|
||||||
failedDir string
|
failedDir string
|
||||||
}
|
}
|
||||||
|
|
||||||
func New(tracker *status.Tracker, log *logger.Logger, store *settings.Store, controls Controls, inputDir, failedDir string) *Server {
|
func New(tracker *status.Tracker, log *logger.Logger, store *settings.Store, controls Controls, probeDuration func(string) (float64, error), inputDir, failedDir string) *Server {
|
||||||
return &Server{tracker: tracker, log: log, store: store, controls: controls, inputDir: inputDir, failedDir: failedDir}
|
return &Server{
|
||||||
|
tracker: tracker,
|
||||||
|
log: log,
|
||||||
|
store: store,
|
||||||
|
controls: controls,
|
||||||
|
durCache: newDurationCache(probeDuration),
|
||||||
|
inputDir: inputDir,
|
||||||
|
failedDir: failedDir,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type durEntry struct {
|
||||||
|
mtime time.Time
|
||||||
|
sec float64
|
||||||
|
}
|
||||||
|
|
||||||
|
// durationCache memoizes source durations (keyed by path+mtime) so the queue
|
||||||
|
// ETA doesn't re-probe every file on every 1s poll. A miss kicks a background
|
||||||
|
// ffprobe and resolves on a later poll; the file shows as "estimating" until.
|
||||||
|
type durationCache struct {
|
||||||
|
probe func(string) (float64, error)
|
||||||
|
mu sync.Mutex
|
||||||
|
m map[string]durEntry
|
||||||
|
inflight map[string]bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func newDurationCache(probe func(string) (float64, error)) *durationCache {
|
||||||
|
return &durationCache{probe: probe, m: map[string]durEntry{}, inflight: map[string]bool{}}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *durationCache) Get(path string, mtime time.Time) (float64, bool) {
|
||||||
|
c.mu.Lock()
|
||||||
|
if e, ok := c.m[path]; ok && e.mtime.Equal(mtime) {
|
||||||
|
c.mu.Unlock()
|
||||||
|
return e.sec, true
|
||||||
|
}
|
||||||
|
if c.inflight[path] {
|
||||||
|
c.mu.Unlock()
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
c.inflight[path] = true
|
||||||
|
c.mu.Unlock()
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
sec, err := c.probe(path)
|
||||||
|
c.mu.Lock()
|
||||||
|
delete(c.inflight, path)
|
||||||
|
if err == nil {
|
||||||
|
c.m[path] = durEntry{mtime, sec}
|
||||||
|
}
|
||||||
|
c.mu.Unlock()
|
||||||
|
}()
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
|
||||||
|
// Retain drops cached entries for paths no longer present, bounding the map on
|
||||||
|
// a long-running daemon.
|
||||||
|
func (c *durationCache) Retain(paths []string) {
|
||||||
|
keep := make(map[string]bool, len(paths))
|
||||||
|
for _, p := range paths {
|
||||||
|
keep[p] = true
|
||||||
|
}
|
||||||
|
c.mu.Lock()
|
||||||
|
for p := range c.m {
|
||||||
|
if !keep[p] {
|
||||||
|
delete(c.m, p)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
c.mu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// queueETASeconds estimates wall-clock time to clear the queue: the current
|
||||||
|
// job's remaining time plus each pending file's duration / current speed.
|
||||||
|
// 0 when there's no speed to extrapolate from (idle/held).
|
||||||
|
func queueETASeconds(currentETASec int, sumQueuedSec, speed float64) int {
|
||||||
|
if speed <= 0 {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
return currentETASec + int(sumQueuedSec/speed)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Handler returns the mux for all status routes. Phase 3 adds "/" (the HTML
|
// Handler returns the mux for all status routes. Phase 3 adds "/" (the HTML
|
||||||
@@ -157,11 +239,13 @@ func (s *Server) handleIndex(w http.ResponseWriter, r *http.Request) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type statusResponse struct {
|
type statusResponse struct {
|
||||||
Running bool `json:"running"`
|
Running bool `json:"running"`
|
||||||
Current status.Snapshot `json:"current"`
|
Current status.Snapshot `json:"current"`
|
||||||
Queue []string `json:"queue"`
|
Queue []string `json:"queue"`
|
||||||
Failed []string `json:"failed"`
|
Failed []string `json:"failed"`
|
||||||
Recent []logger.LogEntry `json:"recent"`
|
QueueETASec int `json:"queue_eta_sec"`
|
||||||
|
QueueETAPartial bool `json:"queue_eta_partial"`
|
||||||
|
Recent []logger.LogEntry `json:"recent"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
|
func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
|
||||||
@@ -171,13 +255,39 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
snap := s.tracker.Snapshot()
|
||||||
|
files := watcher.InputFiles(s.inputDir)
|
||||||
|
|
||||||
|
// Sum durations of pending files (excluding the current job, whose remaining
|
||||||
|
// time is already in snap.ETASec). Unprobed files mark the estimate partial.
|
||||||
|
var sumQueued float64
|
||||||
|
partial := false
|
||||||
|
for _, f := range files {
|
||||||
|
if f == snap.File {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
info, statErr := os.Stat(f)
|
||||||
|
if statErr != nil {
|
||||||
|
partial = true
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if sec, ok := s.durCache.Get(f, info.ModTime()); ok {
|
||||||
|
sumQueued += sec
|
||||||
|
} else {
|
||||||
|
partial = true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
s.durCache.Retain(files)
|
||||||
|
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
_ = json.NewEncoder(w).Encode(statusResponse{
|
_ = json.NewEncoder(w).Encode(statusResponse{
|
||||||
Running: s.controls.Running(),
|
Running: s.controls.Running(),
|
||||||
Current: s.tracker.Snapshot(),
|
Current: snap,
|
||||||
Queue: baseNames(watcher.InputFiles(s.inputDir)),
|
Queue: baseNames(files),
|
||||||
Failed: baseNames(watcher.InputFiles(s.failedDir)),
|
Failed: baseNames(watcher.InputFiles(s.failedDir)),
|
||||||
Recent: recent,
|
QueueETASec: queueETASeconds(snap.ETASec, sumQueued, snap.Speed),
|
||||||
|
QueueETAPartial: partial,
|
||||||
|
Recent: recent,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,62 @@
|
|||||||
|
package server
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestQueueETASeconds(t *testing.T) {
|
||||||
|
// No speed → can't extrapolate.
|
||||||
|
if got := queueETASeconds(100, 3600, 0); got != 0 {
|
||||||
|
t.Errorf("speed 0: got %d, want 0", got)
|
||||||
|
}
|
||||||
|
// current remaining 600s + 1800s of queued source at 0.5x (=3600s) = 4200s.
|
||||||
|
if got := queueETASeconds(600, 1800, 0.5); got != 4200 {
|
||||||
|
t.Errorf("got %d, want 4200", got)
|
||||||
|
}
|
||||||
|
// At 2x, 1800s of source encodes in 900s; + 600 remaining = 1500.
|
||||||
|
if got := queueETASeconds(600, 1800, 2); got != 1500 {
|
||||||
|
t.Errorf("got %d, want 1500", got)
|
||||||
|
}
|
||||||
|
// Empty queue → just the current job's remaining.
|
||||||
|
if got := queueETASeconds(600, 0, 1); got != 600 {
|
||||||
|
t.Errorf("got %d, want 600", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// waitCached polls Get (the probe resolves on a background goroutine).
|
||||||
|
func waitCached(c *durationCache, p string, mt time.Time) (float64, bool) {
|
||||||
|
for i := 0; i < 200; i++ {
|
||||||
|
if sec, ok := c.Get(p, mt); ok {
|
||||||
|
return sec, true
|
||||||
|
}
|
||||||
|
time.Sleep(2 * time.Millisecond)
|
||||||
|
}
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDurationCache(t *testing.T) {
|
||||||
|
c := newDurationCache(func(p string) (float64, error) { return 42, nil })
|
||||||
|
|
||||||
|
mt := time.Unix(1000, 0)
|
||||||
|
if _, ok := c.Get("/a.mkv", mt); ok {
|
||||||
|
t.Fatal("first Get should miss")
|
||||||
|
}
|
||||||
|
if sec, ok := waitCached(c, "/a.mkv", mt); !ok || sec != 42 {
|
||||||
|
t.Fatalf("after probe: got %v ok=%v, want 42 true", sec, ok)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A changed mtime invalidates the entry.
|
||||||
|
if _, ok := c.Get("/a.mkv", time.Unix(2000, 0)); ok {
|
||||||
|
t.Error("changed mtime should miss")
|
||||||
|
}
|
||||||
|
if _, ok := waitCached(c, "/a.mkv", time.Unix(2000, 0)); !ok {
|
||||||
|
t.Fatal("re-probe under new mtime should cache")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Retain drops entries for paths not in the keep set.
|
||||||
|
c.Retain([]string{"/b.mkv"})
|
||||||
|
if _, ok := c.Get("/a.mkv", time.Unix(2000, 0)); ok {
|
||||||
|
t.Error("Retain should have dropped /a.mkv")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user