Adjustable buffer

Decimus.xcodeproj/project.pbxproj

@@ -52,6 +52,7 @@
 		9B131E332B7E4A1C00B29D77 /* ApplicationHEVCSEIs.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9B131E322B7E4A1C00B29D77 /* ApplicationHEVCSEIs.swift */; };
 		9B131E352B7E548700B29D77 /* CMSampleBuffer+Attachments.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9B131E342B7E548700B29D77 /* CMSampleBuffer+Attachments.swift */; };
 		9B165A1529840B610017079F /* CallController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9B165A1429840B610017079F /* CallController.swift */; };
+		9B3D773E2C64B70200858313 /* SlidingWindow.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9B3D773D2C64B70200858313 /* SlidingWindow.swift */; };
 		9B3F86852C58F4B800B5B8BA /* CircularBuffer.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9B3F86842C58F4B800B5B8BA /* CircularBuffer.swift */; };
 		9B4788D42B972F990056CE7E /* TestVideoSubscription.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9B4788D32B972F990056CE7E /* TestVideoSubscription.swift */; };
 		9B480662297EAF170040F5D4 /* InCallView.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9B480661297EAF170040F5D4 /* InCallView.swift */; };
@@ -221,6 +222,7 @@
 		9B131E322B7E4A1C00B29D77 /* ApplicationHEVCSEIs.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = ApplicationHEVCSEIs.swift; sourceTree = "<group>"; };
 		9B131E342B7E548700B29D77 /* CMSampleBuffer+Attachments.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "CMSampleBuffer+Attachments.swift"; sourceTree = "<group>"; };
 		9B165A1429840B610017079F /* CallController.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CallController.swift; sourceTree = "<group>"; };
+		9B3D773D2C64B70200858313 /* SlidingWindow.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = SlidingWindow.swift; sourceTree = "<group>"; };
 		9B3F86842C58F4B800B5B8BA /* CircularBuffer.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CircularBuffer.swift; sourceTree = "<group>"; };
 		9B4788D32B972F990056CE7E /* TestVideoSubscription.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TestVideoSubscription.swift; sourceTree = "<group>"; };
 		9B480661297EAF170040F5D4 /* InCallView.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = InCallView.swift; sourceTree = "<group>"; };
@@ -520,6 +522,7 @@
 				9B131E342B7E548700B29D77 /* CMSampleBuffer+Attachments.swift */,
 				9BC53C9B2BFE075300BB39C6 /* VarianceCalculator.swift */,
 				9B3F86842C58F4B800B5B8BA /* CircularBuffer.swift */,
+				9B3D773D2C64B70200858313 /* SlidingWindow.swift */,
 			);
 			path = Decimus;
 			sourceTree = "<group>";
@@ -940,6 +943,7 @@
 				1867635D2B2B6BB000339421 /* H264Publication+Measurement.swift in Sources */,
 				FF2498B42A55E8CC00C6D66D /* SubscriptionFactory.swift in Sources */,
 				186397A12ADE0B9100FBF877 /* VideoDevices.swift in Sources */,
+				9B3D773E2C64B70200858313 /* SlidingWindow.swift in Sources */,
 				9B52EE332B763157006E9C58 /* PublicationPopover.swift in Sources */,
 				18C89A3C2A13D30B005B333B /* MetricsSubmitter.swift in Sources */,
 				FFD1CFF02A214FA60034CDD0 /* SubscriberDelegate.swift in Sources */,

Decimus/Measurements/VideoJitterBuffer+Measurement.swift

@@ -23,6 +23,14 @@ extension VideoJitterBuffer {
             record(field: "currentDepth", value: UInt32(depth * 1000) as AnyObject, timestamp: timestamp)
         }
 
