From b7861a08508b9c4f8fd861cd7cf2aa4d393501af Mon Sep 17 00:00:00 2001 From: Ihor Ivlev Date: Sat, 31 Oct 2026 15:13:41 +0100 Subject: [PATCH 1/2] decoder: one access unit per MediaCodec input buffer VideoDecoder queued every NAL as its own input buffer with its own timestamp. A frame encoded as several slices is several NALs, and some decoders then show only the first slice of each frame. NALs now gather into one buffer per access unit, flushed on the RTP marker (set on the last packet of each frame) or, for a stream without markers, on the next frame's first slice. A buffer with only parameter sets is queued as CODEC_CONFIG. A single-slice stream is unchanged: one NAL per frame was already one access unit per buffer. --- app/videonative/src/main/cpp/VideoDecoder.cpp | 186 ++++++++++++------ app/videonative/src/main/cpp/VideoDecoder.h | 26 ++- app/videonative/src/main/cpp/VideoPlayer.cpp | 1 + .../src/main/cpp/parser/H26XParser.cpp | 8 + .../src/main/cpp/parser/H26XParser.h | 5 + 5 files changed, 158 insertions(+), 68 deletions(-) diff --git a/app/videonative/src/main/cpp/VideoDecoder.cpp b/app/videonative/src/main/cpp/VideoDecoder.cpp index 0ae7f05d..321ecc7b 100644 --- a/app/videonative/src/main/cpp/VideoDecoder.cpp +++ b/app/videonative/src/main/cpp/VideoDecoder.cpp @@ -35,6 +35,8 @@ void VideoDecoder::setOutputSurface(JNIEnv* env, jobject surface, jint idx) } std::lock_guard lock(mMutexInputPipe); inputPipeClosed = true; + mAuBuffer.clear(); + mAuHasVcl = false; if (decoder.configured[idx]) { AMediaCodec_stop(decoder.codec[idx]); @@ -98,19 +100,56 @@ void VideoDecoder::interpretNALU(const NALU& nalu) // A feedD thread (e.g. file or udp) thread might be running even tough no output surface was set // But at least we can buffer the sps/pps data mKeyFrameFinder.saveIfKeyFrame(nalu); + mAuBuffer.clear(); + mAuHasVcl = false; return; } if (decoder.configured[0] || decoder.configured[1]) { - feedDecoder(nalu, 0); - feedDecoder(nalu, 1); + // One input buffer per access unit, not per NAL (see endAccessUnit + // in the header). The transport's AU end (RTP marker) flushes with no + // added latency; the next first-slice NAL is the safety net for a + // stream without markers (one frame of latency, for those only). + const int type = nalu.get_nal_unit_type(); + const bool isVcl = nalu.IS_H265_PACKET ? (type < 32) : (type >= 1 && type <= 5); + bool firstSlice = false; + // H.264 data partitions B and C (types 3, 4) carry no slice header. + const bool hasSliceHeader = nalu.IS_H265_PACKET ? isVcl : (type == 1 || type == 2 || type == 5); + const int headerLen = nalu.IS_H265_PACKET ? 2 : 1; + if (hasSliceHeader && nalu.getDataSizeWithoutPrefix() > headerLen) + { + // first_slice_segment_in_pic_flag (H.265) / first_mb_in_slice == 0 + // (H.264, ue(v) with a leading 1 bit): the MSB of the first + // slice-header byte. + firstSlice = (nalu.getDataWithoutPrefix()[headerLen] & 0x80) != 0; + } + // Parameter sets, AUD and prefix SEI after a slice also start a new + // access unit (H.264 7.4.1.2.3, H.265 7.4.2.4.4): without a marker + // they belong to the next frame, not the one being gathered. + const bool startsAu = + nalu.IS_H265_PACKET ? (type >= 32 && type <= 39) : ((type >= 6 && type <= 9) || (type >= 14 && type <= 18)); + if (mAuHasVcl && (firstSlice || startsAu)) + { + flushAccessUnitLocked(); + } + if (mAuBuffer.size() + nalu.getSize() > kMaxAccessUnitBytes) + { + // No end of frame in sight (a stream without markers that never + // starts a new frame): drop what was gathered. + MLOGD << "Access unit over " << kMaxAccessUnitBytes << " bytes, dropped"; + mAuBuffer.clear(); + mAuHasVcl = false; + } + if (mAuBuffer.empty()) + { + mAuFirstNaluTime = nalu.creationTime; + } + mAuBuffer.insert(mAuBuffer.end(), nalu.getData(), nalu.getData() + nalu.getSize()); + if (isVcl) + { + mAuHasVcl = true; + } decodingInfo.nNALUSFeeded++; - // manually feeding AUDs doesn't seem to change anything for high latency streams - // Only for the x264 sw encoded example stream it might improve latency slightly - // if(!nalu.IS_H265_PACKET && nalu.get_nal_unit_type()==NAL_UNIT_TYPE_CODED_SLICE_NON_IDR){ - // MLOGD<<"Feeding special AUD"; - // feedDecoder(NALU::createExampleH264_AUD()); - //} } else { @@ -126,6 +165,83 @@ void VideoDecoder::interpretNALU(const NALU& nalu) } } +void VideoDecoder::endAccessUnit() +{ + std::lock_guard lock(mMutexInputPipe); + if (inputPipeClosed || !(decoder.configured[0] || decoder.configured[1])) + { + // Nothing downstream: drop a half-built AU rather than let it prefix + // the next session's first one. + mAuBuffer.clear(); + mAuHasVcl = false; + return; + } + flushAccessUnitLocked(); +} + +void VideoDecoder::flushAccessUnitLocked() +{ + if (mAuBuffer.empty()) + { + return; + } + // A buffer without a VCL NAL (bare parameter sets) is flagged + // CODEC_CONFIG, or the decoder emits a phantom output frame for it. + const bool configOnly = !mAuHasVcl; + feedAccessUnit(mAuBuffer.data(), mAuBuffer.size(), configOnly, 0); + feedAccessUnit(mAuBuffer.data(), mAuBuffer.size(), configOnly, 1); + mAuBuffer.clear(); + mAuHasVcl = false; +} + +void VideoDecoder::feedAccessUnit(const uint8_t* data, size_t size, bool configOnly, int idx) +{ + if (!decoder.codec[idx]) + { + return; + } + const auto now = std::chrono::steady_clock::now(); + const auto deltaParsing = now - mAuFirstNaluTime; + while (true) + { + const auto index = AMediaCodec_dequeueInputBuffer(decoder.codec[idx], BUFFER_TIMEOUT_US); + if (index >= 0) + { + size_t inputBufferSize; + uint8_t* buf = AMediaCodec_getInputBuffer(decoder.codec[idx], (size_t) index, &inputBufferSize); + if (size > inputBufferSize) + { + MLOGD << "Access unit too big " << size << " > " << inputBufferSize; + // Hand the dequeued buffer back empty: an index that is never + // queued is lost to the codec for good. + AMediaCodec_queueInputBuffer(decoder.codec[idx], (size_t) index, 0, 0, 0, 0); + return; + } + const int flag = configOnly ? AMEDIACODEC_BUFFER_FLAG_CODEC_CONFIG : 0; + std::memcpy(buf, data, size); + const uint64_t presentationTimeUS = + (uint64_t) duration_cast(steady_clock::now().time_since_epoch()).count(); + AMediaCodec_queueInputBuffer(decoder.codec[idx], (size_t) index, 0, size, presentationTimeUS, flag); + waitForInputB.add(steady_clock::now() - now); + parsingTime.add(deltaParsing); + return; + } + else if (index == AMEDIACODEC_INFO_TRY_AGAIN_LATER) + { + if (std::chrono::steady_clock::now() - now > std::chrono::seconds(1)) + { + MLOGE << "AMEDIACODEC_INFO_TRY_AGAIN_LATER for more than 1 second, dropping an access unit"; + return; + } + } + else + { + MLOGD << "dequeueInputBuffer idx " << (int) index << " return."; + return; + } + } +} + void VideoDecoder::configureStartDecoder(int idx) { if (decoder.window[idx] == nullptr) return; @@ -205,60 +321,6 @@ void VideoDecoder::configureStartDecoder(int idx) decoder.configured[idx] = true; } -void VideoDecoder::feedDecoder(const NALU& nalu, int idx) -{ - if (!decoder.codec[idx]) return; - const auto now = std::chrono::steady_clock::now(); - const auto deltaParsing = now - nalu.creationTime; - while (true) - { - const auto index = AMediaCodec_dequeueInputBuffer(decoder.codec[idx], BUFFER_TIMEOUT_US); - if (index >= 0) - { - size_t inputBufferSize; - uint8_t* buf = AMediaCodec_getInputBuffer(decoder.codec[idx], (size_t) index, &inputBufferSize); - // I have not seen any case where the input buffer returned by MediaCodec is too small to hold the NALU - // But better be safe than crashing with a memory exception - if (nalu.getSize() > inputBufferSize) - { - MLOGD << "Nalu too big" << nalu.getSize(); - return; - } - - int flag = - (IS_H265 && (nalu.isSPS() || nalu.isPPS() || nalu.isVPS())) ? AMEDIACODEC_BUFFER_FLAG_CODEC_CONFIG : 0; - std::memcpy(buf, nalu.getData(), (size_t) nalu.getSize()); - const uint64_t presentationTimeUS = - (uint64_t) duration_cast(steady_clock::now().time_since_epoch()).count(); - AMediaCodec_queueInputBuffer( - decoder.codec[idx], (size_t) index, 0, (size_t) nalu.getSize(), presentationTimeUS, flag); - waitForInputB.add(steady_clock::now() - now); - parsingTime.add(deltaParsing); - return; - } - else if (index == AMEDIACODEC_INFO_TRY_AGAIN_LATER) - { - // just try again. But if we had no success in the last 1 second,log a warning and return. - const auto elapsedTimeTryingForBuffer = std::chrono::steady_clock::now() - now; - if (elapsedTimeTryingForBuffer > std::chrono::seconds(1)) - { - // Since OpenHD provides a lossy link it is really unlikely, but possible that we somehow 'break' the - // codec by feeding corrupt data. It will probably recover itself as soon as we feed enough valid data - // though; - MLOGE << "AMEDIACODEC_INFO_TRY_AGAIN_LATER for more than 1 second " - << MyTimeHelper::R(elapsedTimeTryingForBuffer) << "return."; - return; - } - } - else - { - // Something went wrong. But we will feed the next NALU soon anyways - MLOGD << "dequeueInputBuffer idx " << (int) index << "return."; - return; - } - } -} - void VideoDecoder::checkOutputLoop(int idx) { NDKThreadHelper::setProcessThreadPriorityAttachDetach(javaVm, -16, "DecoderCheckOutput"); diff --git a/app/videonative/src/main/cpp/VideoDecoder.h b/app/videonative/src/main/cpp/VideoDecoder.h index 20a43849..89abcc1e 100644 --- a/app/videonative/src/main/cpp/VideoDecoder.h +++ b/app/videonative/src/main/cpp/VideoDecoder.h @@ -12,6 +12,7 @@ #include #include #include +#include #include "NALU/KeyFrameFinder.hpp" #include "NALU/NALU.hpp" #include "helper/TimeHelper.hpp" @@ -95,13 +96,20 @@ class VideoDecoder // If the input pipe was closed (surface has been removed or is not set yet), only buffer key frames void interpretNALU(const NALU& nalu); + // End of an access unit, from the RTP marker bit: queue the NALs gathered + // since the last one as one input buffer. Some decoders show only the + // first slice of a multi-slice frame fed one NAL per buffer. + void endAccessUnit(); + private: // Initialize decoder with SPS / PPS data from KeyFrameFinder // Set Decoder.configured to true on success void configureStartDecoder(int idx); - // Wait for input buffer to become available before feeding NALU - void feedDecoder(const NALU& nalu, int idx); + // Queue the gathered access unit as one input buffer (both decoder + // instances) and clear it. Caller holds mMutexInputPipe. + void flushAccessUnitLocked(); + void feedAccessUnit(const uint8_t* data, size_t size, bool configOnly, int idx); // Runs until EOS arrives at output buffer or decoder is stopped void checkOutputLoop(int idx); @@ -119,10 +127,16 @@ class VideoDecoder Decoder decoder{}; DecodingInfo decodingInfo; // The input pipe is closed until we set a valid surface - bool inputPipeClosed = true; - std::mutex mMutexInputPipe; - DECODER_RATIO_CHANGED onDecoderRatioChangedCallback = nullptr; - DECODING_INFO_CHANGED_CALLBACK onDecodingInfoChangedCallback = nullptr; + bool inputPipeClosed = true; + std::mutex mMutexInputPipe; + // Access-unit accumulator (endAccessUnit): the NALs of the current AU, + // flushed on the transport's AU end or on the next first-slice NAL. + std::vector mAuBuffer; + static constexpr size_t kMaxAccessUnitBytes = 4 * 1024 * 1024; + bool mAuHasVcl = false; + std::chrono::steady_clock::time_point mAuFirstNaluTime; + DECODER_RATIO_CHANGED onDecoderRatioChangedCallback = nullptr; + DECODING_INFO_CHANGED_CALLBACK onDecodingInfoChangedCallback = nullptr; // So we can temporarily attach the output thread to the vm and make ndk calls JavaVM* javaVm = nullptr; std::chrono::steady_clock::time_point lastLog = std::chrono::steady_clock::now(); diff --git a/app/videonative/src/main/cpp/VideoPlayer.cpp b/app/videonative/src/main/cpp/VideoPlayer.cpp index 173b794e..a963dc1d 100644 --- a/app/videonative/src/main/cpp/VideoPlayer.cpp +++ b/app/videonative/src/main/cpp/VideoPlayer.cpp @@ -15,6 +15,7 @@ VideoPlayer::VideoPlayer(JNIEnv* env, jobject context) : mParser{std::bind(&VideoPlayer::onNewNALU, this, std::placeholders::_1)}, videoDecoder(env) { env->GetJavaVM(&javaVm); + mParser.onAccessUnitEnd = [this]() { videoDecoder.endAccessUnit(); }; videoDecoder.registerOnDecoderRatioChangedCallback( [this](const VideoRatio ratio) { diff --git a/app/videonative/src/main/cpp/parser/H26XParser.cpp b/app/videonative/src/main/cpp/parser/H26XParser.cpp index b0bd5a48..473101b8 100644 --- a/app/videonative/src/main/cpp/parser/H26XParser.cpp +++ b/app/videonative/src/main/cpp/parser/H26XParser.cpp @@ -39,6 +39,14 @@ void H26XParser::parse_rtp_stream(const uint8_t* rtp_data, const size_t data_len IS_H265 = true; mDecodeRTP.parseRTPH265toNALU(rtp_data, data_length); } + else + { + return; + } + if (rtpPacket.header.marker && onAccessUnitEnd) + { + onAccessUnitEnd(); + } } void H26XParser::onNewNaluDataExtracted( diff --git a/app/videonative/src/main/cpp/parser/H26XParser.h b/app/videonative/src/main/cpp/parser/H26XParser.h index c2c045e5..4cf8d1ee 100644 --- a/app/videonative/src/main/cpp/parser/H26XParser.h +++ b/app/videonative/src/main/cpp/parser/H26XParser.h @@ -32,6 +32,11 @@ class H26XParser void parse_rtp_stream(const uint8_t* rtp_data, const size_t data_len); + // Called after an RTP packet whose marker bit is set (the last packet + // of an access unit), so the decoder can queue AU-aligned buffers with + // no added latency. + std::function onAccessUnitEnd; + void reset(); public: From f6c5b6acdd77106575ebb87d01c0ca1cf34ea729 Mon Sep 17 00:00:00 2001 From: Ihor Ivlev Date: Sat, 31 Oct 2026 17:43:48 +0100 Subject: [PATCH 2/2] decoder: end an access unit only on a marker the RTP decoder accepted The RTP marker flushed the gathered access unit for any H.264/H.265 packet, including ones the RTP decoder then dropped: too short, empty, a duplicate sequence number, an unsupported NAL type. A late copy of the previous frame's last packet could also flush the current frame half-built. parseRTPH264toNALU/parseRTPH265toNALU now say whether they took the packet, and the marker counts only for a taken packet whose sequence number is newer than any seen so far. A jump of more than 512 behind is taken as a restarted sender and resyncs, so a new stream's markers are not ignored until its numbers catch up. --- .../src/main/cpp/parser/H26XParser.cpp | 29 +++++++++++++++---- .../src/main/cpp/parser/H26XParser.h | 4 +++ .../src/main/cpp/parser/ParseRTP.cpp | 21 ++++++++------ .../src/main/cpp/parser/ParseRTP.h | 4 +-- 4 files changed, 41 insertions(+), 17 deletions(-) diff --git a/app/videonative/src/main/cpp/parser/H26XParser.cpp b/app/videonative/src/main/cpp/parser/H26XParser.cpp index 473101b8..727e3085 100644 --- a/app/videonative/src/main/cpp/parser/H26XParser.cpp +++ b/app/videonative/src/main/cpp/parser/H26XParser.cpp @@ -22,6 +22,7 @@ H26XParser::H26XParser(NALU_DATA_CALLBACK onNewNALU) void H26XParser::reset() { mDecodeRTP.reset(); + mNewestSeq = -1; nParsedNALUs = 0; nParsedKonfigurationFrames = 0; } @@ -29,21 +30,37 @@ void H26XParser::reset() void H26XParser::parse_rtp_stream(const uint8_t* rtp_data, const size_t data_length) { const RTP::RTPPacket rtpPacket(rtp_data, data_length); + bool accepted = false; if (rtpPacket.header.payload == RTP_PAYLOAD_TYPE_H264) { - IS_H265 = false; - mDecodeRTP.parseRTPH264toNALU(rtp_data, data_length); + IS_H265 = false; + accepted = mDecodeRTP.parseRTPH264toNALU(rtp_data, data_length); } else if (rtpPacket.header.payload == RTP_PAYLOAD_TYPE_H265) { - IS_H265 = true; - mDecodeRTP.parseRTPH265toNALU(rtp_data, data_length); + IS_H265 = true; + accepted = mDecodeRTP.parseRTPH265toNALU(rtp_data, data_length); } - else + if (!accepted) { return; } - if (rtpPacket.header.marker && onAccessUnitEnd) + // A marker ends the current access unit only from a packet the RTP decoder + // took and that is newer than any seen so far: a late copy of the previous + // frame's last packet, or a second copy of the newest one, must not cut the + // current frame short. A jump far behind is a restarted sender: resync. + const uint16_t seq = rtpPacket.header.getSequence(); + bool newest = true; + if (mNewestSeq >= 0) + { + const uint16_t ahead = static_cast(seq - mNewestSeq); + newest = ahead != 0 && (ahead < 0x8000 || static_cast(mNewestSeq - seq) > kSeqResyncWindow); + } + if (newest) + { + mNewestSeq = seq; + } + if (rtpPacket.header.marker && newest && onAccessUnitEnd) { onAccessUnitEnd(); } diff --git a/app/videonative/src/main/cpp/parser/H26XParser.h b/app/videonative/src/main/cpp/parser/H26XParser.h index 4cf8d1ee..575aaa4d 100644 --- a/app/videonative/src/main/cpp/parser/H26XParser.h +++ b/app/videonative/src/main/cpp/parser/H26XParser.h @@ -60,6 +60,10 @@ class H26XParser int maxFPS = 0; bool IS_H265 = false; + // Sequence number of the newest accepted video RTP packet, -1 before the first. + int mNewestSeq = -1; + // Further behind than this is a new stream, not a late packet. + static constexpr uint16_t kSeqResyncWindow = 512; // First time a NALU was succesfully decoded // std::chrono::steady_clock::time_point timeFirstNALUArrived=std::chrono::steady_clock::time_point(0); }; diff --git a/app/videonative/src/main/cpp/parser/ParseRTP.cpp b/app/videonative/src/main/cpp/parser/ParseRTP.cpp index 08e9d1a8..4918d2eb 100644 --- a/app/videonative/src/main/cpp/parser/ParseRTP.cpp +++ b/app/videonative/src/main/cpp/parser/ParseRTP.cpp @@ -113,13 +113,13 @@ void RTPDecoder::h264_reconstruct_and_forward_one_nalu(const uint8_t* data, cons m_nalu_data_length = 0; } -void RTPDecoder::parseRTPH264toNALU(const uint8_t* rtp_data, const size_t data_length) +bool RTPDecoder::parseRTPH264toNALU(const uint8_t* rtp_data, const size_t data_length) { // 12 rtp header bytes and 1 nalu_header_t type byte if (data_length <= sizeof(rtp_header_t) + sizeof(nalu_header_t)) { MLOGD << "Not enough rtp data"; - return; + return false; } // MLOGD<<"Got h264 rtp data"; const RTP::RTPPacketH264 rtpPacket(rtp_data, data_length); @@ -127,13 +127,13 @@ void RTPDecoder::parseRTPH264toNALU(const uint8_t* rtp_data, const size_t data_l if (rtpPacket.rtpPayloadSize == 0) { MLOGD << "RTP packet is empty"; - return; + return false; } // MLOGD<<"RTP Header: "<asString(); if (!validateRTPPacket(rtpPacket.header)) { - return; + return false; } const auto& nalu_header = rtpPacket.getNALUHeaderH264(); if (nalu_header.type == 28) @@ -223,7 +223,9 @@ void RTPDecoder::parseRTPH264toNALU(const uint8_t* rtp_data, const size_t data_l else { MLOGD << "Got unsupported H264 RTP packet. NALU type:" << (int) nalu_header.type; + return false; } + return true; } // https://github.com/ireader/media-server/blob/master/librtp/payload/rtp-h265-unpack.c @@ -248,13 +250,13 @@ void RTPDecoder::h265_forward_one_nalu(const uint8_t* data, int data_size, bool m_nalu_data_length = 0; } -void RTPDecoder::parseRTPH265toNALU(const uint8_t* rtp_data, const size_t data_length) +bool RTPDecoder::parseRTPH265toNALU(const uint8_t* rtp_data, const size_t data_length) { // 12 rtp header bytes and 1 nalu_header_t type byte if (data_length <= sizeof(rtp_header_t) + sizeof(nal_unit_header_h265_t)) { MLOGD << "Not enough rtp data"; - return; + return false; } // MLOGD<<"Got h265 rtp data"; const RTP::RTPPacketH265 rtpPacket(rtp_data, data_length); @@ -262,13 +264,13 @@ void RTPDecoder::parseRTPH265toNALU(const uint8_t* rtp_data, const size_t data_l if (!validateRTPPacket(rtpPacket.header)) { MLOGD << "Invalid rtp packet"; - return; + return false; } const auto& nal_unit_header_h265 = rtpPacket.getNALUHeaderH265(); if (nal_unit_header_h265.type > 50) { MLOGD << "Unsupported (HEVC) NAL type " << (int) nal_unit_header_h265.type; - return; + return false; } if (nal_unit_header_h265.type == 48) { @@ -296,7 +298,7 @@ void RTPDecoder::parseRTPH265toNALU(const uint8_t* rtp_data, const size_t data_l break; } } - return; + return true; } else if (nal_unit_header_h265.type == 49) { @@ -346,6 +348,7 @@ void RTPDecoder::parseRTPH265toNALU(const uint8_t* rtp_data, const size_t data_l // h265_forward_one_nalu(rtpPacket.rtpPayload,rtpPacket.rtpPayloadSize,false); h265_forward_one_nalu(rtpPacket.rtpPayload, rtpPacket.rtpPayloadSize, true); } + return true; } // MJPEG diff --git a/app/videonative/src/main/cpp/parser/ParseRTP.h b/app/videonative/src/main/cpp/parser/ParseRTP.h index 41573c70..03e9adc3 100644 --- a/app/videonative/src/main/cpp/parser/ParseRTP.h +++ b/app/videonative/src/main/cpp/parser/ParseRTP.h @@ -42,10 +42,10 @@ class RTPDecoder bool validateRTPPacket(const rtp_header_t& rtpHeader); // parse rtp h264 packet to NALU - void parseRTPH264toNALU(const uint8_t* rtp_data, const size_t data_length); + bool parseRTPH264toNALU(const uint8_t* rtp_data, const size_t data_length); // parse rtp h265 packet to NALU - void parseRTPH265toNALU(const uint8_t* rtp_data, const size_t data_length); + bool parseRTPH265toNALU(const uint8_t* rtp_data, const size_t data_length); // exp void parse_rtp_mjpeg(const uint8_t* rtp_data, const size_t data_length);