diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ba692b1..26faae9b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,19 @@ the public-API contract. ## [Unreleased] +### Changed + +- **A live stream on the software path starts near the newest media.** A continuous live stream over + HTTP opens with a burst of what the origin holds of the channel (8 to 18 s on one IPTV panel, at + link speed), and the software host armed its clock on that burst's first packet. Nothing on the + software path catches up, so the picture then ran the whole burst behind the broadcast for as long + as the channel played: measured 17 s behind on a channel whose native-path session ran 6 s behind. + The live feeder now waits for the burst to land (the ring growing no faster than real time over a + quarter second, from 0.3 s on, or 1.5 s at most) and starts 4 s behind the newest media, on a + keyframe; a source that delivers in real time, or a ring too shallow to be worth a later keyframe, + still starts from its first packet. The media behind the start stays in the rewind ring. + `SoftwareLiveJoinPolicy`, `SoftwareLiveJoinPlacementTests`. + ### Fixed - **Profile 7 FEL chroma residual is sampled where the base chroma sits.** With a quarter-resolution diff --git a/Sources/AetherEngine/Native/SoftwareLiveJoinPolicy.swift b/Sources/AetherEngine/Native/SoftwareLiveJoinPolicy.swift new file mode 100644 index 00000000..fd9222e6 --- /dev/null +++ b/Sources/AetherEngine/Native/SoftwareLiveJoinPolicy.swift @@ -0,0 +1,78 @@ +import Foundation + +/// Where a live session on the software host starts playing, once what the source sends at connect +/// has landed. +/// +/// A continuous live stream over HTTP (an IPTV panel's MPEG-TS form of a channel) opens with a burst: +/// the panel sends what it holds of the channel, 8 to 18 s of media measured on one panel, at link +/// speed, and real time after that. The feeder used to arm the clock on the first packet of that +/// burst, and since nothing on the software path ever catches up, the picture then ran the whole +/// burst behind the broadcast for as long as the channel played: 17 s on a channel whose native-path +/// session, which AVPlayer places a holdback behind the newest segment, ran 6 s behind. For live +/// sport that is the difference other players are judged by. +/// +/// So a live join on the software host waits for the burst to land, then starts `leadSeconds` behind +/// the newest media, on the keyframe at or before that moment. The burst is landed when the ring has +/// grown no faster than real time over `calmWindowSeconds` (a source that delivers in real time is +/// calm from its first packets and starts after `minimumSettleSeconds`), or after +/// `maximumSettleSeconds` on a link that is still catching up, where the start takes what has +/// arrived. A ring holding less than `leadSeconds + minimumSkipSeconds` is played from its start: +/// there is nothing worth skipping. The media behind the start stays in the rewind ring. +/// +/// Pure, so the rule is testable without a session; `SoftwarePlaybackHost` feeds it the ring's +/// first and newest PTS from the feeder thread until it decides. +struct SoftwareLiveJoinPolicy: Equatable { + /// How far behind the newest media the picture starts. The audio pump keeps this much decoded + /// audio ahead of the clock (`AudioLookaheadPolicy.targetLeadSeconds`), and a panel that hands + /// over about a second of media at a time leaves the lead swinging within it. + static let leadSeconds: Double = 4 + /// The burst is given at least this long to begin landing before the ring's growth is judged. + static let minimumSettleSeconds: TimeInterval = 0.3 + /// After this long the start takes what has arrived, calm or not. + static let maximumSettleSeconds: TimeInterval = 1.5 + /// Growth is judged over windows this long. + static let calmWindowSeconds: TimeInterval = 0.25 + /// A ring that grew by more than this multiple of the wall clock over a window is still landing + /// the burst. + static let calmGrowthRatio: Double = 1.5 + /// A start is moved only when it skips at least this much; less is not worth a later keyframe. + static let minimumSkipSeconds: Double = 2 + + enum Decision: Equatable { + /// Play from the ring's first packet. + case fromStart + /// Start on the keyframe at or before this source PTS. + case place(targetPts: Double) + } + + private var firstObservedAt: TimeInterval? + private var windowStartPts: Double? + private var windowStartAt: TimeInterval? + + /// Notes the ring's first and newest PTS at `now` (seconds on any monotonic clock); nil while the + /// burst is still landing, then the decision, once. + mutating func observe(firstPts: Double, newestPts: Double, at now: TimeInterval) -> Decision? { + guard let firstObservedAt, let windowStartPts, let windowStartAt else { + self.firstObservedAt = now + self.windowStartPts = newestPts + self.windowStartAt = now + return nil + } + let elapsed = now - firstObservedAt + var settled = elapsed >= Self.maximumSettleSeconds + if !settled, elapsed >= Self.minimumSettleSeconds, now - windowStartAt >= Self.calmWindowSeconds { + let grew = newestPts - windowStartPts + let wall = now - windowStartAt + if grew <= wall * Self.calmGrowthRatio { + settled = true + } else { + self.windowStartPts = newestPts + self.windowStartAt = now + } + } + guard settled else { return nil } + let span = newestPts - firstPts + guard span.isFinite, span >= Self.leadSeconds + Self.minimumSkipSeconds else { return .fromStart } + return .place(targetPts: newestPts - Self.leadSeconds) + } +} diff --git a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift index 33b97a6e..aeea19bd 100644 --- a/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift +++ b/Sources/AetherEngine/Native/SoftwarePlaybackHost.swift @@ -371,6 +371,9 @@ final class SoftwarePlaybackHost { nonisolated(unsafe) private var sessionStartPts: Double = .nan nonisolated(unsafe) private var newestSourcePts: Double = .nan + /// Where a live session's picture starts once the source's connect burst has landed; read and + /// advanced by the live feeder under `liveEdgeLock`, reset with the session. + nonisolated(unsafe) private var liveJoinPolicy = SoftwareLiveJoinPolicy() /// Guards `sessionStartPts` / `newestSourcePts` against the demux /// thread writing while the main-actor time tick reads them. @@ -2057,6 +2060,7 @@ final class SoftwarePlaybackHost { retentionClaims.removeAll() liveEdgeLock.lock() sessionStartPts = .nan + liveJoinPolicy = SoftwareLiveJoinPolicy() newestSourcePts = .nan liveEdgeLock.unlock() @@ -2219,6 +2223,20 @@ final class SoftwarePlaybackHost { let recoverCursor: @Sendable (Int) -> Void = { [weak self] old in self?.recoverEvictedFeedCursor(from: old, ring: ring) } + // Where the live join starts, once the connect burst has landed (see the policy). Nil + // while it is still landing; the feeder asks again a moment later. + let liveJoinPlacement: @Sendable () -> SoftwareLiveJoinPolicy.Decision? = { [weak self] in + guard let self else { return .fromStart } + self.liveEdgeLock.lock() + defer { self.liveEdgeLock.unlock() } + guard self.sessionStartPts.isFinite, self.newestSourcePts.isFinite else { return nil } + return self.liveJoinPolicy.observe(firstPts: self.sessionStartPts, + newestPts: self.newestSourcePts, + at: ProcessInfo.processInfo.systemUptime) + } + let placeCursor: @Sendable (Int) -> Void = { [weak self] seq in + self?.setFeedCursor(seq) + } let setSourceEnded: @Sendable () -> Void = { [weak self] in self?.sourceEnded = true } @@ -2267,6 +2285,8 @@ final class SoftwarePlaybackHost { readCursor: readCursor, advanceCursor: advanceCursor, recoverCursor: recoverCursor, + liveJoinPlacement: liveJoinPlacement, + placeCursor: placeCursor, seekGeneration: getSeekGeneration, currentRate: currentRate, isPlaying: getFeederPlaying, @@ -2541,6 +2561,8 @@ final class SoftwarePlaybackHost { readCursor: @Sendable () -> Int, advanceCursor: @Sendable (Int) -> Void, recoverCursor: @Sendable (Int) -> Void, + liveJoinPlacement: @Sendable () -> SoftwareLiveJoinPolicy.Decision? = { .fromStart }, + placeCursor: @Sendable (Int) -> Void = { _ in }, seekGeneration: @Sendable () -> UInt64, currentRate: @Sendable () -> Float, isPlaying: @Sendable () -> Bool, @@ -2552,6 +2574,9 @@ final class SoftwarePlaybackHost { audioTapSink: @Sendable () -> ((@Sendable (CMSampleBuffer) -> Void)?), noteDecodeGeneration: @Sendable () -> Void ) { + // The join's start is placed once, before the first packet is fed; a seek before then is + // the viewer's own choice of position and retires it. + var liveJoinPending = true // Audio look-ahead pump (#107 audio chopping): feed audio packets ahead of the // combined cursor so the audio renderer holds AudioLookaheadPolicy.targetLeadSeconds // of decoded audio regardless of video decode pace. Without it audio lead is capped @@ -2689,6 +2714,40 @@ final class SoftwarePlaybackHost { preArmPacketsFed = 0 hadLead = false rebuffering = false + liveJoinPending = false + } + if liveJoinPending { + switch liveJoinPlacement() { + case nil: + if sourceEnded() { + // Nothing more will land; play what there is from its start. + liveJoinPending = false + } else { + condition.lock() + autoreleasepool { + _ = condition.wait(until: Date(timeIntervalSinceNow: 0.05)) + } + condition.unlock() + return true + } + case .fromStart?: + liveJoinPending = false + case .place(let targetPts)?: + liveJoinPending = false + let cursor = readCursor() + if let seq = ring.seq(forKeyframeAtOrBefore: targetPts), seq > cursor, + let keyframe = ring.packet(atSeq: seq) { + let skipped = keyframe.pts - (ring.packet(atSeq: cursor)?.pts ?? keyframe.pts) + EngineLog.emit( + "[SWHost] live join placed \(String(format: "%.1f", SoftwareLiveJoinPolicy.leadSeconds))s " + + "behind the newest media: skipped \(String(format: "%.1f", skipped))s of the " + + "connect burst (cursor seq=\(cursor) -> \(seq), keyframe pts=\(String(format: "%.2f", keyframe.pts)))", + category: .swPlayback + ) + placeCursor(seq) + return true + } + } } let cursor = readCursor() let bounds = ring.seqBounds diff --git a/Tests/AetherEngineTests/DocumentedConstantsTests.swift b/Tests/AetherEngineTests/DocumentedConstantsTests.swift index c7688b15..4305539a 100644 --- a/Tests/AetherEngineTests/DocumentedConstantsTests.swift +++ b/Tests/AetherEngineTests/DocumentedConstantsTests.swift @@ -107,6 +107,15 @@ final class DocumentedConstantsTests: XCTestCase { return text } + /// docs/formats.md quotes the lead a software-path live join keeps behind the newest media. + func testSoftwareLiveJoinLeadMatchesDocumentation() throws { + XCTAssertEqual(SoftwareLiveJoinPolicy.leadSeconds, 4) + XCTAssertEqual(SoftwareLiveJoinPolicy.minimumSettleSeconds, 0.3) + XCTAssertEqual(SoftwareLiveJoinPolicy.maximumSettleSeconds, 1.5) + XCTAssertEqual(SoftwareLiveJoinPolicy.calmWindowSeconds, 0.25) + assertDocumented("`SoftwareLiveJoinPolicy.leadSeconds` (4 s) behind the newest media", try documentation()) + } + // MARK: - Documentation corpus private static var repoRoot: URL { diff --git a/Tests/AetherEngineTests/SoftwareLiveJoinPlacementTests.swift b/Tests/AetherEngineTests/SoftwareLiveJoinPlacementTests.swift new file mode 100644 index 00000000..c163e7c9 --- /dev/null +++ b/Tests/AetherEngineTests/SoftwareLiveJoinPlacementTests.swift @@ -0,0 +1,94 @@ +import Testing +import Foundation +@testable import AetherEngine + +/// A live session on the software host starts near the newest media once the source's connect burst +/// has landed, instead of at the burst's first packet, where it would run the whole burst behind the +/// broadcast for the rest of the channel. +@Suite("Software live join placement") +struct SoftwareLiveJoinPlacementTests { + + /// A panel that sends 12 s of media in the first half second, then real time: the join waits + /// for the burst to calm and starts `leadSeconds` behind the newest media. + @Test("a connect burst is landed before the start is placed behind the newest media") + func burstIsLandedThenPlaced() throws { + var policy = SoftwareLiveJoinPolicy() + let first = 1000.0 + var decision: SoftwareLiveJoinPolicy.Decision? + var decidedAt: TimeInterval? + var now: TimeInterval = 0 + while now <= 2, decision == nil { + // 12 s of media over the first 0.5 s, then one second of media per second. + let newest = now < 0.5 ? first + 24 * now : first + 12 + (now - 0.5) + decision = policy.observe(firstPts: first, newestPts: newest, at: now) + if decision != nil { decidedAt = now } + now += 0.05 + } + let at = try #require(decidedAt) + // Calm is judged from 0.3 s on, over a quarter-second window. + #expect(at >= 0.3 && at <= 1.0, "decided at \(at)s") + let newestThen = first + 12 + (at - 0.5) + #expect(decision == .place(targetPts: newestThen - SoftwareLiveJoinPolicy.leadSeconds)) + } + + @Test("a source that delivers in real time starts from its first packet, soon") + func realTimeSourceStartsFromTheFirstPacket() throws { + var policy = SoftwareLiveJoinPolicy() + let first = 50.0 + var decision: SoftwareLiveJoinPolicy.Decision? + var decidedAt: TimeInterval? + var now: TimeInterval = 0 + while now <= 2, decision == nil { + decision = policy.observe(firstPts: first, newestPts: first + now, at: now) + if decision != nil { decidedAt = now } + now += 0.05 + } + #expect(decision == .fromStart) + let at = try #require(decidedAt) + #expect(at >= SoftwareLiveJoinPolicy.minimumSettleSeconds && at <= 0.7, "decided at \(at)s") + } + + @Test("a burst still landing at the deadline is placed on what has arrived") + func slowLinkPlacesAtTheDeadline() throws { + var policy = SoftwareLiveJoinPolicy() + let first = 0.0 + var decision: SoftwareLiveJoinPolicy.Decision? + var decidedAt: TimeInterval? + var now: TimeInterval = 0 + while now <= 3, decision == nil { + // Six seconds of media per second for as long as it takes: never calm. + decision = policy.observe(firstPts: first, newestPts: first + 6 * now, at: now) + if decision != nil { decidedAt = now } + now += 0.05 + } + let at = try #require(decidedAt) + #expect(abs(at - SoftwareLiveJoinPolicy.maximumSettleSeconds) < 0.06, "decided at \(at)s") + #expect(decision == .place(targetPts: 6 * at - SoftwareLiveJoinPolicy.leadSeconds)) + } + + @Test("a short burst is not worth a later start") + func shortBurstPlaysFromTheStart() { + var policy = SoftwareLiveJoinPolicy() + let first = 10.0 + var decision: SoftwareLiveJoinPolicy.Decision? + var now: TimeInterval = 0 + while now <= 2, decision == nil { + // Five seconds of media at once, then real time: a lead of four leaves one to skip. + decision = policy.observe(firstPts: first, newestPts: first + 5 + now, at: now) + now += 0.05 + } + #expect(decision == .fromStart) + } + + @Test("media whose timestamps run backwards is played from the start") + func backwardsTimestampsPlayFromTheStart() { + var policy = SoftwareLiveJoinPolicy() + var decision: SoftwareLiveJoinPolicy.Decision? + var now: TimeInterval = 0 + while now <= 2, decision == nil { + decision = policy.observe(firstPts: 5000, newestPts: 20 + now, at: now) + now += 0.05 + } + #expect(decision == .fromStart) + } +} diff --git a/docs/formats.md b/docs/formats.md index 142e9dcc..c6917216 100644 --- a/docs/formats.md +++ b/docs/formats.md @@ -607,7 +607,7 @@ Server-side ad insertion (SSAI) plays through the direct path instead of bouncin The live path's sliding-window eviction (which bounds resident memory) and DVR rewind are confirmed on Apple TV against a real broadcast feed: `behindLiveSeconds` holds at real-time pacing and the resident footprint stays bounded within the tvOS jetsam budget. The same behavior is exercised off-device through the `aetherctl live` / `hlsfixture` harnesses (sliding-window retention, real-time pacing, mid-stream reconnect, program-boundary discontinuities, DVR timeshift). -Raw live MPEG-TS over HTTP (a tuner or tuner proxy serving the transport stream directly, no HLS) plays on the software path; the forward-only reader routes it there automatically. Load it with `isLive: true` plus a `dvrWindowSeconds` for live semantics and the fastest open (an explicitly live load skips the open-time size probes entirely); `SourceProbe.isLive` flags no-duration network streams so hosts can detect and reload. Mid-stream-joined sources, which deliver their first samples at arbitrary PTS hours past zero, anchor the clock at the first decoded sample on every session shape (live, live+DVR, and a plain VOD open of the same URL or of a mid-broadcast capture file), publishing session-relative positions while `sourceTime` stays on the source axis for subtitles (#107). +Raw live MPEG-TS over HTTP (a tuner, a tuner proxy, or an IPTV panel's continuous form of a channel) is routed like any live source: the native loopback path unless the codec, declared interlace or a host override sends it to the software path. Such a stream opens with whatever the origin holds of the channel, sent at link speed (8 to 18 s on one IPTV panel), and a software session no longer starts on the first packet of that burst, where it would run the whole burst behind the broadcast for as long as it played: once the burst has landed the live feeder places the start `SoftwareLiveJoinPolicy.leadSeconds` (4 s) behind the newest media, on a keyframe, and the burst stays in the rewind ring (the native path's AVPlayer already sits a holdback behind the newest segment). Load it with `isLive: true` plus a `dvrWindowSeconds` for live semantics and the fastest open (an explicitly live load skips the open-time size probes entirely); `SourceProbe.isLive` flags no-duration network streams so hosts can detect and reload. Mid-stream-joined sources, which deliver their first samples at arbitrary PTS hours past zero, anchor the clock at the first decoded sample on every session shape (live, live+DVR, and a plain VOD open of the same URL or of a mid-broadcast capture file), publishing session-relative positions while `sourceTime` stays on the source axis for subtitles (#107). ## Known limitations