\r
#include <core/video_format.h>\r
\r
-#include <common/concurrency/executor.h>\r
#include <common/diagnostics/graph.h>\r
\r
#include <tbb/concurrent_queue.h>\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
#include <boost/range/iterator_range.hpp>\r
\r
extern "C" \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
\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
-\r
- tbb::atomic<size_t> buffer_size_;\r
- boost::condition_variable cond_;\r
- boost::mutex mutex_;\r
-\r
- tbb::concurrent_bounded_queue<std::shared_ptr<AVPacket>> buffer_;\r
+ const std::wstring filename_;\r
+ const bool loop_;\r
+ const int start_; \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
- executor executor_;\r
+ boost::thread thread_;\r
+ tbb::atomic<bool> is_running_;\r
public:\r
explicit implementation(const safe_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop, int start) \r
: graph_(graph)\r
, loop_(loop)\r
, filename_(filename)\r
- , executor_(print())\r
, start_(std::max(start, 0))\r
{ \r
- int errn;\r
+ is_running_ = true;\r
+\r
+ int ret;\r
\r
AVFormatContext* weak_format_context_ = nullptr;\r
- errn = avformat_open_input(&weak_format_context_, narrow(filename).c_str(), nullptr, nullptr);\r
- if(errn < 0 || weak_format_context_ == nullptr)\r
+ ret = avformat_open_input(&weak_format_context_, narrow(filename).c_str(), nullptr, nullptr);\r
+ if(ret < 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
+ msg_info(av_error_str(ret)) <<\r
boost::errinfo_api_function("av_open_input_file") <<\r
- boost::errinfo_errno(AVUNERROR(errn)) <<\r
+ boost::errinfo_errno(AVUNERROR(ret)) <<\r
boost::errinfo_file_name(narrow(filename)));\r
}\r
\r
format_context_.reset(weak_format_context_, av_close_input_file);\r
\r
- errn = avformat_find_stream_info(format_context_.get(), nullptr);\r
- if(errn < 0)\r
+ ret = avformat_find_stream_info(format_context_.get(), nullptr);\r
+ if(ret < 0)\r
{ \r
BOOST_THROW_EXCEPTION(\r
file_read_error() << \r
source_info(narrow(print())) << \r
- msg_info(av_error_str(errn)) <<\r
+ msg_info(av_error_str(ret)) <<\r
boost::errinfo_api_function("av_find_stream_info") <<\r
- boost::errinfo_errno(AVUNERROR(errn)));\r
+ boost::errinfo_errno(AVUNERROR(ret)));\r
}\r
\r
if(start_ != 0) \r
seek_frame(start_);\r
\r
- for(int n = 0; n < 16 && buffer_size_ < MAX_BUFFER_SIZE && buffer_.size() < MAX_BUFFER_COUNT; ++n)\r
+ for(int n = 0; n < 16 && !full(); ++n)\r
read_next_packet();\r
-\r
- buffer_.set_capacity(MAX_BUFFER_COUNT);\r
- \r
+ \r
graph_->set_color("seek", diagnostics::color(0.5f, 1.0f, 0.5f)); \r
graph_->set_color("buffer-count", diagnostics::color(0.2f, 0.8f, 1.0f));\r
graph_->set_color("buffer-size", diagnostics::color(0.2f, 0.4f, 1.0f)); \r
\r
- executor_.begin_invoke([this]{read_file();});\r
- CASPAR_LOG(info) << print() << " Started.";\r
+ thread_ = boost::thread([this]{run();});\r
}\r
\r
~implementation()\r
{\r
- stop();\r
- // Unblock thread.\r
- std::shared_ptr<AVPacket> packet;\r
- buffer_.try_pop(packet);\r
- buffer_size_ = 0;\r
- cond_.notify_all();\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
- bool result = buffer_.try_pop(packet);\r
- graph_->update_value("buffer-count", MAX_BUFFER_SIZE/static_cast<double>(buffer_.size()));\r
+ const bool result = buffer_.try_pop(packet);\r
+\r
if(result)\r
{\r
if(packet)\r
buffer_size_ -= packet->size;\r
- graph_->update_value("buffer-size", MAX_BUFFER_SIZE/static_cast<double>(buffer_size_));\r
- cond_.notify_all();\r
+ buffer_cond_.notify_all();\r
}\r
- return result;\r
- }\r
\r
-private:\r
+ graph_->update_value("buffer-size", MAX_BUFFER_SIZE/static_cast<double>(buffer_size_));\r
+ graph_->update_value("buffer-count", MAX_BUFFER_SIZE/static_cast<double>(buffer_.size()));\r
\r
- void stop()\r
- {\r
- executor_.stop();\r
- \r
- CASPAR_LOG(debug) << print() << " Stopping.";\r
+ return result;\r
}\r
\r
- void read_file()\r
- { \r
- read_next_packet();\r
- executor_.begin_invoke([this]{read_file();});\r
- }\r
- \r
- void read_next_packet()\r
+private:\r
+ \r
+ void run()\r
{ \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
- av_init_packet(read_packet.get());\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(trace) << print() << " Received EOF. Looping."; \r
- } \r
- else\r
- {\r
- stop();\r
- CASPAR_LOG(trace) << print() << " Received EOF. Stopping.";\r
+ boost::unique_lock<boost::mutex> lock(buffer_mutex_);\r
+ buffer_cond_.wait(lock, [this]{return !full();});\r
}\r
+ read_next_packet(); \r
}\r
- else if(errn < 0)\r
+\r
+ CASPAR_LOG(info) << print() << " Thread Stopped.";\r
+ }\r
+ catch(...)\r
+ {\r
+ CASPAR_LOG_CURRENT_EXCEPTION();\r
+ is_running_ = false;\r
+ }\r
+ }\r
+ \r
+ void read_next_packet()\r
+ { \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
+ const int ret = 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(ret)) // Use av_dup_packet to extend its life.\r
+ {\r
+ if(loop_)\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
+ seek_frame(start_, AVSEEK_FLAG_BACKWARD);\r
+ graph_->add_tag("seek"); \r
+ CASPAR_LOG(trace) << print() << " Received EOF. Looping."; \r
+ } \r
else\r
{\r
- av_dup_packet(read_packet.get());\r
+ is_running_ = false;\r
+ CASPAR_LOG(trace) << print() << " Received EOF. Stopping.";\r
+ }\r
+ }\r
+ else if(ret < 0)\r
+ {\r
+ BOOST_THROW_EXCEPTION(\r
+ file_read_error() <<\r
+ msg_info(av_error_str(ret)) <<\r
+ source_info(narrow(print())) << \r
+ boost::errinfo_api_function("av_read_frame") <<\r
+ boost::errinfo_errno(AVUNERROR(ret)));\r
+ }\r
+ else\r
+ { \r
+ av_dup_packet(read_packet.get());\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
+ // 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* pkt)
\r- {\r
- read_packet->size = size;\r
- read_packet->data = data;\r
- });\r
+ read_packet = std::shared_ptr<AVPacket>(read_packet.get(), [=](AVPacket* pkt)\r
+ {\r
+ read_packet->size = size;\r
+ read_packet->data = data;\r
+ });\r
\r
- graph_->update_value("buffer-count", MAX_BUFFER_SIZE/static_cast<double>(buffer_.size()));\r
- \r
- boost::unique_lock<boost::mutex> lock(mutex_);\r
- while(buffer_size_ > MAX_BUFFER_SIZE && buffer_.size() > MAX_BUFFER_COUNT)\r
- cond_.wait(lock);\r
+ buffer_.try_push(read_packet);\r
+ buffer_size_ += read_packet->size;\r
\r
- buffer_.push(read_packet);\r
- buffer_size_ += read_packet->size;\r
+ graph_->update_value("buffer-count", MAX_BUFFER_SIZE/static_cast<double>(buffer_.size()));\r
+ graph_->update_value("buffer-size", MAX_BUFFER_SIZE/static_cast<double>(buffer_size_));\r
+ } \r
+ }\r
\r
- graph_->update_value("buffer-size", MAX_BUFFER_SIZE/static_cast<double>(buffer_size_));\r
- } \r
- }\r
- catch(...)\r
- {\r
- stop();\r
- CASPAR_LOG_CURRENT_EXCEPTION();\r
- return;\r
- } \r
+ bool full() const\r
+ {\r
+ return is_running_ && buffer_size_ > MAX_BUFFER_SIZE && buffer_.size() > MAX_BUFFER_COUNT;\r
}\r
\r
void seek_frame(int64_t frame, int flags = 0)\r
boost::errinfo_errno(AVUNERROR(stream_index)));\r
}\r
\r
- const int errn = av_seek_frame(format_context_.get(), stream_index, frame, flags);\r
- if(errn < 0)\r
+ const int ret = av_seek_frame(format_context_.get(), stream_index, frame, flags);\r
+ if(ret < 0)\r
{ \r
BOOST_THROW_EXCEPTION(\r
invalid_operation() << \r
source_info(narrow(print())) << \r
- msg_info(av_error_str(errn)) <<\r
+ msg_info(av_error_str(ret)) <<\r
boost::errinfo_api_function("av_seek_frame") <<\r
- boost::errinfo_errno(AVUNERROR(errn)));\r
+ boost::errinfo_errno(AVUNERROR(ret)));\r
}\r
\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(errn == AVERROR_EOF)\r
+ if(ret == AVERROR(EIO))\r
+ CASPAR_LOG(debug) << print() << " Received EIO, assuming EOF.";\r
+ //if(ret == AVERROR_EOF)\r
// CASPAR_LOG(info) << print() << " Received EOF.";\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); // av_read_frame doesn't always correctly return AVERROR_EOF;\r
}\r
\r
std::wstring print() const\r
\r
input::input(const safe_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop, int start, int length) \r
: impl_(new implementation(graph, filename, loop, start)){}\r
-bool input::eof() const {return !impl_->executor_.is_running();}\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
std::shared_ptr<AVFormatContext> input::context(){return impl_->format_context_;}\r
}
\ No newline at end of file
projects / casparcg / commitdiff