major remote desktop upgrades + do opus. Hopefully all the tests pass 😭

This commit is contained in:
KingKDot
2026-07-17 04:28:23 -05:00
parent 779f86b259
commit 0219081682
35 changed files with 1304 additions and 91 deletions
+16 -2
View File
@@ -154,6 +154,8 @@ func CaptureAndSend(ctx context.Context, env *rt.Env) error {
slotAcquired = true
}
frameWidth := img.Bounds().Dx()
frameHeight := img.Bounds().Dy()
quality := jpegQuality()
frame, encodeDur, err := buildFrame(img, display, quality)
PutRGBA(img)
@@ -164,6 +166,8 @@ func CaptureAndSend(ctx context.Context, env *rt.Env) error {
}
return err
}
frame.Header.Width = frameWidth
frame.Header.Height = frameHeight
now := time.Now()
fps := frameFPS(now)
if fps <= 0 {
@@ -181,17 +185,21 @@ func CaptureAndSend(ctx context.Context, env *rt.Env) error {
if dur <= 0 {
dur = 33 * time.Millisecond
}
sendStart := time.Now()
if werr := webrtcpub.WriteH264(webrtcpub.KindDesktop, frame.Data, dur); werr != nil {
log.Printf("webrtc: write h264 failed: %v", werr)
}
sendDur := time.Since(sendStart)
if slotAcquired {
ReleaseFrameSlot()
}
statFrames.Add(1)
statCapNs.Add(captureDur.Nanoseconds())
statEncNs.Add(encodeDur.Nanoseconds())
statTotalNs.Add(time.Since(t0).Nanoseconds())
totalDur := time.Since(t0)
statTotalNs.Add(totalDur.Nanoseconds())
statBytes.Add(int64(len(frame.Data)))
emitDesktopStreamStats(ctx, env, frame, fps, captureDur, encodeDur, sendDur, totalDur, "webrtc")
return nil
}
if !slotAcquired {
@@ -233,6 +241,7 @@ func CaptureAndSend(ctx context.Context, env *rt.Env) error {
statSendNs.Add(sendDur.Nanoseconds())
statTotalNs.Add(time.Since(t0).Nanoseconds())
statBytes.Add(int64(len(frame.Data)))
emitDesktopStreamStats(ctx, env, frame, fps, captureDur, encodeDur, sendDur, time.Since(t0), "websocket")
return err
}
@@ -251,14 +260,18 @@ func sendCompletedFrame(ctx context.Context, env *rt.Env, frame wire.Frame, disp
if dur <= 0 {
dur = 33 * time.Millisecond
}
sendStart := time.Now()
if werr := webrtcpub.WriteH264(webrtcpub.KindDesktop, frame.Data, dur); werr != nil {
log.Printf("webrtc: write h264 failed: %v", werr)
}
sendDur := time.Since(sendStart)
statFrames.Add(1)
statCapNs.Add(captureDur.Nanoseconds())
statEncNs.Add(encodeDur.Nanoseconds())
statTotalNs.Add(time.Since(t0).Nanoseconds())
totalDur := time.Since(t0)
statTotalNs.Add(totalDur.Nanoseconds())
statBytes.Add(int64(len(frame.Data)))
emitDesktopStreamStats(ctx, env, frame, fps, captureDur, encodeDur, sendDur, totalDur, "webrtc")
return nil
}
if !AcquireFrameSlot() {
@@ -301,6 +314,7 @@ func sendCompletedFrame(ctx context.Context, env *rt.Env, frame wire.Frame, disp
statSendNs.Add(sendDur.Nanoseconds())
statTotalNs.Add(time.Since(t0).Nanoseconds())
statBytes.Add(int64(len(frame.Data)))
emitDesktopStreamStats(ctx, env, frame, fps, captureDur, encodeDur, sendDur, time.Since(t0), "websocket")
return nil
}
@@ -44,6 +44,7 @@ func TestCaptureAndSend_NoDisplays(t *testing.T) {
}
func TestCaptureAndSend_SendsFrame(t *testing.T) {
previousStatsAt := lastDesktopStreamStatsMs.Swap(0)
originalCount := activeDisplays
originalCapture := captureDisplayFn
activeDisplays = func() int { return 1 }
@@ -56,6 +57,7 @@ func TestCaptureAndSend_SendsFrame(t *testing.T) {
t.Cleanup(func() {
activeDisplays = originalCount
captureDisplayFn = originalCapture
lastDesktopStreamStatsMs.Store(previousStatsAt)
})
writer := &recordingWriter{}
@@ -63,8 +65,8 @@ func TestCaptureAndSend_SendsFrame(t *testing.T) {
if err := CaptureAndSend(context.Background(), env); err != nil {
t.Fatalf("CaptureAndSend returned error: %v", err)
}
if len(writer.msgs) != 1 {
t.Fatalf("expected one message written, got %d", len(writer.msgs))
if len(writer.msgs) != 2 {
t.Fatalf("expected frame and telemetry messages, got %d", len(writer.msgs))
}
var frame wire.Frame
@@ -84,6 +86,17 @@ func TestCaptureAndSend_SendsFrame(t *testing.T) {
if _, err := jpeg.Decode(bytes.NewReader(frame.Data)); err != nil {
t.Fatalf("jpeg payload did not decode: %v", err)
}
var stats wire.DesktopStreamStats
if err := msgpack.Unmarshal(writer.msgs[1], &stats); err != nil {
t.Fatalf("decode desktop stream stats: %v", err)
}
if stats.Type != "desktop_stream_stats" || stats.Format != "jpeg" {
t.Fatalf("unexpected stream stats: %#v", stats)
}
if stats.CaptureMs < 0 || stats.EncodeMs < 0 || stats.TotalMs < 0 {
t.Fatalf("stream stats contained a negative duration: %#v", stats)
}
}
func TestActualScreenCapture(t *testing.T) {
@@ -306,8 +319,13 @@ func TestCaptureAndSend_DisplayOutOfRange(t *testing.T) {
t.Errorf("expected capture from display %d, got %d", tt.expectedDisplay, capturedDisplay)
}
if len(writer.msgs) != 1 {
t.Errorf("expected 1 message, got %d", len(writer.msgs))
if len(writer.msgs) == 0 {
t.Error("expected a frame message")
} else {
var frame wire.Frame
if err := msgpack.Unmarshal(writer.msgs[0], &frame); err != nil || frame.Type != "frame" {
t.Errorf("first message was not a valid frame: type=%q err=%v", frame.Type, err)
}
}
})
}
@@ -0,0 +1,66 @@
package capture
import (
"math"
"sync/atomic"
)
const maxH264Bitrate = 50_000_000
var h264TargetBitrate atomic.Int64
func SetH264TargetBitrate(bps int) int {
if bps < 0 {
bps = 0
}
if bps > maxH264Bitrate {
bps = maxH264Bitrate
}
previous := h264TargetBitrate.Swap(int64(bps))
if previous != int64(bps) {
resetH264Encoder()
resetH264Encoderbackstage()
resetH264TextureEncoderForBitrate()
RequestDesktopFullFrame()
}
return bps
}
func configuredH264Bitrate() int {
return int(h264TargetBitrate.Load())
}
func automaticH264Bitrate(width, height, fps int) int {
pixelsPerSecond := float64(width * height * fps)
bitrate := int(pixelsPerSecond * 0.08)
if bitrate < 1_500_000 {
return 1_500_000
}
if bitrate > 18_000_000 {
return 18_000_000
}
return bitrate
}
func targetH264Bitrate(width, height, fps int) int {
if configured := configuredH264Bitrate(); configured > 0 {
return configured
}
return automaticH264Bitrate(width, height, fps)
}
func targetH264CRF(width, height, fps int) float32 {
configured := configuredH264Bitrate()
if configured <= 0 {
return 23
}
ratio := float64(configured) / float64(automaticH264Bitrate(width, height, fps))
crf := 23 - 6*math.Log2(ratio)
if crf < 12 {
crf = 12
}
if crf > 35 {
crf = 35
}
return float32(crf)
}
@@ -0,0 +1,5 @@
//go:build !windows
package capture
func resetH264TextureEncoderForBitrate() {}
@@ -0,0 +1,7 @@
//go:build windows
package capture
func resetH264TextureEncoderForBitrate() {
resetH264D3D11TextureEncoder()
}
@@ -0,0 +1,38 @@
package capture
import "testing"
func TestH264TargetBitrateAutoAndManual(t *testing.T) {
original := configuredH264Bitrate()
defer SetH264TargetBitrate(original)
SetH264TargetBitrate(0)
if got := targetH264Bitrate(1920, 1080, 60); got != 9_953_280 {
t.Fatalf("automatic 1080p60 bitrate = %d, want 9953280", got)
}
if got := targetH264Bitrate(3840, 2160, 240); got != 18_000_000 {
t.Fatalf("automatic bitrate cap = %d, want 18000000", got)
}
if got := SetH264TargetBitrate(30_000_000); got != 30_000_000 {
t.Fatalf("manual bitrate = %d, want 30000000", got)
}
if got := targetH264Bitrate(1280, 720, 30); got != 30_000_000 {
t.Fatalf("manual target = %d, want 30000000", got)
}
if got := SetH264TargetBitrate(60_000_000); got != maxH264Bitrate {
t.Fatalf("manual cap = %d, want %d", got, maxH264Bitrate)
}
}
func TestH264TargetCRFTracksManualRate(t *testing.T) {
original := configuredH264Bitrate()
defer SetH264TargetBitrate(original)
SetH264TargetBitrate(0)
auto := targetH264CRF(1920, 1080, 60)
SetH264TargetBitrate(20_000_000)
higherRate := targetH264CRF(1920, 1080, 60)
if higherRate >= auto {
t.Fatalf("higher bitrate CRF = %.2f, want below automatic %.2f", higherRate, auto)
}
}
@@ -14,7 +14,7 @@ func tryBuildDirectH264Frame(display int) (wire.Frame, time.Duration, time.Durat
return wire.Frame{}, 0, 0, false, nil
}
forceKeyframe := webrtcpub.ConsumeKeyframeRequest()
data, _, _, captureDur, encodeDur, used, err := captureDisplayDXGIH264(display, forceKeyframe)
data, width, height, captureDur, encodeDur, used, err := captureDisplayDXGIH264(display, forceKeyframe)
if !used || err != nil {
return wire.Frame{}, captureDur, encodeDur, used, err
}
@@ -24,5 +24,5 @@ func tryBuildDirectH264Frame(display int) (wire.Frame, time.Duration, time.Durat
now := time.Now()
lastKeyframe.Store(now.UnixNano())
statFullFrames.Add(1)
return wire.Frame{Type: "frame", Header: wire.FrameHeader{Monitor: display, FPS: 0, Format: "h264"}, Data: data}, captureDur, encodeDur, true, nil
return wire.Frame{Type: "frame", Header: wire.FrameHeader{Monitor: display, FPS: 0, Format: "h264", Width: width, Height: height}, Data: data}, captureDur, encodeDur, true, nil
}
@@ -1296,12 +1296,6 @@ func failedHR(hr uintptr) bool {
return hr&0x80000000 != 0
}
func targetH264Bitrate(width, height, fps int) int {
pixelsPerSecond := float64(width * height * fps)
bitrate := int(pixelsPerSecond * 0.08)
return clampInt(bitrate, 1_500_000, 18_000_000)
}
func h264LevelFor(width, height, fps int) int {
mbsPerFrame := ((width + 15) / 16) * ((height + 15) / 16)
mbsPerSecond := mbsPerFrame * fps
@@ -104,7 +104,7 @@ func ensureH264EncoderLocked(width, height int) error {
Preset: "veryfast",
Profile: "main",
RateControl: "crf",
RateConstant: 23,
RateConstant: targetH264CRF(width, height, fps),
LogLevel: x264.LogError,
}
@@ -201,7 +201,7 @@ func ensurebackstageH264EncoderLocked(width, height int) error {
Preset: "veryfast",
Profile: "main",
RateControl: "crf",
RateConstant: 23,
RateConstant: targetH264CRF(width, height, fps),
LogLevel: x264.LogError,
}
enc, err := x264.NewEncoder(&backstageH264Buf, opts)
@@ -280,7 +280,7 @@ func ensureWebcamH264EncoderLocked(width, height int) error {
Preset: "veryfast",
Profile: "main",
RateControl: "crf",
RateConstant: 23,
RateConstant: targetH264CRF(width, height, fps),
LogLevel: x264.LogError,
}
enc, err := x264.NewEncoder(&webcamH264Buf, opts)
@@ -0,0 +1,37 @@
package capture
import (
"context"
"sync/atomic"
"time"
rt "overlord-client/cmd/agent/runtime"
"overlord-client/cmd/agent/wire"
)
var lastDesktopStreamStatsMs atomic.Int64
func emitDesktopStreamStats(ctx context.Context, env *rt.Env, frame wire.Frame, fps int, captureDur, encodeDur, sendDur, totalDur time.Duration, transport string) {
now := time.Now().UnixMilli()
previous := lastDesktopStreamStatsMs.Load()
if now-previous < 500 || !lastDesktopStreamStatsMs.CompareAndSwap(previous, now) {
return
}
_ = wire.WriteMsg(ctx, env.Conn, wire.DesktopStreamStats{
Type: "desktop_stream_stats",
FPS: fps,
Format: frame.Header.Format,
Bytes: len(frame.Data),
Width: frame.Header.Width,
Height: frame.Header.Height,
CaptureMs: durationMs(captureDur),
EncodeMs: durationMs(encodeDur),
SendMs: durationMs(sendDur),
TotalMs: durationMs(totalDur),
Transport: transport,
})
}
func durationMs(value time.Duration) float64 {
return float64(value.Microseconds()) / 1000
}