+        func targetDepth(depth: TimeInterval, timestamp: Date) {
+            record(field: "targetDepth", value: depth, timestamp: timestamp)
+        }
+
+        func idealTargetDepth(depth: TimeInterval, timestamp: Date) {
+            record(field: "idealDepth", value: depth, timestamp: timestamp)
+        }
+
         func underrun(timestamp: Date?) {
             self.underruns += 1
             record(field: "underruns", value: self.underruns as AnyObject, timestamp: timestamp)

Decimus/SlidingWindow.swift

@@ -0,0 +1,47 @@
+import DequeModule
+
+class SlidingWindow<T: Numeric> {
+    private var values: Deque<(timestamp: Date, value: T)> = []
+    private let length: TimeInterval
+
+    init(length: TimeInterval) {
+        self.length = length
+    }
+
+    func add(timestamp: Date, value: T) {
+        // Remove values that are too old.
+        while let first = self.values.first,
+              timestamp.timeIntervalSince(first.timestamp) > self.length {
+            _ = self.values.popFirst()
+        }
+
+        self.values.append((timestamp, value))
+    }
+
+    func get(from: Date) -> [T] {
+        self.values.compactMap {
+            from.timeIntervalSince($0.timestamp) <= self.length ? $0.value : nil
+        }
+    }
+}
+
+class SlidingTimeWindow {
+    private let window: SlidingWindow<TimeInterval>
+    private var lastSubmit: Date?
+
+    init(length: TimeInterval) {
+        self.window = .init(length: length)
+    }
+
+    func add(timestamp: Date) {
+        if let lastSubmit = self.lastSubmit {
+            let elapsed = timestamp.timeIntervalSince(lastSubmit)
+            self.window.add(timestamp: timestamp, value: elapsed)
+        }
+        self.lastSubmit = timestamp
+    }
+
+    func max(from: Date) -> TimeInterval? {
+        self.window.get(from: from).max()
+    }
+}

Decimus/Subscriptions/VideoHandler.swift

@@ -2,7 +2,7 @@
 import Atomics
 import os
 
 // swiftlint:disable type_body_length
 
 enum DecimusVideoRotation: UInt8 {
     case portrait = 1
@@ -56,6 +56,7 @@
 
     private var duration: TimeInterval? = 0
     private let variances: VarianceCalculator
+    private var currentTargetDepth: TimeInterval
 
     /// Create a new video handler.
     /// - Parameters:
@@ -100,6 +101,7 @@
         self.metricsSubmitter = metricsSubmitter
         self.description = self.namespace
         self.variances = variances
+        self.currentTargetDepth = jitterBufferConfig.minDepth
 
         if jitterBufferConfig.mode != .layer {
             // Create the decoder.
@@ -250,8 +252,9 @@
         self.jitterBuffer = try .init(namespace: self.namespace,
                                       metricsSubmitter: self.metricsSubmitter,
                                       sort: !self.reliable,
-                                      minDepth: self.jitterBufferConfig.minDepth,
-                                      capacity: Int(floor(self.jitterBufferConfig.capacity / duration)))
+                                      minDepth: self.currentTargetDepth,
+                                      capacity: self.jitterBufferConfig.capacity,
+                                      duration: duration)
         self.duration = duration
     }
 
@@ -318,6 +321,17 @@
                                                      ordering: .acquiringAndReleasing)
     }
 
+    /// Update the target depth of this handler's jitter buffer, if any.
+    /// - Parameter depth: Target depth in seconds.
+    func setTargetDepth(_ depth: TimeInterval, from: Date) {
+        self.currentTargetDepth = depth
+        guard let buffer = self.jitterBuffer else {
+            Self.logger.warning("Set target depth on nil buffer!?")
+            return
+        }
+        buffer.setTargetDepth(depth, from: from)
+    }
+
     private func decode(sample: DecimusVideoFrame, from: Date) throws {
         // Should we feed this frame to the decoder?
         // get groupId and objectId from the frame (1st frame)
@@ -442,7 +456,7 @@
             throw "Missing sample format"
         }
         let size = format.dimensions
         return "\(self.namespace): \(String(describing: config.codec)) \(size.width)x\(size.height) \(fps)fps \(Float(config.bitrate) / pow(10, 6))Mbps"
     }
 }
 
