* along with CasparCG. If not, see <http://www.gnu.org/licenses/>.\r
*\r
*/\r
-#include "..\stdafx.h"\r
+#if defined(_MSC_VER)\r
+#pragma warning (disable : 4244)\r
+#endif\r
+\r
+#include "../stdafx.h"\r
\r
#include "input.h"\r
+#include "util.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/queuing_mutex.h>\r
+#include <tbb/atomic.h>\r
\r
-#include <boost/exception/error_info.hpp>\r
-#include <boost/thread/once.hpp>\r
-#include <boost/thread/thread.hpp>\r
+#include <agents.h>\r
+#include <concrt_extras.h>\r
+#include <semaphore.h>\r
\r
-#include <errno.h>\r
-#include <system_error>\r
- \r
#if defined(_MSC_VER)\r
+#pragma warning (push)\r
#pragma warning (disable : 4244)\r
#endif\r
-\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
-struct input::implementation : boost::noncopyable\r
-{ \r
- static const size_t PACKET_BUFFER_COUNT = 50;\r
-\r
- safe_ptr<diagnostics::graph> graph_;\r
+using namespace Concurrency;\r
\r
- std::shared_ptr<AVFormatContext> format_context_; // Destroy this last\r
+namespace caspar { namespace ffmpeg {\r
\r
- std::shared_ptr<AVCodecContext> video_codec_context_;\r
- std::shared_ptr<AVCodecContext> audio_codex_context_;\r
+static const size_t MAX_TOKENS = 32;\r
\r
- const std::wstring filename_;\r
+struct input::implementation : public Concurrency::agent, boost::noncopyable\r
+{\r
+ input::target_t& target_;\r
\r
- bool loop_;\r
- int video_s_index_;\r
- int audio_s_index_;\r
+ const std::wstring filename_;\r
+ const safe_ptr<AVFormatContext> format_context_; // Destroy this last\r
+ int default_stream_index_;\r
+ const boost::iterator_range<AVStream**> streams_;\r
+\r
+ safe_ptr<diagnostics::graph> graph_;\r
\r
- tbb::concurrent_bounded_queue<std::shared_ptr<aligned_buffer>> video_packet_buffer_;\r
- tbb::concurrent_bounded_queue<std::shared_ptr<aligned_buffer>> audio_packet_buffer_;\r
+ const bool loop_;\r
+ const size_t start_; \r
+ const size_t length_;\r
+ size_t frame_number_;\r
+ \r
+ tbb::atomic<size_t> nb_frames_;\r
+ tbb::atomic<size_t> nb_loops_; \r
+ tbb::atomic<size_t> packets_count_;\r
+ tbb::atomic<size_t> packets_size_;\r
\r
- boost::condition_variable cond_;\r
- boost::mutex mutex_;\r
- \r
- executor executor_;\r
+ bool stop_;\r
+ \r
public:\r
- explicit implementation(const safe_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop) \r
- : graph_(graph)\r
- , loop_(loop)\r
- , video_s_index_(-1)\r
- , audio_s_index_(-1)\r
+ explicit implementation(input::target_t& target,\r
+ const safe_ptr<diagnostics::graph>& graph, \r
+ const std::wstring& filename, \r
+ bool loop, \r
+ size_t start,\r
+ size_t length)\r
+ : target_(target)\r
, filename_(filename)\r
- , executor_(print())\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
- int errn;\r
- AVFormatContext* weak_format_context_;\r
- if((errn = -av_open_input_file(&weak_format_context_, narrow(filename).c_str(), nullptr, 0, nullptr)) > 0)\r
- BOOST_THROW_EXCEPTION(\r
- file_read_error() << \r
- source_info(narrow(print())) << \r
- msg_info("No format context found.") << \r
- boost::errinfo_api_function("av_open_input_file") <<\r
- boost::errinfo_errno(errn) <<\r
- boost::errinfo_file_name(narrow(filename)));\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
- BOOST_THROW_EXCEPTION(\r
- file_read_error() << \r
- source_info(narrow(print())) << \r
- boost::errinfo_api_function("av_find_stream_info") <<\r
- msg_info("No stream found.") << \r
- boost::errinfo_errno(errn));\r
-\r
- video_codec_context_ = open_stream(CODEC_TYPE_VIDEO, video_s_index_);\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
- \r
- audio_codex_context_ = open_stream(CODEC_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(video_codec_context_.get());\r
-\r
- if(!video_codec_context_ && !audio_codex_context_)\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
- executor_.start();\r
- executor_.begin_invoke([this]{read_file();});\r
- CASPAR_LOG(info) << print() << " Started.";\r
+ , format_context_(open_input(filename)) \r
+ , default_stream_index_(av_find_default_stream_index(format_context_.get()))\r
+ , streams_(format_context_->streams, format_context_->streams + format_context_->nb_streams)\r
+ , graph_(graph)\r
+ , loop_(loop)\r
+ , start_(start)\r
+ , length_(length)\r
+ , frame_number_(0)\r
+ , stop_(false)\r
+ { \r
+ packets_count_ = 0;\r
+ packets_size_ = 0;\r
+ nb_frames_ = 0;\r
+ nb_loops_ = 0;\r
+ \r
+ av_dump_format(format_context_.get(), 0, narrow(filename).c_str(), 0);\r
+ \r
+ if(start_ > 0) \r
+ seek_frame(start_);\r
+ \r
+ graph_->set_color("seek", diagnostics::color(1.0f, 0.5f, 0.0f));\r
}\r
-\r
- ~implementation()\r
+ \r
+ void stop()\r
{\r
- executor_.clear();\r
- executor_.stop();\r
- cond_.notify_all();\r
- CASPAR_LOG(info) << print() << " Stopped.";\r
+ stop_ = true;\r
+ agent::wait(this);\r
}\r
- \r
- void fix_time_base(AVCodecContext* context) // Some files give an invalid numerator, try to fix it.\r
+ \r
+ virtual void run()\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
+ try\r
+ {\r
+ while(!stop_)\r
+ {\r
+ auto packet = read_next_packet();\r
+ if(!packet)\r
+ break;\r
+\r
+ Concurrency::asend(target_, make_safe_ptr(packet));\r
+ Concurrency::wait(40);\r
+ }\r
+ }\r
+ catch(...)\r
+ {\r
+ CASPAR_LOG_CURRENT_EXCEPTION();\r
+ } \r
+ \r
+ BOOST_FOREACH(auto stream, streams_)\r
+ Concurrency::send(target_, eof_packet(stream->index)); \r
+\r
+ done();\r
}\r
\r
- std::shared_ptr<AVCodecContext> open_stream(int codec_type, int& s_index)\r
+ std::shared_ptr<AVPacket> read_next_packet()\r
{ \r
- AVStream** streams_end = format_context_->streams+format_context_->nb_streams;\r
- AVStream** stream = std::find_if(format_context_->streams, streams_end, \r
- [&](AVStream* stream) { return stream != nullptr && stream->codec->codec_type == codec_type ;});\r
+ auto packet = create_packet();\r
\r
- if(stream == streams_end) \r
- return nullptr;\r
+ int ret = [&]() -> int\r
+ {\r
+ Concurrency::scoped_oversubcription_token oversubscribe;\r
+ return av_read_frame(format_context_.get(), packet.get()); // packet is only valid until next call of av_read_frame. Use av_dup_packet to extend its life. \r
+ }();\r
\r
- s_index = (*stream)->index;\r
- \r
- auto codec = avcodec_find_decoder((*stream)->codec->codec_id); \r
- if(codec == nullptr)\r
- return nullptr;\r
- \r
- if((-avcodec_open((*stream)->codec, codec)) > 0) \r
- return nullptr;\r
+ if(is_eof(ret)) \r
+ {\r
+ ++nb_loops_;\r
+ frame_number_ = 0;\r
\r
- return std::shared_ptr<AVCodecContext>((*stream)->codec, avcodec_close);\r
- }\r
- \r
- void read_file()\r
- { \r
- AVPacket tmp_packet;\r
- safe_ptr<AVPacket> read_packet(&tmp_packet, av_free_packet); \r
+ if(loop_)\r
+ {\r
+ CASPAR_LOG(trace) << print() << " Looping.";\r
+ seek_frame(start_, AVSEEK_FLAG_BACKWARD); \r
+ return read_next_packet();\r
+ } \r
+ else\r
+ {\r
+ CASPAR_LOG(trace) << print() << " Stopping.";\r
+ return nullptr;\r
+ }\r
+ }\r
+\r
+ THROW_ON_ERROR(ret, print(), "av_read_frame");\r
\r
- if (av_read_frame(format_context_.get(), read_packet.get()) >= 0) // NOTE: read_packet is only valid until next call of av_read_frame or av_close_input_file\r
+ if(packet->stream_index == default_stream_index_)\r
{\r
- auto packet = std::make_shared<aligned_buffer>(read_packet->data, read_packet->data + read_packet->size);\r
- if(read_packet->stream_index == video_s_index_) \r
- video_packet_buffer_.try_push(std::move(packet)); \r
- else if(read_packet->stream_index == audio_s_index_) \r
- audio_packet_buffer_.try_push(std::move(packet)); \r
+ if(nb_loops_ == 0)\r
+ ++nb_frames_;\r
+ ++frame_number_;\r
}\r
- else if(!loop_ || av_seek_frame(format_context_.get(), -1, 0, AVSEEK_FLAG_BACKWARD) < 0) // TODO: av_seek_frame does not work for all formats\r
+\r
+ THROW_ON_ERROR2(av_dup_packet(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 = packet->size;\r
+ auto data = packet->data; \r
+\r
+ packet = safe_ptr<AVPacket>(packet.get(), [=](AVPacket*)\r
{\r
- executor_.stop();\r
- CASPAR_LOG(info) << print() << " eof";\r
- } \r
- else\r
- graph_->add_tag("seek"); \r
- \r
- graph_->update_value("input-buffer", static_cast<float>(video_packet_buffer_.size())/static_cast<float>(PACKET_BUFFER_COUNT)); \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
- aligned_buffer get_video_packet()\r
- {\r
- return get_packet(video_packet_buffer_);\r
- }\r
+ packet->size = size;\r
+ packet->data = data;\r
+ --packets_count_;\r
+ });\r
\r
- aligned_buffer get_audio_packet()\r
- {\r
- return get_packet(audio_packet_buffer_);\r
- }\r
- \r
- aligned_buffer get_packet(tbb::concurrent_bounded_queue<std::shared_ptr<aligned_buffer>>& buffer)\r
- {\r
- cond_.notify_all();\r
- std::shared_ptr<aligned_buffer> packet;\r
- return buffer.try_pop(packet) ? std::move(*packet) : aligned_buffer();\r
+ ++packets_count_;\r
+ \r
+ return packet;\r
}\r
\r
- bool is_eof() const\r
- {\r
- return !executor_.is_running() && video_packet_buffer_.empty() && audio_packet_buffer_.empty();\r
- }\r
- \r
- double fps() const\r
+ void seek_frame(int64_t frame, int flags = 0)\r
+ { \r
+ if(flags == AVSEEK_FLAG_BACKWARD)\r
+ {\r
+ // Fix VP6 seeking\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
+\r
+ THROW_ON_ERROR2(av_seek_frame(format_context_.get(), default_stream_index_, frame, flags), print()); \r
+ auto packet = create_packet();\r
+ packet->size = 0;\r
+\r
+ BOOST_FOREACH(auto stream, streams_)\r
+ Concurrency::asend(target_, loop_packet(stream->index)); \r
+\r
+ graph_->add_tag("seek"); \r
+ } \r
+\r
+ bool is_eof(int ret)\r
{\r
- return static_cast<double>(video_codec_context_->time_base.den) / static_cast<double>(video_codec_context_->time_base.num);\r
- }\r
+ if(ret == AVERROR(EIO))\r
+ CASPAR_LOG(trace) << print() << " Received EIO, assuming EOF. " << nb_frames_;\r
+ if(ret == AVERROR_EOF)\r
+ CASPAR_LOG(trace) << print() << " Received EOF. " << nb_frames_;\r
+\r
+ CASPAR_VERIFY(ret >= 0 || ret == AVERROR_EOF || ret == AVERROR(EIO), ffmpeg_error() << source_info(narrow(print())));\r
\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";\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) : impl_(new implementation(graph, filename, loop)){}\r
-const std::shared_ptr<AVCodecContext>& input::get_video_codec_context() const{return impl_->video_codec_context_;}\r
-const std::shared_ptr<AVCodecContext>& input::get_audio_codec_context() const{return impl_->audio_codex_context_;}\r
-bool input::is_eof() const{return impl_->is_eof();}\r
-bool input::is_running() const {return impl_->executor_.is_running();}\r
-aligned_buffer input::get_video_packet(){return impl_->get_video_packet();}\r
-aligned_buffer 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(input::target_t& target,\r
+ const safe_ptr<diagnostics::graph>& graph, \r
+ const std::wstring& filename, \r
+ bool loop, \r
+ size_t start, \r
+ size_t length)\r
+ : impl_(new implementation(target, graph, filename, loop, start, length))\r
+{\r
+}\r
+\r
+safe_ptr<AVFormatContext> input::context()\r
+{\r
+ return safe_ptr<AVFormatContext>(impl_->format_context_);\r
+}\r
+\r
+size_t input::nb_frames() const\r
+{\r
+ return impl_->nb_frames_;\r
+}\r
+\r
+size_t input::nb_loops() const \r
+{\r
+ return impl_->nb_loops_;\r
+}\r
+\r
+void input::start()\r
+{\r
+ impl_->start();\r
+}\r
+\r
+void input::stop()\r
+{\r
+ impl_->stop();\r
+}\r
+\r
+}}
\ No newline at end of file