diff --git a/client/include/interface/audio.h b/client/include/interface/audio.h index 35cd9cc4..96154c6b 100644 --- a/client/include/interface/audio.h +++ b/client/include/interface/audio.h @@ -116,10 +116,11 @@ typedef struct LG_AudioEventOps uint8_t channels, const uint16_t volume[]); void (*playbackMute)(void * opaque, uint32_t generation, bool mute); /* Sample data is borrowed for the duration of the call. When supplied, the - * source clock position and time identify the first frame in this packet. */ + * source clock position and time identify the first frame in this packet. + * Discontinuity remains valid when no source clock is available. */ void (*playbackData)(void * opaque, uint32_t generation, const void * data, size_t frames, - const LG_AudioClock * sourceClock); + const LG_AudioClock * sourceClock, bool discontinuity); /* May repeat with the same generation to update an active stream format. */ void (*recordStart)(void * opaque, uint32_t generation, diff --git a/client/src/audio.c b/client/src/audio.c index 12d1d3f1..df55c3f0 100644 --- a/client/src/audio.c +++ b/client/src/audio.c @@ -2294,7 +2294,8 @@ static void playbackResetRateControl(PlaybackSourceData * sourceData, } static PlaybackDataResult playbackData(const void * data, size_t frameCount, - const LG_AudioClock * sourceClock, int64_t arrivalTime) + const LG_AudioClock * sourceClock, bool streamDiscontinuity, + int64_t arrivalTime) { StreamState state = playbackGetState(); if (state == STREAM_STATE_STOP_PENDING) @@ -2363,8 +2364,9 @@ static PlaybackDataResult playbackData(const void * data, size_t frameCount, const bool bufferUnderrun = STREAM_ACTIVE(playbackGetState()) && ringbuffer_getCount(audio.playback.buffer) < 0; - bool discontinuity = - bufferUnderrun || (sourceClock && sourceClock->discontinuity); + bool discontinuity = bufferUnderrun || + streamDiscontinuity || + (sourceClock && sourceClock->discontinuity); const int64_t previousArrivalTime = sourceData->lastArrivalTime; const int64_t packetTime = playbackMapMediaTime(sourceData, sourceClock, frames, @@ -4039,11 +4041,12 @@ static void eventPlaybackMute(void * opaque, uint32_t generation, bool mute) } static void eventPlaybackData(void * opaque, uint32_t generation, - const void * data, size_t frames, const LG_AudioClock * sourceClock) + const void * data, size_t frames, const LG_AudioClock * sourceClock, + bool discontinuity) { - const int64_t arrivalTime = nanotime(); - AudioBinding * binding = opaque; - AudioBinding active; + const int64_t arrivalTime = nanotime(); + AudioBinding * binding = opaque; + AudioBinding active; if (!eventBegin(binding, &active)) return; @@ -4057,7 +4060,8 @@ static void eventPlaybackData(void * opaque, uint32_t generation, PlaybackDataResult result = PLAYBACK_DATA_DROP; if (liveGeneration == generation) { - result = playbackData(data, frames, sourceClock, arrivalTime); + result = playbackData( + data, frames, sourceClock, discontinuity, arrivalTime); if (result == PLAYBACK_DATA_PROCESSED && atomic_load_explicit(&audio.playback.streamGeneration, diff --git a/client/tests/audio_test.c b/client/tests/audio_test.c index 09380403..c0e2f861 100644 --- a/client/tests/audio_test.c +++ b/client/tests/audio_test.c @@ -593,7 +593,7 @@ static void testPlaybackRetry(void) atomic_store(&b.setupOK, true); uint8_t frames[480 * 2 * 2] = { 0 }; usleep(10000); - p.events->playbackData(p.eventOpaque, 11, frames, 480, NULL); + p.events->playbackData(p.eventOpaque, 11, frames, 480, NULL, false); waitFor(&b.setupN, failed + 1); for (unsigned int i = 0; i < WAIT_MS; ++i) @@ -605,7 +605,7 @@ static void testPlaybackRetry(void) } CHECK(atomic_load(&audio.playback.streamGeneration) == 11); for (unsigned int i = 0; i < 8; ++i) - p.events->playbackData(p.eventOpaque, 11, frames, 480, NULL); + p.events->playbackData(p.eventOpaque, 11, frames, 480, NULL, false); waitFor(&b.playStartN, 1); LG_AudioFailureFn fail = b.playFail; @@ -617,7 +617,7 @@ static void testPlaybackRetry(void) const unsigned int retry = atomic_load(&b.setupN); usleep(10000); - p.events->playbackData(p.eventOpaque, 11, frames, 480, NULL); + p.events->playbackData(p.eventOpaque, 11, frames, 480, NULL, false); waitFor(&b.setupN, retry + 1); p.events->playbackStop(p.eventOpaque, 11); @@ -679,7 +679,7 @@ static void testPlaybackRecovery(void) CHECK(ringbuffer_consume(audio.playback.buffer, NULL, 16) == 16); CHECK(ringbuffer_getCount(audio.playback.buffer) == -16); uint8_t frames[1200 * 2 * 2] = { 0 }; - CHECK(playbackData(frames, 1200, NULL, arrival) == + CHECK(playbackData(frames, 1200, NULL, false, arrival) == PLAYBACK_DATA_PROCESSED); CHECK(source->sourcePacketDurationSec < 0.010); @@ -690,12 +690,12 @@ static void testPlaybackRecovery(void) CHECK(fabs(atomic_load(&audio.playback.backendResampleRatio) - source->lastClockRatio) < 0.000001); - CHECK(playbackData(frames, 1200, NULL, arrival + 1000) == + CHECK(playbackData(frames, 1200, NULL, false, arrival + 1000) == PLAYBACK_DATA_PROCESSED); CHECK(source->sourcePacketDurationSec < 0.010); CHECK(source->sourcePacketRecovering); - CHECK(playbackData(frames, 480, NULL, + CHECK(playbackData(frames, 480, NULL, false, arrival + INT64_C(10000000)) == PLAYBACK_DATA_PROCESSED); CHECK(!source->sourcePacketRecovering); CHECK(fabs(source->sourcePacketDurationSec - 0.010) < 0.000001); @@ -733,7 +733,7 @@ static void testPlaybackStableUnderrun(void) uint8_t frames[64 * 2 * 2]; uint8_t output[64 * 2 * 2]; memset(frames, 0x5a, sizeof(frames)); - CHECK(playbackData(frames, 64, NULL, arrival) == + CHECK(playbackData(frames, 64, NULL, false, arrival) == PLAYBACK_DATA_PROCESSED); const int recoveredOccupancy = @@ -765,7 +765,7 @@ static void testPlaybackStableUnderrun(void) source->devicePositionOffsetFrames; const int64_t secondPacketArrival = arrival + llrint(64.0e9 / format.sampleRate); - CHECK(playbackData(frames, 64, NULL, + CHECK(playbackData(frames, 64, NULL, false, secondPacketArrival) == PLAYBACK_DATA_PROCESSED); CHECK(source->devicePositionOffsetFrames == recoveredOffset); CHECK(ringbuffer_getCount(audio.playback.buffer) == @@ -852,7 +852,7 @@ static void testPlaybackResume(void) CHECK(atomic_load(&b.setupN) == setupN); uint8_t input[480 * 2 * 2] = { 0 }; - p.events->playbackData(p.eventOpaque, 12, input, 480, NULL); + p.events->playbackData(p.eventOpaque, 12, input, 480, NULL, false); CHECK(atomic_load(&audio.playback.streamGeneration) == 12); CHECK(playbackGetState() == STREAM_STATE_RUN); CHECK(atomic_load(&audio.playback.underruns) == 0); @@ -883,7 +883,7 @@ static void testPlaybackResume(void) p.events->playbackStart(p.eventOpaque, 13, &format, NULL); CHECK(atomic_load(&audio.playback.streamGeneration) == 13); - p.events->playbackData(p.eventOpaque, 13, input, 480, NULL); + p.events->playbackData(p.eventOpaque, 13, input, 480, NULL, false); CHECK(playbackGetState() == STREAM_STATE_RUN); CHECK(source->deviceClock.time == arrival); CHECK(source->deviceClock.position == idleFrames); diff --git a/client/tests/lgmp_audio_test.c b/client/tests/lgmp_audio_test.c index 25d8ee2d..7831e988 100644 --- a/client/tests/lgmp_audio_test.c +++ b/client/tests/lgmp_audio_test.c @@ -150,21 +150,23 @@ static unsigned audioSampleBytes(LG_AudioSampleFormat format) } static void playbackData(void * opaque, uint32_t generation, - const void * data, size_t frames, const LG_AudioClock * clock) + const void * data, size_t frames, const LG_AudioClock * clock, + bool discontinuity) { struct Trace * trace = opaque; + (void)discontinuity; pthread_mutex_lock(&trace->mutex); - trace->playGeneration = generation; - trace->playFrames = frames; - const size_t bytes = frames * trace->playFormat.channelCount * + trace->playGeneration = generation; + trace->playFrames = frames; + const size_t bytes = frames * trace->playFormat.channelCount * audioSampleBytes(trace->playFormat.sampleFormat); - const size_t copy = bytes < sizeof(trace->playData) ? + const size_t copy = bytes < sizeof(trace->playData) ? bytes : sizeof(trace->playData); memset(trace->playData, 0, sizeof(trace->playData)); memcpy(trace->playData, data, copy); - trace->playClockValid = clock != NULL; + trace->playClockValid = clock != NULL; if (clock) - trace->playClock = *clock; + trace->playClock = *clock; pthread_mutex_unlock(&trace->mutex); traceEvent(trace, TRACE_PLAY_DATA, generation); atomic_fetch_add_explicit(&trace->playDataCount, 1, diff --git a/client/transports/LGMP/audio.c b/client/transports/LGMP/audio.c index bec638e6..1076d42b 100644 --- a/client/transports/LGMP/audio.c +++ b/client/transports/LGMP/audio.c @@ -1486,13 +1486,15 @@ static void dispatchPlaybackData(LGMPAudio * audio, return; } - const uint32_t generation = audio->playback.generation.local; - const LG_AudioEventOps * events = + const uint32_t generation = + audio->playback.generation.local; + const LG_AudioEventOps * events = audio->attached ? audio->events : NULL; - void * opaque = audio->eventOpaque; - const bool discontinuity = sequenceGap || - audio->playbackDiscontinuity; - const bool clockValid = + void * opaque = audio->eventOpaque; + const bool discontinuity = sequenceGap || + audio->playbackDiscontinuity || + (message->flags & KVMFR_AUDIO_MESSAGE_DISCONTINUITY) != 0; + const bool clockValid = (message->flags & KVMFR_AUDIO_MESSAGE_CLOCK_VALID) != 0; if (!events || !events->playbackData) audio->playbackDiscontinuity = true; @@ -1506,7 +1508,7 @@ static void dispatchPlaybackData(LGMPAudio * audio, clockFromMessage(message, discontinuity); events->playbackData(opaque, generation, silent ? l_silence : payload, message->frames, - clockValid ? &clock : NULL); + clockValid ? &clock : NULL, discontinuity); } LG_UNLOCK(audio->eventLock); } diff --git a/client/transports/Network/audio.c b/client/transports/Network/audio.c index c45d58c7..e04d972d 100644 --- a/client/transports/Network/audio.c +++ b/client/transports/Network/audio.c @@ -46,7 +46,6 @@ #define NETWORK_AUDIO_CHANNEL_MASK UINT32_C(0x0003FFFF) #define NETWORK_AUDIO_PLAYBACK_STREAM_ID 1U #define NETWORK_AUDIO_CAPTURE_STREAM_ID 2U -#define NETWORK_AUDIO_CLOCK_CACHE_SIZE 16U #define NETWORK_AUDIO_DATAGRAM_PAYLOAD 1136U #define NETWORK_AUDIO_DATAGRAM_PCM_BYTES \ (NETWORK_AUDIO_DATAGRAM_PAYLOAD - LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE) @@ -55,8 +54,7 @@ (LG_NET_AUDIO_SUBSCRIBE_LOW_LATENCY | \ LG_NET_AUDIO_SUBSCRIBE_EXCLUSIVE_CAPTURE | \ LG_NET_AUDIO_SUBSCRIBE_CLOCK_FEEDBACK | \ - LG_NET_AUDIO_SUBSCRIBE_DATAGRAM_PCM | \ - LG_NET_AUDIO_SUBSCRIBE_RELIABLE_PCM) + LG_NET_AUDIO_SUBSCRIBE_DATAGRAM_PCM) #define NETWORK_AUDIO_DIRECTIONS \ (LG_NET_AUDIO_DIRECTIONS_PLAYBACK | LG_NET_AUDIO_DIRECTIONS_CAPTURE) @@ -76,12 +74,6 @@ enum NetworkAudioPhase NETWORK_AUDIO_RELEASING, }; -struct NetworkAudioClockEntry -{ - bool valid; - LGNetAudioClockState state; -}; - struct NetworkAudioStream { uint32_t streamID; @@ -110,8 +102,6 @@ struct NetworkAudioStream bool packetKnown; uint64_t lastPacketID; bool discontinuity; - unsigned clockInsert; - struct NetworkAudioClockEntry clocks[NETWORK_AUDIO_CLOCK_CACHE_SIZE]; }; struct NetworkAudio @@ -463,12 +453,6 @@ static void networkAudioEndEvents(NetworkAudio * audio) LG_UNLOCK(audio->eventLock); } -static void networkAudioClearClockNL(struct NetworkAudioStream * stream) -{ - stream->clockInsert = 0; - memset(stream->clocks, 0, sizeof(stream->clocks)); -} - static void networkAudioClearStreamNL(struct NetworkAudioStream * stream) { const uint32_t streamID = stream->streamID; @@ -601,7 +585,7 @@ static LG_TransportStatus networkAudioSendDatagramNL(NetworkAudio * audio, { return networkConnectionSendDatagram(audio->connection, LG_NET_SERVICE_AUDIO, messageType, audio->serviceVersion, 0, - audio->subscriptionEpoch, requestID, payload, payloadLength, 0, false); + audio->subscriptionEpoch, requestID, payload, payloadLength, 0, true); } static bool networkAudioSendSubscribeNL(NetworkAudio * audio) @@ -962,21 +946,12 @@ static void networkAudioDetach(void * opaque) LG_UNLOCK(audio->eventLock); } -static bool networkAudioSendClockAndDataNL(NetworkAudio * audio, - const uint8_t * clockPayload, const uint8_t * dataPayload, - uint32_t dataPayloadLength) +static LG_TransportStatus networkAudioSendPCMNL(NetworkAudio * audio, + uint16_t messageType, const void * payload, uint32_t payloadLength) { - /* CLOCK_STATE and PCM are an ordered pair. Keep both on the same reliable - * stream until a future protocol version provides atomic datagram framing. */ - if (!(audio->grantedRoles & LG_NET_AUDIO_ROLE_RELIABLE_PCM)) - return false; - - return networkAudioSendReliableNL(audio, - LG_NET_AUDIO_MESSAGE_CLOCK_STATE, 0, 0, clockPayload, - LG_NET_AUDIO_CLOCK_STATE_WIRE_SIZE) == LG_TRANSPORT_OK && - networkAudioSendReliableNL(audio, - LG_NET_AUDIO_MESSAGE_CAPTURE_DATA, 0, 0, dataPayload, - dataPayloadLength) == LG_TRANSPORT_OK; + return (audio->grantedRoles & LG_NET_AUDIO_ROLE_DATAGRAM_PCM) ? + networkAudioSendDatagramNL(audio, messageType, 0, + payload, payloadLength) : LG_TRANSPORT_UNAVAILABLE; } static bool networkAudioRecordData(void * opaque, uint32_t generation, @@ -986,10 +961,9 @@ static bool networkAudioRecordData(void * opaque, uint32_t generation, if (!generation || !data || !frames) return false; - uint64_t rateQ32 = 0; - if (sourceClock && - !networkAudioRateQ32(sourceClock->rate, true, &rateQ32)) - return false; + uint64_t rateQ32 = 0; + const bool sourceClockValid = sourceClock && sourceClock->time > 0 && + networkAudioRateQ32(sourceClock->rate, false, &rateQ32); LG_LOCK(audio->stateLock); struct NetworkAudioStream * stream = &audio->capture; @@ -1044,11 +1018,11 @@ static bool networkAudioRecordData(void * opaque, uint32_t generation, return false; } - bool result = true; - bool discontinuity = stream->discontinuity || + bool result = true; + bool discontinuity = stream->discontinuity || (sourceClock && sourceClock->discontinuity); - size_t offset = 0; - const uint8_t * pcm = data; + size_t offset = 0; + const uint8_t * pcm = data; while (offset < frames) { size_t chunkFrames = frames - offset; @@ -1058,10 +1032,11 @@ static bool networkAudioRecordData(void * opaque, uint32_t generation, audio->capturePacketID = networkAudioNext64(audio->capturePacketID); - const uint64_t packetID = audio->capturePacketID; - uint64_t position = firstPosition + offset; - int64_t timeNs = sourceClock ? sourceClock->time : 0; - if (sourceClock && offset) + const uint64_t packetID = audio->capturePacketID; + const uint64_t position = firstPosition + offset; + int64_t timeNs = sourceClockValid ? sourceClock->time : 0; + bool clockValid = sourceClockValid; + if (clockValid && offset) { const long double elapsed = (long double)offset * 1000000000.0L / @@ -1070,29 +1045,13 @@ static bool networkAudioRecordData(void * opaque, uint32_t generation, if (adjusted >= (long double)INT64_MIN && adjusted <= (long double)INT64_MAX) timeNs = (int64_t)adjusted; + else + { + clockValid = false; + timeNs = 0; + } } - uint64_t timestampNs = timeNs > 0 ? (uint64_t)timeNs : nanotime(); - if (!timestampNs) - timestampNs = 1; - - const LGNetAudioClockState clock = - { - .streamID = stream->streamID, - .direction = stream->direction, - .flags = sourceClock ? - LG_NET_AUDIO_CLOCK_VALID | - (sourceClock->stable ? LG_NET_AUDIO_CLOCK_STABLE : 0) : 0, - .formatEpoch = stream->wireFormat.formatEpoch, - .subscriptionEpoch = audio->subscriptionEpoch, - .packetID = packetID, - .framePosition = position, - .timeNs = timeNs, - .rateQ32 = rateQ32, - .targetRateQ32 = 0, - .queuedFrames = 0, - .driftPpm = 0, - }; - const LGNetAudioData packet = + LGNetAudioData packet = { .streamID = stream->streamID, .direction = stream->direction, @@ -1100,31 +1059,50 @@ static bool networkAudioRecordData(void * opaque, uint32_t generation, LG_NET_AUDIO_DATA_DISCONTINUITY : 0, .formatEpoch = stream->wireFormat.formatEpoch, .packetID = packetID, - .timestampNs = timestampNs, + .timestampNs = timeNs > 0 ? (uint64_t)timeNs : 0, .startFrame = position, + .rateQ32 = clockValid ? rateQ32 : 0, .frameCount = (uint32_t)chunkFrames, .dataLength = (uint32_t)chunkBytes, .data = pcm + offset * frameSize, }; - uint8_t clockPayload[LG_NET_AUDIO_CLOCK_STATE_WIRE_SIZE]; + if (clockValid) + { + packet.flags |= LG_NET_AUDIO_DATA_CLOCK_VALID; + if (sourceClock->stable) + packet.flags |= LG_NET_AUDIO_DATA_CLOCK_STABLE; + } + const size_t dataPayloadLength = LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE + chunkBytes; - if (!lgNetAudioClockStateEncode(clockPayload, - sizeof(clockPayload), &clock) || - !lgNetAudioDataEncode(payload, dataPayloadLength, &packet) || - !networkAudioSendClockAndDataNL(audio, clockPayload, payload, - (uint32_t)dataPayloadLength)) + if (!lgNetAudioDataEncode(payload, dataPayloadLength, &packet)) { result = false; break; } - discontinuity = false; + const LG_TransportStatus status = networkAudioSendPCMNL( + audio, LG_NET_AUDIO_MESSAGE_CAPTURE_DATA, + payload, (uint32_t)dataPayloadLength); + if (status != LG_TRANSPORT_OK) + { + if (status == LG_TRANSPORT_UNAVAILABLE) + { + discontinuity = true; + offset += chunkFrames; + continue; + } + + result = false; + break; + } + + discontinuity = false; offset += chunkFrames; } audio->captureFramePosition = firstPosition + offset; - stream->discontinuity = !result; + stream->discontinuity = discontinuity || !result; free(payload); LG_UNLOCK(audio->stateLock); return result; @@ -1173,16 +1151,12 @@ static bool networkAudioClockFeedback(void * opaque, uint32_t generation, const bool encoded = lgNetAudioClockStateEncode( payload, sizeof(payload), &feedback); bool result = false; - if (encoded && - (audio->grantedRoles & LG_NET_AUDIO_ROLE_DATAGRAM_PCM)) - result = networkAudioSendDatagramNL(audio, - LG_NET_AUDIO_MESSAGE_CLOCK_STATE, 0, - payload, sizeof(payload)) == LG_TRANSPORT_OK; - if (encoded && !result && - (audio->grantedRoles & LG_NET_AUDIO_ROLE_RELIABLE_PCM)) - result = networkAudioSendReliableNL(audio, - LG_NET_AUDIO_MESSAGE_CLOCK_STATE, 0, 0, - payload, sizeof(payload)) == LG_TRANSPORT_OK; + if (encoded) + { + const LG_TransportStatus status = networkAudioSendDatagramNL(audio, + LG_NET_AUDIO_MESSAGE_CLOCK_STATE, 0, payload, sizeof(payload)); + result = status == LG_TRANSPORT_OK; + } LG_UNLOCK(audio->stateLock); return result; } @@ -1322,8 +1296,8 @@ static bool networkAudioHandleGrant(NetworkAudio * audio, if (!grant.maxPacketFrames || (grant.grantedRoles & ~audio->requestedRoles) || grant.maxPacketFrames > NETWORK_AUDIO_MAX_PACKET_FRAMES || - (grant.grantedRoles & LG_NET_AUDIO_ROLE_DATAGRAM_PCM && - !datagramAvailable) || + !(grant.grantedRoles & LG_NET_AUDIO_ROLE_DATAGRAM_PCM) || + !datagramAvailable || (grant.grantedRoles & LG_NET_AUDIO_ROLE_CAPTURE && !(grant.grantedRoles & LG_NET_AUDIO_ROLE_EXCLUSIVE_CAPTURE)) || (grant.grantedRoles & LG_NET_AUDIO_ROLE_CLOCK_FEEDBACK && @@ -1430,7 +1404,6 @@ static bool networkAudioHandleStart(NetworkAudio * audio, stream->packetKnown = false; stream->lastPacketID = 0; stream->discontinuity = true; - networkAudioClearClockNL(stream); networkAudioInvalidateBarrierNL(audio, stream); const LG_AudioEventOps * events = audio->attached ? audio->events : NULL; @@ -1504,7 +1477,6 @@ static bool networkAudioHandleStop(NetworkAudio * audio, stream->volumeValid = false; stream->muteValid = false; stream->discontinuity = true; - networkAudioClearClockNL(stream); networkAudioInvalidateBarrierNL(audio, stream); LG_UNLOCK(audio->stateLock); networkAudioCallEventsNL(audio); @@ -1774,7 +1746,6 @@ static bool networkAudioHandleStatus(NetworkAudio * audio, stream->volumeValid = false; stream->muteValid = false; stream->discontinuity = true; - networkAudioClearClockNL(stream); } stream->stateSequence = state.stateSequence; networkAudioInvalidateBarrierNL(audio, stream); @@ -1861,62 +1832,6 @@ static bool networkAudioHandleBarrier(NetworkAudio * audio, return true; } -static struct NetworkAudioClockEntry * networkAudioFindClockNL( - struct NetworkAudioStream * stream, uint64_t packetID) -{ - for (unsigned i = 0; i < NETWORK_AUDIO_CLOCK_CACHE_SIZE; ++i) - if (stream->clocks[i].valid && - stream->clocks[i].state.packetID == packetID) - return &stream->clocks[i]; - return NULL; -} - -static bool networkAudioHandleClock(NetworkAudio * audio, - const LGNetPacketView * packet) -{ - LGNetAudioClockState state; - if (lgNetAudioClockStateDecode(&state, - packet->payload, packet->payloadLength) != LG_NET_PARSE_OK) - return false; - - LG_LOCK(audio->stateLock); - struct NetworkAudioStream * stream = networkAudioStreamNL( - audio, state.streamID, state.direction); - if (!stream || stream != &audio->playback) - { - LG_UNLOCK(audio->stateLock); - return false; - } - if (!networkAudioSubscriptionCurrentNL(audio, packet) || - !networkAudioStreamRequiredNL(audio, stream) || - state.subscriptionEpoch != audio->subscriptionEpoch) - { - LG_UNLOCK(audio->stateLock); - return true; - } - if (!stream->active || !stream->formatValid || - state.formatEpoch != stream->wireFormat.formatEpoch || - (stream->packetKnown && state.packetID <= stream->lastPacketID)) - { - LG_UNLOCK(audio->stateLock); - return true; - } - - struct NetworkAudioClockEntry * entry = networkAudioFindClockNL( - stream, state.packetID); - if (!entry) - { - entry = &stream->clocks[ - stream->clockInsert++ % NETWORK_AUDIO_CLOCK_CACHE_SIZE]; - if (entry->valid) - stream->discontinuity = true; - } - entry->valid = true; - entry->state = state; - LG_UNLOCK(audio->stateLock); - return true; -} - static bool networkAudioHandlePlaybackData(NetworkAudio * audio, const LGNetPacketView * packet) { @@ -1971,32 +1886,23 @@ static bool networkAudioHandlePlaybackData(NetworkAudio * audio, return true; } - struct NetworkAudioClockEntry * entry = networkAudioFindClockNL( - stream, data.packetID); - LG_AudioClock clock; - const LG_AudioClock * clockPtr = NULL; - if (entry && (entry->state.flags & LG_NET_AUDIO_CLOCK_VALID)) + const bool discontinuity = stream->discontinuity || sequenceGap || + (data.flags & LG_NET_AUDIO_DATA_DISCONTINUITY) != 0; + LG_AudioClock clock; + const LG_AudioClock * clockPtr = NULL; + if (data.flags & LG_NET_AUDIO_DATA_CLOCK_VALID) { clock = (LG_AudioClock) { - .position = entry->state.framePosition, - .time = entry->state.timeNs, - .rate = networkAudioRate(entry->state.rateQ32), + .position = data.startFrame, + .time = (int64_t)data.timestampNs, + .rate = networkAudioRate(data.rateQ32), .stable = - (entry->state.flags & LG_NET_AUDIO_CLOCK_STABLE) != 0, - .discontinuity = stream->discontinuity || sequenceGap || - (data.flags & LG_NET_AUDIO_DATA_DISCONTINUITY), + (data.flags & LG_NET_AUDIO_DATA_CLOCK_STABLE) != 0, + .discontinuity = discontinuity, }; - clockPtr = &clock; - stream->discontinuity = false; + clockPtr = &clock; } - else - stream->discontinuity = true; - if (entry) - entry->valid = false; - if (data.flags & LG_NET_AUDIO_DATA_END_OF_STREAM) - stream->discontinuity = true; - const size_t frameSize = (size_t)networkAudioSampleSize( stream->wireFormat.sampleFormat) * stream->wireFormat.channels; const size_t bytes = frameSize * data.frameCount; @@ -2013,11 +1919,13 @@ static bool networkAudioHandlePlaybackData(NetworkAudio * audio, const LG_AudioEventOps * events = audio->attached ? audio->events : NULL; void * opaque = audio->eventOpaque; + stream->discontinuity = !events || !events->playbackData || + (data.flags & LG_NET_AUDIO_DATA_END_OF_STREAM) != 0; LG_UNLOCK(audio->stateLock); networkAudioCallEventsNL(audio); if (events && events->playbackData) events->playbackData(opaque, generation, pcm, - data.frameCount, clockPtr); + data.frameCount, clockPtr, discontinuity); networkAudioEndEvents(audio); return true; } @@ -2078,7 +1986,7 @@ static bool networkAudioHandlePacket(NetworkAudio * audio, audio, packet, &audio->playback); case LG_NET_AUDIO_MESSAGE_PLAYBACK_DATA: - return networkAudioHandlePlaybackData(audio, packet); + return datagram && networkAudioHandlePlaybackData(audio, packet); case LG_NET_AUDIO_MESSAGE_PLAYBACK_VOLUME: return !datagram && networkAudioHandleVolume( @@ -2107,9 +2015,6 @@ static bool networkAudioHandlePacket(NetworkAudio * audio, case LG_NET_AUDIO_MESSAGE_STATE_BARRIER: return !datagram && networkAudioHandleBarrier(audio, packet); - case LG_NET_AUDIO_MESSAGE_CLOCK_STATE: - return networkAudioHandleClock(audio, packet); - default: return false; } @@ -2149,7 +2054,6 @@ bool networkAudioCreate(NetworkAudio ** result, LG_NET_AUDIO_ROLE_CAPTURE | LG_NET_AUDIO_ROLE_CLOCK_FEEDBACK | LG_NET_AUDIO_ROLE_DATAGRAM_PCM | - LG_NET_AUDIO_ROLE_RELIABLE_PCM | LG_NET_AUDIO_ROLE_EXCLUSIVE_CAPTURE; audio->endpointGeneration = 1; audio->generationSerial = 1; @@ -2191,6 +2095,8 @@ void networkAudioConnectionState(NetworkAudio * audio, bool connected) const bool serviceAvailable = connected && networkConnectionServiceAvailable(audio->connection, LG_NET_SERVICE_AUDIO, LG_NET_CAPABILITY_RELIABLE) && + networkConnectionServiceAvailable(audio->connection, + LG_NET_SERVICE_AUDIO, LG_NET_CAPABILITY_DATAGRAM) && networkConnectionServiceVersion(audio->connection, LG_NET_SERVICE_AUDIO, &serviceVersion) && networkAudioVersionSupported(serviceVersion); diff --git a/client/transports/SPICE/audio.c b/client/transports/SPICE/audio.c index df74f780..a396d659 100644 --- a/client/transports/SPICE/audio.c +++ b/client/transports/SPICE/audio.c @@ -845,7 +845,7 @@ void spiceAudio_playbackData(SpiceAudio * audio, if (target.events->playbackData && playbackEventCurrent(audio, &target, generation, true)) target.events->playbackData(target.opaque, generation, - data, frames, &clock); + data, frames, &clock, false); endEvent(audio); } diff --git a/client/transports/SPICE/audio_usb.c b/client/transports/SPICE/audio_usb.c index 52d94096..ea73a4c7 100644 --- a/client/transports/SPICE/audio_usb.c +++ b/client/transports/SPICE/audio_usb.c @@ -592,7 +592,7 @@ static void usbPlaybackData( { const LG_AudioClock clock = makePlaybackClock(state, position); events->playbackData( - target, generation, data, frames, &clock); + target, generation, data, frames, &clock, false); } } endCallback(&frame, &l_playbackFrames, diff --git a/idd/LGIdd/transport/network/CNetworkAudio.cpp b/idd/LGIdd/transport/network/CNetworkAudio.cpp index f26cd01d..b9c28960 100644 --- a/idd/LGIdd/transport/network/CNetworkAudio.cpp +++ b/idd/LGIdd/transport/network/CNetworkAudio.cpp @@ -36,17 +36,19 @@ namespace { - constexpr uint32_t PLAYBACK_STREAM_ID = 1; - constexpr uint32_t CAPTURE_STREAM_ID = 2; - constexpr uint64_t AUDIO_LEASE_MS = 2000; - constexpr uint64_t AUTHORITY_STALL_MS = 250; - constexpr size_t CLOCK_CACHE_SIZE = 8; - constexpr size_t RECENT_PACKET_SIZE = 16; + constexpr uint32_t PLAYBACK_STREAM_ID = 1; + constexpr uint32_t CAPTURE_STREAM_ID = 2; + constexpr uint64_t AUDIO_LEASE_MS = 2000; + constexpr size_t RECENT_PACKET_SIZE = 16; + constexpr uint32_t MIN_DATAGRAM_PAYLOAD = + LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE + + static_cast( + AUDIO_STREAM_MAX_CHANNELS * sizeof(uint32_t)); constexpr LGNetAudioRoles DIRECTION_ROLES = LG_NET_AUDIO_ROLE_PLAYBACK | LG_NET_AUDIO_ROLE_CAPTURE; - constexpr LGNetAudioRoles DELIVERY_ROLES = - LG_NET_AUDIO_ROLE_DATAGRAM_PCM | LG_NET_AUDIO_ROLE_RELIABLE_PCM; + constexpr LGNetAudioRoles DELIVERY_ROLES = + LG_NET_AUDIO_ROLE_DATAGRAM_PCM; bool SamePeer(const NetworkAudioPeer& left, const NetworkAudioPeer& right) @@ -71,12 +73,6 @@ namespace return serial ? serial : 1; } - uint64_t MonotonicNs() - { - const uint64_t result = GetTickCount64() * UINT64_C(1000000); - return result ? result : 1; - } - uint32_t ChannelCount(uint32_t mask) { uint32_t count = 0; @@ -251,19 +247,6 @@ struct CNetworkAudio::Impl BARRIER, }; - enum class BlockDisposition - { - STALE, - RETRY, - WAIT, - }; - - struct ClockRecord - { - bool valid = false; - LGNetAudioClockState state = {}; - }; - struct PacketRecord { bool valid = false; @@ -275,47 +258,45 @@ struct CNetworkAudio::Impl { NetworkAudioPeer peer; NetworkAudioPeerInfo info; - uint64_t serial = 0; - bool hasRequest = false; - bool subscribed = false; - bool expiring = false; - bool feedbackAuthority = false; - bool feedbackRevoked = false; - bool captureOwner = false; - bool captureRevoked = false; - bool ready = false; - bool awaitingAck = false; - bool playbackGap = true; - bool captureGap = true; - bool resetPlayback = false; - bool resetCapture = false; - bool capturePacketKnown = false; - uint64_t deadline = 0; - uint64_t requestID = 0; - uint64_t subscriptionEpoch = 0; - uint64_t sendReadySerial = 0; - LGNetAudioSubscribe request = {}; - LGNetAudioRoles requestedRoles = 0; - LGNetAudioRoles grantedRoles = 0; - LGNetAudioGrantStatus grantStatus = LG_NET_AUDIO_GRANT_BUSY; + uint64_t serial = 0; + bool hasRequest = false; + bool subscribed = false; + bool expiring = false; + bool feedbackAuthority = false; + bool feedbackRevoked = false; + bool captureOwner = false; + bool captureRevoked = false; + bool ready = false; + bool awaitingAck = false; + bool playbackGap = true; + bool captureGap = true; + bool resetPlayback = false; + bool resetCapture = false; + bool capturePacketKnown = false; + uint64_t deadline = 0; + uint64_t requestID = 0; + uint64_t subscriptionEpoch = 0; + uint64_t lastFeedbackPacketID = 0; + LGNetAudioSubscribe request = {}; + LGNetAudioRoles requestedRoles = 0; + LGNetAudioRoles grantedRoles = 0; + LGNetAudioGrantStatus grantStatus = LG_NET_AUDIO_GRANT_BUSY; - uint64_t playbackSequence = 0; - uint64_t captureSequence = 0; - uint64_t playbackPacketID = 0; - uint64_t capturePacketID = 0; - uint64_t barrierID = 0; - uint64_t playbackBarrierID = 0; - uint64_t captureBarrierID = 0; - uint8_t pendingBarriers = 0; - uint64_t announcedPlayback = 0; - uint64_t announcedCapture = 0; - uint64_t pumpToken = 0; - PumpStage pumpStage = PumpStage::DONE; - bool pumpGrant = false; - bool pumpInFlight = false; + uint64_t playbackSequence = 0; + uint64_t captureSequence = 0; + uint64_t playbackPacketID = 0; + uint64_t capturePacketID = 0; + uint64_t barrierID = 0; + uint64_t playbackBarrierID = 0; + uint64_t captureBarrierID = 0; + uint8_t pendingBarriers = 0; + uint64_t announcedPlayback = 0; + uint64_t announcedCapture = 0; + uint64_t pumpToken = 0; + PumpStage pumpStage = PumpStage::DONE; + bool pumpGrant = false; + bool pumpInFlight = false; - std::array captureClock = {}; - size_t captureClockNext = 0; std::array playbackPacket = {}; size_t playbackPacketNext = 0; }; @@ -358,8 +339,6 @@ struct CNetworkAudio::Impl uint32_t frameOffset = 0; uint32_t chunkFrames = 0; uint64_t packetID = 0; - bool clockSent = false; - uint64_t blockedSince = 0; }; struct PendingCapture @@ -386,9 +365,6 @@ struct CNetworkAudio::Impl bool m_activationCommanded = false; bool m_failed = false; bool m_failureNotified = false; - bool m_playbackReadyPending = false; - bool m_playbackWaiting = false; - NetworkAudioPeer m_playbackWaitingPeer; AudioEndpointStatus m_state; std::vector m_peers; uint64_t m_nextPeerSerial = 0; @@ -432,15 +408,9 @@ struct CNetworkAudio::Impl static bool PeerCanDeliver(const PeerState& peer) { - const bool reliable = - (peer.requestedRoles & LG_NET_AUDIO_ROLE_RELIABLE_PCM) && - (peer.info.capabilityFlags & LG_NET_CAPABILITY_RELIABLE) && - peer.info.maxReliablePayload >= LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE; - const bool datagram = - (peer.requestedRoles & LG_NET_AUDIO_ROLE_DATAGRAM_PCM) && + return (peer.requestedRoles & LG_NET_AUDIO_ROLE_DATAGRAM_PCM) && (peer.info.capabilityFlags & LG_NET_CAPABILITY_DATAGRAM) && - peer.info.maxDatagramPayload >= LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE; - return reliable || datagram; + peer.info.maxDatagramPayload >= MIN_DATAGRAM_PAYLOAD; } bool CaptureCandidateNL(const PeerState& peer) const @@ -462,16 +432,7 @@ struct CNetworkAudio::Impl LGNetAudioRoles DeliveryRolesNL(const PeerState& peer) const { - LGNetAudioRoles result = 0; - if ((peer.requestedRoles & LG_NET_AUDIO_ROLE_RELIABLE_PCM) && - (peer.info.capabilityFlags & LG_NET_CAPABILITY_RELIABLE) && - peer.info.maxReliablePayload >= LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE) - result |= LG_NET_AUDIO_ROLE_RELIABLE_PCM; - if ((peer.requestedRoles & LG_NET_AUDIO_ROLE_DATAGRAM_PCM) && - (peer.info.capabilityFlags & LG_NET_CAPABILITY_DATAGRAM) && - peer.info.maxDatagramPayload >= LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE) - result |= LG_NET_AUDIO_ROLE_DATAGRAM_PCM; - return result; + return PeerCanDeliver(peer) ? LG_NET_AUDIO_ROLE_DATAGRAM_PCM : 0; } LGNetAudioRoles DesiredRolesNL(const PeerState& peer) const @@ -510,40 +471,34 @@ struct CNetworkAudio::Impl { if (m_capture.valid && SamePeer(m_capture.peer, peer.peer)) m_capture.valid = false; - if (m_playbackWaiting && SamePeer(m_playbackWaitingPeer, peer.peer)) - { - m_playbackWaiting = false; - m_playbackReadyPending = true; - } - peer.hasRequest = false; - peer.subscribed = false; - peer.expiring = false; - peer.feedbackAuthority = false; - peer.feedbackRevoked = false; - peer.captureOwner = false; - peer.captureRevoked = false; - peer.ready = false; - peer.awaitingAck = false; - peer.pendingBarriers = 0; - peer.deadline = 0; - peer.requestID = 0; - peer.request = {}; - peer.requestedRoles = 0; - peer.grantedRoles = 0; - peer.pumpGrant = false; - peer.pumpStage = PumpStage::DONE; - peer.pumpToken = NextSerial(peer.pumpToken); - peer.playbackGap = true; - peer.captureGap = true; - peer.resetPlayback = false; - peer.resetCapture = false; - peer.capturePacketKnown = false; - peer.capturePacketID = 0; - peer.captureClock = {}; - peer.captureClockNext = 0; - peer.playbackPacket = {}; - peer.playbackPacketNext = 0; + peer.hasRequest = false; + peer.subscribed = false; + peer.expiring = false; + peer.feedbackAuthority = false; + peer.feedbackRevoked = false; + peer.captureOwner = false; + peer.captureRevoked = false; + peer.ready = false; + peer.awaitingAck = false; + peer.pendingBarriers = 0; + peer.deadline = 0; + peer.requestID = 0; + peer.request = {}; + peer.requestedRoles = 0; + peer.grantedRoles = 0; + peer.pumpGrant = false; + peer.pumpStage = PumpStage::DONE; + peer.pumpToken = NextSerial(peer.pumpToken); + peer.playbackGap = true; + peer.captureGap = true; + peer.resetPlayback = false; + peer.resetCapture = false; + peer.capturePacketKnown = false; + peer.capturePacketID = 0; + peer.lastFeedbackPacketID = 0; + peer.playbackPacket = {}; + peer.playbackPacketNext = 0; } void RebalanceRolesNL(uint64_t forcedSerial = 0); @@ -563,6 +518,8 @@ struct CNetworkAudio::Impl bool datagram, uint16_t messageType, LGNetEnvelopeFlags flags, uint64_t componentEpoch, uint64_t requestID, const void * payload, uint32_t payloadLength); + static bool DatagramPCM( + const PlaybackTarget& target, uint32_t payloadLength); NetworkAudioSendResult SendPCM(const PlaybackTarget& target, uint16_t messageType, const void * payload, uint32_t payloadLength); @@ -576,14 +533,9 @@ struct CNetworkAudio::Impl const LGNetPacketView& packet); void ProcessActivation(); - void NotifyPlaybackReady(); void NotifyFailure(); void MarkFailed(const char * reason); void RetryCapture(); - void RevokeStalledAuthority(uint64_t now); - uint64_t PlaybackReadySerial(const PlaybackTarget& target); - BlockDisposition HandlePlaybackBlock( - const PlaybackTarget& target, uint64_t readySerial); AudioResult BeginPlayback(const AudioPacket& packet, size_t size); AudioResult ContinuePlayback(const uint8_t * data); @@ -657,18 +609,17 @@ void CNetworkAudio::Impl::RebalanceRolesNL(uint64_t forcedSerial) if (m_capture.valid && m_capture.peerSerial == peer.serial && SamePeer(m_capture.peer, peer.peer)) m_capture.valid = false; - peer.subscriptionEpoch = NextSerial(m_nextEpoch); - m_nextEpoch = peer.subscriptionEpoch; - peer.grantedRoles = desired; - peer.grantStatus = status; - peer.playbackGap = true; - peer.captureGap = true; - peer.capturePacketKnown = false; - peer.capturePacketID = 0; - peer.captureClock = {}; - peer.captureClockNext = 0; - peer.playbackPacket = {}; - peer.playbackPacketNext = 0; + peer.subscriptionEpoch = NextSerial(m_nextEpoch); + m_nextEpoch = peer.subscriptionEpoch; + peer.grantedRoles = desired; + peer.grantStatus = status; + peer.playbackGap = true; + peer.captureGap = true; + peer.capturePacketKnown = false; + peer.capturePacketID = 0; + peer.lastFeedbackPacketID = 0; + peer.playbackPacket = {}; + peer.playbackPacketNext = 0; SchedulePumpNL(peer, true); } } @@ -1214,13 +1165,6 @@ void CNetworkAudio::Impl::PumpPeer(const NetworkAudioPeer& key) { if (peer->subscribed && !peer->pendingBarriers) peer->ready = true; - if (m_playbackWaiting && - SamePeer(m_playbackWaitingPeer, peer->peer) && - (peer->ready || !peer->subscribed)) - { - m_playbackWaiting = false; - m_playbackReadyPending = true; - } return; } peer->pumpInFlight = true; @@ -1238,13 +1182,6 @@ void CNetworkAudio::Impl::PumpPeer(const NetworkAudioPeer& key) { if (peer->subscribed && !peer->pendingBarriers) peer->ready = true; - if (m_playbackWaiting && - SamePeer(m_playbackWaitingPeer, peer->peer) && - (peer->ready || !peer->subscribed)) - { - m_playbackWaiting = false; - m_playbackReadyPending = true; - } } return; } @@ -1366,7 +1303,31 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleSubscribe( const LGNetAudioRoles requestedRoles = lgNetAudioSubscribeRoles(&subscribe); - if (peer.hasRequest && peer.subscribed && !peer.expiring && + bool superseded = false; + bool displaced = false; + for (PeerState& candidate : m_peers) + { + if (&candidate == &peer || !candidate.hasRequest || + candidate.expiring || + candidate.request.subscriberID != subscribe.subscriberID) + continue; + + if (candidate.serial > peer.serial) + { + superseded = true; + continue; + } + + if (m_capture.valid && m_capture.peerSerial == candidate.serial && + SamePeer(m_capture.peer, candidate.peer)) + m_capture.valid = false; + candidate.expiring = true; + candidate.deadline = 0; + displaced = true; + } + + if (!displaced && !superseded && peer.hasRequest && peer.subscribed && + !peer.expiring && peer.requestID == packet.envelope.requestID && peer.requestedRoles == requestedRoles && SameSubscription(peer.request, subscribe)) @@ -1385,16 +1346,16 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleSubscribe( return NetworkAudioPacketResult::HANDLED; } - peer.hasRequest = true; - peer.expiring = false; - peer.feedbackRevoked = false; - peer.captureRevoked = false; - peer.request = subscribe; - peer.requestID = packet.envelope.requestID; - peer.requestedRoles = requestedRoles; - peer.deadline = now + AUDIO_LEASE_MS; - peer.captureClock = {}; - peer.playbackPacket = {}; + peer.hasRequest = true; + peer.expiring = superseded; + peer.feedbackRevoked = false; + peer.captureRevoked = false; + peer.request = subscribe; + peer.requestID = packet.envelope.requestID; + peer.requestedRoles = requestedRoles; + peer.deadline = superseded ? 0 : now + AUDIO_LEASE_MS; + peer.lastFeedbackPacketID = 0; + peer.playbackPacket = {}; RebalanceRolesNL(peer.serial); return NetworkAudioPacketResult::HANDLED; } @@ -1407,49 +1368,12 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleClock( packet.payloadLength) != LG_NET_PARSE_OK) return NetworkAudioPacketResult::INVALID; - if (clock.direction == LG_NET_AUDIO_DIRECTION_CAPTURE) - { - CSRWExclusiveLock lock(m_lock); - PeerState * peer = FindPeerNL(key); - if (!peer || !peer->hasRequest || - peer->subscriptionEpoch != packet.envelope.componentEpoch) - return NetworkAudioPacketResult::IGNORED; - if (clock.subscriptionEpoch != peer->subscriptionEpoch || - clock.streamID != CAPTURE_STREAM_ID) - return NetworkAudioPacketResult::INVALID; - - peer->deadline = GetTickCount64() + AUDIO_LEASE_MS; - LGNetAudioFormat format = {}; - if (!peer->captureOwner || - !peer->ready || !CaptureCurrentNL(*peer) || - !WireFormatNL(*peer, false, format) || - clock.formatEpoch != format.formatEpoch) - return NetworkAudioPacketResult::IGNORED; - if (peer->capturePacketKnown && - clock.packetID <= peer->capturePacketID) - return NetworkAudioPacketResult::IGNORED; - - ClockRecord * record = nullptr; - for (ClockRecord& candidate : peer->captureClock) - if (candidate.valid && candidate.state.packetID == clock.packetID && - candidate.state.formatEpoch == clock.formatEpoch) - { - record = &candidate; - break; - } - if (!record) - record = &peer->captureClock[ - peer->captureClockNext++ % CLOCK_CACHE_SIZE]; - record->valid = true; - record->state = clock; - return NetworkAudioPacketResult::HANDLED; - } - if (clock.direction != LG_NET_AUDIO_DIRECTION_PLAYBACK || clock.streamID != PLAYBACK_STREAM_ID) return NetworkAudioPacketResult::INVALID; - CSRWSharedLock callbackLock(m_targetCallbackLock); + CSRWExclusiveLock callbackLock(m_targetCallbackLock); + IAudioTarget * target = nullptr; AudioClockFeedback feedback = {}; uint64_t peerSerial = 0; @@ -1462,7 +1386,6 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleClock( return NetworkAudioPacketResult::IGNORED; if (clock.subscriptionEpoch != peer->subscriptionEpoch) return NetworkAudioPacketResult::INVALID; - peer->deadline = GetTickCount64() + AUDIO_LEASE_MS; if (!peer->feedbackAuthority || !peer->ready || !PlaybackCurrentNL(*peer) || @@ -1479,10 +1402,18 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleClock( } if (!knownPacket) return NetworkAudioPacketResult::IGNORED; + if (clock.packetID <= peer->lastFeedbackPacketID) + return NetworkAudioPacketResult::IGNORED; if (!(clock.flags & LG_NET_AUDIO_CLOCK_VALID) || !clock.targetRateQ32) return NetworkAudioPacketResult::HANDLED; + target = m_started ? m_target : nullptr; + if (!target) + return NetworkAudioPacketResult::IGNORED; + peer->lastFeedbackPacketID = clock.packetID; + peer->deadline = GetTickCount64() + AUDIO_LEASE_MS; + feedback.endpointGeneration = m_state.generation; feedback.playbackGeneration = m_state.playback.generation; feedback.position = clock.framePosition; @@ -1493,13 +1424,10 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleClock( feedback.flags |= AUDIO_CLOCK_VALID; if (clock.flags & LG_NET_AUDIO_CLOCK_STABLE) feedback.flags |= AUDIO_CLOCK_STABLE; - target = m_started ? m_target : nullptr; peerSerial = peer->serial; subscriptionEpoch = peer->subscriptionEpoch; } - if (!target) - return NetworkAudioPacketResult::IGNORED; const AudioResult result = target->SetClockFeedback(feedback); if (result == AudioResult::FAILED || result == AudioResult::REJECTED) { @@ -1531,7 +1459,8 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCapture( audio.dataLength > AUDIO_MAX_PCM_BYTES) return NetworkAudioPacketResult::INVALID; - CSRWSharedLock callbackLock(m_targetCallbackLock); + CSRWExclusiveLock callbackLock(m_targetCallbackLock); + IAudioTarget * target = nullptr; AudioPacket capture = {}; uint64_t peerSerial = 0; @@ -1551,7 +1480,6 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCapture( } if (!lgNetAudioDataMatchesFormat(&audio, &format)) return NetworkAudioPacketResult::INVALID; - peer->deadline = GetTickCount64() + AUDIO_LEASE_MS; if (peer->capturePacketKnown && audio.packetID <= peer->capturePacketID) @@ -1577,27 +1505,16 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCapture( sequenceGap) capture.flags |= AUDIO_PACKET_DISCONTINUITY; - bool clockFound = false; - for (ClockRecord& record : peer->captureClock) - if (record.valid && record.state.packetID == audio.packetID && - record.state.formatEpoch == audio.formatEpoch) - { - const LGNetAudioClockState& clock = record.state; - if (clock.flags & LG_NET_AUDIO_CLOCK_VALID) - { - capture.position = clock.framePosition; - capture.timeNs = clock.timeNs; - capture.rateQ32 = clock.rateQ32; - capture.flags |= AUDIO_PACKET_CLOCK_VALID; - } - if (clock.flags & LG_NET_AUDIO_CLOCK_STABLE) - capture.flags |= AUDIO_PACKET_CLOCK_STABLE; - record.valid = false; - clockFound = true; - break; - } - if (!clockFound) - capture.flags |= AUDIO_PACKET_DISCONTINUITY; + if (audio.flags & LG_NET_AUDIO_DATA_CLOCK_VALID) + { + capture.position = audio.startFrame; + capture.timeNs = static_cast(audio.timestampNs); + capture.rateQ32 = audio.rateQ32; + + capture.flags |= AUDIO_PACKET_CLOCK_VALID; + } + if (audio.flags & LG_NET_AUDIO_DATA_CLOCK_STABLE) + capture.flags |= AUDIO_PACKET_CLOCK_STABLE; target = m_started ? m_target : nullptr; peerSerial = peer->serial; @@ -1606,7 +1523,8 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCapture( peer->captureGap = true; return NetworkAudioPacketResult::IGNORED; } - peer->captureGap = false; + peer->captureGap = false; + m_capture.callback = true; } @@ -1621,6 +1539,8 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCapture( return NetworkAudioPacketResult::IGNORED; if (result == AudioResult::BUSY) { + peer->deadline = GetTickCount64() + AUDIO_LEASE_MS; + m_capture.valid = true; m_capture.peer = key; m_capture.peerSerial = peerSerial; @@ -1631,7 +1551,9 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCapture( return NetworkAudioPacketResult::HANDLED; } - if (result == AudioResult::FAILED || result == AudioResult::REJECTED) + if (result == AudioResult::ACCEPTED) + peer->deadline = GetTickCount64() + AUDIO_LEASE_MS; + else if (result == AudioResult::FAILED || result == AudioResult::REJECTED) { peer->captureRevoked = true; peer->captureGap = true; @@ -1649,14 +1571,14 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCurrentPacket( switch (packet.envelope.messageType) { case LG_NET_AUDIO_MESSAGE_CLOCK_STATE: - if (packet.envelope.requestID || + if (!datagram || packet.envelope.requestID || !DeliveryFlags(packet.envelope, datagram, LG_NET_ENVELOPE_FINAL)) return NetworkAudioPacketResult::INVALID; return HandleClock(key, packet); case LG_NET_AUDIO_MESSAGE_CAPTURE_DATA: - if (packet.envelope.requestID || + if (!datagram || packet.envelope.requestID || !DeliveryFlags(packet.envelope, datagram, LG_NET_ENVELOPE_FINAL)) return NetworkAudioPacketResult::INVALID; @@ -1678,7 +1600,6 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCurrentPacket( if (!peer || !peer->hasRequest || peer->subscriptionEpoch != packet.envelope.componentEpoch) return NetworkAudioPacketResult::IGNORED; - peer->deadline = now + AUDIO_LEASE_MS; uint8_t barrierBit; uint64_t barrierID; uint64_t stateSequence; @@ -1705,15 +1626,11 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCurrentPacket( barrier.barrierID != barrierID || barrier.stateSequence != stateSequence) return NetworkAudioPacketResult::IGNORED; - peer->pendingBarriers &= static_cast(~barrierBit); - peer->awaitingAck = peer->pendingBarriers != 0; - peer->ready = !peer->awaitingAck && + peer->deadline = now + AUDIO_LEASE_MS; + peer->pendingBarriers &= static_cast(~barrierBit); + peer->awaitingAck = peer->pendingBarriers != 0; + peer->ready = !peer->awaitingAck && peer->pumpStage == PumpStage::DONE; - if (m_playbackWaiting && SamePeer(m_playbackWaitingPeer, key)) - { - m_playbackWaiting = false; - m_playbackReadyPending = true; - } return NetworkAudioPacketResult::HANDLED; } @@ -1764,17 +1681,21 @@ NetworkAudioPacketResult CNetworkAudio::Impl::HandleCurrentPacket( } } +bool CNetworkAudio::Impl::DatagramPCM( + const PlaybackTarget& target, uint32_t payloadLength) +{ + return (target.roles & LG_NET_AUDIO_ROLE_DATAGRAM_PCM) && + payloadLength <= target.info.maxDatagramPayload; +} + NetworkAudioSendResult CNetworkAudio::Impl::SendPCM( const PlaybackTarget& target, uint16_t messageType, const void * payload, uint32_t payloadLength) { - /* CLOCK_STATE and PCM are an ordered pair. Keep both on the same reliable - * stream until a future protocol version provides atomic datagram framing. */ - if (!(target.roles & LG_NET_AUDIO_ROLE_RELIABLE_PCM) || - payloadLength > target.info.maxReliablePayload) + if (!DatagramPCM(target, payloadLength)) return NetworkAudioSendResult::FAILED; - return Send(target.peer, false, messageType, LG_NET_ENVELOPE_FINAL, + return Send(target.peer, true, messageType, LG_NET_ENVELOPE_FINAL, target.subscriptionEpoch, 0, payload, payloadLength); } @@ -1842,35 +1763,6 @@ AudioResult CNetworkAudio::Impl::BeginPlayback( return AudioResult::BUSY; } -uint64_t CNetworkAudio::Impl::PlaybackReadySerial( - const PlaybackTarget& target) -{ - CSRWSharedLock lock(m_lock); - const PeerState * peer = FindPeerNL(target.peer); - if (!peer || peer->serial != target.peerSerial || - peer->subscriptionEpoch != target.subscriptionEpoch) - return 0; - return peer->sendReadySerial; -} - -CNetworkAudio::Impl::BlockDisposition -CNetworkAudio::Impl::HandlePlaybackBlock( - const PlaybackTarget& target, uint64_t readySerial) -{ - CSRWExclusiveLock lock(m_lock); - PeerState * peer = FindPeerNL(target.peer); - if (!peer || peer->serial != target.peerSerial || - peer->subscriptionEpoch != target.subscriptionEpoch || - !peer->feedbackAuthority) - return BlockDisposition::STALE; - if (peer->sendReadySerial != readySerial) - return BlockDisposition::RETRY; - - m_playbackWaiting = true; - m_playbackWaitingPeer = target.peer; - return BlockDisposition::WAIT; -} - AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) { while (m_playback.valid && @@ -1894,7 +1786,6 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) m_playback.frameOffset = 0; m_playback.chunkFrames = 0; m_playback.packetID = 0; - m_playback.clockSent = false; continue; } ready = peer->ready && peer->pumpStage == PumpStage::DONE; @@ -1904,13 +1795,6 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) if (!ready) { - if (authority) - { - CSRWExclusiveLock lock(m_lock); - m_playbackWaiting = true; - m_playbackWaitingPeer = target.peer; - return AudioResult::BUSY; - } CSRWExclusiveLock lock(m_lock); PeerState * peer = FindPeerNL(target.peer); if (peer && peer->serial == target.peerSerial) @@ -1919,7 +1803,6 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) m_playback.frameOffset = 0; m_playback.chunkFrames = 0; m_playback.packetID = 0; - m_playback.clockSent = false; continue; } @@ -1938,16 +1821,6 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) capacity = (m_playback.packet.flags & AUDIO_PACKET_SILENT) ? target.maxPacketFrames : bytes / blockAlign; } - if (!capacity && - (target.roles & LG_NET_AUDIO_ROLE_RELIABLE_PCM)) - { - const uint32_t bytes = target.info.maxReliablePayload > - LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE ? - target.info.maxReliablePayload - - LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE : 0; - capacity = (m_playback.packet.flags & AUDIO_PACKET_SILENT) ? - target.maxPacketFrames : bytes / blockAlign; - } capacity = (std::min)(capacity, target.maxPacketFrames); capacity = (std::min)(capacity, AUDIO_MAX_FRAMES_PER_PACKET); if (!capacity) @@ -1976,91 +1849,29 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) } } - const uint32_t frameOffset = m_playback.frameOffset; - const uint32_t frames = m_playback.chunkFrames; - const uint64_t timeAdvance = + const uint32_t frameOffset = m_playback.frameOffset; + const uint32_t frames = m_playback.chunkFrames; + const uint64_t timeAdvance = static_cast(frameOffset) * UINT64_C(1000000000) / m_playback.packet.format.sampleRate; - const bool clockValid = + const bool packetClockValid = (m_playback.packet.flags & AUDIO_PACKET_CLOCK_VALID) != 0; - const bool clockStable = + const int64_t packetTime = packetClockValid ? + m_playback.packet.timeNs + static_cast(timeAdvance) : 0; + const bool clockValid = packetClockValid && packetTime > 0 && + m_playback.packet.rateQ32 != 0; + const bool clockStable = clockValid && (m_playback.packet.flags & AUDIO_PACKET_CLOCK_STABLE) != 0; - LGNetAudioClockState clock = {}; - clock.streamID = PLAYBACK_STREAM_ID; - clock.direction = LG_NET_AUDIO_DIRECTION_PLAYBACK; - clock.flags = static_cast( - clockValid ? LG_NET_AUDIO_CLOCK_VALID : 0); - if (clockStable) - clock.flags |= LG_NET_AUDIO_CLOCK_STABLE; - clock.formatEpoch = target.formatEpoch; - clock.subscriptionEpoch = target.subscriptionEpoch; - clock.packetID = m_playback.packetID; - clock.framePosition = m_playback.packet.position + frameOffset; - clock.timeNs = clockValid ? - m_playback.packet.timeNs + static_cast(timeAdvance) : 0; - clock.rateQ32 = clockValid ? m_playback.packet.rateQ32 : 0; - clock.targetRateQ32 = 0; - - std::array clockPayload; - if (!lgNetAudioClockStateEncode(clockPayload.data(), - clockPayload.size(), &clock)) - return AudioResult::FAILED; - if (!m_playback.clockSent) - { - const uint64_t readySerial = PlaybackReadySerial(target); - const NetworkAudioSendResult result = SendPCM(target, - LG_NET_AUDIO_MESSAGE_CLOCK_STATE, clockPayload.data(), - static_cast(clockPayload.size())); - if (result != NetworkAudioSendResult::ACCEPTED) - { - if (authority && result == NetworkAudioSendResult::BLOCKED) - { - const BlockDisposition disposition = - HandlePlaybackBlock(target, readySerial); - if (disposition == BlockDisposition::RETRY) - continue; - if (disposition == BlockDisposition::WAIT) - { - if (!m_playback.blockedSince) - m_playback.blockedSince = GetTickCount64(); - return AudioResult::BUSY; - } - } - - CSRWExclusiveLock lock(m_lock); - PeerState * peer = FindPeerNL(target.peer); - if (peer && peer->serial == target.peerSerial && - peer->subscriptionEpoch == target.subscriptionEpoch) - { - peer->playbackGap = true; - if (authority && peer->feedbackAuthority) - { - peer->feedbackAuthority = false; - peer->feedbackRevoked = true; - RebalanceRolesNL(); - } - } - ++m_playback.target; - m_playback.frameOffset = 0; - m_playback.chunkFrames = 0; - m_playback.packetID = 0; - m_playback.clockSent = false; - continue; - } - m_playback.clockSent = true; - } - - const int64_t packetTime = m_playback.packet.timeNs + - static_cast(timeAdvance); LGNetAudioData audio = {}; audio.streamID = PLAYBACK_STREAM_ID; audio.direction = LG_NET_AUDIO_DIRECTION_PLAYBACK; audio.formatEpoch = target.formatEpoch; audio.packetID = m_playback.packetID; - audio.timestampNs = packetTime > 0 ? - static_cast(packetTime) : MonotonicNs(); + audio.timestampNs = clockValid ? + static_cast(packetTime) : 0; audio.startFrame = m_playback.packet.position + frameOffset; + audio.rateQ32 = clockValid ? m_playback.packet.rateQ32 : 0; audio.frameCount = frames; if (!frameOffset && ((m_playback.packet.flags & AUDIO_PACKET_DISCONTINUITY) || gap)) @@ -2074,6 +1885,10 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) static_cast(frameOffset) * m_playback.packet.format.blockAlign; } + if (clockValid) + audio.flags |= LG_NET_AUDIO_DATA_CLOCK_VALID; + if (clockStable) + audio.flags |= LG_NET_AUDIO_DATA_CLOCK_STABLE; const size_t wireSize = lgNetAudioDataSize(&audio); std::vector payload; @@ -2089,34 +1904,21 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) payload.data(), payload.size(), &audio)) return AudioResult::FAILED; - const uint64_t readySerial = PlaybackReadySerial(target); - const NetworkAudioSendResult result = SendPCM(target, + const uint32_t payloadLength = + static_cast(payload.size()); + const NetworkAudioSendResult result = SendPCM(target, LG_NET_AUDIO_MESSAGE_PLAYBACK_DATA, payload.data(), - static_cast(payload.size())); + payloadLength); if (result != NetworkAudioSendResult::ACCEPTED) { - if (authority && result == NetworkAudioSendResult::BLOCKED) - { - m_playback.clockSent = false; - const BlockDisposition disposition = - HandlePlaybackBlock(target, readySerial); - if (disposition == BlockDisposition::RETRY) - continue; - if (disposition == BlockDisposition::WAIT) - { - if (!m_playback.blockedSince) - m_playback.blockedSince = GetTickCount64(); - return AudioResult::BUSY; - } - } - CSRWExclusiveLock lock(m_lock); PeerState * peer = FindPeerNL(target.peer); if (peer && peer->serial == target.peerSerial && peer->subscriptionEpoch == target.subscriptionEpoch) { peer->playbackGap = true; - if (authority && peer->feedbackAuthority) + if (authority && result == NetworkAudioSendResult::FAILED && + peer->feedbackAuthority) { peer->feedbackAuthority = false; peer->feedbackRevoked = true; @@ -2127,7 +1929,6 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) m_playback.frameOffset = 0; m_playback.chunkFrames = 0; m_playback.packetID = 0; - m_playback.clockSent = false; continue; } @@ -2146,11 +1947,9 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) } } - m_playback.blockedSince = 0; m_playback.frameOffset += frames; m_playback.chunkFrames = 0; m_playback.packetID = 0; - m_playback.clockSent = false; if (m_playback.frameOffset == m_playback.packet.frames) { ++m_playback.target; @@ -2159,10 +1958,6 @@ AudioResult CNetworkAudio::Impl::ContinuePlayback(const uint8_t * data) } m_playback = {}; - { - CSRWExclusiveLock lock(m_lock); - m_playbackWaiting = false; - } return AudioResult::ACCEPTED; } @@ -2171,8 +1966,6 @@ void CNetworkAudio::Impl::MarkFailed(const char * reason) if (!m_failed) DEBUG_ERROR("%s failed", reason); m_failed = true; - if (m_playback.valid) - m_playbackReadyPending = true; } void CNetworkAudio::Impl::NotifyFailure() @@ -2190,21 +1983,6 @@ void CNetworkAudio::Impl::NotifyFailure() target->AudioFailed(); } -void CNetworkAudio::Impl::NotifyPlaybackReady() -{ - CSRWSharedLock callbackLock(m_targetCallbackLock); - IAudioTarget * target = nullptr; - { - CSRWExclusiveLock lock(m_lock); - if (!m_playbackReadyPending) - return; - m_playbackReadyPending = false; - target = m_started ? m_target : nullptr; - } - if (target) - target->PlaybackReceiveReady(); -} - void CNetworkAudio::Impl::ProcessActivation() { CSRWSharedLock callbackLock(m_targetCallbackLock); @@ -2298,33 +2076,6 @@ void CNetworkAudio::Impl::RetryCapture() peer->captureGap = true; } -void CNetworkAudio::Impl::RevokeStalledAuthority(uint64_t now) -{ - CSRWExclusiveLock lock(m_lock); - if (!m_playback.valid || !m_playback.blockedSince || - !DeadlineReached(now, - m_playback.blockedSince + AUTHORITY_STALL_MS) || - m_playback.target >= m_playback.targets.size()) - return; - - const PlaybackTarget& target = - m_playback.targets[m_playback.target]; - PeerState * peer = FindPeerNL(target.peer); - if (peer && peer->serial == target.peerSerial && - peer->feedbackAuthority) - { - DEBUG_WARN("Network audio feedback authority %u:%u stalled", - peer->peer.client, peer->peer.generation); - peer->feedbackAuthority = false; - peer->feedbackRevoked = true; - peer->playbackGap = true; - RebalanceRolesNL(); - } - m_playback.blockedSince = 0; - m_playbackWaiting = false; - m_playbackReadyPending = true; -} - CNetworkAudio::CNetworkAudio() : m_impl(new (std::nothrow) Impl()) { @@ -2359,9 +2110,8 @@ void CNetworkAudio::DetachSender() CSRWExclusiveLock lock(m_impl->m_lock); m_impl->m_sender = nullptr; m_impl->m_peers.clear(); - m_impl->m_playback = {}; - m_impl->m_capture = {}; - m_impl->m_playbackWaiting = false; + m_impl->m_playback = {}; + m_impl->m_capture = {}; } bool CNetworkAudio::PeerConnected(const NetworkAudioPeer& key) @@ -2383,8 +2133,9 @@ bool CNetworkAudio::PeerConnected(const NetworkAudioPeer& key) if (!AudioVersionSupported(info.serviceVersion) || !(info.capabilityFlags & LG_NET_CAPABILITY_RELIABLE) || - info.maxReliablePayload < - LG_NET_AUDIO_DATA_HEADER_WIRE_SIZE + AUDIO_MAX_PCM_BYTES) + !(info.capabilityFlags & LG_NET_CAPABILITY_DATAGRAM) || + info.maxReliablePayload < LG_NET_AUDIO_CLOCK_STATE_WIRE_SIZE || + info.maxDatagramPayload < MIN_DATAGRAM_PAYLOAD) return false; CSRWExclusiveLock lock(m_impl->m_lock); @@ -2416,6 +2167,7 @@ void CNetworkAudio::PeerDisconnected(const NetworkAudioPeer& key) { if (!m_impl) return; + CSRWExclusiveLock callbackLock(m_impl->m_targetCallbackLock); CSRWExclusiveLock lock(m_impl->m_lock); const auto found = std::find_if(m_impl->m_peers.begin(), m_impl->m_peers.end(), [&key](const Impl::PeerState& peer) @@ -2426,12 +2178,6 @@ void CNetworkAudio::PeerDisconnected(const NetworkAudioPeer& key) return; if (m_impl->m_capture.valid && SamePeer(m_impl->m_capture.peer, key)) m_impl->m_capture.valid = false; - if (m_impl->m_playbackWaiting && - SamePeer(m_impl->m_playbackWaitingPeer, key)) - { - m_impl->m_playbackWaiting = false; - m_impl->m_playbackReadyPending = true; - } m_impl->m_peers.erase(found); m_impl->RebalanceRolesNL(); } @@ -2453,6 +2199,7 @@ NetworkAudioPacketResult CNetworkAudio::HandlePacket( !ReliableFlags(packet.envelope, LG_NET_ENVELOPE_ACK_REQUIRED | LG_NET_ENVELOPE_FINAL)) return NetworkAudioPacketResult::INVALID; + CSRWExclusiveLock callbackLock(m_impl->m_targetCallbackLock); CSRWExclusiveLock lock(m_impl->m_lock); Impl::PeerState * peer = m_impl->FindPeerNL(key); if (!peer || peer->info.serviceVersion != @@ -2470,13 +2217,7 @@ NetworkAudioPacketResult CNetworkAudio::HandlePacket( return NetworkAudioPacketResult::IGNORED; } - const NetworkAudioPacketResult result = - m_impl->HandleCurrentPacket(key, packet, datagram, now); - CSRWExclusiveLock processLock = - CSRWExclusiveLock::Try(m_impl->m_processLock); - if (processLock) - m_impl->NotifyPlaybackReady(); - return result; + return m_impl->HandleCurrentPacket(key, packet, datagram, now); } void CNetworkAudio::Process(uint64_t nowMs) @@ -2485,6 +2226,7 @@ void CNetworkAudio::Process(uint64_t nowMs) return; CSRWExclusiveLock processLock(m_impl->m_processLock); { + CSRWExclusiveLock callbackLock(m_impl->m_targetCallbackLock); CSRWExclusiveLock lock(m_impl->m_lock); bool rebalance = false; for (Impl::PeerState& peer : m_impl->m_peers) @@ -2504,10 +2246,8 @@ void CNetworkAudio::Process(uint64_t nowMs) m_impl->RebalanceRolesNL(); } - m_impl->RevokeStalledAuthority(nowMs); m_impl->ProcessActivation(); m_impl->PumpAll(); - m_impl->NotifyPlaybackReady(); m_impl->NotifyFailure(); } @@ -2516,26 +2256,12 @@ void CNetworkAudio::SendReady(const NetworkAudioPeer& key, { if (!m_impl || service != LG_NET_SERVICE_AUDIO) return; - { - CSRWExclusiveLock lock(m_impl->m_lock); - Impl::PeerState * peer = m_impl->FindPeerNL(key); - if (peer) - peer->sendReadySerial = NextSerial(peer->sendReadySerial); - if (m_impl->m_playbackWaiting && - SamePeer(m_impl->m_playbackWaitingPeer, key)) - { - m_impl->m_playbackWaiting = false; - m_impl->m_playbackReadyPending = true; - } - } + if (datagram) + return; CSRWExclusiveLock processLock = CSRWExclusiveLock::Try(m_impl->m_processLock); if (processLock) - { - if (!datagram) - m_impl->PumpPeer(key); - m_impl->NotifyPlaybackReady(); - } + m_impl->PumpPeer(key); } bool CNetworkAudio::Start(IAudioTarget& target) @@ -2574,14 +2300,12 @@ void CNetworkAudio::Stop() deactivate = m_impl->m_activationCommanded; for (Impl::PeerState& peer : m_impl->m_peers) m_impl->ClearSubscriptionNL(peer); - m_impl->m_playback = {}; - m_impl->m_capture = {}; - m_impl->m_started = false; - m_impl->m_target = nullptr; - m_impl->m_stateKnown = false; - m_impl->m_activationCommanded = false; - m_impl->m_playbackReadyPending = false; - m_impl->m_playbackWaiting = false; + m_impl->m_playback = {}; + m_impl->m_capture = {}; + m_impl->m_started = false; + m_impl->m_target = nullptr; + m_impl->m_stateKnown = false; + m_impl->m_activationCommanded = false; } callbackLock.Unlock(); processLock.Unlock(); @@ -2619,11 +2343,7 @@ void CNetworkAudio::AudioState(const AudioEndpointStatus& state) !state.playback.formatValid || !SameFormat(m_impl->m_playback.packet.format, state.playback.format))) - { - m_impl->m_playback = {}; - m_impl->m_playbackReadyPending = true; - m_impl->m_playbackWaiting = false; - } + m_impl->m_playback = {}; if (m_impl->m_capture.valid && (m_impl->m_capture.packet.endpointGeneration != state.generation || m_impl->m_capture.packet.streamGeneration != @@ -2689,7 +2409,6 @@ AudioResult CNetworkAudio::SendPlayback( if (begin != AudioResult::BUSY) return begin; } - m_impl->m_playbackWaiting = false; } const AudioResult result = m_impl->ContinuePlayback(data); diff --git a/idd/LGIdd/transport/network/CNetworkServer.cpp b/idd/LGIdd/transport/network/CNetworkServer.cpp index a951f067..559ced7d 100644 --- a/idd/LGIdd/transport/network/CNetworkServer.cpp +++ b/idd/LGIdd/transport/network/CNetworkServer.cpp @@ -4747,13 +4747,18 @@ retry: HandshakeState handshake; uint64_t sessionEpoch; uint32_t maxPayload; + bool closing; { CSRWSharedLock guard(connection.lock); handshake = connection.handshake; sessionEpoch = connection.sessionEpoch; maxPayload = connection.maxPayload; + closing = connection.closing; } + if (closing) + return true; + if (handshake == HandshakeState::WAITING_HELLO) return packet.envelope.service == LG_NET_SERVICE_CORE && packet.envelope.messageType == LG_NET_CORE_MESSAGE_HELLO && @@ -5188,6 +5193,8 @@ retry: const uint16_t serviceMask = UINT16_C(1) << packet.envelope.service; { CSRWExclusiveLock guard(connection.lock); + if (connection.closing) + return true; const bool serviceReady = packet.envelope.service == LG_NET_SERVICE_CORE || ((connection.selectedServices & serviceMask) && diff --git a/repos/LGProtocol b/repos/LGProtocol index 189c9bc5..a4a7519f 160000 --- a/repos/LGProtocol +++ b/repos/LGProtocol @@ -1 +1 @@ -Subproject commit 189c9bc5b6805dda63dad3f121b8077c2850906d +Subproject commit a4a7519f47d703d575aac1d33418f3390df098b3