Skip to content

Duet stability: packetizer telemetry and recovery notes #78

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mrmidi merged 2 commits into from
Jul 18, 2026
Merged

Duet stability: packetizer telemetry and recovery notes #78

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b83ae3c
Improve Duet stream stability telemetry
mrmidi Jul 18, 2026
5228347
Document Apple TX horizon evidence
mrmidi Jul 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions ASFWDriver/Audio/DriverKit/ASFWAudioDriverIO.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,6 +210,12 @@ kern_return_t InstallIOOperationHandler(IOUserAudioDevice& audioDevice,
hostBuffer,
completionCursor);

const auto packetizerSnapshot =
driverIvars->runtime.txStreamEngine
.PacketizerTelemetrySnapshot();
const auto& txCounters =
driverIvars->runtime.txStreamEngine.Counters();

// Fan out the same host buffer to the secondary stream; its
// payload writer reads channels [16, 32) via sourceChannelOffset.
if (driverIvars->runtime.txSecondaryActive) {
Expand Down Expand Up @@ -246,6 +252,28 @@ kern_return_t InstallIOOperationHandler(IOUserAudioDevice& audioDevice,
cw.underExposureCalls.load(std::memory_order_relaxed);
rec.underExposureFrames =
cw.underExposureFrames.load(std::memory_order_relaxed);
rec.packetizerNextAudioFrame =
packetizerSnapshot.nextAudioFrame;
rec.packetizerLastDataFirstAudioFrame =
packetizerSnapshot.lastDataFirstAudioFrame;
rec.packetizerLastDataEndAudioFrame =
packetizerSnapshot.lastDataEndAudioFrame;
rec.packetizerLastDataPacketIndex =
packetizerSnapshot.lastDataPacketIndex;
rec.packetizerCursorEpoch =
packetizerSnapshot.cursorEpoch;
rec.packetizerFrameCursorAligned =
packetizerSnapshot.frameCursorAligned;
rec.packetizerHasLastDataPacket =
packetizerSnapshot.hasLastDataPacket;
rec.packetsPrepared =
txCounters.packetsPrepared.load(std::memory_order_relaxed);
rec.dataPacketsPrepared =
txCounters.dataPacketsPrepared.load(std::memory_order_relaxed);
rec.noDataPacketsPrepared =
txCounters.noDataPacketsPrepared.load(std::memory_order_relaxed);
rec.slotAcquireFailures =
txCounters.slotAcquireFailures.load(std::memory_order_relaxed);
const uint32_t bits = cw.maxAbsSampleBits.load(std::memory_order_relaxed);
std::memcpy(&rec.maxAbsSample, &bits, sizeof(bits));

Expand Down
14 changes: 14 additions & 0 deletions ASFWDriver/Audio/DriverKit/Runtime/PayloadWriterTelemetry.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,20 @@ struct PayloadWriterTelemetryRecord final {
uint64_t underExposureFrames{0};
float maxAbsSample{0.0f};

// Snapshot of the TX packetizer state. This is published through atomic
// mirrors by the TX preparation side; it is diagnostic only.
uint64_t packetizerNextAudioFrame{0};
uint64_t packetizerLastDataFirstAudioFrame{0};
uint64_t packetizerLastDataEndAudioFrame{0};
uint64_t packetizerLastDataPacketIndex{0};
uint64_t packetizerCursorEpoch{0};
uint64_t packetsPrepared{0};
uint64_t dataPacketsPrepared{0};
uint64_t noDataPacketsPrepared{0};
uint64_t slotAcquireFailures{0};
bool packetizerFrameCursorAligned{false};
bool packetizerHasLastDataPacket{false};

// Additional cursors and addresses for diagnosing buffer mapping / pointer alignment
uint64_t playbackRingReadFrame{0};
uint64_t playbackRingWriteFrame{0};
Expand Down
152 changes: 141 additions & 11 deletions ASFWDriver/Audio/Engine/Direct/Rx/DirectAudioReceiveConsumer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,23 @@

namespace ASFW::AudioEngine::Direct::Rx {

const char* DirectAudioReceiveConsumer::ReplayResetReasonName(
ReplayResetReason reason) noexcept {
switch (reason) {
case ReplayResetReason::kPacketProcessorStatus:
return "packet-status";
case ReplayResetReason::kInvalidReceiveTimestamp:
return "invalid-rx-timestamp";
case ReplayResetReason::kReceiveCycleGap:
return "receive-cycle-gap";
case ReplayResetReason::kSytCadenceRejected:
return "syt-cadence-rejected";
case ReplayResetReason::kClockAnchorRejected:
return "clock-anchor-rejected";
}
return "unknown";
}

DirectAudioReceiveConsumer::DirectAudioReceiveConsumer(
::ASFW::Audio::Runtime::IDirectAudioBindingSource* bindingSource,
Configuration configuration) noexcept
Expand Down Expand Up @@ -162,7 +179,17 @@ void DirectAudioReceiveConsumer::ConsumePacket(
result.status == DirectRxWriteStatus::kInvalidBinding) {
absoluteFrameCursor_ += result.framesDecoded;
} else {
ResetReplayEpochForDiscontinuity();
ResetReplayEpochForDiscontinuity(
ReplayResetReason::kPacketProcessorStatus,
{
.descriptorIndex = packet.descriptorIndex,
.payloadBytes = static_cast<uint32_t>(packet.payload.size()),
.drainCycleTimer = batch.drainCycleTimer,
.receiveCycleTimestamp = result.receiveCycleTimestamp,
.syt = result.syt,
.packetStatus = static_cast<uint32_t>(result.status),
.sampleFrame = absoluteFrameCursor_,
});
return;
}

Expand All @@ -186,7 +213,16 @@ void DirectAudioReceiveConsumer::ConsumePacket(
result.receiveCycleTimestamp, batch.drainCycleTimer, timestamp);
if (!validTimestamp) {
++timestampInvalidCount_;
ResetReplayEpochForDiscontinuity();
ResetReplayEpochForDiscontinuity(
ReplayResetReason::kInvalidReceiveTimestamp,
{
.descriptorIndex = packet.descriptorIndex,
.payloadBytes = static_cast<uint32_t>(packet.payload.size()),
.drainCycleTimer = batch.drainCycleTimer,
.receiveCycleTimestamp = result.receiveCycleTimestamp,
.syt = result.syt,
.sampleFrame = absoluteFrameCursor_,
});
return;
}

Expand All @@ -198,7 +234,18 @@ void DirectAudioReceiveConsumer::ConsumePacket(
::ASFW::Timing::kFWTimeWrapSeconds * ::ASFW::Timing::kCyclesPerSecond;
if (replayCycleInitialized_ &&
cycleOrdinal != (lastReplayCycleOrdinal_ + 1) % kCycleDomain) {
ResetReplayEpochForDiscontinuity();
ResetReplayEpochForDiscontinuity(
ReplayResetReason::kReceiveCycleGap,
{
.descriptorIndex = packet.descriptorIndex,
.payloadBytes = static_cast<uint32_t>(packet.payload.size()),
.drainCycleTimer = batch.drainCycleTimer,
.receiveCycleTimestamp = result.receiveCycleTimestamp,
.syt = result.syt,
.expectedCycleOrdinal = (lastReplayCycleOrdinal_ + 1) % kCycleDomain,
.observedCycleOrdinal = cycleOrdinal,
.sampleFrame = absoluteFrameCursor_,
});
}
lastReplayCycleOrdinal_ = cycleOrdinal;
replayCycleInitialized_ = true;
Expand All @@ -215,7 +262,17 @@ void DirectAudioReceiveConsumer::ConsumePacket(
const bool cadenceAccepted = inputView_.control->rxSytCadence.Observe(
result.syt, timestamp.cycleTimer);
if (!cadenceAccepted && inputView_.control->rxSequenceReplay.IsEstablished()) {
ResetReplayEpochForDiscontinuity();
ResetReplayEpochForDiscontinuity(
ReplayResetReason::kSytCadenceRejected,
{
.descriptorIndex = packet.descriptorIndex,
.payloadBytes = static_cast<uint32_t>(packet.payload.size()),
.drainCycleTimer = batch.drainCycleTimer,
.receiveCycleTimestamp = result.receiveCycleTimestamp,
.syt = result.syt,
.observedCycleOrdinal = cycleOrdinal,
.sampleFrame = absoluteFrameCursor_,
});
}
replayEntry.sytOffset = ::ASFW::Audio::Runtime::ComputeReplaySytOffset(
result.syt, timestamp.cycleTimer,
Expand Down Expand Up @@ -266,7 +323,17 @@ void DirectAudioReceiveConsumer::ConsumePacket(
const auto publish = clockPublisher_.Publish(packetFirstFrame, packetHostTicks,
nanosPerSampleQ8);
if (!publish.accepted) {
ResetReplayEpochForDiscontinuity();
ResetReplayEpochForDiscontinuity(
ReplayResetReason::kClockAnchorRejected,
{
.descriptorIndex = packet.descriptorIndex,
.payloadBytes = static_cast<uint32_t>(packet.payload.size()),
.drainCycleTimer = batch.drainCycleTimer,
.receiveCycleTimestamp = result.receiveCycleTimestamp,
.syt = result.syt,
.observedCycleOrdinal = cycleOrdinal,
.sampleFrame = packetFirstFrame,
});
} else {
++ztsPublishCount_;
if (publish.notifyConsumer && ztsAnchorReadyCallback_) {
Expand Down Expand Up @@ -294,7 +361,9 @@ void DirectAudioReceiveConsumer::ConsumePacket(
}
}

void DirectAudioReceiveConsumer::ResetReplayEpochForDiscontinuity() noexcept {
void DirectAudioReceiveConsumer::ResetReplayEpochForDiscontinuity(
ReplayResetReason reason,
const ReplayResetContext& context) noexcept {
auto* control = inputView_.control;
if (!control || !replayResetForStart_) {
replayCycleInitialized_ = false;
Expand All @@ -303,11 +372,23 @@ void DirectAudioReceiveConsumer::ResetReplayEpochForDiscontinuity() noexcept {
const bool wasEstablished = control->rxSequenceReplay.IsEstablished();
control->rxSytCadence.Reset();
control->rxSequenceReplay.Reset();
control->rxReplayEpochResets.fetch_add(1, std::memory_order_relaxed);
const uint64_t resetEpoch =
control->rxReplayEpochResets.fetch_add(1, std::memory_order_relaxed) + 1;
replayReadyNotified_ = false;
cadenceEstablishedLogged_ = false;
replayCycleInitialized_ = false;
dbcInitialized_ = false;
if (wasEstablished) {
ASFW_LOG_ERROR(
DirectAudio,
"[RxReplayReset] epoch=%llu reason=%{public}s desc=%u bytes=%u "
"drain=0x%08x rawTs=0x%04x syt=0x%04x expectedCycle=%u observedCycle=%u "
"status=%u frame=%llu validTs=%llu invalidTs=%llu",
resetEpoch, ReplayResetReasonName(reason), context.descriptorIndex,
context.payloadBytes, context.drainCycleTimer, context.receiveCycleTimestamp,
context.syt, context.expectedCycleOrdinal, context.observedCycleOrdinal,
context.packetStatus, context.sampleFrame, timestampValidCount_, timestampInvalidCount_);
}
if (wasEstablished && timingLossCallback_) {
timingLossCallback_();
}
Expand Down Expand Up @@ -366,27 +447,76 @@ void DirectAudioReceiveConsumer::DrainPayloadTelemetry() {
}

payloadWriterTelemetryAggregator_.BeginDrain();
::ASFW::Audio::Runtime::PayloadWriterTelemetryRecord firstDeficitRecord{};
::ASFW::Audio::Runtime::PayloadWriterTelemetryRecord lastRecord{};
bool haveFirstDeficitRecord = false;
bool haveLastRecord = false;
const uint64_t dropped = control->payloadWriterTelemetry.Drain(
[this, &lastRecord, &haveLastRecord](
[this, &firstDeficitRecord, &haveFirstDeficitRecord,
&lastRecord, &haveLastRecord](
const ::ASFW::Audio::Runtime::PayloadWriterTelemetryRecord& record) {
payloadWriterTelemetryAggregator_.Observe(record);
if (!haveFirstDeficitRecord && record.exposureDeficitFrames != 0) {
firstDeficitRecord = record;
haveFirstDeficitRecord = true;
}
lastRecord = record;
haveLastRecord = true;
});
const auto& summary = payloadWriterTelemetryAggregator_.Summary();
if (haveLastRecord && summary.HasAnomaly()) {
ASFW_LOG(DirectAudio,
"[PayloadWriter] anomaly lastSample=%llu completion=%llu deficitMax=%llu "
"[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",
"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);
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);
}
if (dropped != 0) {
ASFW_LOG(DirectAudio, "[PayloadWriter] drain overflow: dropped=%llu (capacity=%u)",
Expand Down
24 changes: 23 additions & 1 deletion ASFWDriver/Audio/Engine/Direct/Rx/DirectAudioReceiveConsumer.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,29 @@ class DirectAudioReceiveConsumer final : public ::ASFW::Isoch::IIsochReceiveCons
void LogTransmitTimingTrace() override;

private:
void ResetReplayEpochForDiscontinuity() noexcept;
enum class ReplayResetReason : uint8_t {
kPacketProcessorStatus,
kInvalidReceiveTimestamp,
kReceiveCycleGap,
kSytCadenceRejected,
kClockAnchorRejected,
};

struct ReplayResetContext final {
uint32_t descriptorIndex{0};
uint32_t payloadBytes{0};
uint32_t drainCycleTimer{0};
uint16_t receiveCycleTimestamp{0};
uint16_t syt{0xffff};
uint32_t expectedCycleOrdinal{0};
uint32_t observedCycleOrdinal{0};
uint32_t packetStatus{0};
uint64_t sampleFrame{0};
};

[[nodiscard]] static const char* ReplayResetReasonName(ReplayResetReason reason) noexcept;
void ResetReplayEpochForDiscontinuity(ReplayResetReason reason,
const ReplayResetContext& context) noexcept;

::ASFW::Audio::Runtime::IDirectAudioBindingSource* bindingSource_{nullptr};
uint64_t lastBindingGeneration_{0};
Expand Down
5 changes: 5 additions & 0 deletions ASFWDriver/Audio/Engine/Direct/Tx/DiceTxStreamEngine.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,11 @@ const AMDTP::AmdtpStreamConfig& DiceTxStreamEngine::StreamConfig() const noexcep
return packetizer_.StreamConfig();
}

AMDTP::AmdtpTxPacketizerTelemetrySnapshot
DiceTxStreamEngine::PacketizerTelemetrySnapshot() const noexcept {
return packetizer_.TelemetrySnapshot();
}

const DiceTxEngineCounters& DiceTxStreamEngine::Counters() const noexcept {
return counters_;
}
Expand Down
3 changes: 3 additions & 0 deletions ASFWDriver/Audio/Engine/Direct/Tx/DiceTxStreamEngine.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,9 @@ class DiceTxStreamEngine final {

[[nodiscard]] const AMDTP::AmdtpStreamConfig& StreamConfig() const noexcept;

[[nodiscard]] AMDTP::AmdtpTxPacketizerTelemetrySnapshot
PacketizerTelemetrySnapshot() const noexcept;

[[nodiscard]] const DiceTxEngineCounters& Counters() const noexcept;

[[nodiscard]] const AMDTP::AmdtpPayloadWriterCounters&
Expand Down
5 changes: 4 additions & 1 deletion ASFWDriver/Audio/Model/ASFWAudioDevice.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,10 +76,12 @@ struct ASFWAudioDevice {
auto inputPlugNameStr = OSSharedPtr(OSString::withCString(inputPlugName.c_str()), OSNoRetain);
auto outputPlugNameStr = OSSharedPtr(OSString::withCString(outputPlugName.c_str()), OSNoRetain);
auto currentRateNum = OSSharedPtr(OSNumber::withNumber(currentSampleRate, 32), OSNoRetain);
auto streamModeNum = OSSharedPtr(
OSNumber::withNumber(static_cast<uint32_t>(streamMode), 32), OSNoRetain);

if (!deviceNameStr || !channelCountNum || !guidNum || !vendorIdNum || !modelIdNum ||
!inputChannelCountNum || !outputChannelCountNum || !sampleRatesArray ||
!inputPlugNameStr || !outputPlugNameStr || !currentRateNum) {
!inputPlugNameStr || !outputPlugNameStr || !currentRateNum || !streamModeNum) {
return false;
}

Expand All @@ -101,6 +103,7 @@ struct ASFWAudioDevice {
properties->setObject(PropertyKeys::kInputPlugName, inputPlugNameStr.get());
properties->setObject(PropertyKeys::kOutputPlugName, outputPlugNameStr.get());
properties->setObject(PropertyKeys::kCurrentSampleRate, currentRateNum.get());
properties->setObject(PropertyKeys::kStreamMode, streamModeNum.get());

// Sample rates advertised to CoreAudio. The HAL builds a stream format
// per entry (ASFWAudioDriverGraph), so this is what the user can select.
Expand Down
Loading
Loading