From 5b1eee2699771abbdef2c34aca1d6e71c7888de4 Mon Sep 17 00:00:00 2001 From: flamingcow Date: Mon, 31 Aug 2026 08:58:31 -0700 Subject: [PATCH] Split the harness stream into capture, encode, and per-client send loops --- harness/grab.go | 12 +-- harness/stream.go | 202 +++++++++++++++++++++++++++++++++------------- 2 files changed, 154 insertions(+), 60 deletions(-) diff --git a/harness/grab.go b/harness/grab.go index 6b78ba2..12ad01f 100644 --- a/harness/grab.go +++ b/harness/grab.go @@ -36,6 +36,9 @@ func openGrabber() (*grabber, error) { } crtc, index, err := activeCrtc(fd) if err == nil { + // Opening the node while no master exists hands this fd implicit + // drm master; give it back so cabletest can take it. + drm.Ioctl(fd, drm.DropMaster, nil) return &grabber{fd: fd, crtc: crtc, index: index}, nil } unix.Close(fd) @@ -129,7 +132,7 @@ const overrunLimit = 10 // sequences differ by at most one therefore provably finished before the // rewrite could begin. Overrunning that is a timing miss, not corruption: // sample again, and treat only systematic misses as an error. -func (g *grabber) rawFrame(dst []byte) ([]byte, int, int, int, error) { +func (g *grabber) rawFrame() ([]byte, int, int, int, error) { for tries := 0; tries < overrunLimit; tries++ { s0, err := g.vblank(1) if err != nil { @@ -139,10 +142,7 @@ func (g *grabber) rawFrame(dst []byte) ([]byte, int, int, int, error) { if err != nil { return nil, 0, 0, 0, err } - if len(dst) < len(mem) { - dst = make([]byte, len(mem)) - } - dst = dst[:len(mem)] + dst := make([]byte, len(mem)) copy(dst, mem) unix.Munmap(mem) s1, err := g.vblank(0) @@ -157,7 +157,7 @@ func (g *grabber) rawFrame(dst []byte) ([]byte, int, int, int, error) { } func (g *grabber) frame() (*image.NRGBA, error) { - raw, pitch, pw, ph, err := g.rawFrame(nil) + raw, pitch, pw, ph, err := g.rawFrame() if err != nil { return nil, err } diff --git a/harness/stream.go b/harness/stream.go index af1e81f..fae75c1 100644 --- a/harness/stream.go +++ b/harness/stream.go @@ -10,6 +10,7 @@ import ( "net" "net/http" "os" + "sync" "g.fc.run/theater/cabletest/internal/drm" ) @@ -54,7 +55,28 @@ const page = ` ` +type raw struct { + buf []byte + pitch int + pw int + ph int +} + +var ( + captured = make(chan raw, 8) + + latestMu sync.RWMutex + latest raw + + clientsMu sync.RWMutex + clients = make(map[chan []byte]bool) +) + func serveStream() error { + g, err := openGrabber() + if err != nil { + return err + } ln, err := net.Listen("tcp", streamAddr) if err != nil { return err @@ -64,6 +86,8 @@ func serveStream() error { host = "localhost" } fmt.Printf("panel stream at http://%s%s/\n", host, streamAddr) + go captureLoop(g) + go encodeLoop() mux := http.NewServeMux() mux.HandleFunc("/{$}", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, page) @@ -73,73 +97,143 @@ func serveStream() error { return nil } -func frames(w http.ResponseWriter, r *http.Request) { - g, err := openGrabber() - if err != nil { - http.Error(w, err.Error(), http.StatusServiceUnavailable) - return - } - defer g.close() - - cur, pitch, pw, ph, err := g.rawFrame(nil) - if err != nil { - http.Error(w, err.Error(), http.StatusServiceUnavailable) - return - } - - w.Header().Set("Content-Type", "application/octet-stream") - if err := send(w, image.Pt(0, 0), decode(cur, pitch, pw, ph)); err != nil { - return - } - - ctx := r.Context() - prev := make([]byte, len(cur)) +// Between runs the crtc scans out the console framebuffer, which is not a +// dumb buffer and fails to map on every frame; each error reads once, not at +// frame rate. +func captureLoop(g *grabber) { + lastErr := "" for { - prev, cur = cur, prev - for { - if ctx.Err() != nil { - return + buf, pitch, pw, ph, err := g.rawFrame() + if err != nil { + if err.Error() != lastErr { + lastErr = err.Error() + fmt.Fprintf(os.Stderr, "capture: %v\n", err) } - if cur, pitch, pw, ph, err = g.rawFrame(cur); err != nil { - fmt.Fprintf(os.Stderr, "stream %s: %v\n", r.RemoteAddr, err) - return - } - rects := diffRects(prev, cur, pitch, pw, ph) - if len(rects) == 0 { - continue - } - for _, rect := range rects { - if err := send(w, rect.Min, maskRect(prev, cur, pitch, ph, rect)); err != nil { - return - } - } - break + continue + } + lastErr = "" + captured <- raw{buf, pitch, pw, ph} + } +} + +// latest advances before the delta fans out, so a handler that registers and +// then snapshots latest can only ever replay a delta its keyframe already +// includes, never miss one. +func encodeLoop() { + for cur := range captured { + prev := latest + if prev.buf != nil && (prev.pitch != cur.pitch || prev.pw != cur.pw || prev.ph != cur.ph) { + panic(fmt.Sprintf("framebuffer geometry changed: %dx%d pitch %d to %dx%d pitch %d", + prev.pw, prev.ph, prev.pitch, cur.pw, cur.ph, cur.pitch)) + } + latestMu.Lock() + latest = cur + latestMu.Unlock() + if prev.buf != nil { + broadcast(prev, cur) } } } -// Encoding happens inside the vblank budget, and on rect-sized parts the -// dense level trades 3-6x the cpu for a few percent of size — enough to spill -// past a vblank on the once-per-second bursts and stutter the stream. +// A client that stops draining its channel is closed so its page reconnects +// into a fresh keyframe. +func broadcast(prev, cur raw) { + var delta []byte + for _, rect := range diffRects(prev.buf, cur.buf, cur.pitch, cur.pw, cur.ph) { + p, err := pack(rect.Min, maskRect(prev.buf, cur.buf, cur.pitch, cur.ph, rect)) + if err != nil { + fmt.Fprintf(os.Stderr, "encode: %v\n", err) + return + } + delta = append(delta, p...) + } + var dead []chan []byte + clientsMu.RLock() + for ch := range clients { + select { + case ch <- delta: + default: + dead = append(dead, ch) + } + } + clientsMu.RUnlock() + if len(dead) == 0 { + return + } + clientsMu.Lock() + for _, ch := range dead { + if clients[ch] { + delete(clients, ch) + close(ch) + } + } + clientsMu.Unlock() +} + +// Frames are never written again once captured, so the lock only guards the +// swap of latest itself; deltas queue in the channel while the keyframe +// encodes, and replaying one the keyframe already includes is idempotent. +func frames(w http.ResponseWriter, r *http.Request) { + ch := make(chan []byte, 16) + clientsMu.Lock() + clients[ch] = true + clientsMu.Unlock() + defer func() { + clientsMu.Lock() + delete(clients, ch) + clientsMu.Unlock() + }() + latestMu.RLock() + k := latest + latestMu.RUnlock() + if k.buf == nil { + http.Error(w, "no frame captured yet", http.StatusServiceUnavailable) + return + } + key, err := pack(image.Pt(0, 0), decode(k.buf, k.pitch, k.pw, k.ph)) + if err != nil { + fmt.Fprintf(os.Stderr, "keyframe %s: %v\n", r.RemoteAddr, err) + return + } + w.Header().Set("Content-Type", "application/octet-stream") + if _, err := w.Write(key); err != nil { + return + } + w.(http.Flusher).Flush() + ctx := r.Context() + for { + select { + case <-ctx.Done(): + return + case payload, ok := <-ch: + if !ok { + return + } + if _, err := w.Write(payload); err != nil { + return + } + w.(http.Flusher).Flush() + } + } +} + +// On rect-sized parts the dense level trades 3-6x the cpu for a few percent +// of size — enough to fall behind the capture loop on the once-per-second +// bursts and stutter the stream. var encoder = png.Encoder{CompressionLevel: png.DefaultCompression} -func send(w http.ResponseWriter, at image.Point, img *image.NRGBA) error { +func pack(at image.Point, img *image.NRGBA) ([]byte, error) { var buf bytes.Buffer - if err := encoder.Encode(&buf, img); err != nil { - return err - } var hdr [12]byte binary.LittleEndian.PutUint32(hdr[0:], uint32(at.X)) binary.LittleEndian.PutUint32(hdr[4:], uint32(at.Y)) - binary.LittleEndian.PutUint32(hdr[8:], uint32(buf.Len())) - if _, err := w.Write(hdr[:]); err != nil { - return err + buf.Write(hdr[:]) + if err := encoder.Encode(&buf, img); err != nil { + return nil, err } - if _, err := w.Write(buf.Bytes()); err != nil { - return err - } - w.(http.Flusher).Flush() - return nil + b := buf.Bytes() + binary.LittleEndian.PutUint32(b[8:], uint32(len(b)-12)) + return b, nil } // Distant regions changing together (a spinner up top, counters at the foot)