[executor] changed default to unbounded like in 2.0.7 and fixed a deadlock when capac...

[casparcg] / modules / ffmpeg / ffmpeg_pipeline_backend_internal.cpp

diff --git a/modules/ffmpeg/ffmpeg_pipeline_backend_internal.cpp b/modules/ffmpeg/ffmpeg_pipeline_backend_internal.cpp
index 6fe13ee7e5c02b01d072d23c1ad07ca7e10793d5..0329a1662bb012a92595ffb278a2aecbd7bd81f5 100644 (file)
--- a/modules/ffmpeg/ffmpeg_pipeline_backend_internal.cpp
+++ b/modules/ffmpeg/ffmpeg_pipeline_backend_internal.cpp
@@ -182,7 +182,7 @@ public:
                {
                        try
                        {
-                               audio_decoders_.push_back(spl::make_shared<audio_decoder>(*input_, core::video_format_desc(), i));
+                               audio_decoders_.push_back(spl::make_shared<audio_decoder>(*input_, core::video_format::invalid, i));
                        }
                        catch (...)
                        {
@@ -357,7 +357,7 @@ public:
                                {
                                        frame = (*a_decoder)();
 
-                                       if (frame && frame->data[0])
+                                       if (frame == flush() || (frame && frame->data[0]))
                                                break;
                                        else
                                                frame.reset();
@@ -376,7 +376,7 @@ public:
                        {
                                frame = (*v_decoder)();
 
-                               if (frame && frame->data[0])
+                               if (frame == flush() || (frame && frame->data[0]))
                                        return { frame };
                        }
                }
@@ -604,11 +604,12 @@ struct sink
        virtual void                                                    framerate(boost::rational<int> framerate)                                                                               { CASPAR_THROW_EXCEPTION(invalid_operation() << msg_info(print() + L" not an encoder.")); }
        virtual void                                                    start(bool has_audio, bool has_video)                                                                                   { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
        virtual void                                                    stop()                                                                                                                                                  { }
+       virtual void                                                    flush_all()                                                                                                                                             { }
        virtual std::vector<AVSampleFormat>             supported_sample_formats() const                                                                                                { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
        virtual std::vector<int>                                supported_samplerates() const                                                                                                   { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
        virtual std::vector<AVPixelFormat>              supported_pixel_formats() const                                                                                                 { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
        virtual int                                                             wanted_num_audio_streams() const                                                                                                { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
-       virtual boost::optional<int>                    wanted_num_channels_per_stream() const                                                                          { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
+       virtual boost::optional<int>                    wanted_num_channels_per_stream() const                                                                                  { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
        virtual boost::optional<AVMediaType>    try_push(AVMediaType type, int stream_index, spl::shared_ptr<AVFrame> frame)    { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
        virtual void                                                    eof()                                                                                                                                                   { CASPAR_THROW_EXCEPTION(not_implemented() << msg_info(print())); }
 };
@@ -653,6 +654,7 @@ class memory_sink : public sink
        core::mutable_audio_buffer                                              audio_samples_;
 
        std::queue<std::shared_ptr<AVFrame>>                    video_frames_;
+       std::shared_ptr<AVFrame>                                                last_video_frame_;
 
        tbb::concurrent_bounded_queue<core::draw_frame> output_frames_;
        tbb::atomic<bool>                                                               running_;
@@ -700,8 +702,7 @@ public:
        void stop() override
        {
                running_ = false;
-               try_pop_frame();
-               try_pop_frame();
+               output_frames_.set_capacity(4);
        }
 
        std::vector<AVSampleFormat> supported_sample_formats() const override
@@ -739,6 +740,14 @@ public:
                return boost::none;
        }
 
+       void flush_all() override
+       {
+               audio_samples_.clear();
+
+               while (!video_frames_.empty())
+                       video_frames_.pop();
+       }
+
        boost::optional<AVMediaType> try_push(AVMediaType type, int stream_index, spl::shared_ptr<AVFrame> av_frame) override
        {
                if (!has_audio_ && !has_video_)
@@ -808,6 +817,7 @@ public:
                        }
 
                        auto output_frame = make_frame(this, spl::make_shared_ptr(video_frames_.front()), *factory_, channel_layout_);
+                       last_video_frame_ = video_frames_.front();
                        video_frames_.pop();
                        output_frame.audio_data() = std::move(audio_data);
 
@@ -824,9 +834,15 @@ public:
 
                        audio_data.swap(audio_samples_);
 
+                       if (video_frames_.empty() && !audio_data.empty() && last_video_frame_) // More audio samples than video
+                       {
+                               video_frames_.push(last_video_frame_);
+                       }
+
                        if (!video_frames_.empty())
                        {
                                auto output_frame = make_frame(this, spl::make_shared_ptr(video_frames_.front()), *factory_, channel_layout_);
+                               last_video_frame_ = video_frames_.front();
                                video_frames_.pop();
                                output_frame.audio_data() = std::move(audio_data);
 
@@ -1111,8 +1127,21 @@ private:
                        {
                                auto needed                                             = *result;
                                auto input_frames_for_streams   = source_->get_input_frames_for_streams(needed);
+                               bool flush_all                                  = !input_frames_for_streams.empty() && input_frames_for_streams.at(0) == flush();
 
-                               if (!input_frames_for_streams.empty() && input_frames_for_streams.at(0))
+                               if (flush_all)
+                               {
+                                       sink_->flush_all();
+                                       
+                                       if (source_->has_audio() && source_->has_video())
+                                               result = needed == AVMediaType::AVMEDIA_TYPE_AUDIO ? AVMediaType::AVMEDIA_TYPE_VIDEO : AVMediaType::AVMEDIA_TYPE_AUDIO;
+
+                                       continue;
+                               }
+
+                               bool got_requested_media_type   = !input_frames_for_streams.empty() && input_frames_for_streams.at(0);
+
+                               if (got_requested_media_type)
                                {
                                        for (int input_stream_index = 0; input_stream_index < input_frames_for_streams.size(); ++input_stream_index)
                                        {
@@ -1195,9 +1224,13 @@ private:
                        auto& av_frame  = *av_frames_per_stream.at(i);
                        auto& stream    = source_audio_streams_.at(i);
 
+                       auto channel_layout = av_frame.channel_layout == 0
+                                       ? av_get_default_channel_layout(av_frame.channels)
+                                       : av_frame.channel_layout;
+
                        set_if_changed(changed, stream.sampleformat, static_cast<AVSampleFormat>(av_frame.format));
                        set_if_changed(changed, stream.num_channels, av_frame.channels);
-                       set_if_changed(changed, stream.channel_layout, av_frame.channel_layout);
+                       set_if_changed(changed, stream.channel_layout, channel_layout);
                }
 
                if (changed)