decoder: one access unit per MediaCodec input buffer

app/videonative/src/main/cpp/VideoDecoder.cpp

@@ -35,6 +35,8 @@ void VideoDecoder::setOutputSurface(JNIEnv* env, jobject surface, jint idx)
         }
         std::lock_guard<std::mutex> 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<std::mutex> 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<microseconds>(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<microseconds>(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");

app/videonative/src/main/cpp/VideoDecoder.h

@@ -12,6 +12,7 @@
 #include <atomic>
 #include <iostream>
 #include <thread>
+#include <vector>
 #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<uint8_t>                  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();

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)
         {

app/videonative/src/main/cpp/parser/H26XParser.cpp

@@ -22,22 +22,47 @@ H26XParser::H26XParser(NALU_DATA_CALLBACK onNewNALU)
 void H26XParser::reset()
 {
     mDecodeRTP.reset();
+    mNewestSeq                 = -1;
     nParsedNALUs               = 0;
     nParsedKonfigurationFrames = 0;
 }
 
 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);
+    }
+    if (!accepted)
+    {
+        return;
+    }
+    // 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<uint16_t>(seq - mNewestSeq);
+        newest = ahead != 0 && (ahead < 0x8000 || static_cast<uint16_t>(mNewestSeq - seq) > kSeqResyncWindow);
+    }
+    if (newest)
+    {
+        mNewestSeq = seq;
+    }
+    if (rtpPacket.header.marker && newest && onAccessUnitEnd)
+    {
+        onAccessUnitEnd();
     }
 }
 

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<void()> onAccessUnitEnd;
+
     void reset();
 
   public:
@@ -55,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);
 };