From d76fb7498430930ba4519c06ef9736ac0b25dd26 Mon Sep 17 00:00:00 2001 From: anjali-syna <206662904+anjali-syna@users.noreply.github.com> Date: Tue, 14 Jul 2026 11:32:48 +0530 Subject: [PATCH 1/7] Addling logic to inject video first for muxed ts In direct rialto, audio will wait for video attach, but muxed content will have only one inject thread for adio and video, so the wait can cause deadlock --- tsprocessor.cpp | 54 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/tsprocessor.cpp b/tsprocessor.cpp index d354345ab8..478423e7b4 100644 --- a/tsprocessor.cpp +++ b/tsprocessor.cpp @@ -1502,6 +1502,47 @@ bool TSProcessor::demuxAndSend(const void *ptr, size_t len, double position, dou } AAMPLOG_INFO("demuxAndSend : len %d videoPid %d audioPid %d m_pcrPid %d videoComponentCount %d m_demuxInitialized = %d", (int)len, videoPid, audioPid, m_pcrPid, videoComponentCount, m_demuxInitialized); + // In eStreamOp_DEMUX_ALL (muxed TS), video and audio are injected from a single + // thread. Audio PES packets often complete before the first video frame; + // sending audio first causes audio's waitForAttach() to block the thread + // indefinitely in Direct Rialto mode, since the video frame that would trigger + // AttachSource(video) can never be sent from the same blocked thread. + // Wrap the processor to buffer audio frames until the first video frame of + // each demuxAndSend call is dispatched, guaranteeing video-first ordering. + struct MuxedAudioFrame + { + AampMediaType type; + SegmentInfo_t info; + std::vector buf; + }; + std::vector muxedAudioQueue; + bool muxedVideoDispatched = false; + MediaProcessor::process_fcn_t origProcessor; + if (m_streamOperation == eStreamOp_DEMUX_ALL && processor) + { + origProcessor = processor; + processor = [&muxedAudioQueue, &muxedVideoDispatched, &origProcessor] + (AampMediaType type, SegmentInfo_t info, std::vector buf) + { + if (type == eMEDIATYPE_VIDEO && !muxedVideoDispatched) + { + muxedVideoDispatched = true; + origProcessor(type, info, std::move(buf)); + for (auto& f : muxedAudioQueue) + origProcessor(f.type, f.info, std::move(f.buf)); + muxedAudioQueue.clear(); + } + else if (type == eMEDIATYPE_AUDIO && !muxedVideoDispatched) + { + muxedAudioQueue.push_back({type, info, std::move(buf)}); + } + else + { + origProcessor(type, info, std::move(buf)); + } + }; + } + std::unordered_set updated_demuxers{}; const unsigned char * packetStart = (const unsigned char *)ptr; while (len >= PACKET_SIZE) @@ -1651,6 +1692,19 @@ bool TSProcessor::demuxAndSend(const void *ptr, size_t len, double position, dou } } + // Flush any audio that was buffered by the muxed reorder wrapper but had no + // corresponding video frame in this batch (e.g. trailing audio data after the + // last video PES in the segment). Calling origProcessor directly avoids + // re-entering the wrapper's buffer path. + if (!muxedAudioQueue.empty()) + { + AAMPLOG_WARN("demuxAndSend: flushing %zu buffered audio frame(s) with no prior video frame in this batch", + muxedAudioQueue.size()); + for (auto& f : muxedAudioQueue) + origProcessor(f.type, f.info, std::move(f.buf)); + muxedAudioQueue.clear(); + } + return ret; } From 7ad1fff7e455a17126a40bbde820946658ced928 Mon Sep 17 00:00:00 2001 From: anjali-syna <206662904+anjali-syna@users.noreply.github.com> Date: Tue, 14 Jul 2026 15:20:32 +0530 Subject: [PATCH 2/7] Revert tsprocessor changes and attach all the sources from SendCopy --- direct-rialto/AampRialtoPlayer.cpp | 17 ++++++++++ tsprocessor.cpp | 54 ------------------------------ 2 files changed, 17 insertions(+), 54 deletions(-) diff --git a/direct-rialto/AampRialtoPlayer.cpp b/direct-rialto/AampRialtoPlayer.cpp index d59d8a4508..58318c7ae3 100644 --- a/direct-rialto/AampRialtoPlayer.cpp +++ b/direct-rialto/AampRialtoPlayer.cpp @@ -804,6 +804,7 @@ bool AampRialtoPlayer::SendCopy( } else { +#if 0//anj if (!source->isAttached()) { AAMPLOG_INFO("Setting stream caps for mediaType=%d", static_cast(mediaType)); @@ -812,6 +813,22 @@ bool AampRialtoPlayer::SendCopy( codecInfo.mCodecFormat = toGstStreamOutputFormat(source->format()); SetStreamCaps(mediaType, std::move(codecInfo)); } +#else + if (!source->isAttached()) + { + // For HLS-TS the codec format is all that Rialto requires to set the stream caps. + for (auto source2: m_sources) + { + if (source2 && !source2->isAttached()) + { + AAMPLOG_INFO("Setting stream caps for mediaType=%d", static_cast(source2->mediaType())); + MediaCodecInfo codecInfo{}; + codecInfo.mCodecFormat = toGstStreamOutputFormat(source2->format()); + SetStreamCaps(source2->mediaType(), std::move(codecInfo)); + } + } + } +#endif auto sharedBuffer = std::make_shared>(std::move(buffer)); diff --git a/tsprocessor.cpp b/tsprocessor.cpp index 478423e7b4..d354345ab8 100644 --- a/tsprocessor.cpp +++ b/tsprocessor.cpp @@ -1502,47 +1502,6 @@ bool TSProcessor::demuxAndSend(const void *ptr, size_t len, double position, dou } AAMPLOG_INFO("demuxAndSend : len %d videoPid %d audioPid %d m_pcrPid %d videoComponentCount %d m_demuxInitialized = %d", (int)len, videoPid, audioPid, m_pcrPid, videoComponentCount, m_demuxInitialized); - // In eStreamOp_DEMUX_ALL (muxed TS), video and audio are injected from a single - // thread. Audio PES packets often complete before the first video frame; - // sending audio first causes audio's waitForAttach() to block the thread - // indefinitely in Direct Rialto mode, since the video frame that would trigger - // AttachSource(video) can never be sent from the same blocked thread. - // Wrap the processor to buffer audio frames until the first video frame of - // each demuxAndSend call is dispatched, guaranteeing video-first ordering. - struct MuxedAudioFrame - { - AampMediaType type; - SegmentInfo_t info; - std::vector buf; - }; - std::vector muxedAudioQueue; - bool muxedVideoDispatched = false; - MediaProcessor::process_fcn_t origProcessor; - if (m_streamOperation == eStreamOp_DEMUX_ALL && processor) - { - origProcessor = processor; - processor = [&muxedAudioQueue, &muxedVideoDispatched, &origProcessor] - (AampMediaType type, SegmentInfo_t info, std::vector buf) - { - if (type == eMEDIATYPE_VIDEO && !muxedVideoDispatched) - { - muxedVideoDispatched = true; - origProcessor(type, info, std::move(buf)); - for (auto& f : muxedAudioQueue) - origProcessor(f.type, f.info, std::move(f.buf)); - muxedAudioQueue.clear(); - } - else if (type == eMEDIATYPE_AUDIO && !muxedVideoDispatched) - { - muxedAudioQueue.push_back({type, info, std::move(buf)}); - } - else - { - origProcessor(type, info, std::move(buf)); - } - }; - } - std::unordered_set updated_demuxers{}; const unsigned char * packetStart = (const unsigned char *)ptr; while (len >= PACKET_SIZE) @@ -1692,19 +1651,6 @@ bool TSProcessor::demuxAndSend(const void *ptr, size_t len, double position, dou } } - // Flush any audio that was buffered by the muxed reorder wrapper but had no - // corresponding video frame in this batch (e.g. trailing audio data after the - // last video PES in the segment). Calling origProcessor directly avoids - // re-entering the wrapper's buffer path. - if (!muxedAudioQueue.empty()) - { - AAMPLOG_WARN("demuxAndSend: flushing %zu buffered audio frame(s) with no prior video frame in this batch", - muxedAudioQueue.size()); - for (auto& f : muxedAudioQueue) - origProcessor(f.type, f.info, std::move(f.buf)); - muxedAudioQueue.clear(); - } - return ret; } From dee3035be51d936d2f7e637a4150ce0580256925 Mon Sep 17 00:00:00 2001 From: anjali-syna <206662904+anjali-syna@users.noreply.github.com> Date: Tue, 14 Jul 2026 15:54:06 +0530 Subject: [PATCH 3/7] Fixing compilation issue --- direct-rialto/AampRialtoPlayer.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/direct-rialto/AampRialtoPlayer.cpp b/direct-rialto/AampRialtoPlayer.cpp index 58318c7ae3..0ece5445e6 100644 --- a/direct-rialto/AampRialtoPlayer.cpp +++ b/direct-rialto/AampRialtoPlayer.cpp @@ -817,7 +817,7 @@ bool AampRialtoPlayer::SendCopy( if (!source->isAttached()) { // For HLS-TS the codec format is all that Rialto requires to set the stream caps. - for (auto source2: m_sources) + for (const auto& source2: m_sources) { if (source2 && !source2->isAttached()) { From f9c7422c931e1d5abb4ba51ac5036bb9aa14622c Mon Sep 17 00:00:00 2001 From: anjali-syna <206662904+anjali-syna@users.noreply.github.com> Date: Thu, 16 Jul 2026 10:31:30 +0530 Subject: [PATCH 4/7] Adding more logs to trace subtitle flow --- middleware/subtec/subtecparser/WebVttSubtecParser.cpp | 1 + streamabstraction.cpp | 5 +++++ 2 files changed, 6 insertions(+) diff --git a/middleware/subtec/subtecparser/WebVttSubtecParser.cpp b/middleware/subtec/subtecparser/WebVttSubtecParser.cpp index 92e9d71db1..f4c6a41ca6 100644 --- a/middleware/subtec/subtecparser/WebVttSubtecParser.cpp +++ b/middleware/subtec/subtecparser/WebVttSubtecParser.cpp @@ -82,6 +82,7 @@ void WebVTTSubtecParser::setPtsOffset(double ptsOffsetSec) // convention shared with Rialto SetSubtitlePtsOffset). The HLS // restamped video PTS is media_PTS + ptsOffsetSec, so we negate // here to make subtec add the offset on the subtitle path. + MW_LOG_INFO("ANJ: WebVTTSubtecParser::setPtsOffset(): ptsOffsetSec=%f", ptsOffsetSec); time_offset_ms_ = -static_cast(std::llround(ptsOffsetSec * 1000.0)); } diff --git a/streamabstraction.cpp b/streamabstraction.cpp index 7707a0ade0..12a5ebef6f 100644 --- a/streamabstraction.cpp +++ b/streamabstraction.cpp @@ -1298,14 +1298,18 @@ void MediaTrack::ProcessAndInjectFragment(CachedFragment *cachedFragment, bool f // Only for DASH streams ClearMediaHeaderDuration(cachedFragment); } + AAMPLOG_INFO("ANJ: mSubtitleParser = %p, aamp->IsGstreamerSubsEnabled=%d, type = %d", mSubtitleParser, aamp->IsGstreamerSubsEnabled(), type); if ((mSubtitleParser || (aamp->IsGstreamerSubsEnabled())) && type == eTRACK_SUBTITLE) { + AAMPLOG_INFO("ANJ:Inside if"); auto ptr = reinterpret_cast(cachedFragment->fragment.data()); auto len = cachedFragment->fragment.size(); if( ISCONFIGSET(eAAMPConfig_HlsTsEnablePTSReStamp) ) { + AAMPLOG_INFO("ANJ:Inside if ISCONFIGSET(eAAMPConfig_HlsTsEnablePTSReStamp)"); while( aamp->mDownloadsEnabled ) { + AAMPLOG_INFO("ANJ:Inside if aamp->mDownloadsEnabled)"); if( pContext->mPtsOffsetMap.count(cachedFragment->discontinuityIndex)==0 ) { AAMPLOG_WARN( "blocking subtitle track injection waiting for pts_offset[%" PRIu64 "]", cachedFragment->discontinuityIndex ); @@ -1316,6 +1320,7 @@ void MediaTrack::ProcessAndInjectFragment(CachedFragment *cachedFragment, bool f // Video and subtitle segments from the same discontinuity share the same // firstPts (same CDN stream). Apply ptsOffset[N] directly — no normalisation. cachedFragment->PTSOffsetSec = pContext->mPtsOffsetMap[cachedFragment->discontinuityIndex]; + AAMPLOG_INFO("ANJ:cachedFragment->PTSOffsetSec = %f", cachedFragment->PTSOffsetSec); if(mSubtitleParser) { // DASH-style PTS-offset propagation: rather than rewriting MPEGTS in From d77752d453f731668288ad242876ed450db86197 Mon Sep 17 00:00:00 2001 From: anjali-syna <206662904+anjali-syna@users.noreply.github.com> Date: Thu, 16 Jul 2026 11:25:09 +0530 Subject: [PATCH 5/7] Fixing compilation issue --- streamabstraction.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/streamabstraction.cpp b/streamabstraction.cpp index 12a5ebef6f..6f395f66e0 100644 --- a/streamabstraction.cpp +++ b/streamabstraction.cpp @@ -1298,7 +1298,8 @@ void MediaTrack::ProcessAndInjectFragment(CachedFragment *cachedFragment, bool f // Only for DASH streams ClearMediaHeaderDuration(cachedFragment); } - AAMPLOG_INFO("ANJ: mSubtitleParser = %p, aamp->IsGstreamerSubsEnabled=%d, type = %d", mSubtitleParser, aamp->IsGstreamerSubsEnabled(), type); + AAMPLOG_INFO("ANJ: aamp->IsGstreamerSubsEnabled=%d, type = %d", aamp->IsGstreamerSubsEnabled(), type); + //AAMPLOG_INFO("ANJ: mSubtitleParser = %p, aamp->IsGstreamerSubsEnabled=%d, type = %d", mSubtitleParser, aamp->IsGstreamerSubsEnabled(), type); if ((mSubtitleParser || (aamp->IsGstreamerSubsEnabled())) && type == eTRACK_SUBTITLE) { AAMPLOG_INFO("ANJ:Inside if"); @@ -1323,6 +1324,7 @@ void MediaTrack::ProcessAndInjectFragment(CachedFragment *cachedFragment, bool f AAMPLOG_INFO("ANJ:cachedFragment->PTSOffsetSec = %f", cachedFragment->PTSOffsetSec); if(mSubtitleParser) { + AAMPLOG_INFO("ANJ:Inside if mSubtitleParser"); // DASH-style PTS-offset propagation: rather than rewriting MPEGTS in // the VTT header (RestampSubtitle), push the per-fragment pts offset // down into the subtec parser and forward the buffer unchanged. The From b13513ffd0dd957bd7ea8cfd0961f1e7520fc14d Mon Sep 17 00:00:00 2001 From: anjali-syna <206662904+anjali-syna@users.noreply.github.com> Date: Thu, 16 Jul 2026 15:26:09 +0530 Subject: [PATCH 6/7] Removed unwanted code --- direct-rialto/AampRialtoPlayer.cpp | 11 ----------- middleware/subtec/subtecparser/WebVttSubtecParser.cpp | 1 - streamabstraction.cpp | 7 ------- 3 files changed, 19 deletions(-) diff --git a/direct-rialto/AampRialtoPlayer.cpp b/direct-rialto/AampRialtoPlayer.cpp index 0ece5445e6..75419e3d36 100644 --- a/direct-rialto/AampRialtoPlayer.cpp +++ b/direct-rialto/AampRialtoPlayer.cpp @@ -804,16 +804,6 @@ bool AampRialtoPlayer::SendCopy( } else { -#if 0//anj - if (!source->isAttached()) - { - AAMPLOG_INFO("Setting stream caps for mediaType=%d", static_cast(mediaType)); - // For HLS-TS the codec format is all that Rialto requires to set the stream caps. - MediaCodecInfo codecInfo{}; - codecInfo.mCodecFormat = toGstStreamOutputFormat(source->format()); - SetStreamCaps(mediaType, std::move(codecInfo)); - } -#else if (!source->isAttached()) { // For HLS-TS the codec format is all that Rialto requires to set the stream caps. @@ -828,7 +818,6 @@ bool AampRialtoPlayer::SendCopy( } } } -#endif auto sharedBuffer = std::make_shared>(std::move(buffer)); diff --git a/middleware/subtec/subtecparser/WebVttSubtecParser.cpp b/middleware/subtec/subtecparser/WebVttSubtecParser.cpp index f4c6a41ca6..92e9d71db1 100644 --- a/middleware/subtec/subtecparser/WebVttSubtecParser.cpp +++ b/middleware/subtec/subtecparser/WebVttSubtecParser.cpp @@ -82,7 +82,6 @@ void WebVTTSubtecParser::setPtsOffset(double ptsOffsetSec) // convention shared with Rialto SetSubtitlePtsOffset). The HLS // restamped video PTS is media_PTS + ptsOffsetSec, so we negate // here to make subtec add the offset on the subtitle path. - MW_LOG_INFO("ANJ: WebVTTSubtecParser::setPtsOffset(): ptsOffsetSec=%f", ptsOffsetSec); time_offset_ms_ = -static_cast(std::llround(ptsOffsetSec * 1000.0)); } diff --git a/streamabstraction.cpp b/streamabstraction.cpp index 6f395f66e0..7707a0ade0 100644 --- a/streamabstraction.cpp +++ b/streamabstraction.cpp @@ -1298,19 +1298,14 @@ void MediaTrack::ProcessAndInjectFragment(CachedFragment *cachedFragment, bool f // Only for DASH streams ClearMediaHeaderDuration(cachedFragment); } - AAMPLOG_INFO("ANJ: aamp->IsGstreamerSubsEnabled=%d, type = %d", aamp->IsGstreamerSubsEnabled(), type); - //AAMPLOG_INFO("ANJ: mSubtitleParser = %p, aamp->IsGstreamerSubsEnabled=%d, type = %d", mSubtitleParser, aamp->IsGstreamerSubsEnabled(), type); if ((mSubtitleParser || (aamp->IsGstreamerSubsEnabled())) && type == eTRACK_SUBTITLE) { - AAMPLOG_INFO("ANJ:Inside if"); auto ptr = reinterpret_cast(cachedFragment->fragment.data()); auto len = cachedFragment->fragment.size(); if( ISCONFIGSET(eAAMPConfig_HlsTsEnablePTSReStamp) ) { - AAMPLOG_INFO("ANJ:Inside if ISCONFIGSET(eAAMPConfig_HlsTsEnablePTSReStamp)"); while( aamp->mDownloadsEnabled ) { - AAMPLOG_INFO("ANJ:Inside if aamp->mDownloadsEnabled)"); if( pContext->mPtsOffsetMap.count(cachedFragment->discontinuityIndex)==0 ) { AAMPLOG_WARN( "blocking subtitle track injection waiting for pts_offset[%" PRIu64 "]", cachedFragment->discontinuityIndex ); @@ -1321,10 +1316,8 @@ void MediaTrack::ProcessAndInjectFragment(CachedFragment *cachedFragment, bool f // Video and subtitle segments from the same discontinuity share the same // firstPts (same CDN stream). Apply ptsOffset[N] directly — no normalisation. cachedFragment->PTSOffsetSec = pContext->mPtsOffsetMap[cachedFragment->discontinuityIndex]; - AAMPLOG_INFO("ANJ:cachedFragment->PTSOffsetSec = %f", cachedFragment->PTSOffsetSec); if(mSubtitleParser) { - AAMPLOG_INFO("ANJ:Inside if mSubtitleParser"); // DASH-style PTS-offset propagation: rather than rewriting MPEGTS in // the VTT header (RestampSubtitle), push the per-fragment pts offset // down into the subtec parser and forward the buffer unchanged. The From 885a4f3c5f4b4daca542044274241cb5b31e156f Mon Sep 17 00:00:00 2001 From: anjali-syna <206662904+anjali-syna@users.noreply.github.com> Date: Thu, 16 Jul 2026 15:39:17 +0530 Subject: [PATCH 7/7] Adding comments --- direct-rialto/AampRialtoPlayer.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/direct-rialto/AampRialtoPlayer.cpp b/direct-rialto/AampRialtoPlayer.cpp index 75419e3d36..725cc96015 100644 --- a/direct-rialto/AampRialtoPlayer.cpp +++ b/direct-rialto/AampRialtoPlayer.cpp @@ -806,7 +806,8 @@ bool AampRialtoPlayer::SendCopy( { if (!source->isAttached()) { - // For HLS-TS the codec format is all that Rialto requires to set the stream caps. + // Attaching all sources to avoid deadlock with muxed HLS-TS which uses only one injection thread + // For HLS-TS, the codec format is all that Rialto requires to set the stream caps. for (const auto& source2: m_sources) { if (source2 && !source2->isAttached())