#include "../stdafx.h"\r
\r
#include "input.h"\r
+#include "util.h"\r
#include "../ffmpeg_error.h"\r
#include "../tbb_avcodec.h"\r
\r
#include <core/video_format.h>\r
\r
+#include <common/concrt/scoped_oversubscription_token.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/atomic.h>\r
\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
+#include <agents.h>\r
\r
#if defined(_MSC_VER)\r
#pragma warning (push)\r
#pragma warning (pop)\r
#endif\r
\r
+using namespace Concurrency;\r
+\r
namespace caspar { namespace ffmpeg {\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
+struct input::implementation : public Concurrency::agent, boost::noncopyable\r
{ \r
std::shared_ptr<AVFormatContext> format_context_; // Destroy this last\r
int default_stream_index_;\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
+ input::token_t& active_token_;\r
+ input::target_t& video_target_;\r
+ input::target_t& audio_target_;\r
\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
+ int video_index_;\r
+ int audio_index_;\r
+\r
public:\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
+ explicit implementation(input::token_t& active_token,\r
+ input::target_t& video_target,\r
+ input::target_t& audio_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
+ : active_token_(active_token)\r
+ , video_target_(video_target)\r
+ , audio_target_(audio_target)\r
+ , graph_(graph)\r
, loop_(loop)\r
, filename_(filename)\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
THROW_ON_ERROR2(avformat_find_stream_info(format_context_.get(), nullptr), print());\r
\r
default_stream_index_ = THROW_ON_ERROR2(av_find_default_stream_index(format_context_.get()), print());\r
-\r
+ video_index_ = av_find_best_stream(format_context_.get(), AVMEDIA_TYPE_VIDEO, -1, -1, 0, 0);\r
+ audio_index_ = av_find_best_stream(format_context_.get(), AVMEDIA_TYPE_AUDIO, -1, -1, 0, 0);\r
+ \r
if(start_ > 0) \r
seek_frame(start_);\r
\r
- for(int n = 0; n < 16 && !full(); ++n)\r
+ for(int n = 0; n < 16; ++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
+ agent::start();\r
}\r
\r
~implementation()\r
{\r
- is_running_ = false;\r
- buffer_cond_.notify_all();\r
- thread_.join();\r
- }\r
- \r
- bool try_pop(std::shared_ptr<AVPacket>& packet)\r
- {\r
- const bool result = buffer_.try_pop(packet);\r
-\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
- size_t nb_frames() const\r
- {\r
- return nb_frames_;\r
- }\r
-\r
- size_t nb_loops() const\r
- {\r
- return nb_loops_;\r
+ agent::wait(this);\r
}\r
-\r
-private:\r
\r
- void run()\r
- { \r
- caspar::win32_exception::install_handler();\r
-\r
+ virtual void run()\r
+ {\r
try\r
{\r
- CASPAR_LOG(info) << print() << " Thread Started.";\r
-\r
- while(is_running_)\r
+ while(Concurrency::receive(active_token_))\r
{\r
+ if(!read_next_packet())\r
{\r
- boost::unique_lock<boost::mutex> lock(buffer_mutex_);\r
- while(full())\r
- buffer_cond_.timed_wait(lock, boost::posix_time::millisec(20));\r
+ Concurrency::send(video_target_, eof_packet());\r
+ Concurrency::send(audio_target_, eof_packet());\r
+ break;\r
}\r
- read_next_packet(); \r
- }\r
-\r
- CASPAR_LOG(info) << print() << " Thread Stopped.";\r
+ } \r
}\r
catch(...)\r
{\r
CASPAR_LOG_CURRENT_EXCEPTION();\r
- is_running_ = false;\r
- }\r
+ } \r
+ \r
+ done();\r
}\r
- \r
- void read_next_packet()\r
+\r
+ bool read_next_packet()\r
{ \r
int ret = 0;\r
\r
- std::shared_ptr<AVPacket> read_packet(new AVPacket, [](AVPacket* p)\r
+ auto read_packet = create_packet();\r
+\r
{\r
- av_free_packet(p);\r
- delete p;\r
- });\r
- av_init_packet(read_packet.get());\r
+ Concurrency::scoped_oversubcription_token oversubscribe;\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
\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
} \r
else\r
{\r
- is_running_ = false;\r
CASPAR_LOG(trace) << print() << " Stopping.";\r
+ return false;\r
}\r
}\r
- else\r
+ else if(read_packet->stream_index == video_index_ || read_packet->stream_index == audio_index_)\r
{ \r
THROW_ON_ERROR(ret, print(), "av_read_frame");\r
\r
read_packet->size = size;\r
read_packet->data = data;\r
});\r
+ \r
+ if(read_packet->stream_index == video_index_)\r
+ Concurrency::send(video_target_, read_packet);\r
+ else if(read_packet->stream_index == audio_index_)\r
+ Concurrency::send(audio_target_, read_packet);\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
+ return true;\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
+ 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
+ Concurrency::send(video_target_, loop_packet()); \r
+ Concurrency::send(audio_target_, loop_packet());\r
} \r
\r
bool is_eof(int ret)\r
}\r
};\r
\r
-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
+input::input(token_t& active_token, \r
+ target_t& video_target, \r
+ target_t& audio_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(active_token, video_target, audio_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
}}
\ No newline at end of file
projects / casparcg / blobdiff