From 56f3305fe8c0b32244f6868d294f0db7eaa4e7a8 Mon Sep 17 00:00:00 2001 From: Geoffrey McRae Date: Sat, 5 Sep 2026 07:55:23 +1000 Subject: [PATCH] [client/idd] audio: prevent recording transport stalls Give microphone data priority over replaceable clock feedback and reduce feedback traffic on the shared client-to-host FIFO. Retry short FIFO stalls outside the realtime thread and coalesce consecutive packets in the existing fixed Windows buffers. Avoid one broker write per tiny PipeWire quantum without adding allocations or intentional capture latency. --- client/src/audio.c | 2 +- client/transports/LGMP/audio.c | 55 ++++++++----- idd/LGIdd/audio/CLGAudioDevice.cpp | 122 +++++++++++++++++------------ idd/LGIdd/audio/CLGAudioDevice.h | 8 +- idd/LGIdd/transport/CAudioHub.h | 2 +- 5 files changed, 116 insertions(+), 73 deletions(-) diff --git a/client/src/audio.c b/client/src/audio.c index a17277d0..3e199e30 100644 --- a/client/src/audio.c +++ b/client/src/audio.c @@ -74,7 +74,7 @@ #define PLAYBACK_DEVICE_RATE_STABLE_DELTA_PPM 50.0 #define PLAYBACK_DEVICE_RATE_MAX_ACQUIRE_SEC 20.0 #define PLAYBACK_MAX_SOURCE_PACKET_MS 100 -#define PLAYBACK_FEEDBACK_INTERVAL_NS INT64_C(1000000) +#define PLAYBACK_FEEDBACK_INTERVAL_NS INT64_C(10000000) #define PLAYBACK_FEEDBACK_FAILURE_NS INT64_C(100000000) #define PLAYBACK_GRAPH_INTERVAL_NS INT64_C(25000000) diff --git a/client/transports/LGMP/audio.c b/client/transports/LGMP/audio.c index c7c41a5d..e60b02c1 100644 --- a/client/transports/LGMP/audio.c +++ b/client/transports/LGMP/audio.c @@ -41,6 +41,8 @@ #define AUDIO_CLAIM_RETRY_US 200000U #define AUDIO_OWNER_RETRY_US 1200000U #define AUDIO_DRAIN_MAX 32U +#define AUDIO_CAPTURE_WRITE_TIMEOUT_US 5000U +#define AUDIO_CAPTURE_WRITE_MAX_WAIT_US 250U #define AUDIO_CAPABILITIES \ (KVMFR_AUDIO_CAP_PLAYBACK | KVMFR_AUDIO_CAP_CAPTURE | \ @@ -2226,12 +2228,7 @@ static bool recordData(void * opaque, uint32_t generation, if (!generation || !data || !frames || frames > KVMFR_AUDIO_MAX_FRAMES_PER_PACKET) return false; - if (!tryWriteGate(audio)) - { - atomic_store_explicit( - &audio->captureDiscontinuity, true, memory_order_release); - return false; - } + lockWriteGate(audio); if (!audio->publicDataAllowed || !audio->writeBound || generation != audio->writeCaptureLocalGeneration || @@ -2271,10 +2268,43 @@ static bool recordData(void * opaque, uint32_t generation, &audio->captureDiscontinuity, false, memory_order_acq_rel)) flags |= KVMFR_AUDIO_MESSAGE_DISCONTINUITY; - const enum WriteResult result = writeMessageLocked(audio, + enum WriteResult result = writeMessageLocked(audio, KVMFR_AUDIO_MESSAGE_CAPTURE_DATA, audio->writeCaptureGeneration, flags, position, timeNs, rateQ32, (uint32_t)frames, data, (uint32_t)bytes); + if (result == WRITE_FULL) + { + LGMPStreamPollState streamPoll; + const LGMP_STATUS pollStatus = lgmpStreamPollInit(&streamPoll, + (struct LGMPStreamPollConfig) + { + .spinCount = 64U, + .minWaitUs = AUDIO_ACTIVE_POLL_US, + .maxWaitUs = AUDIO_CAPTURE_WRITE_MAX_WAIT_US, + }); + DEBUG_ASSERT(pollStatus == LGMP_OK); + + const uint64_t deadline = + microtime() + AUDIO_CAPTURE_WRITE_TIMEOUT_US; + while (result == WRITE_FULL && pollStatus == LGMP_OK) + { + const uint64_t now = microtime(); + if (now >= deadline) + break; + + uint32_t waitUs = lgmpStreamPollIdle(&streamPoll); + const uint64_t remainingUs = deadline - now; + if (waitUs > remainingUs) + waitUs = (uint32_t)remainingUs; + if (waitUs) + nsleep((uint64_t)waitUs * 1000U); + + result = writeMessageLocked(audio, + KVMFR_AUDIO_MESSAGE_CAPTURE_DATA, + audio->writeCaptureGeneration, flags, position, timeNs, rateQ32, + (uint32_t)frames, data, (uint32_t)bytes); + } + } if (result != WRITE_OK) { if (result == WRITE_FULL) @@ -2298,11 +2328,7 @@ static bool clockFeedback(void * opaque, uint32_t generation, return false; if (!tryWriteGate(audio)) - { - atomic_store_explicit( - &audio->captureDiscontinuity, true, memory_order_release); return false; - } LG_LOCK(audio->lock); const bool supported = audio->phase == AUDIO_PHASE_BOUND && !audio->releaseRequested && audio->statusAvailable && @@ -2338,13 +2364,6 @@ static bool clockFeedback(void * opaque, uint32_t generation, audio->writePlaybackGeneration, flags, playbackClock->position, playbackClock->time, rateQ32, 0, &feedback, sizeof(feedback)); - if (result != WRITE_OK) - { - if (result == WRITE_FULL) - ++audio->clientSequence; - atomic_store_explicit( - &audio->captureDiscontinuity, true, memory_order_release); - } unlockWriteGate(audio); return result == WRITE_OK; } diff --git a/idd/LGIdd/audio/CLGAudioDevice.cpp b/idd/LGIdd/audio/CLGAudioDevice.cpp index 7270accf..28b1fee1 100644 --- a/idd/LGIdd/audio/CLGAudioDevice.cpp +++ b/idd/LGIdd/audio/CLGAudioDevice.cpp @@ -283,7 +283,7 @@ private: }; mutable CSRWLock m_lock; - ILGAudioDeviceEvents * m_events = nullptr; + IAudioDeviceEvents * m_events = nullptr; HANDLE m_stopEvent = nullptr; HANDLE m_wakeEvent = nullptr; HANDLE m_controlEvent = nullptr; @@ -340,7 +340,7 @@ private: return WaitForSingleObject(m_stopEvent, 0) == 0; } - void ClearPlaybackBlock() + void UnblockPlayback() { CSRWExclusiveLock lock(m_lock); m_playbackBlocked = false; @@ -352,28 +352,28 @@ private: return m_playbackBlocked; } - void ClearClockFeedbackLocked() + void ClearFeedback() { m_clockFeedback = {}; m_clockFeedbackPending = false; } - void ClearClockControlFailureLocked() + void ClearFaultLocked() { m_clockControlFailureActive = false; m_clockControlFailureSince = 0; } - void ResetClockControlLocked() + void ResetClockLocked() { - ClearClockFeedbackLocked(); - ClearClockControlFailureLocked(); + ClearFeedback(); + ClearFaultLocked(); } - void ResetClockControl() + void ResetClock() { CSRWExclusiveLock lock(m_lock); - ResetClockControlLocked(); + ResetClockLocked(); } static DWORD CALLBACK ThreadProc(void * opaque) @@ -435,7 +435,7 @@ private: void Publish(AudioEndpointStatus next, bool advance) { - ILGAudioDeviceEvents * events = nullptr; + IAudioDeviceEvents * events = nullptr; { CSRWExclusiveLock lock(m_lock); if (advance) @@ -447,7 +447,7 @@ private: next.active != m_status.active || !SameStream(next.playback, m_status.playback)) { - ResetClockControlLocked(); + ResetClockLocked(); } if (SameStatus(next, m_status)) return; @@ -458,7 +458,7 @@ private: events->LGAudioState(next); } - void PublishUnavailable(bool advance) + void PublishOffline(bool advance) { AudioEndpointStatus next; next.capabilities = 0; @@ -477,7 +477,7 @@ private: return result; } - void PublishBrokerState(bool forceAdvance = false) + void PublishBroker(bool forceAdvance = false) { AudioEndpointStatus previous = Status(); AudioEndpointStatus next; @@ -519,7 +519,7 @@ private: if (previous.playback.generation != next.playback.generation || previous.active != next.active || bindingChanged) { - ClearPlaybackBlock(); + UnblockPlayback(); m_renderSequenceValid = false; m_renderDiscontinuity = true; } @@ -628,12 +628,12 @@ private: m_writeBytes = 0; ResetEvent(m_writeOverlapped.hEvent); m_captureDiscontinuity = false; - ClearPlaybackBlock(); + UnblockPlayback(); m_renderSequenceValid = false; m_renderDiscontinuity = true; } - void ClearCaptureQueue() + void ClearCapture() { CSRWExclusiveLock lock(m_lock); m_captureHead = 0; @@ -642,7 +642,7 @@ private: void CloseBroker(bool publish = true) { - ResetClockControl(); + ResetClock(); if (m_handle == INVALID_HANDLE_VALUE) return; @@ -668,14 +668,14 @@ private: m_captureStreamGeneration = 0; m_nextCaptureSequence = 0; m_captureDiscontinuity = false; - ClearCaptureQueue(); + ClearCapture(); if (!Stopping()) m_nextOpenAt = GetTickCount64() + OPEN_RETRY_MS; if (publish || inactiveBarrier) - PublishUnavailable(true); + PublishOffline(true); } - bool FinishAbsentDeactivation() + bool FinishAbsentStop() { { CSRWExclusiveLock lock(m_lock); @@ -683,11 +683,11 @@ private: m_activation != ActivationState::DEACTIVATING) return false; m_activation = ActivationState::IDLE; - ResetClockControlLocked(); + ResetClockLocked(); m_captureHead = 0; m_captureCount = 0; } - PublishUnavailable(true); + PublishOffline(true); return true; } @@ -724,7 +724,7 @@ private: DEBUG_WARN("Looking Glass Audio broker query failed: %lu", error); CloseBroker(false); m_nextOpenAt = now + OPEN_RETRY_MS; - PublishUnavailable(false); + PublishOffline(false); return false; } @@ -738,7 +738,7 @@ private: } m_nextPollAt = 0; DEBUG_INFO("Looking Glass Audio broker connected"); - PublishBrokerState(); + PublishBroker(); return true; } @@ -760,16 +760,16 @@ private: else if (!bound && m_activation == ActivationState::DEACTIVATING) { m_activation = ActivationState::IDLE; - ResetClockControlLocked(); + ResetClockLocked(); return true; } return false; } - void DropQueuedCapture() + void DropCapture() { bool ready = false; - ILGAudioDeviceEvents * events = nullptr; + IAudioDeviceEvents * events = nullptr; { CSRWExclusiveLock lock(m_lock); ready = m_captureCount == CAPTURE_SLOT_COUNT; @@ -820,7 +820,7 @@ private: else if (!desired && initiallyBound) { CancelDataIo(); - DropQueuedCapture(); + DropCapture(); if (!Unbind(error) && !Stopping()) DEBUG_WARN("Looking Glass Audio broker unbind failed: %lu", error); if (!QueryState(error)) @@ -837,13 +837,13 @@ private: LGAUDIO_BROKER_STATE_DATA_PLANE_ACTIVE) != 0; const bool inactiveBarrier = FinishActivation(bound, dataPlaneActive); - PublishBrokerState(inactiveBarrier); + PublishBroker(inactiveBarrier); m_nextPollAt = now + (desired ? ACTIVE_POLL_MS : IDLE_POLL_MS); return true; } - void ProcessClockControl() + void ProcessClock() { AudioClockFeedback feedback; { @@ -852,7 +852,7 @@ private: return; feedback = m_clockFeedback; - ClearClockFeedbackLocked(); + ClearFeedback(); if (!m_started || m_stopping || m_activation != ActivationState::ACTIVE || !(m_status.capabilities & AUDIO_CAP_CLOCK_FEEDBACK) || @@ -892,7 +892,7 @@ private: returned == 0) { CSRWExclusiveLock lock(m_lock); - ClearClockControlFailureLocked(); + ClearFaultLocked(); return; } @@ -927,9 +927,9 @@ private: { CSRWExclusiveLock lock(m_lock); m_lastError = error; - ResetClockControlLocked(); + ResetClockLocked(); } - PublishBrokerState(true); + PublishBroker(true); } bool ValidateRender(const LGAudioBrokerPacketHeader& header, @@ -1027,7 +1027,7 @@ private: if (m_renderDiscontinuity) packet.flags |= AUDIO_PACKET_DISCONTINUITY; - ILGAudioDeviceEvents * events = nullptr; + IAudioDeviceEvents * events = nullptr; uint64_t readySerial = 0; { CSRWSharedLock lock(m_lock); @@ -1148,7 +1148,7 @@ private: bool PopCapture(CaptureSlot& result, bool& becameReady) { - ILGAudioDeviceEvents * events = nullptr; + IAudioDeviceEvents * events = nullptr; { CSRWExclusiveLock lock(m_lock); if (!m_captureCount) @@ -1289,7 +1289,7 @@ private: while (!Stopping()) { if (m_handle == INVALID_HANDLE_VALUE) - FinishAbsentDeactivation(); + FinishAbsentStop(); if (m_handle == INVALID_HANDLE_VALUE && !OpenBroker()) { const HANDLE waits[] = { m_stopEvent, m_wakeEvent }; @@ -1310,7 +1310,7 @@ private: if (!Reconcile(false)) continue; - ProcessClockControl(); + ProcessClock(); DeliverRender(); StartWriteIo(); StartRenderIo(); @@ -1328,7 +1328,7 @@ private: { CSRWExclusiveLock lock(m_lock); m_activation = ActivationState::DEACTIVATING; - ResetClockControlLocked(); + ResetClockLocked(); } if (m_handle != INVALID_HANDLE_VALUE) { @@ -1340,7 +1340,7 @@ private: } CloseBroker(false); } - FinishAbsentDeactivation(); + FinishAbsentStop(); if (avTaskHandle) AvRevertMmThreadCharacteristics(avTaskHandle); } @@ -1380,7 +1380,7 @@ public: Stop(); } - bool Start(ILGAudioDeviceEvents& events) + bool Start(IAudioDeviceEvents& events) { CSRWExclusiveLock lock(m_lock); if (m_started) @@ -1410,7 +1410,7 @@ public: m_events = &events; m_started = true; m_stopping = false; - ResetClockControlLocked(); + ResetClockLocked(); m_thread = CreateThread(nullptr, 0, ThreadProc, this, 0, nullptr); if (!m_thread) { @@ -1430,7 +1430,7 @@ public: if (!m_started) return; m_stopping = true; - ResetClockControlLocked(); + ResetClockLocked(); thread = m_thread; SetEvent(m_stopEvent); SetEvent(m_wakeEvent); @@ -1451,7 +1451,7 @@ public: m_captureCount = 0; m_activation = ActivationState::IDLE; m_controlDirty = true; - ResetClockControlLocked(); + ResetClockLocked(); m_status = {}; m_status.generation = m_endpointGeneration; CloseEvents(); @@ -1495,7 +1495,7 @@ public: return AudioResult::BUSY; if (m_activation == ActivationState::IDLE) { - ResetClockControlLocked(); + ResetClockLocked(); m_activation = ActivationState::ACTIVATING; } } @@ -1503,7 +1503,7 @@ public: m_activation == ActivationState::ACTIVE) { m_activation = ActivationState::DEACTIVATING; - ResetClockControlLocked(); + ResetClockLocked(); m_controlDirty = true; SetEvent(m_wakeEvent); } @@ -1548,6 +1548,30 @@ public: m_status.capture.state != AudioStreamState::RUNNING || !SameFormat(packet.format, m_status.capture.format)) return AudioResult::UNAVAILABLE; + + if (m_captureCount && + !(packet.flags & AUDIO_PACKET_DISCONTINUITY)) + { + CaptureSlot& tail = m_capture[ + (m_captureHead + m_captureCount - 1) % CAPTURE_SLOT_COUNT]; + if (tail.packet.endpointGeneration == packet.endpointGeneration && + tail.packet.streamGeneration == packet.streamGeneration && + SameFormat(tail.packet.format, packet.format) && + packet.frames <= + AUDIO_MAX_FRAMES_PER_PACKET - tail.packet.frames && + pcmBytes <= AUDIO_MAX_PCM_BYTES - tail.size) + { + if (silent) + memset(tail.data + tail.size, 0, pcmBytes); + else + memcpy(tail.data + tail.size, data, pcmBytes); + tail.packet.frames += packet.frames; + tail.size += pcmBytes; + SetEvent(m_wakeEvent); + return AudioResult::ACCEPTED; + } + } + if (m_captureCount == CAPTURE_SLOT_COUNT) return AudioResult::BUSY; @@ -1595,7 +1619,7 @@ public: return AudioResult::ACCEPTED; } - void PlaybackReceiveReady() + void PlaybackReady() { bool wake = false; { @@ -1622,7 +1646,7 @@ CLGAudioDevice::~CLGAudioDevice() Stop(); } -bool CLGAudioDevice::Start(ILGAudioDeviceEvents& events) +bool CLGAudioDevice::Start(IAudioDeviceEvents& events) { return m_state && m_state->Start(events); } @@ -1661,5 +1685,5 @@ AudioResult CLGAudioDevice::SetClockFeedback( void CLGAudioDevice::PlaybackReceiveReady() { if (m_state) - m_state->PlaybackReceiveReady(); + m_state->PlaybackReady(); } diff --git a/idd/LGIdd/audio/CLGAudioDevice.h b/idd/LGIdd/audio/CLGAudioDevice.h index 062374a0..8d099eff 100644 --- a/idd/LGIdd/audio/CLGAudioDevice.h +++ b/idd/LGIdd/audio/CLGAudioDevice.h @@ -39,10 +39,10 @@ struct LGAudioDeviceSnapshot uint32_t lastError = 0; }; -class ILGAudioDeviceEvents +class IAudioDeviceEvents { public: - virtual ~ILGAudioDeviceEvents() = default; + virtual ~IAudioDeviceEvents() = default; virtual void LGAudioState(const AudioEndpointStatus& state) = 0; virtual AudioResult LGAudioPlayback(const AudioPacket& packet, @@ -57,7 +57,7 @@ public: // Methods are thread-safe and nonblocking except Stop. Stop is a callback- // quiescence barrier; after it returns no event callback can still run. - virtual bool Start(ILGAudioDeviceEvents& events) = 0; + virtual bool Start(IAudioDeviceEvents& events) = 0; virtual void Stop() = 0; virtual LGAudioDeviceSnapshot Snapshot() const = 0; virtual AudioResult SetActive(bool active) = 0; @@ -81,7 +81,7 @@ public: CLGAudioDevice(const CLGAudioDevice&) = delete; CLGAudioDevice& operator=(const CLGAudioDevice&) = delete; - bool Start(ILGAudioDeviceEvents& events) override; + bool Start(IAudioDeviceEvents& events) override; void Stop() override; LGAudioDeviceSnapshot Snapshot() const override; diff --git a/idd/LGIdd/transport/CAudioHub.h b/idd/LGIdd/transport/CAudioHub.h index cbc17f72..e71a7292 100644 --- a/idd/LGIdd/transport/CAudioHub.h +++ b/idd/LGIdd/transport/CAudioHub.h @@ -28,7 +28,7 @@ #include class CAudioHub final : public IAudioTarget, - private ILGAudioDeviceEvents + private IAudioDeviceEvents { private: CSRWLock m_lifecycleLock;