#pragma warning (disable : 4244)\r
#endif\r
\r
-#include "..\stdafx.h"\r
+#include "../stdafx.h"\r
\r
#include "input.h"\r
#include "../ffmpeg_error.h"\r
\r
#include <core/video_format.h>\r
\r
-#include <common/concurrency/executor.h>\r
#include <common/diagnostics/graph.h>\r
+#include <common/exception/exceptions.h>\r
+#include <common/exception/win32_exception.h>\r
\r
#include <tbb/concurrent_queue.h>\r
-#include <tbb/mutex.h>\r
+#include <tbb/atomic.h>\r
\r
-#include <boost/range/iterator_range.hpp>\r
#include <boost/range/algorithm.hpp>\r
+#include <boost/thread/condition_variable.hpp>\r
+#include <boost/thread/mutex.hpp>\r
+#include <boost/thread/thread.hpp>\r
\r
+#if defined(_MSC_VER)\r
+#pragma warning (push)\r
+#pragma warning (disable : 4244)\r
+#endif\r
extern "C" \r
{\r
#define __STDC_CONSTANT_MACROS\r
#define __STDC_LIMIT_MACROS\r
#include <libavformat/avformat.h>\r
}\r
+#if defined(_MSC_VER)\r
+#pragma warning (pop)\r
+#endif\r
\r
-namespace caspar {\r
- \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
-class stream\r
-{\r
- std::shared_ptr<AVCodecContext> ctx_;\r
- int index_;\r
- tbb::concurrent_bounded_queue<std::shared_ptr<AVPacket>> buffer_;\r
-\r
-public:\r
-\r
- stream()\r
- : index_(-1)\r
- {\r
- buffer_.set_capacity(PACKET_BUFFER_COUNT);\r
- }\r
-\r
- int open(std::shared_ptr<AVFormatContext>& fctx, AVMediaType media_type)\r
- { \r
- const auto streams = boost::iterator_range<AVStream**>(fctx->streams, fctx->streams+fctx->nb_streams);\r
- const auto stream = boost::find_if(streams, [&](AVStream* stream) \r
- {\r
- return stream && stream->codec->codec_type == media_type;\r
- });\r
- \r
- if(stream == streams.end()) \r
- return AVERROR_STREAM_NOT_FOUND;\r
- \r
- auto codec = avcodec_find_decoder((*stream)->codec->codec_id); \r
- if(!codec)\r
- return AVERROR_DECODER_NOT_FOUND;\r
- \r
- index_ = (*stream)->index;\r
-\r
- int errn = tbb_avcodec_open((*stream)->codec, codec);\r
- if(errn >= 0)\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<AVPacket> pop()\r
- {\r
- std::shared_ptr<AVPacket> pkt;\r
- buffer_.try_pop(pkt);\r
- return pkt;\r
- }\r
-\r
- void push(const std::shared_ptr<AVPacket>& pkt)\r
- {\r
- if(pkt->stream_index != index_)\r
- return;\r
-\r
- av_dup_packet(pkt.get());\r
- buffer_.push(pkt); \r
- }\r
-\r
- const std::shared_ptr<AVCodecContext>& ctx() { return ctx_; }\r
-\r
- operator bool(){return ctx_ != nullptr;}\r
-\r
- double fps() const { return !ctx_ ? -1.0 : static_cast<double>(ctx_->time_base.den) / static_cast<double>(ctx_->time_base.num); }\r
+namespace caspar { namespace ffmpeg {\r
\r
- bool empty() const { return buffer_.empty();}\r
- int size() const { return buffer_.size();}\r
-};\r
- \r
+static const size_t MAX_BUFFER_COUNT = 100;\r
+static const size_t MIN_BUFFER_COUNT = 4;\r
+static const size_t MAX_BUFFER_SIZE = 16 * 1000000;\r
+ \r
struct input::implementation : boost::noncopyable\r
{ \r
- safe_ptr<diagnostics::graph> graph_;\r
+ std::shared_ptr<AVFormatContext> format_context_; // Destroy this last\r
+ int default_stream_index_;\r
\r
- std::shared_ptr<AVFormatContext> format_context_; // Destroy this last\r
+ safe_ptr<diagnostics::graph> graph_;\r
\r
- const std::wstring filename_;\r
- const bool loop_;\r
- const int start_; \r
- double fps_;\r
-\r
- stream video_stream_;\r
- stream audio_stream_;\r
+ const std::wstring filename_;\r
+ const bool loop_;\r
+ const size_t start_; \r
+ const size_t length_;\r
+ size_t frame_number_;\r
+ \r
+ tbb::concurrent_bounded_queue<std::shared_ptr<AVPacket>> buffer_;\r
+ tbb::atomic<size_t> buffer_size_;\r
+ boost::condition_variable buffer_cond_;\r
+ boost::mutex buffer_mutex_;\r
\r
- std::exception_ptr exception_;\r
- executor executor_;\r
+ boost::thread thread_;\r
+ tbb::atomic<bool> is_running_;\r
+\r
+ tbb::atomic<size_t> nb_frames_;\r
+ tbb::atomic<size_t> nb_loops_;\r
+\r
public:\r
- explicit implementation(const safe_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop, int start) \r
+ explicit implementation(const safe_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop, size_t start, size_t length) \r
: graph_(graph)\r
, loop_(loop)\r
, filename_(filename)\r
- , executor_(print())\r
- , start_(std::max(start, 0))\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
+ , start_(start)\r
+ , length_(length)\r
+ , frame_number_(0)\r
+ { \r
+ is_running_ = true;\r
+ nb_frames_ = 0;\r
+ nb_loops_ = 0;\r
\r
- int errn;\r
-\r
AVFormatContext* weak_format_context_ = nullptr;\r
- 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
- msg_info(av_error_str(errn)) <<\r
- boost::errinfo_api_function("av_open_input_file") <<\r
- boost::errinfo_errno(AVUNERROR(errn)) <<\r
- boost::errinfo_file_name(narrow(filename)));\r
- }\r
+ THROW_ON_ERROR2(avformat_open_input(&weak_format_context_, narrow(filename).c_str(), nullptr, nullptr), print());\r
\r
format_context_.reset(weak_format_context_, av_close_input_file);\r
+\r
+ av_dump_format(weak_format_context_, 0, narrow(filename).c_str(), 0);\r
\r
- errn = av_find_stream_info(format_context_.get());\r
- if(errn < 0)\r
- { \r
- BOOST_THROW_EXCEPTION(\r
- file_read_error() << \r
- source_info(narrow(print())) << \r
- msg_info(av_error_str(errn)) <<\r
- boost::errinfo_api_function("av_find_stream_info") <<\r
- boost::errinfo_errno(AVUNERROR(errn)));\r
- }\r
- \r
- errn = video_stream_.open(format_context_, AVMEDIA_TYPE_VIDEO);\r
- if(errn < 0)\r
- CASPAR_LOG(warning) << print() << L" Could not open video stream: " << widen(av_error_str(errn));\r
- \r
- errn = audio_stream_.open(format_context_, AVMEDIA_TYPE_AUDIO);\r
- if(errn < 0)\r
- CASPAR_LOG(warning) << print() << L" Could not open audio stream: " << widen(av_error_str(errn));\r
+ THROW_ON_ERROR2(avformat_find_stream_info(format_context_.get(), nullptr), print());\r
\r
- if(!video_stream_ && !audio_stream_)\r
- { \r
- BOOST_THROW_EXCEPTION(\r
- file_read_error() << \r
- source_info(narrow(print())) << \r
- msg_info("No video or audio codec context found.")); \r
- }\r
-\r
- fps_ = video_stream_ ? video_stream_.fps() : audio_stream_.fps();\r
+ default_stream_index_ = THROW_ON_ERROR2(av_find_default_stream_index(format_context_.get()), print());\r
\r
- if(start_ != 0) \r
+ if(start_ > 0) \r
seek_frame(start_);\r
- \r
- executor_.start();\r
- executor_.begin_invoke([this]{read_file();});\r
- CASPAR_LOG(info) << print() << " Started.";\r
+ \r
+ for(int n = 0; n < 16 && !full(); ++n)\r
+ read_next_packet();\r
+ \r
+ graph_->set_color("seek", diagnostics::color(1.0f, 0.5f, 0.0f)); \r
+ graph_->set_color("buffer-count", diagnostics::color(0.7f, 0.4f, 0.4f));\r
+ graph_->set_color("buffer-size", diagnostics::color(1.0f, 1.0f, 0.0f)); \r
+\r
+ thread_ = boost::thread([this]{run();});\r
}\r
\r
~implementation()\r
{\r
- stop();\r
+ is_running_ = false;\r
+ buffer_cond_.notify_all();\r
+ thread_.join();\r
}\r
\r
- std::shared_ptr<AVPacket> get_video_packet()\r
+ bool try_pop(std::shared_ptr<AVPacket>& packet)\r
{\r
- return video_stream_.pop();\r
- }\r
+ const bool result = buffer_.try_pop(packet);\r
\r
- std::shared_ptr<AVPacket> get_audio_packet()\r
- {\r
- return audio_stream_.pop();\r
+ if(result)\r
+ {\r
+ if(packet)\r
+ buffer_size_ -= packet->size;\r
+ buffer_cond_.notify_all();\r
+ }\r
+\r
+ graph_->update_value("buffer-size", (static_cast<double>(buffer_size_)+0.001)/MAX_BUFFER_SIZE);\r
+ graph_->update_value("buffer-count", (static_cast<double>(buffer_.size()+0.001)/MAX_BUFFER_COUNT));\r
+\r
+ return result;\r
}\r
\r
- bool has_packet() const\r
+ size_t nb_frames() const\r
{\r
- return !video_stream_.empty() || !audio_stream_.empty();\r
+ return nb_frames_;\r
}\r
- \r
- double fps()\r
+\r
+ size_t nb_loops() const\r
{\r
- return fps_;\r
+ return nb_loops_;\r
}\r
\r
private:\r
-\r
- void stop()\r
- {\r
- executor_.stop();\r
- get_video_packet();\r
- get_audio_packet();\r
- CASPAR_LOG(info) << print() << " Stopping.";\r
- }\r
- \r
- void read_file()\r
+ \r
+ void run()\r
{ \r
- if(audio_stream_.size() > 4 && video_stream_.size() > 4)\r
- Sleep(5); // There are enough packets, no hurry.\r
+ caspar::win32_exception::install_handler();\r
\r
try\r
{\r
- std::shared_ptr<AVPacket> read_packet(new AVPacket(), [](AVPacket* p)\r
- {\r
- av_free_packet(p);\r
- delete p;\r
- });\r
+ CASPAR_LOG(info) << print() << " Thread Started.";\r
\r
- const int errn = av_read_frame(format_context_.get(), read_packet.get()); // read_packet is only valid until next call of av_read_frame.\r
- if(is_eof(errn)) // Use av_dup_packet to extend its life.\r
+ while(is_running_)\r
{\r
- if(loop_)\r
- {\r
- seek_frame(start_, AVSEEK_FLAG_BACKWARD);\r
- graph_->add_tag("seek"); \r
- CASPAR_LOG(info) << print() << " Received EOF. Looping."; \r
- } \r
- else\r
{\r
- stop();\r
- CASPAR_LOG(info) << print() << " Received EOF. Stopping.";\r
+ boost::unique_lock<boost::mutex> lock(buffer_mutex_);\r
+ while(full())\r
+ buffer_cond_.timed_wait(lock, boost::posix_time::millisec(20));\r
}\r
+ read_next_packet(); \r
}\r
- else if(errn < 0)\r
- {\r
- BOOST_THROW_EXCEPTION(\r
- file_read_error() <<\r
- msg_info(av_error_str(errn)) <<\r
- source_info(narrow(print())) << \r
- boost::errinfo_api_function("av_read_frame") <<\r
- boost::errinfo_errno(AVUNERROR(errn)));\r
- }\r
- else\r
- {\r
- video_stream_.push(read_packet);\r
- audio_stream_.push(read_packet);\r
- }\r
- \r
- graph_->update_value("input-buffer", static_cast<float>(std::max(video_stream_.size(), audio_stream_.size()))/static_cast<float>(PACKET_BUFFER_COUNT)); \r
+\r
+ CASPAR_LOG(info) << print() << " Thread Stopped.";\r
}\r
catch(...)\r
{\r
- stop();\r
CASPAR_LOG_CURRENT_EXCEPTION();\r
- return;\r
+ is_running_ = false;\r
}\r
- \r
- executor_.begin_invoke([this]{read_file();}); \r
}\r
+ \r
+ void read_next_packet()\r
+ { \r
+ int ret = 0;\r
\r
- void seek_frame(int64_t frame, int flags = 0)\r
- { \r
- // Convert from frames into seconds.\r
- const auto ts = frame*static_cast<int64_t>(AV_TIME_BASE/fps_);\r
-\r
- const int errn = av_seek_frame(format_context_.get(), -1, ts, flags | AVSEEK_FLAG_FRAME);\r
- if(errn < 0)\r
- { \r
- BOOST_THROW_EXCEPTION(\r
- invalid_operation() << \r
- source_info(narrow(print())) << \r
- msg_info(av_error_str(errn)) <<\r
- boost::errinfo_api_function("av_seek_frame") <<\r
- boost::errinfo_errno(AVUNERROR(errn)));\r
+ std::shared_ptr<AVPacket> read_packet(new AVPacket, [](AVPacket* p)\r
+ {\r
+ av_free_packet(p);\r
+ delete p;\r
+ });\r
+ av_init_packet(read_packet.get());\r
+\r
+ ret = av_read_frame(format_context_.get(), read_packet.get()); // read_packet is only valid until next call of av_read_frame. Use av_dup_packet to extend its life. \r
+ \r
+ if(is_eof(ret)) \r
+ {\r
+ ++nb_loops_;\r
+ frame_number_ = 0;\r
+\r
+ if(loop_)\r
+ {\r
+ int flags = AVSEEK_FLAG_BACKWARD;\r
+\r
+ int vid_stream_index = av_find_best_stream(format_context_.get(), AVMEDIA_TYPE_VIDEO, -1, -1, 0, 0);\r
+ if(vid_stream_index >= 0)\r
+ {\r
+ auto codec_id = format_context_->streams[vid_stream_index]->codec->codec_id;\r
+ if(codec_id == CODEC_ID_VP6A || codec_id == CODEC_ID_VP6F || codec_id == CODEC_ID_VP6)\r
+ flags |= AVSEEK_FLAG_BYTE;\r
+ }\r
+\r
+ seek_frame(start_, flags);\r
+ graph_->add_tag("seek"); \r
+ CASPAR_LOG(debug) << print() << " Looping."; \r
+ } \r
+ else\r
+ {\r
+ is_running_ = false;\r
+ CASPAR_LOG(debug) << print() << " Stopping.";\r
+ }\r
}\r
+ else\r
+ { \r
+ THROW_ON_ERROR(ret, print(), "av_read_frame");\r
+\r
+ if(read_packet->stream_index == default_stream_index_)\r
+ {\r
+ if(nb_loops_ == 0)\r
+ ++nb_frames_;\r
+ ++frame_number_;\r
+ }\r
+\r
+ THROW_ON_ERROR2(av_dup_packet(read_packet.get()), print());\r
+ \r
+ // Make sure that the packet is correctly deallocated even if size and data is modified during decoding.\r
+ auto size = read_packet->size;\r
+ auto data = read_packet->data;\r
+\r
+ read_packet = std::shared_ptr<AVPacket>(read_packet.get(), [=](AVPacket*)\r
+ {\r
+ read_packet->size = size;\r
+ read_packet->data = data;\r
+ });\r
+\r
+ buffer_.try_push(read_packet);\r
+ buffer_size_ += read_packet->size;\r
+ \r
+ graph_->update_value("buffer-size", (static_cast<double>(buffer_size_)+0.001)/MAX_BUFFER_SIZE);\r
+ graph_->update_value("buffer-count", (static_cast<double>(buffer_.size()+0.001)/MAX_BUFFER_COUNT));\r
+ } \r
+ }\r
+\r
+ bool full() const\r
+ {\r
+ return is_running_ && (buffer_size_ > MAX_BUFFER_SIZE || buffer_.size() > MAX_BUFFER_COUNT) && buffer_.size() > MIN_BUFFER_COUNT;\r
+ }\r
+\r
+ void seek_frame(int64_t frame, int flags = 0)\r
+ { \r
+ THROW_ON_ERROR2(av_seek_frame(format_context_.get(), default_stream_index_, frame, flags), print()); \r
+ buffer_.push(nullptr);\r
} \r
\r
- bool is_eof(int errn)\r
+ bool is_eof(int ret)\r
{\r
- if(errn == AVERROR(EIO))\r
- CASPAR_LOG(warning) << print() << " Received EIO, assuming EOF";\r
+ if(ret == AVERROR(EIO))\r
+ CASPAR_LOG(trace) << print() << " Received EIO, assuming EOF. " << nb_frames_;\r
+ if(ret == AVERROR_EOF)\r
+ CASPAR_LOG(debug) << print() << " Received EOF. " << nb_frames_;\r
\r
- return errn == AVERROR_EOF || errn == AVERROR(EIO); // av_read_frame doesn't always correctly return AVERROR_EOF;\r
+ return ret == AVERROR_EOF || ret == AVERROR(EIO) || frame_number_ >= length_; // av_read_frame doesn't always correctly return AVERROR_EOF;\r
}\r
\r
std::wstring print() const\r
{\r
- return L"ffmpeg_input[" + filename_ + L"]";\r
+ return L"ffmpeg_input[" + filename_ + L")]";\r
}\r
};\r
\r
-input::input(const safe_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop, int start) \r
- : impl_(new implementation(graph, filename, loop, start)){}\r
-const std::shared_ptr<AVCodecContext>& input::get_video_codec_context() const{return impl_->video_stream_.ctx();}\r
-const std::shared_ptr<AVCodecContext>& input::get_audio_codec_context() const{return impl_->audio_stream_.ctx();}\r
-bool input::has_packet() const{return impl_->has_packet();}\r
-bool input::is_running() const {return impl_->executor_.is_running();}\r
-std::shared_ptr<AVPacket> input::get_video_packet(){return impl_->get_video_packet();}\r
-std::shared_ptr<AVPacket> input::get_audio_packet(){return impl_->get_audio_packet();}\r
-double input::fps() const { return impl_->fps(); }\r
-}
\ No newline at end of file
+input::input(const safe_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop, size_t start, size_t length) \r
+ : impl_(new implementation(graph, filename, loop, start, length)){}\r
+bool input::eof() const {return !impl_->is_running_;}\r
+bool input::try_pop(std::shared_ptr<AVPacket>& packet){return impl_->try_pop(packet);}\r
+safe_ptr<AVFormatContext> input::context(){return make_safe(impl_->format_context_);}\r
+size_t input::nb_frames() const {return impl_->nb_frames();}\r
+size_t input::nb_loops() const {return impl_->nb_loops();}\r
+}}
\ No newline at end of file
projects / casparcg / blobdiff