Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 95 additions & 23 deletions Stream/ScreenCaptureController.swift
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,15 @@ final class ScreenCaptureController: NSObject {
/// ~1s telemetry poll, live only. Cancelled on every teardown path (they all
/// funnel through `stopCapture`).
@ObservationIgnored private var statsTask: Task<Void, Never>?
/// Shared frame-drop / achieved-fps counters (issue #23 / M8), created per
/// broadcast and handed to BOTH the capture output and the publisher so the four
/// shed sites and the encoded rate land in one place. Read once per stats poll to
/// fold measured fps + congestion drops into `liveStats`.
@ObservationIgnored private var frameTelemetry: FrameTelemetry?
/// The previous telemetry snapshot + its capture time, so each poll derives
/// achieved fps from the delta over the real elapsed window.
@ObservationIgnored private var lastTelemetrySnapshot: FrameTelemetrySnapshot?
@ObservationIgnored private var lastTelemetryAt: UInt64 = 0

override init() {
super.init()
Expand Down Expand Up @@ -121,24 +130,56 @@ final class ScreenCaptureController: NSObject {
// Write through INCLUDING nil: the publishers return nil during a
// reconnect, which must clear the card to its "—" placeholders rather
// than freezing on stale last-good telemetry while the stream is down.
self?.updateLiveStats(snapshot)
self?.applyPolledStats(snapshot)
try? await Task.sleep(for: .seconds(1))
}
}
}

/// One stats tick: fold the measured frame telemetry (achieved fps + cumulative
/// congestion drops) into the publisher's target-based snapshot, then store it.
/// Advances the achieved-fps delta baseline as a side effect, so it must run
/// exactly once per poll.
private func applyPolledStats(_ base: LiveStats?) {
updateLiveStats(enrichWithTelemetry(base))
}

/// Merges the shared `FrameTelemetry` into the publisher's snapshot: the measured
/// achieved fps (from the encoded-frame delta over the real elapsed window) and
/// the cumulative congestion drops. Returns `base` untouched when telemetry isn't
/// available (pre-live / torn down). Mutates the delta baseline.
private func enrichWithTelemetry(_ base: LiveStats?) -> LiveStats? {
guard let telemetry = frameTelemetry else { return base }
let current = telemetry.snapshot()
let now = DispatchTime.now().uptimeNanoseconds
var achieved = 0
if let previous = lastTelemetrySnapshot, lastTelemetryAt > 0, now > lastTelemetryAt {
let elapsed = Double(now &- lastTelemetryAt) / 1_000_000_000
achieved = FrameTelemetry.rate(from: previous, to: current, elapsed: elapsed).achievedFrameRate
}
// Advance the baseline every tick — even when `base` is nil (reconnecting) —
// so the next window measures against a fresh, ~1s-old sample.
lastTelemetrySnapshot = current
lastTelemetryAt = now
guard var stats = base else { return nil }
stats.achievedFrameRate = achieved
stats.droppedFrames = current.congestionDrops
return stats
}

/// Assigns the latest telemetry, skipping a redundant write (and the 1 Hz leaf
/// re-render it would otherwise trigger) when the snapshot is unchanged.
private func updateLiveStats(_ stats: LiveStats?) {
if liveStats != stats { liveStats = stats }
}

