StreamTranscoder: flush filter graph buffers after decoding process

Valentin NOEL · Valentin NOEL · commit 82382fc08192 · 2017-09-06T14:53:06.000+02:00
Add utility methods, documentation and logs into FilterGraph
and FrameBuffer classes
diff --git a/src/AvTranscoder/filter/FilterGraph.cpp b/src/AvTranscoder/filter/FilterGraph.cpp
@@ -38,6 +38,8 @@ FrameBuffer::~FrameBuffer()
 
 void FrameBuffer::addFrame(IFrame* frame)
 {
+    LOG_DEBUG("Add a new frame to frame buffer. New buffer size: " << _frameQueue.size() + 1);
+    // Copy the input frame to store it into the queue
     AudioFrame* newAudioFrame = new AudioFrame(_audioFrameDesc, false);
     const size_t expectedNbSamples = frame->getDataSize() / (newAudioFrame->getNbChannels() * newAudioFrame->getBytesPerSample());
     newAudioFrame->setNbSamplesPerChannel(expectedNbSamples);
@@ -51,16 +53,19 @@ void FrameBuffer::addFrame(IFrame* frame)
 void FrameBuffer::popFrame()
 {
     _frameQueue.pop();
+    LOG_DEBUG("Pop frame from buffer. Remaining frames in buffer: " << _frameQueue.size());
 }
 
 IFrame* FrameBuffer::getFrame(const size_t size)
 {
+    LOG_DEBUG("Get a " << size << " bytes frame from a " << _totalDataSize << " bytes frame buffer");
     IFrame* next = _frameQueue.front();
     const size_t nextFrameSize = next->getDataSize();
 
     // If no expected size, or if the expected size equals the front frame of the queue (with no offset)
     if(size == 0 || (size == nextFrameSize && _positionInFrontFrame == 0))
     {
+        // Directly return the front frame of the queue
         _totalDataSize -= nextFrameSize;
         popFrame();
         return next;
@@ -77,30 +82,34 @@ IFrame* FrameBuffer::getFrame(const size_t size)
     unsigned char* outputData = new unsigned char[size];
     while(extractedDataSize != size && _frameQueue.size() != 0)
     {
+        // Get the front frame from queue
         next = _frameQueue.front();
-        size_t dataToGet = size - extractedDataSize;
-        size_t remainingDataInNextFrame = next->getDataSize() - _positionInFrontFrame;
+        size_t remainingDataInFrontFrame = next->getDataSize() - _positionInFrontFrame;
 
-        if(dataToGet > remainingDataInNextFrame)
-            dataToGet = remainingDataInNextFrame;
+        // Compute the data size to get from the frame
+        size_t dataToGet = size - extractedDataSize;
+        if(dataToGet > remainingDataInFrontFrame)
+            dataToGet = remainingDataInFrontFrame;
 
+        // Copy the data from the frame to temporal buffer
         for(size_t i = 0; i < dataToGet; i++)
             outputData[extractedDataSize++] = next->getData()[0][_positionInFrontFrame + i];
 
-        if(dataToGet < remainingDataInNextFrame)
+        if(dataToGet < remainingDataInFrontFrame)
         {
+            // Set new position into front frame
             _positionInFrontFrame += dataToGet;
         }
         else
         {
+            // The whole front frame has been read, so pop it from queue
             popFrame();
             _positionInFrontFrame = 0;
         }
     }
 
     _totalDataSize -= extractedDataSize;
     newAudioFrame->assignBuffer(outputData);
-
     return newAudioFrame;
 }
 
@@ -155,6 +164,27 @@ size_t FilterGraph::getMinInputFrameSize(const std::vector<IFrame*>& inputs)
     return minFrameSize;
 }
 
+bool FilterGraph::hasBufferedFrames()
+{
+    if(!_inputFramesBuffer.size())
+        return false;
+
+    for(std::vector<FrameBuffer>::iterator it = _inputFramesBuffer.begin(); it != _inputFramesBuffer.end(); ++it)
+    {
+        if(it->isEmpty())
+            return false;
+    }
+    return true;
+}
+
+bool FilterGraph::hasBufferedFrames(const size_t index)
+{
+    if(index >= _inputFramesBuffer.size())
+        return false;
+
+    return !_inputFramesBuffer.at(index).isEmpty();
+}
+
 bool FilterGraph::areInputFrameSizeEqual(const std::vector<IFrame*>& inputs)
 {
     if(!inputs.size() || inputs.size() == 1)
@@ -196,7 +226,14 @@ void FilterGraph::process(const std::vector<IFrame*>& inputs, IFrame& output)
     {
         // Fill the frame buffer with inputs
         for(size_t index = 0; index < inputs.size(); ++index)
+        {
+            if(!inputs.at(index)->getDataSize())
+            {
+                LOG_DEBUG("Empty frame from filter graph input " << index << ". Remaining frames in buffer: " << _inputFramesBuffer.at(index).getBufferSize());
+                continue;
+            }
             _inputFramesBuffer.at(index).addFrame(inputs.at(index));
+        }
 
         // Get the minimum input frames size
         minInputFrameSize = getMinInputFrameSize(inputs);
@@ -206,12 +243,8 @@ void FilterGraph::process(const std::vector<IFrame*>& inputs, IFrame& output)
     // Setup input frames into the filter graph
     for(size_t index = 0; index < inputs.size(); ++index)
     {
-        IFrame* inputFrame = NULL;
-        if(bypassBuffers)
-            inputFrame = inputs.at(index);
-        else
-            inputFrame = _inputFramesBuffer.at(index).getFrame(minInputFrameSize);
-
+        // Retrieve frame from buffer or directly from input
+        IFrame* inputFrame = (bypassBuffers)? inputs.at(index) : _inputFramesBuffer.at(index).getFrame(minInputFrameSize);
         const int ret = av_buffersrc_add_frame_flags(_filters.at(index)->getAVFilterContext(), &inputFrame->getAVFrame(), AV_BUFFERSRC_FLAG_PUSH);
 
         if(ret < 0)
diff --git a/src/AvTranscoder/filter/FilterGraph.hpp b/src/AvTranscoder/filter/FilterGraph.hpp
@@ -35,6 +35,10 @@ class FrameBuffer
      * @brief Return the total amount of data contained in the frames of the buffer.
      */
     size_t getDataSize() const { return _totalDataSize; }
+    /**
+     * @brief Return the number of frames contained in the buffer.
+     */
+    size_t getBufferSize() const { return _frameQueue.size(); }
 
     /**
      * @brief Push a frame at the end of the buffer.
@@ -107,6 +111,9 @@ class AvExport FilterGraph
      */
     bool hasFilters() const { return !_filters.empty(); }
 
+    bool hasBufferedFrames();
+    bool hasBufferedFrames(const size_t index);
+
 private:
     /**
      * @brief Initialize the graph of filters to process.
diff --git a/src/AvTranscoder/transcoder/StreamTranscoder.cpp b/src/AvTranscoder/transcoder/StreamTranscoder.cpp
@@ -533,7 +533,7 @@ bool StreamTranscoder::processTranscode()
 
     // Decode
     LOG_DEBUG("Decode next frame")
-    bool decodingStatus = true;
+    std::vector<bool> decodingStatus(_generators.size(), true);
     for(size_t index = 0; index < _generators.size(); ++index)
     {
         if(getProcessCase() == eProcessCaseTranscode)
@@ -542,15 +542,17 @@ bool StreamTranscoder::processTranscode()
             _currentDecoder = _generators.at(index);
 
         if(! _inputStreamDesc.empty() && _inputStreamDesc.at(index).demultiplexing())
-            decodingStatus = decodingStatus && _currentDecoder->decodeNextFrame(*_decodedData.at(index), _inputStreamDesc.at(index)._channelIndexArray);
+            decodingStatus.at(index) = decodingStatus.at(index) && _currentDecoder->decodeNextFrame(*_decodedData.at(index), _inputStreamDesc.at(index)._channelIndexArray);
         else
-            decodingStatus = decodingStatus && _currentDecoder->decodeNextFrame(*_decodedData.at(index));
+            decodingStatus.at(index) = decodingStatus.at(index) && _currentDecoder->decodeNextFrame(*_decodedData.at(index));
     }
 
     // check the next data buffers in case of audio frames
     if(_decodedData.at(0)->isAudioFrame())
     {
         const int nbInputSamplesPerChannel = _decodedData.at(0)->getAVFrame().nb_samples;
+
+        // Reallocate output frame
         if(nbInputSamplesPerChannel > _filteredData->getAVFrame().nb_samples)
         {
             LOG_WARN("The buffer of filtered data corresponds to a frame of " << _filteredData->getAVFrame().nb_samples << " samples. The decoded buffer contains " << nbInputSamplesPerChannel << " samples. Reallocate it.")
@@ -569,7 +571,21 @@ bool StreamTranscoder::processTranscode()
 
     // Transform
     CodedData data;
-    if(decodingStatus)
+    bool continueProcess = true;
+    for(size_t index = 0; index < decodingStatus.size(); ++index)
+    {
+        if(!decodingStatus.at(index))
+        {
+            if(!_filterGraph->hasFilters() || !_filterGraph->hasBufferedFrames(index))
+            {
+                continueProcess = false;
+                continue;
+            }
+            LOG_DEBUG("Some frames remain into filter graph buffer " << index);
+        }
+    }
+
+    if(continueProcess)
     {
         IFrame* dataToTransform = NULL;
         if(_filterGraph->hasFilters())
