2.0.0.2: ffmpeg_input: Fixed memory leak wheen looping files with large audio vs...

git-svn-id: https://casparcg.svn.sourceforge.net/svnroot/casparcg/server/branches/2.0.0.2@796 362d55ac-95cf-4e76-9f9a-cbaa9c17b72d

diff --git a/common/concurrency/executor.h b/common/concurrency/executor.h
index 8b3d087a74dfa21042925f2679122daf0edc9b37..32ac8e76fc2dbfed16fe60f9a56c0280dc27e7af 100644 (file)
--- a/common/concurrency/executor.h
+++ b/common/concurrency/executor.h
@@ -102,6 +102,11 @@ public:
                is_running_ = false;    \r
                execution_queue_.try_push([]{});\r
        }\r
+\r
+       void wait()\r
+       {\r
+               invoke([]{});\r
+       }\r
        \r
        void clear() // noexcept\r
        {\r

diff --git a/modules/ffmpeg/producer/input.cpp b/modules/ffmpeg/producer/input.cpp
index aa1200a5b29716fec24e579720ba6d9a48d040ae..2fa87fb0bb7c71616b0168d945cabe3694cce7c3 100644 (file)
--- a/modules/ffmpeg/producer/input.cpp
+++ b/modules/ffmpeg/producer/input.cpp
@@ -49,7 +49,7 @@ namespace caspar {
                \r
 struct input::implementation : boost::noncopyable\r
 {              \r
-       static const size_t PACKET_BUFFER_COUNT = 50;\r
+       static const size_t PACKET_BUFFER_COUNT = 100; // Assume that av_read_frame distance between audio and video packets is less than PACKET_BUFFER_COUNT.\r
 \r
        safe_ptr<diagnostics::graph> graph_;\r
 \r
@@ -68,10 +68,7 @@ struct input::implementation : boost::noncopyable
                \r
        tbb::concurrent_bounded_queue<packet> video_packet_buffer_;\r
        tbb::concurrent_bounded_queue<packet> audio_packet_buffer_;\r
-\r
-       boost::condition_variable       cond_;\r
-       boost::mutex                            mutex_;\r
-       \r
+               \r
        std::exception_ptr exception_;\r
        executor executor_;\r
 public:\r
@@ -85,13 +82,16 @@ public:
                , start_(std::max(start, 0))\r
                , length_(length)\r
                , eof_count_(length)\r
-       {                       \r
+       {               \r
                graph_->set_color("input-buffer", diagnostics::color(1.0f, 1.0f, 0.0f));\r
                graph_->set_color("seek", diagnostics::color(0.5f, 1.0f, 0.5f));        \r
-\r
+               \r
                int errn;\r
+\r
                AVFormatContext* weak_format_context_ = nullptr;\r
-               if((errn = av_open_input_file(&weak_format_context_, narrow(filename).c_str(), nullptr, 0, nullptr)) < 0 || weak_format_context_ == nullptr)\r
+               errn = errn = av_open_input_file(&weak_format_context_, narrow(filename).c_str(), nullptr, 0, nullptr);\r
+               if(errn < 0 || weak_format_context_ == nullptr)\r
+               {       \r
                        BOOST_THROW_EXCEPTION(\r
                                file_read_error() << \r
                                source_info(narrow(print())) << \r
@@ -99,10 +99,12 @@ public:
                                boost::errinfo_api_function("av_open_input_file") <<\r
                                boost::errinfo_errno(AVUNERROR(errn)) <<\r
                                boost::errinfo_file_name(narrow(filename)));\r
+               }\r
 \r
                format_context_.reset(weak_format_context_, av_close_input_file);\r
                        \r
-               if((errn = av_find_stream_info(format_context_.get())) < 0)\r
+               errn = errn = av_find_stream_info(format_context_.get());\r
+               if(errn < 0)\r
                {       \r
                        BOOST_THROW_EXCEPTION(\r
                                file_read_error() << \r
@@ -112,26 +114,24 @@ public:
                                boost::errinfo_errno(AVUNERROR(errn)));\r
                }\r
                \r
-               video_codec_context_ = open_stream(AVMEDIA_TYPE_VIDEO, video_s_index_, true);\r
-               if(!video_codec_context_)\r
-                       CASPAR_LOG(warning) << print() << " Could not open any video stream.";\r
-               else\r
-                       fix_time_base(video_codec_context_.get());\r
+               errn = open_stream(video_codec_context_, AVMEDIA_TYPE_VIDEO, video_s_index_);\r
+               if(errn < 0 || !video_codec_context_)\r
+                       CASPAR_LOG(warning) << print() << L" Could not open video stream: " << widen(av_error_str(errn));\r
                \r
-               audio_codex_context_ = open_stream(AVMEDIA_TYPE_AUDIO, audio_s_index_);\r
-               if(!audio_codex_context_)\r
-                       CASPAR_LOG(warning) << print() << " Could not open any audio stream.";\r
-               else\r
-                       fix_time_base(audio_codex_context_.get());\r
+               errn = open_stream(audio_codex_context_, AVMEDIA_TYPE_AUDIO, audio_s_index_);\r
+               if(errn < 0 || !audio_codex_context_)\r
+                       CASPAR_LOG(warning) << print() << L" Could not open audio stream: " << widen(av_error_str(errn));\r
                \r
                if(!video_codec_context_ && !audio_codex_context_)\r
                {       \r
                        BOOST_THROW_EXCEPTION(\r
                                file_read_error() << \r
-                               msg_info(av_error_str(errn)) <<\r
                                source_info(narrow(print())) << \r
                                msg_info("No video or audio codec context found."));    \r
                }\r
+\r
+               video_packet_buffer_.set_capacity(PACKET_BUFFER_COUNT);\r
+               audio_packet_buffer_.set_capacity(PACKET_BUFFER_COUNT);\r
                \r
                if(start_ != 0)                 \r
                        seek_frame(start_);\r
@@ -143,9 +143,7 @@ public:
 \r
        ~implementation()\r
        {\r
-               executor_.clear();\r
-               executor_.stop();\r
-               cond_.notify_all();\r
+               stop();\r
        }\r
                \r
        packet get_video_packet()\r
@@ -173,16 +171,12 @@ private:
        void stop()\r
        {\r
                executor_.stop();\r
-               CASPAR_LOG(info) << print() << " Stopped.";\r
+               get_video_packet();\r
+               get_audio_packet();\r
+               CASPAR_LOG(info) << print() << " Stopping.";\r
        }\r
                        \r
-       void fix_time_base(AVCodecContext* context) const // Some files give an invalid numerator, try to fix it.\r
-       {\r
-               if(context && context->time_base.num == 1)\r
-                       context->time_base.num = static_cast<int>(std::pow(10.0, static_cast<int>(std::log10(static_cast<float>(context->time_base.den)))-1));\r
-       }\r
-\r
-       std::shared_ptr<AVCodecContext> open_stream(int codec_type, int& s_index, bool tbb = false) const\r
+       int open_stream(std::shared_ptr<AVCodecContext>& ctx, int codec_type, int& s_index) const\r
        {               \r
                const auto streams = boost::iterator_range<AVStream**>(format_context_->streams, format_context_->streams+format_context_->nb_streams);\r
                const auto stream = boost::find_if(streams, [&](AVStream* stream) \r
@@ -191,27 +185,24 @@ private:
                });\r
                \r
                if(stream == streams.end()) \r
-                       return nullptr;\r
+                       return AVERROR_STREAM_NOT_FOUND;\r
                \r
                auto codec = avcodec_find_decoder((*stream)->codec->codec_id);                  \r
                if(codec == nullptr)\r
-                       return nullptr;\r
+                       return AVERROR_DECODER_NOT_FOUND;\r
                        \r
                s_index = (*stream)->index;\r
 \r
-               if(!tbb)\r
-               {\r
-                       if(avcodec_open((*stream)->codec, codec) < 0)           \r
-                               return nullptr;\r
-                       return std::shared_ptr<AVCodecContext>((*stream)->codec, avcodec_close);\r
-               }\r
-               else\r
+               int errn = tbb_avcodec_open((*stream)->codec, codec);\r
+               if(errn >= 0)\r
                {\r
-                       if(tbb_avcodec_open((*stream)->codec, codec) < 0)               \r
-                               return nullptr;\r
-                       return std::shared_ptr<AVCodecContext>((*stream)->codec, tbb_avcodec_close);\r
-               }               \r
+                       ctx.reset((*stream)->codec, tbb_avcodec_close);\r
 \r
+                       // Some files give an invalid time_base numerator, try to fix it.\r
+                       if(ctx && ctx->time_base.num == 1)\r
+                               ctx->time_base.num = static_cast<int>(std::pow(10.0, static_cast<int>(std::log10(static_cast<float>(ctx->time_base.den)))-1));\r
+               }\r
+               return errn;    \r
        }\r
        \r
        std::shared_ptr<AVCodecContext>& get_default_context()\r
@@ -222,7 +213,7 @@ private:
        void read_file()\r
        {               \r
                if(audio_packet_buffer_.size() > 4 && video_packet_buffer_.size() > 4)\r
-                       boost::this_thread::yield(); // There is enough packets, no hurry.\r
+                       boost::this_thread::yield(); // There are enough packets, no hurry.\r
 \r
                try\r
                {\r
@@ -260,15 +251,9 @@ private:
                        else\r
                        {\r
                                if(read_packet->stream_index == video_s_index_)                 \r
-                               {                                       \r
-                                       av_dup_packet(read_packet.get());\r
-                                       video_packet_buffer_.try_push(packet(read_packet));     \r
-                               }\r
-                               else if(read_packet->stream_index)      \r
-                               {                                       \r
-                                       av_dup_packet(read_packet.get());\r
-                                       audio_packet_buffer_.try_push(packet(read_packet));     \r
-                               }\r
+                                       push_packet(video_packet_buffer_, read_packet);\r
+                               else if(read_packet->stream_index == audio_s_index_)    \r
+                                       push_packet(audio_packet_buffer_, read_packet);\r
                        }\r
                                                \r
                        graph_->update_value("input-buffer", static_cast<float>(video_packet_buffer_.size())/static_cast<float>(PACKET_BUFFER_COUNT));          \r
@@ -279,11 +264,8 @@ private:
                        CASPAR_LOG_CURRENT_EXCEPTION();\r
                        return;\r
                }\r
-                               \r
+                                               \r
                executor_.begin_invoke([this]{read_file();});           \r
-               boost::unique_lock<boost::mutex> lock(mutex_);\r
-               while(executor_.is_running() && audio_packet_buffer_.size() > PACKET_BUFFER_COUNT && video_packet_buffer_.size() > PACKET_BUFFER_COUNT)\r
-                       cond_.wait(lock);               \r
        }\r
 \r
        void seek_frame(int64_t frame, int flags = 0)\r
@@ -292,7 +274,6 @@ private:
                const auto ts = frame*static_cast<int64_t>((AV_TIME_BASE*get_default_context()->time_base.num) / get_default_context()->time_base.den);\r
 \r
                const int errn = av_seek_frame(format_context_.get(), -1, ts, flags | AVSEEK_FLAG_FRAME);\r
-\r
                if(errn < 0)\r
                {       \r
                        BOOST_THROW_EXCEPTION(\r
@@ -313,17 +294,22 @@ private:
                if(length_ != -1)\r
                        return get_default_context()->frame_number > eof_count_;                \r
 \r
-               if(-errn == EIO)\r
+               if(errn == AVERROR(EIO))\r
                        CASPAR_LOG(warning) << print() << " Received EIO, assuming EOF";\r
 \r
-               return errn == AVERROR_EOF || -errn == EIO; // av_read_frame doesn't always correctly return AVERROR_EOF;\r
+               return errn == AVERROR_EOF || errn == AVERROR(EIO); // av_read_frame doesn't always correctly return AVERROR_EOF;\r
        }\r
-               \r
+       \r
+       void push_packet(tbb::concurrent_bounded_queue<packet>& buffer, const std::shared_ptr<AVPacket>& read_packet)\r
+       {\r
+               av_dup_packet(read_packet.get());\r
+               buffer.push(packet(read_packet));       \r
+       }\r
+\r
        packet get_packet(tbb::concurrent_bounded_queue<packet>& buffer)\r
        {\r
                packet packet;\r
-               if(buffer.try_pop(packet))\r
-                       cond_.notify_all();\r
+               buffer.try_pop(packet);\r
                return packet;\r
        }\r
 \r