Share the desktop stream rate controller across host platforms

.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/

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)

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 <stdlib.h>
+#include <string.h>
+#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(n<s->n && s->a[n].t<oldest) n++;
+    remove_first(s,n);
+}
+static int64_t base(FCController *c) {
+    int64_t v=0;
+    for(int i=0;i<c->rtts.n;i++) if(!i || c->rtts.a[i].v<v) v=c->rtts.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)-(x<y);
+}
+static int64_t quantile(Samples *s,int64_t since,int numerator,int denominator) {
+    int64_t a[CAP]; int n=0;
+    for(int i=0;i<s->n;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;i<s->n;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.n<window && now-c->flight.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;i<c->flight.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->target<c->ceiling &&
+              (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 && share<floors[c->tier] && sending>=c->target*7/10) {
+        if(!c->below)c->below=now;
+        if(now-c->below>500000) {
+            for(c->tier=0;c->tier<4 && share<floors[c->tier];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;
+    }
+}

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 <stdint.h>
+#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

mac/frame-mac-view/Sources/CGVirtualDisplay.h

@@ -26,3 +26,5 @@
 - (instancetype)initWithDescriptor:(CGVirtualDisplayDescriptor *)descriptor;
 - (BOOL)applySettings:(CGVirtualDisplaySettings *)settings;
 @end
+
+#include "../../../desktop/controller.h"

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<T>(_ 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 } }
 }

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"