Split the harness stream into capture, encode, and per-client send loops

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 = make([]byte, len(mem))
-		}
-		dst = dst[: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
 	}

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 = `<!doctype html>
 </script>
 `
 
+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) {
+// Between runs the crtc scans out the console framebuffer, which is not a
-	g, err := openGrabber()
+// dumb buffer and fails to map on every frame; each error reads once, not at
-	if err != nil {
+// frame rate.
-		http.Error(w, err.Error(), http.StatusServiceUnavailable)
+func captureLoop(g *grabber) {
-		return
+	lastErr := ""
-	}
-	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))
 	for {
-		prev, cur = cur, prev
+		buf, pitch, pw, ph, err := g.rawFrame()
-		for {
+		if err != nil {
-			if ctx.Err() != nil {
+			if err.Error() != lastErr {
-				return
+				lastErr = err.Error()
+				fmt.Fprintf(os.Stderr, "capture: %v\n", err)
 			}
-			if cur, pitch, pw, ph, err = g.rawFrame(cur); err != nil {
+			continue
-				fmt.Fprintf(os.Stderr, "stream %s: %v\n", r.RemoteAddr, err)
+		}
-				return
+		lastErr = ""
-			}
+		captured <- raw{buf, pitch, pw, ph}
-			rects := diffRects(prev, cur, pitch, pw, ph)
+	}
-			if len(rects) == 0 {
+}
-				continue
+
-			}
+// latest advances before the delta fans out, so a handler that registers and
-			for _, rect := range rects {
+// then snapshots latest can only ever replay a delta its keyframe already
-				if err := send(w, rect.Min, maskRect(prev, cur, pitch, ph, rect)); err != nil {
+// includes, never miss one.
-					return
+func encodeLoop() {
-				}
+	for cur := range captured {
-			}
+		prev := latest
-			break
+		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
+// A client that stops draining its channel is closed so its page reconnects
-// dense level trades 3-6x the cpu for a few percent of size — enough to spill
+// into a fresh keyframe.
-// past a vblank on the once-per-second bursts and stutter the stream.
+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()))
+	buf.Write(hdr[:])
-	if _, err := w.Write(hdr[:]); err != nil {
+	if err := encoder.Encode(&buf, img); err != nil {
-		return err
+		return nil, err
 	}
-	if _, err := w.Write(buf.Bytes()); err != nil {
+	b := buf.Bytes()
-		return err
+	binary.LittleEndian.PutUint32(b[8:], uint32(len(b)-12))
-	}
+	return b, nil
-	w.(http.Flusher).Flush()
-	return nil
 }
 
 // Distant regions changing together (a spinner up top, counters at the foot)