From 559caa2dd0b98e262a308c9bd482d575c4e21cef Mon Sep 17 00:00:00 2001 From: saphid <4596216+saphid@users.noreply.github.com> Date: Mon, 28 Sep 2026 22:23:10 +1000 Subject: [PATCH] Share the desktop stream rate controller across host platforms --- .gitignore | 10 + desktop/build-controller.py | 16 ++ desktop/controller.c | 127 +++++++++ desktop/controller.h | 29 ++ mac/frame-mac-view/Sources/CGVirtualDisplay.h | 2 + mac/frame-mac-view/Sources/Controller.swift | 249 +++--------------- mac/frame-mac-view/build.sh | 5 +- 7 files changed, 218 insertions(+), 220 deletions(-) create mode 100644 desktop/build-controller.py create mode 100644 desktop/controller.c create mode 100644 desktop/controller.h diff --git a/.gitignore b/.gitignore index c2019bf..546d780 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,13 @@ compat-db/.env.lakebed.server compat-db/.lakebed/ tests/smoke/results/ mac/bin/ + +# Shared desktop streaming controller build products +desktop/controller.dll +desktop/controller.dylib +desktop/controller.so +desktop/*.obj +desktop/*.lib +desktop/*.exp +desktop/bundle/ +app/build/desktop/ diff --git a/desktop/build-controller.py b/desktop/build-controller.py new file mode 100644 index 0000000..33c83af --- /dev/null +++ b/desktop/build-controller.py @@ -0,0 +1,16 @@ +#!/usr/bin/env python3 +"""Build the common controller for the Python PC host (a C11 compiler needed).""" +import os +from pathlib import Path +import subprocess +import sys + +root = Path(__file__).resolve().parent +suffix = '.dll' if sys.platform == 'win32' else '.dylib' if sys.platform == 'darwin' else '.so' +out = root / ('controller' + suffix) +if sys.platform == 'win32' and os.environ.get('CC', 'cl') == 'cl': + cmd = ['cl', '/nologo', '/O2', '/LD', '/std:c11', str(root / 'controller.c'), '/link', '/OUT:' + str(out)] +else: + cmd = [os.environ.get('CC', 'cc'), '-O2', '-std=c11', '-shared', '-fPIC', str(root / 'controller.c'), '-o', str(out)] +subprocess.run(cmd, cwd=root, check=True) +print(out) diff --git a/desktop/controller.c b/desktop/controller.c new file mode 100644 index 0000000..2ab7aad --- /dev/null +++ b/desktop/controller.c @@ -0,0 +1,127 @@ +/* Extracted from the Mac agent's RateController. Same gate, demand protection, + * bitrate recovery and tier hysteresis for every host. Bounded sample buffers. */ +#include "controller.h" +#include +#include +#define CAP 1024 +#define MIN(a,b) ((a)<(b)?(a):(b)) +#define MAX(a,b) ((a)>(b)?(a):(b)) +typedef struct { int64_t t, v; uint32_t seq; } Sample; +typedef struct { Sample a[CAP]; int n; } Samples; +struct FCController { + int max_fps, enabled, ceiling, target, tier, frame_bytes, held, signal, saw_ack; + int64_t decrease, increase, below, above, slack; + Samples flight, rtts, acked, sent, captures; +}; +static const int fps[] = {60,45,30,30,30}, scale[] = {100,100,100,75,50}; +static const double floors[] = {.45,.28,.16,.08,0}; +static void remove_first(Samples *s, int n) { + s->n -= n; memmove(s->a, s->a+n, (size_t)s->n*sizeof(Sample)); +} +static void add(Samples *s, Sample v, int cap) { + if (s->n >= cap) remove_first(s, 1); + s->a[s->n++] = v; +} +static void expire(Samples *s, int64_t oldest) { + int n=0; while(nn && s->a[n].trtts.n;i++) if(!i || c->rtts.a[i].vrtts.a[i].v; + return v; +} +static int compare(const void *a, const void *b) { + int64_t x=*(const int64_t*)a, y=*(const int64_t*)b; + return (x>y)-(xn;i++) if(s->a[i].t>since) a[n++]=s->a[i].v; + if(!n) return 0; + qsort(a,(size_t)n,sizeof(int64_t),compare); + return a[n*numerator/denominator]; +} +static int rate(Samples *s) { + int64_t total=0; for(int i=0;in;i++) total+=s->a[i].v; + return (int)MIN(total*16,2147483647); +} +FCController *fc_new(int max_fps,int enabled) { + FCController *c=calloc(1,sizeof(*c)); + if(c) {c->max_fps=MAX(1,max_fps);c->enabled=enabled;c->slack=40000;} + return c; +} +void fc_free(FCController *c) {free(c);} +void fc_ceiling(FCController *c,int bps) { + if(!c->target || c->target>bps) c->target=bps; + c->ceiling=bps; +} +int fc_gate(FCController *c,int64_t now,int counts) { + if(!c->enabled || !c->saw_ack) return 1; + expire(&c->flight,now-2000000); + if(!c->flight.n) return 1; + int64_t interval=1000000/MIN(c->max_fps,fps[c->tier]); + int window=MAX(3,(int)((base(c)+c->slack)/interval)+1); + if(c->flight.nflight.a[0].t<=base(c)+c->slack) return 1; + if(counts) c->held++; + return 0; +} +void fc_capture(FCController *c,int64_t now) {add(&c->captures,(Sample){now,0,0},256);} +void fc_sent(FCController *c,uint32_t seq,int bytes,int64_t now) { + Sample s={now,bytes,seq};add(&c->flight,s,512);add(&c->sent,s,CAP); +} +int fc_ack(FCController *c,uint32_t seq,int64_t now) { + c->saw_ack=1; + for(int i=0;iflight.n;i++) if(c->flight.a[i].seq==seq) { + Sample s=c->flight.a[i];remove_first(&c->flight,i+1); + add(&c->rtts,(Sample){now,now-s.t,0},CAP); + add(&c->acked,(Sample){now,s.v,0},CAP);return 1; + } + return 0; +} +int fc_update(FCController *c,int64_t now) { + expire(&c->rtts,now-10000000);expire(&c->acked,now-500000); + expire(&c->sent,now-500000);expire(&c->flight,now-2000000); + expire(&c->captures,now-1000000); + if(!c->enabled || c->ceiling<=0) return 0; + int64_t baseline=base(c), spread=quantile(&c->rtts,now-2000000,9,10); + int64_t jitter=spread ? spread-baseline : 0; + c->slack=1000000/MIN(c->max_fps,fps[c->tier])+MIN(MAX(jitter*3/2,25000),80000); + int64_t recent=quantile(&c->rtts,now-300000,1,2); + int64_t queue=recent ? recent-baseline : 0; + int64_t age=c->flight.n ? now-c->flight.a[0].t : 0; + int delivered=rate(&c->acked),sending=rate(&c->sent); + if(c->sent.n) c->frame_bytes=sending/16/c->sent.n; + int demand=MIN(c->captures.n,MIN(c->max_fps,fps[c->tier]))*c->frame_bytes*8; + int signal=(sending>=c->target/2 && queue>40000)||c->held>=3||age>baseline+100000; + int congested=signal && c->signal;c->signal=signal;c->held=0; + if(congested && now-c->decrease>300000) { + int next=MAX(300000,MIN(c->target*4/5,MAX(delivered*9/10,c->target/2))); + if(demand>0 && demand*2<=c->target*5/4) next=MAX(next,MIN(c->target,demand*2)); + c->target=next;c->decrease=now; + } else if(!congested && now-c->decrease>1000000 && now-c->increase>250000 && c->targetceiling && + (sending>c->target*6/10 || now-c->decrease>3000000)) { + c->target=MIN(c->ceiling,(int)(c->target*1.1)+50000);c->increase=now; + } + double share=(double)c->target/MAX(c->ceiling,1); + if(c->tier<4 && sharetier] && sending>=c->target*7/10) { + if(!c->below)c->below=now; + if(now-c->below>500000) { + for(c->tier=0;c->tier<4 && sharetier];c->tier++) {} + c->below=0; + } + } else c->below=0; + if(c->tier>0 && share>floors[c->tier-1]*1.25) { + if(!c->above)c->above=now; + if(now-c->above>2000000) {c->tier--;c->above=0;} + } else c->above=0; + return c->target; +} +int64_t fc_value(FCController *c,int field) { + switch(field) { + case 0:return c->target;case 1:return c->ceiling;case 2:return c->tier; + case 3:return MIN(c->max_fps,fps[c->tier]);case 4:return scale[c->tier]; + case 5:return base(c);case 6:return c->flight.n;case 7:return c->slack; + default:return 0; + } +} diff --git a/desktop/controller.h b/desktop/controller.h new file mode 100644 index 0000000..5df7212 --- /dev/null +++ b/desktop/controller.h @@ -0,0 +1,29 @@ +/* Shared latency controller. Times are monotonic host microseconds. + * Callers serialize access. Capture is gated BEFORE encoding: dropping an + * encoded inter frame would break the decoder's reference chain. */ +#ifndef FRAME_CONTROLLER_H +#define FRAME_CONTROLLER_H +#include +#ifdef _WIN32 +#define FC_API __declspec(dllexport) +#else +#define FC_API +#endif +#ifdef __cplusplus +extern "C" { +#endif +typedef struct FCController FCController; +FC_API FCController *fc_new(int fps, int enabled); +FC_API void fc_free(FCController *c); +FC_API void fc_ceiling(FCController *c, int bps); +FC_API int fc_gate(FCController *c, int64_t now, int counts); +FC_API void fc_capture(FCController *c, int64_t now); +FC_API void fc_sent(FCController *c, uint32_t seq, int bytes, int64_t now); +FC_API int fc_ack(FCController *c, uint32_t seq, int64_t now); +FC_API int fc_update(FCController *c, int64_t now); +/* target, ceiling, tier, fps, scale percent, baseline us, in flight, slack us */ +FC_API int64_t fc_value(FCController *c, int field); +#ifdef __cplusplus +} +#endif +#endif diff --git a/mac/frame-mac-view/Sources/CGVirtualDisplay.h b/mac/frame-mac-view/Sources/CGVirtualDisplay.h index f67f628..444401e 100644 --- a/mac/frame-mac-view/Sources/CGVirtualDisplay.h +++ b/mac/frame-mac-view/Sources/CGVirtualDisplay.h @@ -26,3 +26,5 @@ - (instancetype)initWithDescriptor:(CGVirtualDisplayDescriptor *)descriptor; - (BOOL)applySettings:(CGVirtualDisplaySettings *)settings; @end + +#include "../../../desktop/controller.h" diff --git a/mac/frame-mac-view/Sources/Controller.swift b/mac/frame-mac-view/Sources/Controller.swift index 2cb4b3b..b40bd6e 100644 --- a/mac/frame-mac-view/Sources/Controller.swift +++ b/mac/frame-mac-view/Sources/Controller.swift @@ -1,234 +1,45 @@ -// Adapts each stream to its network, latency first. The viewer acknowledges -// every frame as it arrives ("rx"); from those acks the controller knows how -// long frames take to get through and how fast the link delivers them. -// -// - The gate: a new frame is sent only while the oldest unacknowledged one is -// younger than the path's usual round trip plus a little slack. So frames -// never queue up in SSH, TCP or the Wi-Fi driver; while the link is stuck -// the newest picture waits and goes out as soon as it moves again. -// - The bitrate: when frames start queueing (the round trip grows) or the -// gate has to hold frames back, it drops to a bit under what the link -// actually delivered; once things are clear it probes up again slowly, never -// above the quality setting's bitrate (the ceiling). -// - The tier: as the bitrate falls, fewer frames per second (60, 45, 30), then -// a smaller picture (75%, then 50% of the panel's pixels). -// See docs/mac-in-headset.md ("Adapting to the network"). Thread-safe. +// Thin, serialized binding to the same controller used by the PC host. +// Algorithm and bounded state live in desktop/controller.c. import Foundation final class RateController { - struct Tier: Equatable { - let fps: Int - let scale: Double // of the picture's long side - } - - static let tiers = [Tier(fps: 60, scale: 1), Tier(fps: 45, scale: 1), Tier(fps: 30, scale: 1), - Tier(fps: 30, scale: 0.75), Tier(fps: 30, scale: 0.5)] - /// A tier is used while the target bitrate is at least this share of the ceiling. - static let floors = [0.45, 0.28, 0.16, 0.08, 0] static let enabled = ProcessInfo.processInfo.environment["FRAME_MAC_VIEW_ADAPT"] != "0" - let maxFps: Int private let lock = NSLock() - private var ceiling = 0 // bits/s at full size and frame rate - private(set) var target = 0 - private(set) var tier = 0 - private var unacked: [(seq: UInt32, sent: Int64, bytes: Int)] = [] - /// Round trips (send -> ack arrives here), for the baseline: the lowest - /// in the last 10 s is the path without any queue. - private var rtts: [(t: Int64, v: Int64)] = [] - private var acked: [(t: Int64, bytes: Int)] = [] // the last second - private var sentLog: [(t: Int64, bytes: Int)] = [] - private var captures: [Int64] = [] // the last second - private var frameBytes = 0 // average recent frame, kept while the gate holds everything back - private var held = 0 // frames the gate held back since the last update - private var lastSignal = false - private var sawAck = false - private var lastDecrease: Int64 = 0 - private var lastIncrease: Int64 = 0 - private var belowSince: Int64 = 0, aboveSince: Int64 = 0 - /// What changed, for the timeline: (time, event). - private(set) var events: [(Int64, String)] = [] - - init(maxFps: Int) { self.maxFps = maxFps } - + private let core: OpaquePointer + private var events: [[String: Any]] = [] + init(maxFps: Int) { + self.maxFps = maxFps + core = fc_new(Int32(maxFps), Self.enabled ? 1 : 0)! + } + deinit { fc_free(core) } private func locked(_ f: () -> T) -> T { lock.lock(); defer { lock.unlock() }; return f() } - - /// The quality setting's bitrate at full size; the first call also starts there. - func setCeiling(_ bps: Int) { - locked { - if target == 0 || target > bps { target = bps } - ceiling = bps - } - } - - var fps: Int { locked { fpsLocked } } - private var fpsLocked: Int { min(maxFps, RateController.tiers[tier].fps) } - var scale: Double { locked { RateController.tiers[tier].scale } } - var baseRtt: Int64 { locked { baseline() } } - - private func baseline() -> Int64 { rtts.map(\.v).min() ?? 0 } - - /// How late a frame may be before the gate holds the next one: one frame - /// interval, plus room for the jitter this link normally has (1.5 times - /// its recent spread), so ordinary Wi-Fi jitter doesn't cost frames but a - /// real queue does. Updated in update(). - private var slack: Int64 = 40_000 - - /// Whether a frame may be sent now without queueing behind earlier ones. - /// `counts`: a held frame is a sign of congestion (not when merely - /// re-checking whether a held frame can go yet). - func maySend(now: Int64, counts: Bool = true) -> Bool { - locked { - guard RateController.enabled, sawAck else { return true } // not heard from the viewer yet - // Unacknowledged for 2 s: gone with a reconnection, not in a queue. - unacked.removeAll { now - $0.sent > 2_000_000 } - guard let oldest = unacked.first else { return true } - // Age is what bounds latency. The count only stops a burst, and it - // allows a full round trip of frames, so a long but clear path - // (100 ms away) still gets every frame. - let interval = Int64(1_000_000 / max(1, fpsLocked)) - let window = max(3, Int((baseline() + slack) / interval) + 1) - if unacked.count < window, now - oldest.sent <= baseline() + slack { return true } - if counts { held += 1 } - return false - } - } - - /// A picture was captured (sent or not): with the frame sizes, what this - /// stream would send if the link allowed. - func captured(at t: Int64) { - locked { - captures.append(t) - if captures.count > 256 { captures.removeFirst(captures.count - 256) } - } - } - - func sent(seq: UInt32, bytes: Int, at t: Int64) { - locked { - // Bounded even if the viewer never acknowledges (an old viewer, or - // the controller is off). - unacked.append((seq, t, bytes)) - if unacked.count > 512 { unacked.removeFirst(unacked.count - 512) } - sentLog.append((t, bytes)) - if sentLog.count > 1024 { sentLog.removeFirst(sentLog.count - 1024) } - } - } - - /// The viewer has frame `seq`. Returns true if that may let a held frame go. - func acked(seq: UInt32, at now: Int64) -> Bool { - locked { - sawAck = true - guard let i = unacked.firstIndex(where: { $0.seq == seq }) else { return false } - let f = unacked[i] - unacked.removeSubrange(0...i) // TCP delivers in order: earlier ones arrived too - rtts.append((now, now - f.sent)) - acked.append((now, f.bytes)) - return true - } - } - - /// Called every 100 ms. Returns the new bitrate target, or nil if the - /// controller is off. + private func value(_ field: Int32) -> Int { Int(fc_value(core, field)) } + var target: Int { locked { value(0) } } + var tier: Int { locked { value(2) } } + var fps: Int { locked { value(3) } } + var scale: Double { locked { Double(value(4)) / 100 } } + var baseRtt: Int64 { locked { fc_value(core, 5) } } + func setCeiling(_ bps: Int) { locked { fc_ceiling(core, Int32(bps)) } } + func maySend(now: Int64, counts: Bool = true) -> Bool { locked { fc_gate(core, now, counts ? 1 : 0) != 0 } } + func captured(at t: Int64) { locked { fc_capture(core, t) } } + func sent(seq: UInt32, bytes: Int, at t: Int64) { locked { fc_sent(core, seq, Int32(bytes), t) } } + func acked(seq: UInt32, at t: Int64) -> Bool { locked { fc_ack(core, seq, t) != 0 } } func update(now: Int64) -> Int? { locked { - rtts.removeAll { now - $0.t > 10_000_000 } - acked.removeAll { now - $0.t > 500_000 } - sentLog.removeAll { now - $0.t > 500_000 } - unacked.removeAll { now - $0.sent > 2_000_000 } - captures.removeAll { now - $0 > 1_000_000 } - guard RateController.enabled, ceiling > 0 else { return nil } - let base = baseline() - let spread = rtts.filter { now - $0.t < 2_000_000 }.map(\.v).sorted() - let jitter = spread.isEmpty ? 0 : spread[spread.count * 9 / 10] - base - let interval = Int64(1_000_000 / max(1, fpsLocked)) - slack = interval + min(max(jitter * 3 / 2, 25_000), 80_000) - let recent = rtts.filter { now - $0.t < 300_000 }.map(\.v).sorted() - let queueing = recent.isEmpty ? 0 : recent[recent.count / 2] - base - let oldestAge = unacked.first.map { now - $0.sent } ?? 0 - let stuck = oldestAge > base + 100_000 - let delivered = acked.reduce(0) { $0 + $1.bytes } * 16 // bits/s over the last half second - let sending = sentLog.reduce(0) { $0 + $1.bytes } * 16 - // Demand: captures per second (up to the tier's rate) times the - // average frame. A test card or a mostly still window wants far - // less than its budget; when the link hiccups, cutting its bitrate - // can't help, and it would only look link-limited afterwards. - if !sentLog.isEmpty { frameBytes = sentLog.reduce(0) { $0 + $1.bytes } / sentLog.count } - let demand = min(captures.count, fpsLocked) * frameBytes * 8 - // Delay alone isn't our queue: Wi-Fi jitters by itself. It only - // counts while this stream uses a good part of its budget (so its - // own data could be what's queueing). Frames the gate had to hold, - // or one stuck in flight, show demand the link isn't carrying - // whatever was sent (the gate itself keeps what's sent low). - let busy = sending >= target / 2 - // Twice in a row (200 ms), so one late ack doesn't count. - let signal = (busy && queueing > 40_000) || held >= 3 || stuck - let congested = signal && lastSignal - lastSignal = signal - let heldNow = held - held = 0 - let floorBps = 300_000 - if congested, now - lastDecrease > 300_000 { - // Down to a bit under what got through: at least a fifth off, at - // most half (a stall delivers nothing, but the link is still there). - let measured = Int(Double(delivered) * 0.9) - var next = max(floorBps, min(target * 4 / 5, max(measured, target / 2))) - // App-limited (it wants about half its budget or less): never - // below twice what it wants, however many cuts in a row. The - // extra quarter is hysteresis, so frame sizes wobbling at the - // floor don't switch the protection off. - if demand > 0, demand * 2 <= target * 5 / 4 { next = max(next, min(target, demand * 2)) } - target = next - lastDecrease = now - events.append((now, "down to \(target / 1000) kbit/s: queue \(queueing / 1000) ms, held \(heldNow), " - + "oldest \(oldestAge / 1000) ms, base \(base / 1000) ms, sent \(sending / 1000) got \(delivered / 1000) " - + "wants \(demand / 1000)")) - } else if !congested, now - lastDecrease > 1_000_000, now - lastIncrease > 250_000, target < ceiling, - sending > target * 6 / 10 || now - lastDecrease > 3_000_000 { - // Clear for a second and using what it has: probe up. - target = min(ceiling, Int(Double(target) * 1.1) + 50_000) - lastIncrease = now + let old = value(0), tier = value(2) + let result = Int(fc_update(core, now)) + if value(0) < old || value(2) != tier { + events.append(["t": now, "e": "target \(value(0)) bit/s; tier \(value(2))"]) + if events.count > 200 { events.removeFirst() } } - // Fewer frames or pixels only help a stream that fills its budget; - // a small one (a still window, a test card) keeps its tier. - retier(now: now, linkLimited: sending >= target * 7 / 10) - if events.count > 200 { events.removeFirst(events.count - 200) } - return target + return result == 0 ? nil : result } } - - /// Steps down quickly, straight to the tier the bitrate supports, and back - /// up one tier at a time only when there's clearly room (hysteresis). - private func retier(now: Int64, linkLimited: Bool) { - let share = Double(target) / Double(max(ceiling, 1)) - if tier < RateController.tiers.count - 1, share < RateController.floors[tier], linkLimited { - if belowSince == 0 { belowSince = now } - if now - belowSince > 500_000 { - tier = RateController.floors.firstIndex { share >= $0 } ?? RateController.tiers.count - 1 - belowSince = 0 - events.append((now, "tier \(tier)")) - } - } else { - belowSince = 0 - } - if tier > 0, share > RateController.floors[tier - 1] * 1.25 { - if aboveSince == 0 { aboveSince = now } - if now - aboveSince > 2_000_000 { - tier -= 1 - aboveSince = 0 - events.append((now, "tier \(tier)")) - } - } else { - aboveSince = 0 - } - } - func state() -> [String: Any] { - locked { - ["target": target, "ceiling": ceiling, "tier": tier, "fps": min(maxFps, RateController.tiers[tier].fps), - "scale": RateController.tiers[tier].scale, "baseRtt": Double(baseline()) / 1000, - "inFlight": unacked.count, "slack": Double(slack) / 1000, "adapt": RateController.enabled] - } + locked { ["target": value(0), "ceiling": value(1), "tier": value(2), "fps": value(3), + "scale": Double(value(4)) / 100, "baseRtt": Double(value(5)) / 1000, + "inFlight": value(6), "slack": Double(value(7)) / 1000, "adapt": Self.enabled] } } - - func eventList() -> [[String: Any]] { locked { events.map { ["t": $0.0, "e": $0.1] } } } + func eventList() -> [[String: Any]] { locked { events } } } diff --git a/mac/frame-mac-view/build.sh b/mac/frame-mac-view/build.sh index f391a58..b7b19c7 100755 --- a/mac/frame-mac-view/build.sh +++ b/mac/frame-mac-view/build.sh @@ -5,7 +5,10 @@ set -eu here=$(cd "$(dirname "$0")" && pwd) out=${1:-$here/../bin/frame-mac-view} mkdir -p "$(dirname "$out")" +obj=$(mktemp /tmp/frame-controller.XXXXXX) +trap 'rm -f "$obj"' EXIT +xcrun clang -O2 -target arm64-apple-macos14.0 -c "$here/../../desktop/controller.c" -o "$obj" xcrun swiftc -O -swift-version 5 -target arm64-apple-macos14.0 \ -import-objc-header "$here/Sources/CGVirtualDisplay.h" \ - -o "$out" "$here"/Sources/*.swift + -o "$out" "$here"/Sources/*.swift "$obj" echo "built $out"