@@ -489,4 +503,4 @@
     public static func == (lhs: CMVideoDimensions, rhs: CMVideoDimensions) -> Bool {
         lhs.width == rhs.width && lhs.height == rhs.height
     }
 }

Decimus/Subscriptions/VideoSubscription.swift

@@ -50,8 +50,10 @@
     private let decodedVariances: VarianceCalculator
     private var formats: [QuicrNamespace: CMFormatDescription?] = [:]
     private var timestampTimeDiff: TimeInterval?
+    private var suspension = SlidingTimeWindow(length: 60)
+    private var currentMax: TimeInterval?
 
     init(sourceId: SourceIDType,
          profileSet: QClientProfileSet,
          participants: VideoParticipants,
          metricsSubmitter: MetricsSubmitter?,
@@ -121,7 +123,7 @@
                 guard let self = self else { return }
                 self.handlerLock.withLock {
                     // Remove any expired handlers.
                     for handler in self.lastUpdateTimes where Date.now.timeIntervalSince(handler.value) >= self.cleanupTimer {
                         self.lastUpdateTimes.removeValue(forKey: handler.key)
                         if let video = self.videoHandlers.removeValue(forKey: handler.key),
                            let last = self.last,
@@ -163,7 +165,7 @@
         return SubscriptionError.noDecoder.rawValue
     }
 
     func subscribedObject(_ name: String!,
                           data: UnsafeRawPointer!,
                           length: Int,
                           groupId: UInt32,
@@ -211,6 +213,13 @@
             startRenderTask()
         }
 
+        // Suspension tracking.
+        if self.currentMax == nil {
+            // Ensure that we take into account our configured value.
+            self.suspension.add(timestamp: now.addingTimeInterval(-self.jitterBufferConfig.minDepth))
+        }
+        self.suspension.add(timestamp: now)
+
         let handler: VideoHandler
         do {
             handler = try self.handlerLock.withLock {
@@ -221,15 +230,33 @@
                     self.videoHandlers[name] = handler
                     return handler
                 }
+
+                // While we're here, set depth for everyone.
+                if self.jitterBufferConfig.adaptive {
+                    if let thisMax = self.suspension.max(from: now) {
+                        if let currentMax = self.currentMax,
+                           thisMax != currentMax {
+                            for handler in self.videoHandlers {
+                                handler.value.setTargetDepth(thisMax, from: now)
+                            }
+                        }
+                        self.currentMax = thisMax
+                    }
+                }
+
                 return lookup
             }
         } catch {
             Self.logger.error("Failed to fetch/create handler: \(error.localizedDescription)")
             return SubscriptionError.none.rawValue
         }
+
+        // Set timestamp diff.
         if let diff = self.timestampTimeDiff {
             handler.setTimeDiff(diff: diff)
         }
+
+        // Submit the data.
         do {
             try handler.submitEncodedData(frame, from: now)
         } catch {
@@ -293,7 +320,7 @@
         case highestRes(item: SimulreceiveItem, pristine: Bool)
     }
 
     internal static func makeSimulreceiveDecision(choices: inout any Collection<SimulreceiveItem>) -> SimulreceiveReason? {
         // Early return.
         guard choices.count > 1 else {
             if let first = choices.first {
@@ -310,7 +337,7 @@
 
         // We want the highest non-discontinous frame.
         // If all are non-discontinous, we'll take the highest quality.
         let sorted = choices.sorted { $0.image.image.formatDescription!.dimensions.width > $1.image.image.formatDescription!.dimensions.width }
         let pristine = sorted.filter { !$0.image.discontinous }
         if let pristine = pristine.first {
             return .highestRes(item: pristine, pristine: true)
@@ -324,7 +351,7 @@
     // Caller must lock handlerLock.
     // swiftlint:disable cyclomatic_complexity
     // swiftlint:disable function_body_length
     private func makeSimulreceiveDecision(at: Date) throws -> TimeInterval {
         guard !self.videoHandlers.isEmpty else {
             throw "No handlers"
         }
@@ -389,7 +416,7 @@
         let incomingWidth = selectedSample.formatDescription!.dimensions.width
         var wouldStepDown = false
         if let last = self.lastImage,
            incomingWidth < last.image.formatDescription!.dimensions.width || selected.image.discontinous && !last.discontinous {
             wouldStepDown = true
         }
 
@@ -409,7 +436,7 @@
                 }
 
                 let newValue = pauseCandidateCount.value + 1
                 Self.logger.warning("Incremented pause count for: \(pauseCandidate.config.width), now: \(newValue)/\(self.pauseMissThreshold)")
                 if newValue >= self.pauseMissThreshold {
                     // Pause this subscription.
                     Self.logger.warning("Pausing subscription: \(pauseCandidate.config.width)")
@@ -435,13 +462,13 @@
                 case .highestRes(let item, let pristine):
                     if choice.namespace == item.namespace {
                         assert(choice.namespace == selected.namespace)
                         report.append(.init(item: choice, selected: true, reason: "Highest \(pristine ? "Pristine" : "Discontinous")", displayed: !qualitySkip))
                         continue
                     }
                 case .onlyChoice(let item):
                     if choice.namespace == item.namespace {
                         assert(choice.namespace == selected.namespace)
                         report.append(.init(item: choice, selected: true, reason: "Only choice", displayed: !qualitySkip))
                     }
                     continue
                 }
@@ -532,7 +559,7 @@
         let seiData: ApplicationSeiData
     }
 
     private func depacketize(namespace: QuicrNamespace,
                              data: Data,
                              groupId: UInt32,
                              objectId: UInt16) throws -> DecimusVideoFrame? {
@@ -619,4 +646,4 @@
                      captureDate: captureDate)
     }
 }
 // swiftlint:enable type_body_length

Decimus/VideoJitterBuffer.swift

@@ -3,7 +3,7 @@
 import AVFoundation
 import Atomics
 
 // swiftlint:disable force_cast
 
 /// A very simplified jitter buffer designed to contain compressed video frames in order.
 class VideoJitterBuffer {
@@ -12,6 +12,7 @@
         var mode: Mode = .none
         var minDepth: TimeInterval = 0.2
         var capacity: TimeInterval = 2
+        var adaptive: Bool = false
     }
 
     enum Mode: CaseIterable, Identifiable, Codable {
@@ -23,24 +24,27 @@
     }
 
     private static let logger = DecimusLogger(VideoJitterBuffer.self)
-    private let minDepth: TimeInterval
+    private var minDepth: TimeInterval
     private var buffer: CMBufferQueue
     private let measurement: MeasurementRegistration<VideoJitterBufferMeasurement>?
     private var play: Bool = false
     private var lastSequenceRead = ManagedAtomic<UInt64>(0)
     private var lastSequenceSet = ManagedAtomic<Bool>(false)
+    private let capacity: TimeInterval
+    private let originalDepth: TimeInterval
 
     /// Create a new video jitter buffer.
     /// - Parameter namespace The namespace of the video this buffer is used for, for identification purposes.
     /// - Parameter metricsSubmitter Optionally, an object to submit metrics through.
     /// - Parameter sort True to actually sort on sequence number, false if they're already in order.
     /// - Parameter minDepth Fixed initial delay in seconds.
-    /// - Parameter capacity Capacity in buffers / elements.
+    /// - Parameter capacity Capacity in time.
     init(namespace: QuicrNamespace,
          metricsSubmitter: MetricsSubmitter?,
          sort: Bool,
          minDepth: TimeInterval,
-         capacity: Int) throws {
+         capacity: TimeInterval,
+         duration: TimeInterval) throws {
         let handlers = CMBufferQueue.Handlers { builder in
             builder.compare {
                 if !sort {
@@ -76,18 +80,27 @@
                 ($0 as! DecimusVideoFrame).samples.reduce(0) { $0 + $1.totalSampleSize }
             }
             builder.isDataReady {
                 ($0 as! DecimusVideoFrame).samples.reduce(true) { $0 && $1.dataReadiness == .ready }
             }
         }
-        self.buffer = try .init(capacity: capacity, handlers: handlers)
+        let elementCapacity = CMItemCount(round(capacity / duration))
+        self.buffer = try .init(capacity: elementCapacity, handlers: handlers)
+        guard minDepth <= capacity else {
+            throw "Target depth cannot be > capacity"
+        }
+        self.minDepth = minDepth
+        self.originalDepth = minDepth
+        self.capacity = capacity
         if let metricsSubmitter = metricsSubmitter {
             let measurement = VideoJitterBufferMeasurement(namespace: namespace)
             self.measurement = .init(measurement: measurement, submitter: metricsSubmitter)
+            let now = Date.now
+            Task(priority: .utility) {
+                await measurement.targetDepth(depth: self.minDepth, timestamp: now)
+            }
         } else {
             self.measurement = nil
         }
-
-        self.minDepth = minDepth
     }
 
     /// Write a video frame into the jitter buffer.
@@ -112,6 +125,18 @@
         }
     }
 
+    func setTargetDepth(_ depth: TimeInterval, from: Date) {
+        // TODO(RichLogan): For now, don't exceed original declared depth.
+        let clampedDepth = min(depth, self.originalDepth)
+        self.minDepth = clampedDepth
+        if let measurement = self.measurement {
+            Task(priority: .utility) {
+                await measurement.measurement.targetDepth(depth: clampedDepth, timestamp: from)
+                await measurement.measurement.idealTargetDepth(depth: depth, timestamp: from)
+            }
+        }
+    }
+
     /// Attempt to read a frame from the front of the buffer.
     /// - Returns Either the oldest available frame, or nil.
     func read(from: Date) -> DecimusVideoFrame? {
@@ -164,7 +189,7 @@
                            offset: TimeInterval,
                            since: Date = .init(timeIntervalSinceReferenceDate: 0)) -> TimeInterval {
         guard let sample = frame.samples.first else {
             // TODO: Ensure we cannot enqueue a frame with no samples to make this a valid error.
             fatalError()
         }
         let timestamp = sample.presentationTimeStamp

Decimus/Views/Settings/SubscriptionSettingsView.swift

@@ -30,7 +30,7 @@
                         "Depth (s)",
                         value: $subscriptionConfig.value.jitterDepthTime,
                         format: .number)
                         .onChange(of: subscriptionConfig.value.jitterDepthTime) {
                             subscriptionConfig.value.videoJitterBuffer.minDepth = $0
                             subscriptionConfig.value.jitterMaxTime = $0
                         }
@@ -41,6 +41,10 @@
                         value: $subscriptionConfig.value.videoJitterBuffer.capacity,
                         format: .number)
                 }