private func makePublisher(for transport: StreamCore.StreamProtocol) -> any Publisher {
private func makePublisher(for transport: StreamCore.StreamProtocol,
telemetry: FrameTelemetry) -> any Publisher {
switch transport {
case .rtmp, .rtmps:
RTMPPublisher()
RTMPPublisher(telemetry: telemetry)
case .srt, .whip:
SessionPublisher(protocol: transport)
SessionPublisher(protocol: transport, telemetry: telemetry)
}
}

Expand All @@ -152,8 +193,15 @@ final class ScreenCaptureController: NSObject {
return
}

let publisher = makePublisher(for: settings.selectedProtocol)
let output = ScreenCaptureOutput(publisher: publisher, settings: settings) { [weak self] error in
// One telemetry instance per broadcast, shared by the capture output (the
// four shed sites) and the publisher (admission + encoded rate).
let telemetry = FrameTelemetry()
frameTelemetry = telemetry
lastTelemetrySnapshot = nil
lastTelemetryAt = 0
let publisher = makePublisher(for: settings.selectedProtocol, telemetry: telemetry)
let output = ScreenCaptureOutput(publisher: publisher, settings: settings,
telemetry: telemetry) { [weak self] error in
Task { @MainActor [weak self] in await self?.captureDidStop(error: error) }
}

Expand Down Expand Up @@ -228,6 +276,9 @@ final class ScreenCaptureController: NSObject {
heartbeatTask = nil
statsTask?.cancel()
statsTask = nil
frameTelemetry = nil
lastTelemetrySnapshot = nil
lastTelemetryAt = 0
isLive = false
thermalNotice = nil
liveStats = nil
Expand Down Expand Up @@ -408,6 +459,10 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
/// the capture pacing so a 1080p60 pick never asks a device for more than it
/// can sustain. The thermal governor tightens this further at runtime.
private let capability = StreamCapability.current
/// Shared frame telemetry (issue #23 / M8): this output records the capture,
/// pacing (PTS-deadline), backpressure (`bufferingNewest(1)`), and compositor
/// (pool-exhaustion) sites on the serial sample queue.
private let telemetry: FrameTelemetry
private let onStopped: @Sendable (Error) -> Void
private let micLevelMeter: ScreenCaptureMicrophoneMeter
private let micLevelChannel = MicrophoneLevelChannel()
Expand All @@ -427,9 +482,11 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg

init(publisher: any Publisher,
settings: StreamSettings,
telemetry: FrameTelemetry,
onStopped: @escaping @Sendable (Error) -> Void) {
self.publisher = publisher
self.settings = settings
self.telemetry = telemetry
self.onStopped = onStopped
micLevelMeter = ScreenCaptureMicrophoneMeter(gain: settings.micVolume)
targetFrameInterval = CMTime(value: 1,
Expand All @@ -454,22 +511,28 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
await publisher.setOutputSize(target, nativeShortEdge: min(width, height))
targetSize = target
}
if settings.pipEnabled, let targetSize,
let camera = self.facecam.latest.take(),
let composited = self.compositor.composite(
screen: image,
camera: camera,
targetSize: targetSize,
orientation: .up,
corner: settings.pipCorner,
scale: settings.pipScale,
cameraPosition: settings.cameraPosition
),
let output = self.compositor.makeSampleBuffer(
from: composited,
timingSource: sampleBuffer
) {
await publisher.appendVideo(output)
if settings.pipEnabled, let targetSize, let camera = self.facecam.latest.take() {
if let composited = self.compositor.composite(
screen: image,
camera: camera,
targetSize: targetSize,
orientation: .up,
corner: settings.pipCorner,
scale: settings.pipScale,
cameraPosition: settings.cameraPosition
),
let output = self.compositor.makeSampleBuffer(
from: composited,
timingSource: sampleBuffer
) {
await publisher.appendVideo(output)
} else {
// Pool exhausted (a slow encoder holding surfaces) or the wrap
// failed: skip the overlay for this frame and send the raw
// screen. The frame itself is NOT lost — the overlay is.
self.telemetry.recordDrop(.compositor)
await publisher.appendVideo(sampleBuffer)
}
} else {
await publisher.appendVideo(sampleBuffer)
}
Expand All @@ -491,12 +554,16 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
guard sampleBuffer.isValid, sampleBuffer.dataReadiness == .ready else { return }
switch type {
case .screen:
telemetry.recordCaptured()
// Downsample the native-refresh feed to the target frame rate by PTS
// deadline: emit the first frame at/after each deadline, then advance
// by one interval; re-anchor after a stall so a gap doesn't burst.
let pts = sampleBuffer.presentationTimeStamp
if pts.isValid {
if nextVideoDeadline.isValid, CMTimeCompare(pts, nextVideoDeadline) < 0 {
// Intentional pacing shed (e.g. 120 Hz → 30 fps), not a fault —
// counted separately from the congestion drops the HUD surfaces.
telemetry.recordDrop(.pacing)
return // arrived before the next target-fps slot — drop it
}
let advanced = CMTimeAdd(nextVideoDeadline.isValid ? nextVideoDeadline : pts,
Expand All @@ -505,7 +572,12 @@ private final class ScreenCaptureOutput: NSObject, SCStreamOutput, SCStreamDeleg
? advanced
: CMTimeAdd(pts, targetFrameInterval)
}
videoContinuation.yield(sampleBuffer)
// `bufferingNewest(1)`: if the consumer (compositor/append) hasn't drained
// the previous frame, this yield evicts it — a backpressure shed the
// pipeline couldn't keep up with. `.dropped` carries that evicted frame.
if case .dropped = videoContinuation.yield(sampleBuffer) {
telemetry.recordDrop(.backpressure)
}
case .audio:
if settings.includeAppAudio { publisher.enqueueApp(sampleBuffer) }
case .microphone:
Expand Down
24 changes: 20 additions & 4 deletions Stream/SettingsView.swift
Original file line number Diff line number Diff line change
Expand Up @@ -1022,7 +1022,8 @@ private final class MicrophoneLevelMonitor {
/// A leaf view so its ~1s telemetry ticks invalidate only this subtree (not the
/// whole launcher List), and every value is monospaced + single-line so the card's
/// height is invariant per tick — the content-sized drawer detent never springs on
/// a number change (only an occasional thermal-notice appearance resizes it).
/// a number change (only an occasional thermal- or dropped-frames notice appearing
/// resizes it).
private struct LiveStatsCard: View {
var capture: ScreenCaptureController

Expand All @@ -1037,6 +1038,7 @@ private struct LiveStatsCard: View {
.lineLimit(1)
.minimumScaleFactor(0.8)
}
droppedNotice
}
.padding(14)
.frame(maxWidth: .infinity, alignment: .leading)
Expand All @@ -1062,18 +1064,32 @@ private struct LiveStatsCard: View {
}
}

/// Bitrate (ABR target) · effective fps · uplink health. Values are the
/// encoder's applied targets, not measured throughput (measured fps needs M8).
/// Bitrate (ABR target) · achieved fps · uplink health. Bitrate is the encoder's
/// applied target; FPS is the MEASURED achieved rate (M8), falling back to the
/// target for the first second before a rate is available.
private var metricsRow: some View {
HStack(spacing: 0) {
metric("Bitrate", value: capture.liveStats?.bitRateLabel ?? "—")
divider
metric("FPS", value: capture.liveStats.map { "\($0.frameRate)" } ?? "—")
metric("FPS", value: capture.liveStats.map { "\($0.displayFrameRate)" } ?? "—")
divider
metric("Uplink", value: capture.liveStats?.linkHealth.label ?? "—", tint: uplinkTint)
}
}

/// A subtle count of the frames congestion has shed this session (backpressure +
/// admission). Hidden until the first drop, so a healthy stream shows nothing;
/// the coloured Uplink metric already carries the live severity signal.
@ViewBuilder private var droppedNotice: some View {
if let stats = capture.liveStats, stats.droppedFrames > 0 {
Label("\(stats.droppedLabel) frames dropped", systemImage: "square.stack.3d.up.slash")
.font(.caption2)
.foregroundStyle(.secondary)
.lineLimit(1)
.minimumScaleFactor(0.8)
}
}

private var divider: some View { Divider().frame(height: 26) }

private var uplinkTint: Color {
Expand Down
15 changes: 13 additions & 2 deletions StreamBroadcast/RTMPPublisher.swift
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,12 @@ actor RTMPPublisher: Publisher {
/// Sheds input video frames when the outbound queue is deep, to bound latency
/// (HaishinKit's send queue is otherwise unbounded). Refreshed by checkNetworkHealth.
private let videoAdmission = VideoFrameAdmission()
/// Shared frame-drop / achieved-fps counters (issue #23 / M8). This actor records
/// the admission shed + every encoded append; the capture side records the
/// capture/pacing/backpressure/compositor sites into the same instance, and the
/// controller reads one snapshot per telemetry poll. Reachable here so the
/// adaptive logic (M9) can consume the congestion-drop signal.
private let telemetry: FrameTelemetry

// Ordered, lossless audio ingress: ScreenCaptureKit yields synchronously, a
// single consumer per track awaits each append — preserving PTS order so
Expand All @@ -80,7 +86,8 @@ actor RTMPPublisher: Publisher {
/// Shared outbound-queue stall watchdog state (stalledTicks + lastQueueBytes).
private var watchdog = WatchdogState()

init() {
init(telemetry: FrameTelemetry = FrameTelemetry()) {
self.telemetry = telemetry
(micStream, micCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded)
(appStream, appCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded)
}
Expand Down Expand Up @@ -770,10 +777,14 @@ actor RTMPPublisher: Publisher {
lastVideoBuffer = sb
// Under outbound congestion, drop a proportion of video frames (audio is
// never dropped) so latency stays bounded instead of the queue growing.
guard videoAdmission.admit() else { return }
guard videoAdmission.admit() else {
telemetry.recordDrop(.admission)
return
}
let duration = CMTime(value: 1,
timescale: CMTimeScale(encodeFrameRate))
let normalized = timeline.normalize(sb, kind: .video, fallbackDuration: duration)
telemetry.recordEncoded()
await mixer.append(enforceMonotonicVideo(normalized, minStep: duration))
}

Expand Down
14 changes: 12 additions & 2 deletions StreamBroadcast/SessionPublisher.swift
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,10 @@ actor SessionPublisher: Publisher {

private var timeline = MediaTimelineNormalizer()
private let videoAdmission = VideoFrameAdmission()
/// Shared frame-drop / achieved-fps counters (issue #23 / M8). See RTMPPublisher:
/// records the admission shed + every encoded append into the same instance the
/// capture side and controller share.
private let telemetry: FrameTelemetry

// Ordered, lossless audio ingress: ScreenCaptureKit yields synchronously, a
// single consumer per track awaits each append — preserving PTS order.
Expand Down Expand Up @@ -123,8 +127,10 @@ actor SessionPublisher: Publisher {
/// The reconnect loop's currently-sleeping backoff; cancelling = "retry now".
private var backoffSleepTask: Task<Void, any Error>?

init(protocol streamProtocol: StreamCore.StreamProtocol) {
init(protocol streamProtocol: StreamCore.StreamProtocol,
telemetry: FrameTelemetry = FrameTelemetry()) {
self.transport = streamProtocol
self.telemetry = telemetry
(micStream, micCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded)
(appStream, appCont) = AsyncStream.makeStream(of: CMSampleBuffer.self, bufferingPolicy: .unbounded)
}
Expand Down Expand Up @@ -644,9 +650,13 @@ actor SessionPublisher: Publisher {
lastMediaAt = now
lastVideoAppendAt = now
lastVideoBuffer = sb
guard videoAdmission.admit() else { return }
guard videoAdmission.admit() else {
telemetry.recordDrop(.admission)
return
}
let duration = CMTime(value: 1, timescale: CMTimeScale(encodeFrameRate))
let normalized = timeline.normalize(sb, kind: .video, fallbackDuration: duration)
telemetry.recordEncoded()
await mixer.append(enforceMonotonicVideo(normalized, minStep: duration))
}

Expand Down
Loading
Loading