diff --git a/ASFWDriver/Audio/DriverKit/ASFWAudioDriverIO.cpp b/ASFWDriver/Audio/DriverKit/ASFWAudioDriverIO.cpp index be4c59f2..c0a34751 100644 --- a/ASFWDriver/Audio/DriverKit/ASFWAudioDriverIO.cpp +++ b/ASFWDriver/Audio/DriverKit/ASFWAudioDriverIO.cpp @@ -186,6 +186,32 @@ kern_return_t InstallIOOperationHandler(IOUserAudioDevice& audioDevice, control->client.PublishWriteEnd(sampleTime, hostTime, ioBufferFrameSize); PublishPlaybackRingWriteEnd(driverIvars->runtime.directAudioGraph, *control); + // Keep packet preparation driven by the CoreAudio write + // frontier as well as the OHCI refill path. The target is a + // 400-cycle content horizon (about 50 ms at 48 kHz), not a + // request for transport to manipulate audio cursors. The + // coalescing latch ensures this RT callback produces at most + // one outstanding action. + const uint64_t writeEndFrame = sampleTime + ioBufferFrameSize; + const uint64_t targetFrameEnd = + writeEndFrame + + ASFW::IsochTransport::AudioTimingGeometry:: + TxDataHorizonFrames( + driverIvars->runtime.txStreamEngine.StreamConfig() + .sampleRate); + const uint64_t requestGeneration = + control->txPreparationRequests.PublishRequest( + hostTime, targetFrameEnd); + if (driverIvars->device.audioNub && + control->txPreparationRequests.TryScheduleWake()) { + const kern_return_t requestKr = + driverIvars->device.audioNub->RequestTxPreparation( + requestGeneration); + if (requestKr != kIOReturnSuccess) { + control->txPreparationRequests.FinishWake(); + } + } + const uint32_t channels = driverIvars->runtime.directAudioGraph.memory.outputChannels; const auto& memory = driverIvars->runtime.directAudioGraph.memory; @@ -200,8 +226,6 @@ kern_return_t InstallIOOperationHandler(IOUserAudioDevice& audioDevice, const uint64_t completionCursor = driverIvars->runtime.txSlotProvider.queueControl ? driverIvars->runtime.txSlotProvider.queueControl->completionCursor.load(std::memory_order_acquire) : 0; - const uint64_t writeEndFrame = - sampleTime + ioBufferFrameSize; const uint64_t exposedFrameEnd = driverIvars->runtime.txStreamEngine.Timeline() .ExposedFrameEnd(); @@ -266,6 +290,17 @@ kern_return_t InstallIOOperationHandler(IOUserAudioDevice& audioDevice, packetizerSnapshot.frameCursorAligned; rec.packetizerHasLastDataPacket = packetizerSnapshot.hasLastDataPacket; + rec.txPreparationTargetFrameEnd = + control->txPreparationRequests.requestedTargetFrameEnd.load( + std::memory_order_acquire); + rec.txPreparationRequestedGeneration = + control->txPreparationRequests.RequestedGeneration(); + rec.txPreparationHandledGeneration = + control->txPreparationRequests.handledGeneration.load( + std::memory_order_acquire); + rec.txPreparationWakeScheduled = + control->txPreparationRequests.wakeScheduled.load( + std::memory_order_acquire); rec.packetsPrepared = txCounters.packetsPrepared.load(std::memory_order_relaxed); rec.dataPacketsPrepared = diff --git a/ASFWDriver/Audio/DriverKit/ASFWAudioDriverZts.cpp b/ASFWDriver/Audio/DriverKit/ASFWAudioDriverZts.cpp index 0ceed1e0..4177e2fa 100644 --- a/ASFWDriver/Audio/DriverKit/ASFWAudioDriverZts.cpp +++ b/ASFWDriver/Audio/DriverKit/ASFWAudioDriverZts.cpp @@ -239,13 +239,40 @@ uint32_t PrepareTransmitSlots(ASFWAudioDriver_IVars& ivars, } ASFW::Audio::Runtime::RxSequenceEntry replay{}; - if (ivars.runtime.txReplayReader.IsActive() && - ivars.runtime.txReplayReader.TryRead( - directControl->rxSequenceReplay, replay)) { + ASFW::Audio::Runtime::RxSequenceReplayReadDiagnostic replayDiagnostic{}; + if (ivars.runtime.txReplayReader.TryRead( + directControl->rxSequenceReplay, replay, + &replayDiagnostic)) { directControl->txReplayEntries.fetch_add( 1, std::memory_order_relaxed); timing.replayDataBlocks = replay.dataBlocks; } else { + const int64_t replayDistance = + replayDiagnostic.readerCursor >= replayDiagnostic.producerCursor + ? static_cast(replayDiagnostic.readerCursor - + replayDiagnostic.producerCursor) + : -static_cast(replayDiagnostic.producerCursor - + replayDiagnostic.readerCursor); + // This is the primary discriminator for a TX silence: it says + // whether RX had not produced this entry yet, had overwritten it, + // reset its epoch, or changed the slot while it was being read. + ASFW_LOG_RING_ONLY_RL( + DirectAudio, + "tx-replay-read", + 1000u, + ::ASFW::Logging::LogLevel::Warning, + "[TxReplay] fail=%s pkt=%llu cur=%llu prod=%llu d=%lld ep=%u/%u slot=%llu/%u est=%u", + ASFW::Audio::Runtime::RxSequenceReplayReadFailureName( + replayDiagnostic.failure), + nextPacketToPrepare, + replayDiagnostic.readerCursor, + replayDiagnostic.producerCursor, + replayDistance, + replayDiagnostic.readerEpoch, + replayDiagnostic.replayEpoch, + replayDiagnostic.slotSequence, + replayDiagnostic.slotEpoch, + replayDiagnostic.replayEstablished ? 1u : 0u); // Reader stale (epoch moved) or ahead of the producer: count it, // drop the reader so the next packet re-Begins on the live epoch, // and fall through with the NO-DATA disposition set above. Also @@ -490,11 +517,10 @@ void PrefillTxRingBeforeStart(ASFWAudioDriver_IVars& ivars) noexcept { return; } - // Commit one complete shared-ring lap before IT RUN. TxPreparationReady - // has a dedicated queue, but action delivery and the startup handoff can - // still be delayed. A lead-only prefill can therefore be consumed before - // the producer gets its first turn. Steady state still targets completion - // + kTxPreparationLeadPackets. + // Commit one complete shared-ring lap before IT RUN. The transport's arm + // contract validates this exact prefill so that a delayed first producer + // action cannot expose an uncommitted slot to IT DMA. Steady state still + // targets completion + kTxPreparationLeadPackets. ASFW::Protocols::Audio::AMDTP::AmdtpTimingState timing{}; timing.replayValid = true; timing.txClockValid = false; @@ -545,6 +571,7 @@ void IMPL(ASFWAudioDriver, ZtsAnchorReady) void IMPL(ASFWAudioDriver, TxPreparationReady) { (void)action; + (void)generation; if (!ivars || !ivars->runtime.txActive.load( std::memory_order_acquire)) { @@ -560,9 +587,10 @@ void IMPL(ASFWAudioDriver, TxPreparationReady) const uint64_t requested = txControl->refillRequestGeneration.load( std::memory_order_acquire); - if (generation > requested) { - return; - } + const uint64_t refillHandled = + txControl->refillHandledGeneration.load( + std::memory_order_acquire); + const bool hardwareWakePending = requested != refillHandled; const uint64_t completionCursor = txControl->completionCursor.load(std::memory_order_acquire); @@ -580,15 +608,28 @@ void IMPL(ASFWAudioDriver, TxPreparationReady) auto* directControl = ivars->runtime.directAudioGraph.control; const bool replayEstablished = directControl && directControl->rxSequenceReplay.IsEstablished(); + const uint64_t audioRequested = directControl + ? directControl->txPreparationRequests.RequestedGeneration() + : 0; + const uint64_t requestedAudioTarget = directControl + ? directControl->txPreparationRequests.requestedTargetFrameEnd.load( + std::memory_order_acquire) + : 0; const uint64_t outputWrittenEndFrame = directControl ? directControl->client.OutputWrittenEndFrame() : 0; - const uint64_t targetFrameEnd = + const uint32_t dataHorizonFrames = + ASFW::IsochTransport::AudioTimingGeometry::TxDataHorizonFrames( + ivars->runtime.txStreamEngine.StreamConfig().sampleRate); + const uint64_t outputTargetFrameEnd = outputWrittenEndFrame != 0 ? ASFW::Audio::DriverKit::SaturatingAdd( outputWrittenEndFrame, - ASFW::IsochTransport::AudioTimingGeometry:: - kTxExposureLeadFrames) + dataHorizonFrames) : 0; + const uint64_t targetFrameEnd = + requestedAudioTarget > outputTargetFrameEnd + ? requestedAudioTarget + : outputTargetFrameEnd; const uint64_t exposedFrameEndBefore = ivars->runtime.txStreamEngine.Timeline().ExposedFrameEnd(); const uint32_t slotsPrepared = @@ -608,20 +649,13 @@ void IMPL(ASFWAudioDriver, TxPreparationReady) // question: did the producer's range reach `target` this wake, or stop // short and leave a hole the IT refill ISR will later trip on? The producer // loop is linear in absolute packet index, so `prepareUntil` is exactly - // `base + slotsPrepared`; `firstMissingAbs` is the first slot NOT covered - // when it stopped short of target (UINT64_MAX = fully covered). + // `base + slotsPrepared`. { const uint64_t prepareBaseAbs = exposeCursor; const uint64_t prepareUntilAbs = exposeCursor + slotsPrepared; - const uint64_t requestedSpan = - packetLimitTarget > prepareBaseAbs - ? packetLimitTarget - prepareBaseAbs - : 0; const bool stoppedShort = prepareUntilAbs < packetCoverageTarget; const bool frameShort = targetFrameEnd != 0 && exposedFrameEndAfter < targetFrameEnd; - const uint64_t firstMissingAbs = - stoppedShort ? prepareUntilAbs : UINT64_MAX; const uint64_t committedMargin = prepareUntilAbs > completionCursor ? prepareUntilAbs - completionCursor @@ -647,63 +681,69 @@ void IMPL(ASFWAudioDriver, TxPreparationReady) "tx-prep-range", stoppedShort ? 0u : 1000u, ::ASFW::Logging::LogLevel::Warning, - "[TxPrepRange] retiredAbs=%llu prepareBaseAbs=%llu " - "prepareUntilAbs=%llu coverageTargetAbs=%llu limitTargetAbs=%llu reqSpan=%llu " - "prepared=%u short=%d firstMissingAbs=%lld marginAfter=%llu " - "frameTarget=%llu exposedBefore=%llu exposedAfter=%llu " - "frameShort=%d frameDeficit=%llu writeEnd=%llu replay=%d", + "[TxPrepRange] short=%u frame=%u ret=%llu base=%llu until=%llu cov=%llu lim=%llu n=%u margin=%llu", + stoppedShort ? 1u : 0u, + frameShort ? 1u : 0u, completionCursor, prepareBaseAbs, prepareUntilAbs, packetCoverageTarget, packetLimitTarget, - requestedSpan, slotsPrepared, - stoppedShort ? 1 : 0, - stoppedShort ? static_cast(firstMissingAbs) - : static_cast(-1), - committedMargin, - targetFrameEnd, - exposedFrameEndBefore, - exposedFrameEndAfter, - frameShort ? 1 : 0, - frameDeficit, - outputWrittenEndFrame, - replayEstablished ? 1 : 0); + committedMargin); + if (frameShort) { + ASFW_LOG_RING_ONLY_RL( + DirectAudio, + "tx-prep-frame", + 1000u, + ::ASFW::Logging::LogLevel::Warning, + "[TxPrepFrame] target=%llu before=%llu after=%llu deficit=%llu write=%llu replay=%u", + targetFrameEnd, + exposedFrameEndBefore, + exposedFrameEndAfter, + frameDeficit, + outputWrittenEndFrame, + replayEstablished ? 1u : 0u); + } } } + bool scheduleAudioFollowUp = false; if (directControl) { const uint64_t now = mach_absolute_time(); const uint64_t requestedAt = - txControl->refillRequestHostTicks.load( - std::memory_order_relaxed); + hardwareWakePending + ? txControl->refillRequestHostTicks.load( + std::memory_order_relaxed) + : now; const uint64_t latency = now >= requestedAt ? now - requestedAt : 0; const uint64_t latencyNanos = ASFW::Timing::hostTicksToNanos(latency); - directControl->txLastPreparationLatencyTicks.store( - latency, std::memory_order_relaxed); - directControl->txPreparationLatencySamples.fetch_add( - 1, std::memory_order_relaxed); - if (latencyNanos <= 750000) { - directControl->txPreparationAtMost750Us.fetch_add( + if (hardwareWakePending) { + directControl->txLastPreparationLatencyTicks.store( + latency, std::memory_order_relaxed); + directControl->txPreparationLatencySamples.fetch_add( 1, std::memory_order_relaxed); - } - if (latencyNanos >= 1500000) { - directControl->txPreparationAtLeast1500Us.fetch_add( - 1, std::memory_order_relaxed); - } - uint64_t previousMax = - directControl->txMaxPreparationLatencyTicks.load( - std::memory_order_relaxed); - while (latency > previousMax && - !directControl->txMaxPreparationLatencyTicks - .compare_exchange_weak( - previousMax, - latency, - std::memory_order_relaxed, - std::memory_order_relaxed)) { + if (latencyNanos <= 750000) { + directControl->txPreparationAtMost750Us.fetch_add( + 1, std::memory_order_relaxed); + } + if (latencyNanos >= 1500000) { + directControl->txPreparationAtLeast1500Us.fetch_add( + 1, std::memory_order_relaxed); + } + uint64_t previousMax = + directControl->txMaxPreparationLatencyTicks.load( + std::memory_order_relaxed); + while (latency > previousMax && + !directControl->txMaxPreparationLatencyTicks + .compare_exchange_weak( + previousMax, + latency, + std::memory_order_relaxed, + std::memory_order_relaxed)) { + } } const uint64_t distance = packetLimitTarget > exposeCursor @@ -766,7 +806,7 @@ void IMPL(ASFWAudioDriver, TxPreparationReady) boundedMargin < committedMarginFloorBefore; const bool slackBudgetExceeded = latencyNanos >= 1500000; if (newCommittedMarginLow || slackBudgetExceeded || - (wakeSamples % 1024) == 0) { + (wakeSamples != 0 && (wakeSamples % 1024) == 0)) { ASFW_LOG( DirectAudio, "[TxPrep] margin=%u min=%u lead=%u distMin=%u lastLatUs=%llu " @@ -783,18 +823,13 @@ void IMPL(ASFWAudioDriver, TxPreparationReady) directControl->txPreparationAtLeast1500Us.load( std::memory_order_relaxed), wakeSamples, - ASFW::IsochTransport::AudioTimingGeometry:: - kTxExposureLeadFrames, + dataHorizonFrames, ASFW::IsochTransport::AudioTimingGeometry:: kTxCoverageLeadPackets, boundedMargin <= kCommittedMarginDangerPackets ? " DANGER" : ""); } - directControl->txPreparationRequests.requestedGeneration.store( - requested, std::memory_order_relaxed); - directControl->txPreparationRequests.requestHostTicks.store( - requestedAt, std::memory_order_relaxed); directControl->counters.txPreparationWakeRequests.store( txControl->refillRequestCount.load( std::memory_order_relaxed), @@ -807,9 +842,29 @@ void IMPL(ASFWAudioDriver, TxPreparationReady) std::memory_order_relaxed); directControl->counters.txPreparationDrainPasses.fetch_add( 1, std::memory_order_relaxed); - directControl->txPreparationRequests.MarkHandled( - requested, now); + const bool audioTargetSatisfied = + targetFrameEnd == 0 || exposedFrameEndAfter >= targetFrameEnd; + if (audioTargetSatisfied) { + directControl->txPreparationRequests.MarkHandled( + audioRequested, now); + } + directControl->txPreparationRequests.FinishWake(); + // A CoreAudio callback can publish while this action is preparing + // slots. It saw wakeScheduled=true and deliberately did not enqueue a + // second action; hand it one now after draining the latest target. + scheduleAudioFollowUp = audioTargetSatisfied && + directControl->txPreparationRequests.NeedsHandling() && + directControl->txPreparationRequests.TryScheduleWake(); } txControl->MarkRefillHandled(requested); + + if (scheduleAudioFollowUp && ivars->device.audioNub) { + const kern_return_t requestKr = + ivars->device.audioNub->RequestTxPreparation( + directControl->txPreparationRequests.RequestedGeneration()); + if (requestKr != kIOReturnSuccess) { + directControl->txPreparationRequests.FinishWake(); + } + } } diff --git a/ASFWDriver/Audio/DriverKit/ASFWAudioNub.cpp b/ASFWDriver/Audio/DriverKit/ASFWAudioNub.cpp index d3b1554e..adb3df49 100644 --- a/ASFWDriver/Audio/DriverKit/ASFWAudioNub.cpp +++ b/ASFWDriver/Audio/DriverKit/ASFWAudioNub.cpp @@ -380,6 +380,15 @@ kern_return_t IMPL(ASFWAudioNub, RegisterTxPreparationAction) return kIOReturnSuccess; } +kern_return_t IMPL(ASFWAudioNub, RequestTxPreparation) +{ + if (!ivars || !ivars->txPreparationAction) { + return kIOReturnNotReady; + } + TxPreparationReady(ivars->txPreparationAction, generation); + return kIOReturnSuccess; +} + void IMPL(ASFWAudioNub, TxPreparationReady) { (void)action; diff --git a/ASFWDriver/Audio/DriverKit/ASFWAudioNub.iig b/ASFWDriver/Audio/DriverKit/ASFWAudioNub.iig index 8f6832e6..fb2fa737 100644 --- a/ASFWDriver/Audio/DriverKit/ASFWAudioNub.iig +++ b/ASFWDriver/Audio/DriverKit/ASFWAudioNub.iig @@ -88,6 +88,10 @@ public: uint64_t generation); virtual kern_return_t RegisterTxPreparationAction( OSAction* action TYPE(TxPreparationReady)); + // Audio-owned write-frontier request. It reuses the registered action so + // packet preparation is dispatched on ASFWAudioDriver's TxPreparation + // queue rather than on the CoreAudio real-time callback. + virtual kern_return_t RequestTxPreparation(uint64_t generation); virtual void ZtsAnchorReady(OSAction* action TARGET, uint64_t generation); diff --git a/ASFWDriver/Audio/DriverKit/Runtime/AudioTransportControlBlock.hpp b/ASFWDriver/Audio/DriverKit/Runtime/AudioTransportControlBlock.hpp index f0bff42f..0e80ed6f 100644 --- a/ASFWDriver/Audio/DriverKit/Runtime/AudioTransportControlBlock.hpp +++ b/ASFWDriver/Audio/DriverKit/Runtime/AudioTransportControlBlock.hpp @@ -170,9 +170,19 @@ struct TxPreparationRequestState final { std::atomic handledGeneration{0}; std::atomic requestHostTicks{0}; std::atomic handledHostTicks{0}; - - [[nodiscard]] uint64_t PublishRequest(uint64_t hostTicks) noexcept { + // Audio-owned target: the packetizer must expose content through this + // absolute host frame before the request is considered drained. Transport + // only carries packet cursors and never interprets or resets this value. + std::atomic requestedTargetFrameEnd{0}; + // CoreAudio can publish every IO period while TxPreparation runs on a + // different queue. This latch makes action delivery edge-triggered and + // coalesces those writes into one follow-up action. + std::atomic wakeScheduled{false}; + + [[nodiscard]] uint64_t PublishRequest(uint64_t hostTicks, + uint64_t targetFrameEnd = 0) noexcept { requestHostTicks.store(hostTicks, std::memory_order_relaxed); + requestedTargetFrameEnd.store(targetFrameEnd, std::memory_order_relaxed); return requestedGeneration.fetch_add(1, std::memory_order_release) + 1; } @@ -185,9 +195,22 @@ struct TxPreparationRequestState final { requestedGeneration.load(std::memory_order_acquire); } + [[nodiscard]] bool TryScheduleWake() noexcept { + return !wakeScheduled.exchange(true, std::memory_order_acq_rel); + } + + void FinishWake() noexcept { + wakeScheduled.store(false, std::memory_order_release); + } + void MarkHandled(uint64_t generation, uint64_t hostTicks) noexcept { + uint64_t handled = handledGeneration.load(std::memory_order_relaxed); + while (handled < generation && + !handledGeneration.compare_exchange_weak( + handled, generation, std::memory_order_release, + std::memory_order_relaxed)) { + } handledHostTicks.store(hostTicks, std::memory_order_relaxed); - handledGeneration.store(generation, std::memory_order_release); } void Reset() noexcept { @@ -195,6 +218,8 @@ struct TxPreparationRequestState final { handledGeneration.store(0, std::memory_order_relaxed); requestHostTicks.store(0, std::memory_order_relaxed); handledHostTicks.store(0, std::memory_order_relaxed); + requestedTargetFrameEnd.store(0, std::memory_order_relaxed); + wakeScheduled.store(false, std::memory_order_relaxed); } }; diff --git a/ASFWDriver/Audio/DriverKit/Runtime/PayloadWriterTelemetry.hpp b/ASFWDriver/Audio/DriverKit/Runtime/PayloadWriterTelemetry.hpp index 1bd2aa1e..3f367e18 100644 --- a/ASFWDriver/Audio/DriverKit/Runtime/PayloadWriterTelemetry.hpp +++ b/ASFWDriver/Audio/DriverKit/Runtime/PayloadWriterTelemetry.hpp @@ -39,6 +39,14 @@ struct PayloadWriterTelemetryRecord final { bool packetizerFrameCursorAligned{false}; bool packetizerHasLastDataPacket{false}; + // Audio-frontier preparation state. This identifies whether a deficit + // occurred before the 400-cycle target was requested, while it was queued, + // or after a supposedly drained request. + uint64_t txPreparationTargetFrameEnd{0}; + uint64_t txPreparationRequestedGeneration{0}; + uint64_t txPreparationHandledGeneration{0}; + bool txPreparationWakeScheduled{false}; + // Additional cursors and addresses for diagnosing buffer mapping / pointer alignment uint64_t playbackRingReadFrame{0}; uint64_t playbackRingWriteFrame{0}; diff --git a/ASFWDriver/Audio/Engine/Direct/Rx/DirectAudioReceiveConsumer.cpp b/ASFWDriver/Audio/Engine/Direct/Rx/DirectAudioReceiveConsumer.cpp index 7cb4d8cf..9299d996 100644 --- a/ASFWDriver/Audio/Engine/Direct/Rx/DirectAudioReceiveConsumer.cpp +++ b/ASFWDriver/Audio/Engine/Direct/Rx/DirectAudioReceiveConsumer.cpp @@ -447,15 +447,22 @@ void DirectAudioReceiveConsumer::DrainPayloadTelemetry() { } payloadWriterTelemetryAggregator_.BeginDrain(); + ::ASFW::Audio::Runtime::PayloadWriterTelemetryRecord firstRecord{}; ::ASFW::Audio::Runtime::PayloadWriterTelemetryRecord firstDeficitRecord{}; ::ASFW::Audio::Runtime::PayloadWriterTelemetryRecord lastRecord{}; + bool haveFirstRecord = false; bool haveFirstDeficitRecord = false; bool haveLastRecord = false; const uint64_t dropped = control->payloadWriterTelemetry.Drain( - [this, &firstDeficitRecord, &haveFirstDeficitRecord, + [this, &firstRecord, &haveFirstRecord, + &firstDeficitRecord, &haveFirstDeficitRecord, &lastRecord, &haveLastRecord]( const ::ASFW::Audio::Runtime::PayloadWriterTelemetryRecord& record) { payloadWriterTelemetryAggregator_.Observe(record); + if (!haveFirstRecord) { + firstRecord = record; + haveFirstRecord = true; + } if (!haveFirstDeficitRecord && record.exposureDeficitFrames != 0) { firstDeficitRecord = record; haveFirstDeficitRecord = true; @@ -465,58 +472,61 @@ void DirectAudioReceiveConsumer::DrainPayloadTelemetry() { }); const auto& summary = payloadWriterTelemetryAggregator_.Summary(); if (haveLastRecord && summary.HasAnomaly()) { - ASFW_LOG(DirectAudio, - "[PayloadWriter] anomaly firstSample=%llu firstRange=[%llu,%llu) " - "firstExposedEnd=%llu firstDeficit=%llu firstCompletion=%llu " - "firstPacketizer={next=%llu aligned=%u epoch=%llu lastPacket=%llu lastRange=[%llu,%llu) valid=%u} " - "lastSample=%llu completion=%llu deficitMax=%llu " - "visitedDelta=%llu writtenDelta=%llu withoutPktDelta=%llu outsidePktDelta=%llu " - "racedDelta=%llu transmittedDelta=%llu underExpCallsDelta=%llu underExpFramesDelta=%llu " - "lastPacketizer={next=%llu aligned=%u epoch=%llu lastPacket=%llu lastRange=[%llu,%llu) valid=%u} " - "prepared={all=%llu data=%llu noData=%llu acquireFail=%llu} playbackRange=[%llu,%llu)", - haveFirstDeficitRecord ? firstDeficitRecord.sampleTime : 0, - haveFirstDeficitRecord ? firstDeficitRecord.sampleTime : 0, - haveFirstDeficitRecord ? firstDeficitRecord.writeEndFrame : 0, - haveFirstDeficitRecord ? firstDeficitRecord.exposedFrameEnd : 0, - haveFirstDeficitRecord ? firstDeficitRecord.exposureDeficitFrames : 0, - haveFirstDeficitRecord ? firstDeficitRecord.completionCursor : 0, - haveFirstDeficitRecord - ? firstDeficitRecord.packetizerNextAudioFrame - : 0, - haveFirstDeficitRecord && - firstDeficitRecord.packetizerFrameCursorAligned - ? 1u - : 0u, - haveFirstDeficitRecord ? firstDeficitRecord.packetizerCursorEpoch : 0, - haveFirstDeficitRecord - ? firstDeficitRecord.packetizerLastDataPacketIndex - : 0, - haveFirstDeficitRecord - ? firstDeficitRecord.packetizerLastDataFirstAudioFrame - : 0, - haveFirstDeficitRecord - ? firstDeficitRecord.packetizerLastDataEndAudioFrame - : 0, - haveFirstDeficitRecord && - firstDeficitRecord.packetizerHasLastDataPacket - ? 1u - : 0u, - lastRecord.sampleTime, lastRecord.completionCursor, - summary.maxExposureDeficitFrames, summary.visitedDelta, - summary.writtenDelta, summary.withoutPacketDelta, - summary.outsidePacketDelta, summary.racedReuseDelta, - summary.wroteIntoTransmittedDelta, summary.underExposureCallsDelta, - summary.underExposureFramesDelta, - lastRecord.packetizerNextAudioFrame, - lastRecord.packetizerFrameCursorAligned ? 1u : 0u, - lastRecord.packetizerCursorEpoch, - lastRecord.packetizerLastDataPacketIndex, - lastRecord.packetizerLastDataFirstAudioFrame, - lastRecord.packetizerLastDataEndAudioFrame, - lastRecord.packetizerHasLastDataPacket ? 1u : 0u, - lastRecord.packetsPrepared, lastRecord.dataPacketsPrepared, - lastRecord.noDataPacketsPrepared, lastRecord.slotAcquireFailures, - lastRecord.playbackRingReadFrame, lastRecord.playbackRingWriteFrame); + ASFW_LOG_RING_ONLY( + DirectAudio, + ::ASFW::Logging::LogLevel::Warning, + "[PayloadWriter] delta v=%llu w=%llu noPkt=%llu outside=%llu race=%llu tx=%llu under=%llu/%llu maxDef=%llu", + summary.visitedDelta, + summary.writtenDelta, + summary.withoutPacketDelta, + summary.outsidePacketDelta, + summary.racedReuseDelta, + summary.wroteIntoTransmittedDelta, + summary.underExposureCallsDelta, + summary.underExposureFramesDelta, + summary.maxExposureDeficitFrames); + ASFW_LOG_RING_ONLY( + DirectAudio, + ::ASFW::Logging::LogLevel::Warning, + "[PayloadWriter] last sample=%llu comp=%llu pkt=%llu aligned=%u epoch=%llu prepared=%llu/%llu/%llu acquire=%llu ring=%llu/%llu", + lastRecord.sampleTime, + lastRecord.completionCursor, + lastRecord.packetizerNextAudioFrame, + lastRecord.packetizerFrameCursorAligned ? 1u : 0u, + lastRecord.packetizerCursorEpoch, + lastRecord.dataPacketsPrepared, + lastRecord.noDataPacketsPrepared, + lastRecord.packetsPrepared, + lastRecord.slotAcquireFailures, + lastRecord.playbackRingReadFrame, + lastRecord.playbackRingWriteFrame); + if (haveFirstDeficitRecord) { + ASFW_LOG_RING_ONLY( + DirectAudio, + ::ASFW::Logging::LogLevel::Warning, + "[PayloadWriter] deficit sample=%llu write=%llu exposed=%llu d=%llu comp=%llu target=%llu gen=%llu/%llu wake=%u", + firstDeficitRecord.sampleTime, + firstDeficitRecord.writeEndFrame, + firstDeficitRecord.exposedFrameEnd, + firstDeficitRecord.exposureDeficitFrames, + firstDeficitRecord.completionCursor, + firstDeficitRecord.txPreparationTargetFrameEnd, + firstDeficitRecord.txPreparationRequestedGeneration, + firstDeficitRecord.txPreparationHandledGeneration, + firstDeficitRecord.txPreparationWakeScheduled ? 1u : 0u); + } else if (haveFirstRecord) { + ASFW_LOG_RING_ONLY( + DirectAudio, + ::ASFW::Logging::LogLevel::Warning, + "[PayloadWriter] first sample=%llu write=%llu exposed=%llu comp=%llu pkt=%llu aligned=%u epoch=%llu", + firstRecord.sampleTime, + firstRecord.writeEndFrame, + firstRecord.exposedFrameEnd, + firstRecord.completionCursor, + firstRecord.packetizerNextAudioFrame, + firstRecord.packetizerFrameCursorAligned ? 1u : 0u, + firstRecord.packetizerCursorEpoch); + } } if (dropped != 0) { ASFW_LOG(DirectAudio, "[PayloadWriter] drain overflow: dropped=%llu (capacity=%u)", diff --git a/ASFWDriver/Audio/Wire/AMDTP/RxSequenceReplay.hpp b/ASFWDriver/Audio/Wire/AMDTP/RxSequenceReplay.hpp index 35ebe2fe..9c0b45c4 100644 --- a/ASFWDriver/Audio/Wire/AMDTP/RxSequenceReplay.hpp +++ b/ASFWDriver/Audio/Wire/AMDTP/RxSequenceReplay.hpp @@ -23,6 +23,47 @@ inline constexpr uint8_t kValidSyt = 1u << 1; inline constexpr uint8_t kDiscontinuity = 1u << 2; } // namespace RxSequenceFlags +// A false replay read is recoverable, but its cause is not interchangeable: +// future consumption means TX is ahead of RX; a stale cursor means history was +// overwritten; an epoch change is an RX discontinuity. Preserve that boundary +// at the reader so the TX recovery path can report what it actually saw. +enum class RxSequenceReplayReadFailure : uint8_t { + kNone, + kReaderInactive, + kEpochChanged, + kAheadOfProducer, + kHistoryOverwritten, + kSlotSequenceMismatch, + kSlotEpochMismatch, + kSlotChanged, +}; + +[[nodiscard]] constexpr const char* RxSequenceReplayReadFailureName( + RxSequenceReplayReadFailure failure) noexcept { + switch (failure) { + case RxSequenceReplayReadFailure::kNone: return "none"; + case RxSequenceReplayReadFailure::kReaderInactive: return "inactive"; + case RxSequenceReplayReadFailure::kEpochChanged: return "epoch"; + case RxSequenceReplayReadFailure::kAheadOfProducer: return "ahead"; + case RxSequenceReplayReadFailure::kHistoryOverwritten: return "overwritten"; + case RxSequenceReplayReadFailure::kSlotSequenceMismatch: return "slot-seq"; + case RxSequenceReplayReadFailure::kSlotEpochMismatch: return "slot-epoch"; + case RxSequenceReplayReadFailure::kSlotChanged: return "slot-changed"; + } + return "unknown"; +} + +struct RxSequenceReplayReadDiagnostic final { + RxSequenceReplayReadFailure failure{RxSequenceReplayReadFailure::kNone}; + uint64_t readerCursor{0}; + uint64_t producerCursor{0}; + uint64_t slotSequence{0}; + uint32_t readerEpoch{0}; + uint32_t replayEpoch{0}; + uint32_t slotEpoch{0}; + bool replayEstablished{false}; +}; + [[nodiscard]] inline uint32_t ComputeReplaySytOffset( uint16_t syt, uint32_t sourceCycleTimer, @@ -157,15 +198,32 @@ class RxSequenceReplayState final { [[nodiscard]] bool Read(uint64_t cursor, uint32_t expectedEpoch, - RxSequenceEntry& out) const noexcept { + RxSequenceEntry& out, + RxSequenceReplayReadDiagnostic* diagnostic = nullptr) const noexcept { const Slot& slot = slots_[cursor % kCapacity]; const uint64_t expectedSequence = cursor + 1; - if (slot.sequence.load(std::memory_order_acquire) != - expectedSequence) { + const uint64_t firstSequence = + slot.sequence.load(std::memory_order_acquire); + if (diagnostic) { + diagnostic->slotSequence = firstSequence; + } + if (firstSequence != expectedSequence) { + if (diagnostic) { + diagnostic->failure = + RxSequenceReplayReadFailure::kSlotSequenceMismatch; + } return false; } - if (slot.epoch.load(std::memory_order_relaxed) != - expectedEpoch) { + const uint32_t firstEpoch = + slot.epoch.load(std::memory_order_relaxed); + if (diagnostic) { + diagnostic->slotEpoch = firstEpoch; + } + if (firstEpoch != expectedEpoch) { + if (diagnostic) { + diagnostic->failure = + RxSequenceReplayReadFailure::kSlotEpochMismatch; + } return false; } @@ -181,10 +239,22 @@ class RxSequenceReplayState final { entry.dbc = slot.dbc.load(std::memory_order_relaxed); entry.flags = slot.flags.load(std::memory_order_relaxed); std::atomic_thread_fence(std::memory_order_acquire); - if (slot.sequence.load(std::memory_order_relaxed) != - expectedSequence || - slot.epoch.load(std::memory_order_relaxed) != - expectedEpoch) { + const uint64_t finalSequence = + slot.sequence.load(std::memory_order_relaxed); + const uint32_t finalEpoch = + slot.epoch.load(std::memory_order_relaxed); + if (diagnostic) { + diagnostic->slotSequence = finalSequence; + diagnostic->slotEpoch = finalEpoch; + } + if (finalSequence != expectedSequence || + finalEpoch != expectedEpoch) { + if (diagnostic) { + diagnostic->failure = + finalEpoch != expectedEpoch + ? RxSequenceReplayReadFailure::kSlotEpochMismatch + : RxSequenceReplayReadFailure::kSlotChanged; + } return false; } @@ -224,9 +294,36 @@ class RxSequenceReplayReader final { [[nodiscard]] bool TryRead( const RxSequenceReplayState& replay, - RxSequenceEntry& out) noexcept { - if (!active_ || replay.Epoch() != epoch_ || - !replay.Read(nextCursor_, epoch_, out)) { + RxSequenceEntry& out, + RxSequenceReplayReadDiagnostic* diagnostic = nullptr) noexcept { + RxSequenceReplayReadDiagnostic ignoredDiagnostic{}; + RxSequenceReplayReadDiagnostic* const observed = + diagnostic ? diagnostic : &ignoredDiagnostic; + *observed = { + .readerCursor = nextCursor_, + .producerCursor = replay.ProducerCursor(), + .readerEpoch = epoch_, + .replayEpoch = replay.Epoch(), + .replayEstablished = replay.IsEstablished(), + }; + if (!active_) { + observed->failure = RxSequenceReplayReadFailure::kReaderInactive; + return false; + } + if (observed->replayEpoch != epoch_) { + observed->failure = RxSequenceReplayReadFailure::kEpochChanged; + return false; + } + if (nextCursor_ >= observed->producerCursor) { + observed->failure = RxSequenceReplayReadFailure::kAheadOfProducer; + return false; + } + if (observed->producerCursor - nextCursor_ > + RxSequenceReplayState::kCapacity) { + observed->failure = RxSequenceReplayReadFailure::kHistoryOverwritten; + return false; + } + if (!replay.Read(nextCursor_, epoch_, out, observed)) { return false; } ++nextCursor_; diff --git a/ASFWDriver/Shared/Isoch/AudioTimingGeometry.hpp b/ASFWDriver/Shared/Isoch/AudioTimingGeometry.hpp index dd4ccc96..e6a90149 100644 --- a/ASFWDriver/Shared/Isoch/AudioTimingGeometry.hpp +++ b/ASFWDriver/Shared/Isoch/AudioTimingGeometry.hpp @@ -92,12 +92,22 @@ struct AudioTimingGeometry final { // RequiredInputSafetyFrames cushion: RX had it, TX did not -- which is why // TX shipped silence when the writer ran beyond ExposedFrameEnd() // (Defect B = under-exposure, W > E). Conservative form = one full - // client-IO budget + scheduling jitter; tighten toward the actual IO size - // once measured. Frames. + // AppleFWAudio's AM824NuDCLWrite keeps its CIP insertion target roughly + // 400 FireWire cycles ahead of the client write frontier at 48 kHz. This + // is content lead, not the much smaller OHCI descriptor/refill lead. Keep + // the invariant in packet time so it remains a 50 ms horizon at every 1x + // sample rate (the runtime converts it to frames for the active stream). + // See AVC_RECOVERY_AND_SYNC_ALGO_AND_BUGS.md, "Apple reference target". + static constexpr uint32_t kTxDataHorizonPackets = 400; static constexpr uint32_t kTxExposureLeadFrames = - kHalIoPeriodFrames + kSchedulingJitterFrames; // 512 + 64 = 576 + (kTxDataHorizonPackets * kSampleRateHz) / 8000; // 2,400 @ 48 kHz + + [[nodiscard]] static constexpr uint32_t TxDataHorizonFrames( + uint32_t sampleRateHz) noexcept { + return (kTxDataHorizonPackets * sampleRateHz + 7999) / 8000; + } // Packet lead deep enough to expose that many frames at the worst-case - // (44.1k) average cadence: ceil(576 / 5.5125) = 105 packets, rounded up + // (44.1k) average cadence: ceil(2400 / 5.5125) = 436 packets, rounded up // to a whole interrupt group (108) so every budget derived from it keeps // the group- and cadence-block-aligned ring-wrap asserts below. static constexpr uint32_t kTxExposureLeadPacketsRaw = @@ -130,23 +140,33 @@ struct AudioTimingGeometry final { kTxHardwareRingPackets + kTxPreparationSlackPackets; // Covers a full client write window plus the output exposure cushion when // the producer target is expressed as WriteEnd + kTxExposureLeadFrames. - // 2 * 108 packets = 1190 worst-case (44.1k) frames, enough for - // 512 + 576 = 1088 frames while preserving cadence/group-friendly packet - // counts at every supported rate. + // The producer needs to preserve a whole maximum CoreAudio write window + // in addition to the packet-time data horizon. Round the result to an + // interrupt group: ceil((512 + 2400) / 5.5125) = 529 -> 534 packets. + static constexpr uint32_t kTxFrameExposureWindowPacketsRaw = + ((kHalIoPeriodFrames + kTxExposureLeadFrames) * + kMinAvgCadencePackets + + kMinAvgCadenceFrames - 1) / + kMinAvgCadenceFrames; static constexpr uint32_t kTxFrameExposureWindowPackets = - 2 * kTxExposureLeadPackets; + ((kTxFrameExposureWindowPacketsRaw + kTxPacketsPerGroup - 1) / + kTxPacketsPerGroup) * + kTxPacketsPerGroup; static constexpr uint32_t kTxPreparationLeadPackets = kTxCoverageLeadPackets + kTxFrameExposureWindowPackets; - // Shared backing leaves one hardware-ring depth before slot reuse. + // A 912-packet (114 ms) backing ring lets the 400-cycle content target + // occupy less than half the ring while retaining one OHCI ring depth + // before reuse. The 48-packet hardware descriptor ring remains a separate + // low-latency transport concern. static constexpr uint32_t kTxSharedSlotPackets = - kTxPreparationLeadPackets + kTxHardwareRingPackets; + 912; // Largest single coalesced deltaConsumed a refill can absorb without holing. static constexpr uint32_t kTxMaxCoveredDeltaConsumedPackets = kTxPreparationLeadPackets - kTxHardwareRingPackets; // Backing packet-ring / timeline slot array length // (AmdtpPacketTimeline, DiceTxStreamEngine::timelineSlots_). Packets. - static constexpr uint32_t kTimelineSlots = 512; + static constexpr uint32_t kTimelineSlots = 1024; }; static_assert(AudioTimingGeometry::kRxDescriptorPackets % @@ -218,12 +238,12 @@ static_assert(AudioTimingGeometry::kTxExposureLeadFrames >= AudioTimingGeometry::kSchedulingJitterFrames, "TX exposure lead must cover one full IO window plus scheduling " "jitter (the cushion whose absence was Defect B)"); -static_assert(AudioTimingGeometry::kTxExposureLeadFrames < - AudioTimingGeometry::kFrameRingFrames, - "TX exposure lead cannot exceed the host frame ring"); static_assert(AudioTimingGeometry::kTxExposureLeadPackets <= AudioTimingGeometry::kTxSharedSlotPackets, "TX packet lead must be able to hold the required exposure frames"); +static_assert(AudioTimingGeometry::kTxSharedSlotPackets >= + 2 * AudioTimingGeometry::kTxExposureLeadPackets, + "TX backing ring must keep the 400-cycle content target below half ring"); static_assert(AudioTimingGeometry::kTxFrameExposureWindowPackets * AudioTimingGeometry::kMinAvgCadenceFrames >= (AudioTimingGeometry::kHalIoPeriodFrames + diff --git a/AVC_RECOVERY_AND_SYNC_ALGO_AND_BUGS.md b/AVC_RECOVERY_AND_SYNC_ALGO_AND_BUGS.md index cac9cc48..ae92407c 100644 --- a/AVC_RECOVERY_AND_SYNC_ALGO_AND_BUGS.md +++ b/AVC_RECOVERY_AND_SYNC_ALGO_AND_BUGS.md @@ -239,11 +239,62 @@ Linux and FFADO, without copying their mechanisms. | ID | Status | Finding | Required action | | --- | --- | --- | --- | | `AVC-RECOVERY-001` | **Confirmed** | Stale `committedEnd` makes IT prime reject timing-loss recovery. | Implement the producer-owned recovery epoch above, then test a forced RX-cadence loss. | +| `AUDIO-RECOVERY-ROUTING-001` | **Confirmed critical defect** | FW-64 / merge `e9ea6ce` added a second registration to `IsochService`'s single `timingLossCallback_`. `AudioCoordinator` constructs `dice_` before `avc_`; AV/C therefore overwrites DICE's callback. A DICE/Focusrite replay reset is routed to AV/C, whose `activeGuid_` does not match and which drops the event. | Replace the single backend-owned callback with one AudioCoordinator-owned router which selects the backend by GUID. Carry the fault reason and session epoch; do not register directly from either backend. | +| `AVC-RECOVERY-002` | **Confirmed critical defect** | FW-64 treats every *established* replay reset as an eventual destructive `RecoverStreaming()`. A reset can mean a malformed RX packet, one cycle gap, rejected cadence, or rejected clock anchor—not necessarily a device outage. The coordinator stops/restarts transport while CoreAudio remains running with its old TX producer state. | Disable AV/C automatic transport restart until recovery atomically quiesces/resets/re-primes the audio producer and transport under one recovery epoch. Report the fault and request an audio-side xrun/controlled restart instead. | | `AVC-PROPERTY-001` | **Fixed; requires hardware confirmation** | `AudioNubPublisher` sets stream mode on the nub, but `ASFWAudioDevice::PopulateNubProperties()` did not publish `ASFWStreamMode`; the driver parser consequently defaulted to non-blocking. The model default is also non-blocking ([`ASFWAudioDevice.hpp:52-60`](ASFWDriver/Audio/Model/ASFWAudioDevice.hpp#L52-L60)); parsing only changes it if the property exists ([`AudioDriverConfig.cpp:67-84`](ASFWDriver/Audio/DriverKit/Config/AudioDriverConfig.cpp#L67-L84)). The matching dictionary now publishes the selected mode. | On the next Duet/BeBoB start, confirm the AudioDriver reports `Stream mode from nub: blocking`; the property path is shared by all published audio nubs. | | `AVC-TX-EXPOSURE-001` | **Confirmed audio-path defect; root scheduling cause open** | The payload writer skips CoreAudio frames which arrive beyond `Timeline().ExposedFrameEnd()` (`framesWithoutPacket`); `TxAlign` later repositions the producer cursor to close the deficit. This can produce audible corruption while DMA/CMP and the TX descriptor lead remain healthy, then self-heal without a stream restart ([`AmdtpPayloadWriter.cpp:90-126`](ASFWDriver/Audio/Wire/AMDTP/AmdtpPayloadWriter.cpp#L90-L126), [`ASFWAudioDriverZts.cpp:360-389`](ASFWDriver/Audio/DriverKit/ASFWAudioDriverZts.cpp#L360-L389)). | Retain pending host frames until slots exist and maintain a data-bearing packet horizon beyond the maximum IO callback plus measured scheduling jitter; never discard a running stream's host frames because their packet is not yet exposed. | | `AVC-SYNC-001` | **Not proven** | Treating raw RX/TX SYT inequality as a fault would be wrong for the captured duplex stream. | Add the normalized, per-direction trace below before adjusting delays or cadence. | | `AVC-HEALTH-001` | **Instrumented; root cause open** | The original RX timing loss occurred without a bus reset in the observed driver ring; successful restart was then masked by `AVC-RECOVERY-001`. | Reproduce with the first-fault record below; it distinguishes a local RX validation failure from a real no-RX/CMP outage. | +### FW-64 callback overwrite and unsafe recovery escalation + +Commit [`e9ea6ce`](https://github.com/mrmidi/ASFireWire/commit/e9ea6ce2fd60fc1e76efaf47fe9fa6feb3ddfaf4) +introduced AV/C timing-loss recovery. Its fault signal is global, but its +handlers are backend-specific: + +```text +AudioCoordinator construction order + dice_ constructor -> IsochService::SetTimingLossCallback(DICE handler) + avc_ constructor -> IsochService::SetTimingLossCallback(AV/C handler) + ^ overwrites the DICE handler + +RX replay reset -> IsochService::OnReceiveTimingLossDetected(active GUID) + -> AV/C handler only + -> DICE GUID is not AV/C active GUID -> event dropped +``` + +This is a direct DICE stream-killer *after* an RX replay reset: the receive +side has discarded its timing history, but the DICE backend never receives the +recovery event. It does not itself create the initial RX reset. The independent +TX replay-horizon failure described below can also produce NODATA/clicks/silence +without any `[RxReplayReset]`; both failures must be fixed. + +The AV/C path is unsafe in the opposite direction. An established replay reset +is raised for packet-processor failure, invalid RX timestamp, receive-cycle +gap, rejected SYT cadence, or rejected clock anchor +([`DirectAudioReceiveConsumer.cpp:178-336`](ASFWDriver/Audio/Engine/Direct/Rx/DirectAudioReceiveConsumer.cpp#L178-L336)). +FW-64 waits 256 ms and, if the same replay has not re-established, calls +`RecoverStreaming()` ([`AVCAudioBackend.cpp:213-288`](ASFWDriver/Audio/Protocols/Backends/AVCAudioBackend.cpp#L213-L288)). +That operation stops and starts the duplex transport +([`AudioDuplexCoordinator.cpp:560-635`](ASFWDriver/Audio/Protocols/Backends/AudioDuplexCoordinator.cpp#L560-L635)), +but does not run the CoreAudio `StopIO`/`StartIO` producer reset and prefill +protocol. It can therefore turn a local timing anomaly into a stale-cursor, +half-running stream. + +The repair boundary is architectural: + +1. `AudioCoordinator` must own the only timing-loss subscription and dispatch + by active GUID/backend; `IsochService` must not be a last-writer-wins + backend callback slot. +2. The event must carry reset reason, RX epoch, and the active duplex session + epoch so a delayed callback cannot act on a newer session. +3. Until the audio-side and transport-side recovery are one atomic epoch, + AV/C timing loss is telemetry plus controlled xrun/explicit restart—not a + background `RecoverStreaming()` call. +4. Callback replace/clear/invoke must be serialized with service teardown; + the current unsynchronized `std::function` slot risks a callback into a + destroying backend. + ## Implemented first-fault MCP instrumentation The driver log ring is the source of truth for this fault. It retained all @@ -500,6 +551,47 @@ This separates a genuine RX interruption from the current, more likely design error: future TX packet preparation consuming a finite history of past RX SYT observations. +### The actual TX failure: content reaches an unprepared region + +The primary fault model is a cursor/geometry violation, not necessarily an +OHCI DMA read of random memory. The transport ring is deliberately prefilled +with legal NODATA packets, so OHCI can remain healthy while it transmits no +audio content. Four independent cursors currently lack one enforced contract: + +```text +W = CoreAudio write end host frames which exist +E = exposed frame end host frames mapped into prepared TX packets +C = completion/committed end packet slots safe for OHCI ownership/reuse +R = RX replay read/producer observed timing history available for TX SYT +``` + +The invalid states are: + +```text +W > E payload writer reaches a host frame with no packet slot +R.read >= R.producer TX requests an RX timing observation that does not exist +``` + +Today both violations are treated as recoverable branches: the payload writer +counts `withoutPacket` and permanently skips the samples; `TryRead()` emits a +legal NODATA packet and re-arms `TxAlign`. When replay becomes readable again, +alignment jumps to a newer frame range. The transport survives, but the +unprepared interval is lost; repetition sounds like clicks/periodic silence and +continuous failure is complete silence. + +The required contract is non-negotiable: + +1. Never consume a CoreAudio frame until an exposed TX packet owns that frame. + Retain it pending, or declare an explicit xrun only after its host-ring + storage is proven overwritten. +2. Never consume RX replay beyond valid observed history. Future TX packet + reservation requires a timing cache/predictor with an explicit validity + horizon, not direct consumption of a finite historical ring. +3. `C` protects DMA ownership only; NODATA prefill is not evidence that audio + content is prepared. +4. On recovery, reset/re-prime W/E/C/R under one epoch before transport runs; + do not use `TxAlign` to silently abandon the old interval. + ## Minimal anomaly-only telemetry for the next hardware run Keep the MCP ring as the primary evidence path; no Console logging is needed diff --git a/tests/audio/AmdtpRateGeometryTests.cpp b/tests/audio/AmdtpRateGeometryTests.cpp index 3661e32d..75941dac 100644 --- a/tests/audio/AmdtpRateGeometryTests.cpp +++ b/tests/audio/AmdtpRateGeometryTests.cpp @@ -53,14 +53,18 @@ TEST(AudioTimingGeometryTests, SaffireGeometryIsUnified) { EXPECT_EQ(Geometry::kRxDescriptorPackets, 504U); // TX budgets are sized for the worst-case (44.1k) average cadence of // 441 frames / 80 packets, exposure lead rounded to a whole interrupt - // group: ceil(576 / 5.5125) = 105 -> 108 packets. - EXPECT_EQ(Geometry::kTxSharedSlotPackets, 408U); + // Apple-comparable 400-cycle content horizon: ceil(2400 / 5.5125) = + // 436 -> 438 packets, plus a full 512-frame write window. + EXPECT_EQ(Geometry::kTxDataHorizonPackets, 400U); + EXPECT_EQ(Geometry::TxDataHorizonFrames(48000), 2400U); + EXPECT_EQ(Geometry::TxDataHorizonFrames(44100), 2205U); + EXPECT_EQ(Geometry::kTxSharedSlotPackets, 912U); EXPECT_EQ(Geometry::kTxHardwareRingPackets, 48U); EXPECT_EQ(Geometry::kTxPreparationSlackPackets, 96U); EXPECT_EQ(Geometry::kTxCoverageLeadPackets, 144U); - EXPECT_EQ(Geometry::kTxExposureLeadPackets, 108U); - EXPECT_EQ(Geometry::kTxFrameExposureWindowPackets, 216U); - EXPECT_EQ(Geometry::kTxPreparationLeadPackets, 360U); + EXPECT_EQ(Geometry::kTxExposureLeadPackets, 438U); + EXPECT_EQ(Geometry::kTxFrameExposureWindowPackets, 534U); + EXPECT_EQ(Geometry::kTxPreparationLeadPackets, 678U); // DMA completion cadence and the ZTS grid are intentionally independent. EXPECT_NE(Geometry::kHalZeroTimestampPeriodFrames, diff --git a/tests/audio/AudioRuntime/AudioTransportControlBlockTests.cpp b/tests/audio/AudioRuntime/AudioTransportControlBlockTests.cpp index 8dd25e5b..10f43bdb 100644 --- a/tests/audio/AudioRuntime/AudioTransportControlBlockTests.cpp +++ b/tests/audio/AudioRuntime/AudioTransportControlBlockTests.cpp @@ -18,18 +18,24 @@ TEST(AudioTransportControlBlockTests, PreparationRequestsAreMonotonicAndCoalesci TxPreparationRequestState requests{}; EXPECT_FALSE(requests.NeedsHandling()); - EXPECT_EQ(requests.PublishRequest(100), 1U); - EXPECT_EQ(requests.PublishRequest(200), 2U); + EXPECT_EQ(requests.PublishRequest(100, 600), 1U); + EXPECT_EQ(requests.PublishRequest(200, 1200), 2U); EXPECT_TRUE(requests.NeedsHandling()); EXPECT_EQ(requests.RequestedGeneration(), 2U); EXPECT_EQ(requests.requestHostTicks.load(std::memory_order_relaxed), 200U); + EXPECT_EQ(requests.requestedTargetFrameEnd.load(std::memory_order_acquire), 1200U); + EXPECT_TRUE(requests.TryScheduleWake()); + EXPECT_FALSE(requests.TryScheduleWake()); requests.MarkHandled(2, 250); EXPECT_FALSE(requests.NeedsHandling()); EXPECT_EQ(requests.handledGeneration.load(std::memory_order_acquire), 2U); EXPECT_EQ(requests.handledHostTicks.load(std::memory_order_relaxed), 250U); + requests.FinishWake(); + EXPECT_TRUE(requests.TryScheduleWake()); + requests.FinishWake(); - EXPECT_EQ(requests.PublishRequest(300), 3U); + EXPECT_EQ(requests.PublishRequest(300, 1800), 3U); EXPECT_TRUE(requests.NeedsHandling()); } diff --git a/tests/audio/RxDrivenTimingTests.cpp b/tests/audio/RxDrivenTimingTests.cpp index cdd1cea1..fc7ca32e 100644 --- a/tests/audio/RxDrivenTimingTests.cpp +++ b/tests/audio/RxDrivenTimingTests.cpp @@ -12,6 +12,8 @@ namespace { using ASFW::Audio::Runtime::RxSequenceEntry; using ASFW::Audio::Runtime::RxSequenceReplayReader; +using ASFW::Audio::Runtime::RxSequenceReplayReadDiagnostic; +using ASFW::Audio::Runtime::RxSequenceReplayReadFailure; using ASFW::Audio::Runtime::RxSequenceReplayState; using ASFW::IsochTransport::AudioTimingGeometry; @@ -175,10 +177,39 @@ TEST(RxDrivenTimingTests, ReplayEpochChangeInvalidatesActiveReader) { replay.Reset(); RxSequenceEntry entry{}; - EXPECT_FALSE(reader.TryRead(replay, entry)); + RxSequenceReplayReadDiagnostic diagnostic{}; + EXPECT_FALSE(reader.TryRead(replay, entry, &diagnostic)); + EXPECT_EQ(diagnostic.failure, RxSequenceReplayReadFailure::kEpochChanged); + EXPECT_EQ(diagnostic.readerEpoch, 1U); + EXPECT_EQ(diagnostic.replayEpoch, 2U); EXPECT_FALSE(replay.IsEstablished()); } +TEST(RxDrivenTimingTests, ReplayReadReportsInactiveAndAheadOfProducer) { + RxSequenceReplayState replay{}; + replay.Reset(); + + RxSequenceReplayReader reader{}; + RxSequenceEntry entry{}; + RxSequenceReplayReadDiagnostic diagnostic{}; + EXPECT_FALSE(reader.TryRead(replay, entry, &diagnostic)); + EXPECT_EQ(diagnostic.failure, RxSequenceReplayReadFailure::kReaderInactive); + EXPECT_FALSE(diagnostic.replayEstablished); + + for (uint64_t cycle = 0; + cycle < RxSequenceReplayState::kReadDelay; + ++cycle) { + replay.Publish(RxSequenceEntry{}); + } + ASSERT_TRUE(replay.MarkEstablished()); + ASSERT_TRUE(reader.Begin(replay)); + while (reader.TryRead(replay, entry)) { + } + EXPECT_FALSE(reader.TryRead(replay, entry, &diagnostic)); + EXPECT_EQ(diagnostic.failure, RxSequenceReplayReadFailure::kAheadOfProducer); + EXPECT_EQ(diagnostic.readerCursor, diagnostic.producerCursor); +} + TEST(RxDrivenTimingTests, ReplayCannotEstablishWithoutHalfRingHistory) { RxSequenceReplayState replay{}; replay.Reset(); @@ -204,10 +235,12 @@ TEST(RxDrivenTimingTests, GeometryUsesSixCycleInterruptsAndCurrentTxDepths) { EXPECT_EQ(AudioTimingGeometry::kTxHardwareRingPackets, 48U); EXPECT_EQ(AudioTimingGeometry::kTxPreparationSlackPackets, 96U); EXPECT_EQ(AudioTimingGeometry::kTxCoverageLeadPackets, 144U); - // Worst-case (44.1k) cadence sizing: exposure lead 108 packets. - EXPECT_EQ(AudioTimingGeometry::kTxFrameExposureWindowPackets, 216U); - EXPECT_EQ(AudioTimingGeometry::kTxPreparationLeadPackets, 360U); - EXPECT_EQ(AudioTimingGeometry::kTxSharedSlotPackets, 408U); + // 400-cycle content horizon at worst-case 44.1k cadence, plus one full + // 512-frame client write window. + EXPECT_EQ(AudioTimingGeometry::kTxExposureLeadPackets, 438U); + EXPECT_EQ(AudioTimingGeometry::kTxFrameExposureWindowPackets, 534U); + EXPECT_EQ(AudioTimingGeometry::kTxPreparationLeadPackets, 678U); + EXPECT_EQ(AudioTimingGeometry::kTxSharedSlotPackets, 912U); } TEST(RxDrivenTimingTests, InputSafetyIsVisibilityMarginNotClientWindow) { diff --git a/tests/audio/TxRefillCoverageTests.cpp b/tests/audio/TxRefillCoverageTests.cpp index 8fa99fa8..db145c1b 100644 --- a/tests/audio/TxRefillCoverageTests.cpp +++ b/tests/audio/TxRefillCoverageTests.cpp @@ -34,10 +34,10 @@ namespace { using ASFW::IsochTransport::AudioTimingGeometry; using ASFW::Isoch::ExpectedTxCommitGeneration; -constexpr uint32_t kNumSlots = AudioTimingGeometry::kTxSharedSlotPackets; // 408 +constexpr uint32_t kNumSlots = AudioTimingGeometry::kTxSharedSlotPackets; // 912 constexpr uint32_t kHwRing = AudioTimingGeometry::kTxHardwareRingPackets; // 48 constexpr uint32_t kCoverageLead = AudioTimingGeometry::kTxCoverageLeadPackets; // 144 -constexpr uint32_t kLead = AudioTimingGeometry::kTxPreparationLeadPackets; // 360 +constexpr uint32_t kLead = AudioTimingGeometry::kTxPreparationLeadPackets; // 678 constexpr uint32_t kGroup = AudioTimingGeometry::kTxPacketsPerGroup; // 6 // The historical pre-fix lead (slack == 2*group) the hardware IT FATAL was @@ -282,7 +282,8 @@ TEST(TxRefillCoverage, CoverageBoundMatchesGeometryConstants) { AudioTimingGeometry::kTxFrameExposureWindowPackets); // Current geometry tolerates sixteen groups without a producer wake. EXPECT_EQ((kCoverageLead - kHwRing) / kGroup, 16u); - EXPECT_EQ(kLead + kHwRing, kNumSlots); + EXPECT_LE(kLead + kHwRing, kNumSlots); + EXPECT_GE(kNumSlots, 2 * AudioTimingGeometry::kTxExposureLeadPackets); } } // namespace diff --git a/tools/buffer_geometry_verify.py b/tools/buffer_geometry_verify.py index 4e0566d2..971e5824 100644 --- a/tools/buffer_geometry_verify.py +++ b/tools/buffer_geometry_verify.py @@ -4,9 +4,9 @@ PURPOSE ------- -Before we enlarge the shared audio ring (to advertise a larger CoreAudio buffer -size than the current 576-frame ceiling) or change the HAL safety offsets, the -proposed numbers must satisfy EVERY compile-time invariant baked into +Before changing the 400-cycle TX content horizon, packet backing, advertised +CoreAudio buffer size, or HAL safety offsets, the proposed numbers must satisfy +EVERY compile-time invariant baked into ASFWDriver/Shared/Isoch/AudioTimingGeometry.hpp as a static_assert. A bump that "looks fine" routinely breaks one of: @@ -25,12 +25,9 @@ It does NOT touch any source. Run it, read the table, THEN edit the header. The exposure cushion is the Defect-B guard: the producer's exposure frontier E -must lead CoreAudio's write frontier W by one full IO window + scheduling -jitter. If CoreAudio is allowed to pick an IO size larger than the budget the -cushion was built on, W jumps past E => silence. So the advertised max IO must -equal the io-budget the cushion is sized for: bumping the buffer ceiling and -sizing the cushion are the SAME change, which is the whole point of verifying -them together. +must lead CoreAudio's write frontier W by the 400-cycle content horizon. The +packet window also has to contain one maximum IO burst; the host frame ring is +not the limit for this packet-domain horizon. Usage: python3 buffer_geometry_verify.py verify # current + 1024 candidate @@ -57,36 +54,36 @@ JITTER_FRAMES = 64 # kSchedulingJitterFrames MAX_FRAMES_PER_INTERRUPT = 40 # kMaximumNominalFramesPerInterrupt MIN_DISPATCH_GROUPS = 16 # coverage target: 16 six-packet groups (~12 ms) +MIN_CADENCE_PACKETS = 80 # 44.1 kHz average cadence numerator +MIN_CADENCE_FRAMES = 441 # 44.1 kHz average cadence denominator +CONTENT_HORIZON_PACKETS = 400 # kTxDataHorizonPackets @dataclass(frozen=True) class Geometry: """One candidate HAL buffer profile + the derived TX packet geometry. - io_budget_frames == kHalIoPeriodFrames is BOTH the advertised CoreAudio max - IO size and the basis of the exposure cushion (see module docstring). + io_budget_frames == kHalIoPeriodFrames is the advertised CoreAudio max IO. + The content horizon is independent and remains packet-time based. """ name: str ring_frames: int # frameRingFrames (shared IOMemoryDescriptor) io_budget_frames: int # kHalIoPeriodFrames == advertised max IO zts_frames: int # kHalZeroTimestampPeriodFrames timeline_slots: int # kTimelineSlots (AmdtpPacketTimeline length) + shared_slots: int = 912 # kTxSharedSlotPackets # --- derived: exposure cushion (frames) --------------------------------- @property def exposure_lead_frames(self) -> int: # kTxExposureLeadFrames - return self.io_budget_frames + JITTER_FRAMES + return CONTENT_HORIZON_PACKETS * 48_000 // 8_000 @property def exposure_lead_pkts(self) -> int: # kTxExposureLeadPackets - # ceil(frames * blockPkts / blockFrames) == ceil(frames / 6), THEN round - # up to a whole timing group so the downstream sharedSlot stays a - # multiple of both 6 (group) and 4 (cadence block). The shipping header - # omits this round-up and is only accidentally aligned at ioBudget=512 - # (576/6 == 96 exactly); any other budget needs it. This is part of the - # proposed header change. - raw = (self.exposure_lead_frames * CADENCE_BLOCK_PKTS + - CADENCE_BLOCK_FRAMES - 1) // CADENCE_BLOCK_FRAMES + # Exact header derivation: worst supported cadence is 44.1 kHz, + # ceil(frames * 80 / 441), then round to a six-packet group. + raw = (self.exposure_lead_frames * MIN_CADENCE_PACKETS + + MIN_CADENCE_FRAMES - 1) // MIN_CADENCE_FRAMES return ((raw + PKTS_PER_GROUP - 1) // PKTS_PER_GROUP) * PKTS_PER_GROUP # --- derived: packet-domain TX ownership -------------------------------- @@ -100,7 +97,9 @@ def coverage_lead_pkts(self) -> int: # kTxCoverageLeadPackets @property def frame_exposure_window_pkts(self) -> int: # kTxFrameExposureWindowPackets - return 2 * self.exposure_lead_pkts + raw = ((self.io_budget_frames + self.exposure_lead_frames) * + MIN_CADENCE_PACKETS + MIN_CADENCE_FRAMES - 1) // MIN_CADENCE_FRAMES + return ((raw + PKTS_PER_GROUP - 1) // PKTS_PER_GROUP) * PKTS_PER_GROUP @property def preparation_lead_pkts(self) -> int: # kTxPreparationLeadPackets @@ -108,7 +107,7 @@ def preparation_lead_pkts(self) -> int: # kTxPreparationLeadPackets @property def shared_slot_pkts(self) -> int: # kTxSharedSlotPackets - return self.preparation_lead_pkts + HW_RING_PKTS + return self.shared_slots @property def max_covered_delta_pkts(self) -> int: # kTxMaxCoveredDeltaConsumedPackets @@ -152,12 +151,12 @@ def max_covered_delta_pkts(self) -> int: # kTxMaxCoveredDeltaConsumedPack ("exposureLead >= ioBudget + jitter", lambda g: g.exposure_lead_frames >= g.io_budget_frames + JITTER_FRAMES, lambda g: f"{g.exposure_lead_frames} >= {g.io_budget_frames + JITTER_FRAMES}"), - ("exposureLead < ring", - lambda g: g.exposure_lead_frames < g.ring_frames, - lambda g: f"{g.exposure_lead_frames} < {g.ring_frames}"), ("exposureLeadPkts <= sharedSlot", lambda g: g.exposure_lead_pkts <= g.shared_slot_pkts, lambda g: f"{g.exposure_lead_pkts} <= {g.shared_slot_pkts}"), + ("sharedSlot keeps content target below half ring", + lambda g: g.shared_slot_pkts >= 2 * g.exposure_lead_pkts, + lambda g: f"{g.shared_slot_pkts} >= {2 * g.exposure_lead_pkts}"), ("frameExposureWindow covers WriteEnd + cushion", lambda g: g.frame_exposure_window_pkts * CADENCE_BLOCK_FRAMES >= (g.io_budget_frames + g.exposure_lead_frames) * CADENCE_BLOCK_PKTS, @@ -176,7 +175,8 @@ def check(g: Geometry) -> List[Tuple[str, bool, str]]: def derived_table(g: Geometry) -> str: return (f" ring={g.ring_frames}fr ioBudget/maxIO={g.io_budget_frames}fr " f"zts={g.zts_frames}fr timeline={g.timeline_slots}slots\n" - f" exposureLead={g.exposure_lead_frames}fr ({g.exposure_lead_pkts}pkt) " + f" contentHorizon={CONTENT_HORIZON_PACKETS}pkt " + f"exposureLead={g.exposure_lead_frames}fr ({g.exposure_lead_pkts}pkt) " f"frameWindow={g.frame_exposure_window_pkts}pkt " f"prepLead={g.preparation_lead_pkts}pkt " f"sharedSlot={g.shared_slot_pkts}pkt " @@ -205,38 +205,35 @@ def report(g: Geometry) -> bool: ring_frames=1536, io_budget_frames=512, zts_frames=1536, - timeline_slots=512, + timeline_slots=1024, + shared_slots=912, ) def solve(max_io: int, zts: int = 1536) -> Geometry: """Smallest cadence-aligned profile that advertises `max_io` frames. - Strategy: io_budget = max_io (cushion basis). Round the derived shared-slot - packet count up to a multiple of lcm(6,4)=12 by nudging the timeline; size - the ring to the least common multiple structure that holds maxIO + cushion. + Strategy: io_budget = max_io. Keep the 400-cycle content horizon, choose a + shared packet backing that covers the preparation lead and keeps that + horizon below half ring, then size the timeline to contain it. We do not mutate the header derivation — we just pick a timeline_slots and ring_frames large enough that every invariant passes. """ g = Geometry(name=f"CANDIDATE (max IO {max_io})", ring_frames=zts, io_budget_frames=max_io, - zts_frames=zts, timeline_slots=512) - - # Timeline must hold the derived shared-slot count (which is already a - # multiple of 12 because frameExposureWindow = 2*ceil(.) and the +192 base - # is %12==0 only when exposure_lead_pkts is %6==0). If the derived - # shared_slot is not %12, the header's own asserts would fail, so we round - # the timeline up to the next multiple of the group size and report the - # shared-slot alignment separately. - timeline = ((g.shared_slot_pkts + PKTS_PER_GROUP - 1) // PKTS_PER_GROUP) * PKTS_PER_GROUP - timeline = max(timeline, 512) - g = replace(g, timeline_slots=timeline) - - # Ring: multiple of both io_budget and zts, strictly greater than the - # exposure lead, and >= maxIO. lcm(maxIO, zts) stepped until > exposureLead. + zts_frames=zts, timeline_slots=1024) + + required_shared = max( + g.preparation_lead_pkts + HW_RING_PKTS, + 2 * g.exposure_lead_pkts) + shared = max(912, ((required_shared + 11) // 12) * 12) + timeline = max(1024, ((shared + 11) // 12) * 12) + g = replace(g, shared_slots=shared, timeline_slots=timeline) + + # Frame ring: multiple of both IO and ZTS periods, and holds one max IO. step = math.lcm(max_io, zts) ring = step - while ring <= g.exposure_lead_frames or ring < max_io: + while ring < max_io: ring += step ring = max(ring, zts) while ring % FRAME_ALIGNMENT != 0: @@ -302,8 +299,9 @@ def cmd_verify(args) -> None: f"clientIoBudgetFrames = {cand.io_budget_frames} " f"zeroTimestampPeriodFrames = {cand.zts_frames}") print(f" AudioTimingGeometry kTimelineSlots = {cand.timeline_slots}") - print(f" (exposure lead, packet windows, shared-slot ring all derive " - f"automatically and stay green)") + print(f" AudioTimingGeometry kTxSharedSlotPackets = {cand.shared_slot_pkts}") + print(" (content horizon stays packet-time based; verify burst behavior with " + "tx_data_horizon_burst_sim.py)") def cmd_solve(args) -> None: diff --git a/tools/tx_data_horizon_burst_sim.py b/tools/tx_data_horizon_burst_sim.py new file mode 100644 index 00000000..71fe64dc --- /dev/null +++ b/tools/tx_data_horizon_burst_sim.py @@ -0,0 +1,242 @@ +#!/usr/bin/env python3 +"""Model ASFW's coalesced TX data-horizon scheduler under CoreAudio bursts. + +This is deliberately narrower than ``tx_payload_ownership_sim.py``. It +models the implementation introduced for Duet stability: + +* CoreAudio emits discrete WriteEnd bursts (normally 512 frames). +* Each burst updates W, then publishes one coalesced preparation request. +* The preparation queue advances E to W + the 400-cycle content horizon when + it is allowed to run. It also restores packet preparation to completion + + the 678-packet lead. +* While the action is delayed, W can cross E (a PayloadWriter + ``framesWithoutPacket`` loss) independently of descriptor margin. + +The model is 48 kHz blocking AMDTP only: D,D,D,N = 8,8,8,0 frames/cycle. +It is a scheduler proof, not a wire/SYT simulator. + +Examples: + python3 tools/tx_data_horizon_burst_sim.py suite + python3 tools/tx_data_horizon_burst_sim.py run --stall-at 1 --stall-cycles 400 + python3 tools/tx_data_horizon_burst_sim.py run --io-frames 1536 --stall-cycles 250 +""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass, replace + + +CYCLES_PER_SECOND = 8_000 +SAMPLE_RATE = 48_000 +CADENCE = (8, 8, 8, 0) + +# Keep these in lock-step with AudioTimingGeometry.hpp. +HW_RING_PACKETS = 48 +PREPARATION_LEAD_PACKETS = 678 +SHARED_SLOT_PACKETS = 912 +CONTENT_HORIZON_PACKETS = 400 +DEFAULT_IO_FRAMES = 512 + + +def frames_at_cycle(cycle: int) -> int: + """Number of 48 kHz blocking frames carried before ``cycle``.""" + blocks, remainder = divmod(cycle, len(CADENCE)) + return blocks * sum(CADENCE) + sum(CADENCE[:remainder]) + + +def horizon_frames(sample_rate: int) -> int: + return (CONTENT_HORIZON_PACKETS * sample_rate + CYCLES_PER_SECOND - 1) // CYCLES_PER_SECOND + + +@dataclass(frozen=True) +class Config: + duration_cycles: int = 8_000 + io_frames: int = DEFAULT_IO_FRAMES + wake_latency_cycles: int = 0 + stall_at_cycle: int = 1 + stall_cycles: int = 0 + content_horizon_packets: int = CONTENT_HORIZON_PACKETS + preparation_lead_packets: int = PREPARATION_LEAD_PACKETS + shared_slot_packets: int = SHARED_SLOT_PACKETS + + +@dataclass +class Result: + write_callbacks: int = 0 + preparation_actions: int = 0 + coalesced_requests: int = 0 + dropped_frames: int = 0 + max_exposure_deficit: int = 0 + max_pending_requests: int = 0 + min_descriptor_margin: int = PREPARATION_LEAD_PACKETS + descriptor_fatal_cycle: int | None = None + target_limited_actions: int = 0 + + @property + def ok(self) -> bool: + return self.dropped_frames == 0 and self.descriptor_fatal_cycle is None + + +def write_cycles(cfg: Config): + """Yield the integer cycle at which each periodic CoreAudio write lands.""" + write_index = 0 + while True: + # ceil(write_index * ioFrames / rate * 8000). The first WriteEnd is + # at cycle zero, matching the start of the host timeline. + cycle = (write_index * cfg.io_frames * CYCLES_PER_SECOND + SAMPLE_RATE - 1) // SAMPLE_RATE + if cycle >= cfg.duration_cycles: + return + yield cycle + write_index += 1 + + +def run(cfg: Config) -> Result: + if cfg.io_frames <= 0: + raise ValueError("io_frames must be positive") + if cfg.preparation_lead_packets + HW_RING_PACKETS > cfg.shared_slot_packets: + raise ValueError("preparation lead must leave one hardware-ring depth before reuse") + + result = Result(min_descriptor_margin=cfg.preparation_lead_packets) + writes = iter(write_cycles(cfg)) + next_write = next(writes, None) + horizon = (cfg.content_horizon_packets * SAMPLE_RATE + CYCLES_PER_SECOND - 1) // CYCLES_PER_SECOND + + write_end = 0 + exposed_end = horizon + prepared_packet_end = cfg.preparation_lead_packets + requested_generation = 0 + handled_generation = 0 + wake_pending = False + wake_due = 0 + stall_end = cfg.stall_at_cycle + cfg.stall_cycles + + for cycle in range(cfg.duration_cycles): + descriptor_margin = prepared_packet_end - cycle + result.min_descriptor_margin = min(result.min_descriptor_margin, descriptor_margin) + if descriptor_margin < HW_RING_PACKETS and result.descriptor_fatal_cycle is None: + result.descriptor_fatal_cycle = cycle + + # The writer runs first, exactly as WriteEnd observes the *current* + # exposure. Its request can only improve later writes, never recover a + # frame that this callback already classified as without-packet. + while next_write == cycle: + write_end += cfg.io_frames + result.write_callbacks += 1 + deficit = max(0, write_end - exposed_end) + result.dropped_frames += deficit + result.max_exposure_deficit = max(result.max_exposure_deficit, deficit) + + requested_generation += 1 + if wake_pending: + result.coalesced_requests += 1 + else: + wake_pending = True + wake_due = cycle + cfg.wake_latency_cycles + result.max_pending_requests = max( + result.max_pending_requests, + requested_generation - handled_generation) + next_write = next(writes, None) + + stalled = cfg.stall_cycles != 0 and cfg.stall_at_cycle <= cycle < stall_end + if wake_pending and cycle >= wake_due and not stalled: + # The producer can only prepare through its packet lead. At 48 kHz + # this is exact cadence rather than a 6-frame average. + capacity_end = frames_at_cycle(cycle + cfg.preparation_lead_packets) + target = write_end + horizon + if target > capacity_end: + result.target_limited_actions += 1 + exposed_end = max(exposed_end, min(target, capacity_end)) + prepared_packet_end = max( + prepared_packet_end, cycle + cfg.preparation_lead_packets) + handled_generation = requested_generation + wake_pending = False + result.preparation_actions += 1 + + return result + + +def print_result(label: str, cfg: Config, result: Result) -> None: + horizon = (cfg.content_horizon_packets * SAMPLE_RATE) / CYCLES_PER_SECOND + print(label) + print( + f" geometry: horizon={cfg.content_horizon_packets}pkt/{horizon:.0f}fr " + f"({cfg.content_horizon_packets / 8:.1f}ms) prepLead={cfg.preparation_lead_packets}pkt " + f"shared={cfg.shared_slot_packets}pkt io={cfg.io_frames}fr") + print( + f" stall: at={cfg.stall_at_cycle} cycles for {cfg.stall_cycles} cycles " + f"({cfg.stall_cycles / 8:.3f}ms), wakeLatency={cfg.wake_latency_cycles} cycles") + print( + f" result: {'PASS' if result.ok else 'FAIL'} callbacks={result.write_callbacks} " + f"actions={result.preparation_actions} coalesced={result.coalesced_requests} " + f"dropFrames={result.dropped_frames} maxDeficit={result.max_exposure_deficit} " + f"minDescriptorMargin={result.min_descriptor_margin} " + f"descriptorFatal={result.descriptor_fatal_cycle} targetLimited={result.target_limited_actions}") + + +def cmd_run(args: argparse.Namespace) -> int: + cfg = Config( + duration_cycles=args.duration, + io_frames=args.io_frames, + wake_latency_cycles=args.wake_latency, + stall_at_cycle=args.stall_at, + stall_cycles=args.stall_cycles, + content_horizon_packets=args.horizon_packets, + preparation_lead_packets=args.preparation_lead, + shared_slot_packets=args.shared_slots, + ) + result = run(cfg) + print_result("TX data-horizon burst simulation", cfg, result) + return 0 if result.ok else 1 + + +def cmd_suite(_args: argparse.Namespace) -> int: + # The old 576-frame cushion equals 96 packet cycles at 48 kHz. A 22.5 ms + # queue stall crosses a second 512-frame WriteEnd and loses PCM. The new + # 400-cycle horizon absorbs a 50 ms queue stall, but correctly does not + # claim to survive an unbounded stall. + cases = ( + ("old 96-cycle horizon, 180-cycle stall (must lose PCM)", + replace(Config(), content_horizon_packets=96, stall_cycles=180), False), + ("new 400-cycle horizon, 400-cycle stall (must survive)", + replace(Config(), stall_cycles=400), True), + ("new 400-cycle horizon, 450-cycle stall (bounded loss)", + replace(Config(), stall_cycles=450), False), + ("new horizon, 700-cycle stall (descriptor deadline still wins)", + replace(Config(), stall_cycles=700), False), + ) + failed = False + for label, cfg, expected_ok in cases: + result = run(cfg) + print_result(label, cfg, result) + if result.ok != expected_ok: + print(" !! unexpected result") + failed = True + print() + return 1 if failed else 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + sub = parser.add_subparsers(dest="command", required=True) + + run_parser = sub.add_parser("run", help="run one deterministic burst/stall scenario") + run_parser.add_argument("--duration", type=int, default=8_000) + run_parser.add_argument("--io-frames", type=int, default=DEFAULT_IO_FRAMES) + run_parser.add_argument("--wake-latency", type=int, default=0) + run_parser.add_argument("--stall-at", type=int, default=1) + run_parser.add_argument("--stall-cycles", type=int, default=0) + run_parser.add_argument("--horizon-packets", type=int, default=CONTENT_HORIZON_PACKETS) + run_parser.add_argument("--preparation-lead", type=int, default=PREPARATION_LEAD_PACKETS) + run_parser.add_argument("--shared-slots", type=int, default=SHARED_SLOT_PACKETS) + run_parser.set_defaults(func=cmd_run) + + suite_parser = sub.add_parser("suite", help="run old/new burst-boundary regression cases") + suite_parser.set_defaults(func=cmd_suite) + + args = parser.parse_args() + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main())