+                HStack {
+                    Text("Video Jitter - Adaptive")
+                    Toggle(isOn: $subscriptionConfig.value.videoJitterBuffer.adaptive) {}
+                }
                 Picker("Opus Window Size (s)", selection: $subscriptionConfig.value.opusWindowSize) {
                     ForEach(OpusWindowSize.allCases) {
                         Text(String(describing: $0))
@@ -68,7 +72,7 @@
                         ForEach(devices.cameras, id: \.uniqueID) {
                             Text($0.localizedName)
                                 .tag($0.uniqueID)
                         }.onChange(of: preferredCamera) { _ in
                             guard self.preferredCamera != self.noPreference else {
                                 AVCaptureDevice.userPreferredCamera = nil
                                 return
@@ -103,13 +107,13 @@
                         }
                     }
                     .pickerStyle(.segmented)
                     .onChange(of: subscriptionConfig.value.videoJitterBuffer.mode) {
                         if $0 == .layer && subscriptionConfig.value.simulreceive != .none {
                             subscriptionConfig.value.simulreceive = .none
                         }
                     }
                     .onAppear {
                         if subscriptionConfig.value.videoJitterBuffer.mode == .layer && subscriptionConfig.value.simulreceive != .none {
                             subscriptionConfig.value.simulreceive = .none
                         }
                     }