git.sesse.net Git - casparcg/blob - modules/ffmpeg/producer/input/input.cpp

   1 /*
   2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>
   3 *
   4 * This file is part of CasparCG (www.casparcg.com).
   5 *
   6 * CasparCG is free software: you can redistribute it and/or modify
   7 * it under the terms of the GNU General Public License as published by
   8 * the Free Software Foundation, either version 3 of the License, or
   9 * (at your option) any later version.
  10 *
  11 * CasparCG is distributed in the hope that it will be useful,
  12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 * GNU General Public License for more details.
  15 *
  16 * You should have received a copy of the GNU General Public License
  17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.
  18 *
  19 * Author: Robert Nagy, ronag89@gmail.com
  20 */
  21
  22 #include "../../StdAfx.h"
  23
  24 #include "input.h"
  25
  26 #include "../util/util.h"
  27 #include "../../ffmpeg_error.h"
  28 #include "../../ffmpeg.h"
  29
  30 #include <common/diagnostics/graph.h>
  31 #include <common/executor.h>
  32 #include <common/lock.h>
  33 //#include <common/except.h>
  34 #include <common/os/general_protection_fault.h>
  35 #include <common/log.h>
  36
  37 #include <core/video_format.h>
  38
  39 #include <tbb/concurrent_queue.h>
  40 #include <tbb/atomic.h>
  41 #include <tbb/recursive_mutex.h>
  42
  43 #include <boost/thread/condition_variable.hpp>
  44 #include <boost/thread/mutex.hpp>
  45 #include <boost/thread/thread.hpp>
  46
  47 #if defined(_MSC_VER)
  48 #pragma warning (push)
  49 #pragma warning (disable : 4244)
  50 #endif
  51 extern "C"
  52 {
  53         #define __STDC_CONSTANT_MACROS
  54         #define __STDC_LIMIT_MACROS
  55         #include <libavformat/avformat.h>
  56 }
  57 #if defined(_MSC_VER)
  58 #pragma warning (pop)
  59 #endif
  60
  61 namespace caspar { namespace ffmpeg {
  62
  63 static const int MAX_PUSH_WITHOUT_POP = 200;
  64 static const int MIN_FRAMES = 25;
  65
  66 class stream
  67 {
  68         stream(const stream&);
  69         stream& operator=(const stream&);
  70
  71         typedef tbb::concurrent_bounded_queue<std::shared_ptr<AVPacket>>::size_type size_type;
  72
  73         int                                                                                                                     index_;
  74         tbb::concurrent_bounded_queue<std::shared_ptr<AVPacket>>        packets_;
  75         tbb::atomic<int>                                                                                        push_since_pop_;
  76 public:
  77
  78         stream(int index)
  79                 : index_(index)
  80         {
  81                 push_since_pop_ = 0;
  82         }
  83
  84         stream(stream&&) = default;
  85
  86         bool is_available() const
  87         {
  88                 return index_ >= 0;
  89         }
  90
  91         int index() const
  92         {
  93                 return index_;
  94         }
  95
  96         void push(const std::shared_ptr<AVPacket>& packet)
  97         {
  98                 if(packet && packet->data && packet->stream_index != index_)
  99                         return;
 100
 101                 if (++push_since_pop_ > MAX_PUSH_WITHOUT_POP) // Out of memory protection for streams never being used.
 102                 {
 103                         return;
 104                 }
 105
 106                 packets_.push(packet);
 107         }
 108
 109         bool try_pop(std::shared_ptr<AVPacket>& packet)
 110         {
 111                 push_since_pop_ = 0;
 112
 113                 return packets_.try_pop(packet);
 114         }
 115
 116         void clear()
 117         {
 118                 std::shared_ptr<AVPacket> packet;
 119                 push_since_pop_ = 0;
 120                 while(packets_.try_pop(packet));
 121         }
 122
 123         size_type size() const
 124         {
 125                 return is_available() ? packets_.size() : std::numeric_limits<size_type>::max();
 126         }
 127 };
 128
 129 struct input::impl : boost::noncopyable
 130 {
 131         const spl::shared_ptr<diagnostics::graph>       graph_;
 132
 133         const std::wstring                                                      filename_;
 134         const spl::shared_ptr<AVFormatContext>          format_context_                 = open_input(filename_); // Destroy this last
 135         const int                                                                       default_stream_index_   = av_find_default_stream_index(format_context_.get());
 136
 137         tbb::atomic<uint32_t>                                           start_;
 138         tbb::atomic<uint32_t>                                           length_;
 139         tbb::atomic<bool>                                                       loop_;
 140         tbb::atomic<bool>                                                       eof_;
 141         double                                                                          fps_                                    = read_fps(*format_context_, 0.0);
 142         uint32_t                                                                        frame_number_                   = 0;
 143
 144         stream                                                                          video_stream_                   {                                                       av_find_best_stream(format_context_.get(), AVMEDIA_TYPE_VIDEO, -1, -1, 0, 0) };
 145         std::vector<stream>                                                     audio_streams_;
 146
 147         boost::optional<uint32_t>                                       seek_target_;
 148
 149         tbb::atomic<bool>                                                       is_running_;
 150         boost::mutex                                                            mutex_;
 151         boost::condition_variable                                       cond_;
 152         boost::thread                                                           thread_;
 153
 154         impl(
 155                         const spl::shared_ptr<diagnostics::graph> graph,
 156                         const std::wstring& filename,
 157                         const bool loop,
 158                         const uint32_t start,
 159                         const uint32_t length,
 160                         bool thumbnail_mode)
 161                 : graph_(graph)
 162                 , filename_(filename)
 163         {
 164                 start_                  = start;
 165                 length_                 = length;
 166                 loop_                   = loop;
 167                 eof_                    = false;
 168                 is_running_             = true;
 169
 170                 if(start_ != 0)
 171                         seek_target_ = start_;
 172
 173                 graph_->set_color("seek", diagnostics::color(1.0f, 0.5f, 0.0f));
 174
 175                 if (!thumbnail_mode)
 176                         for (unsigned i = 0; i < format_context_->nb_streams; ++i)
 177                                 if (format_context_->streams[i]->codec->codec_type == AVMediaType::AVMEDIA_TYPE_AUDIO)
 178                                         audio_streams_.emplace_back(i);
 179
 180                 for (int i = 0; i < audio_streams_.size(); ++i)
 181                         graph_->set_color("audio-buffer" + boost::lexical_cast<std::string>(i + 1), diagnostics::color(0.7f, 0.4f, 0.4f));
 182
 183                 if (video_stream_.is_available())
 184                         graph_->set_color("video-buffer", diagnostics::color(1.0f, 1.0f, 0.0f));
 185
 186                 for(int n = 0; n < 8; ++n)
 187                         tick();
 188
 189                 thread_ = boost::thread([this, thumbnail_mode]{run(thumbnail_mode);});
 190         }
 191
 192         ~impl()
 193         {
 194                 is_running_ = false;
 195                 cond_.notify_one();
 196                 thread_.join();
 197         }
 198
 199         bool try_pop_video(std::shared_ptr<AVPacket>& packet)
 200         {
 201                 if (!video_stream_.is_available())
 202                         return false;
 203
 204                 bool result = video_stream_.try_pop(packet);
 205
 206                 if(result)
 207                         cond_.notify_one();
 208
 209                 graph_->set_value("video-buffer", std::min(1.0, static_cast<double>(video_stream_.size())/MIN_FRAMES));
 210
 211                 return result;
 212         }
 213
 214         bool try_pop_audio(std::shared_ptr<AVPacket>& packet, int audio_stream_index)
 215         {
 216                 if (audio_streams_.size() < audio_stream_index + 1)
 217                         return false;
 218
 219                 auto& audio_stream = audio_streams_.at(audio_stream_index);
 220                 bool result = audio_stream.try_pop(packet);
 221                 if(result)
 222                         cond_.notify_one();
 223
 224                 auto buffer_nr = boost::lexical_cast<std::string>(audio_stream_index + 1);
 225                 graph_->set_value("audio-buffer" + buffer_nr, std::min(1.0, static_cast<double>(audio_stream.size())/MIN_FRAMES));
 226
 227                 return result;
 228         }
 229
 230         void seek(uint32_t target)
 231         {
 232                 {
 233                         boost::lock_guard<boost::mutex> lock(mutex_);
 234
 235                         seek_target_ = target;
 236                         video_stream_.clear();
 237
 238                         for (auto& audio_stream : audio_streams_)
 239                                 audio_stream.clear();
 240                 }
 241
 242                 cond_.notify_one();
 243         }
 244
 245         int get_actual_audio_stream_index(int audio_stream_index) const
 246         {
 247                 if (audio_stream_index + 1 > audio_streams_.size())
 248                         CASPAR_THROW_EXCEPTION(averror_stream_not_found());
 249
 250                 return audio_streams_.at(audio_stream_index).index();
 251         }
 252
 253         std::wstring print() const
 254         {
 255                 return L"ffmpeg_input[" + filename_ + L")]";
 256         }
 257
 258 private:
 259         void internal_seek(uint32_t target)
 260         {
 261                 eof_ = false;
 262                 graph_->set_tag(diagnostics::tag_severity::INFO, "seek");
 263
 264                 if (is_logging_quiet_for_thread())
 265                         CASPAR_LOG(trace) << print() << " Seeking: " << target;
 266                 else
 267                         CASPAR_LOG(debug) << print() << " Seeking: " << target;
 268
 269                 int flags = AVSEEK_FLAG_FRAME;
 270                 if(target == 0)
 271                 {
 272                         // Fix VP6 seeking
 273                         int vid_stream_index = av_find_best_stream(format_context_.get(), AVMEDIA_TYPE_VIDEO, -1, -1, 0, 0);
 274                         if(vid_stream_index >= 0)
 275                         {
 276                                 auto codec_id = format_context_->streams[vid_stream_index]->codec->codec_id;
 277                                 if(codec_id == CODEC_ID_VP6A || codec_id == CODEC_ID_VP6F || codec_id == CODEC_ID_VP6)
 278                                         flags = AVSEEK_FLAG_BYTE;
 279                         }
 280                 }
 281
 282                 auto stream                             = format_context_->streams[default_stream_index_];
 283                 auto fps                                = read_fps(*format_context_, 0.0);
 284                 auto target_timestamp   = static_cast<int64_t>((target / fps * stream->time_base.den) / stream->time_base.num);
 285
 286                 THROW_ON_ERROR2(avformat_seek_file(
 287                                 format_context_.get(),
 288                                 default_stream_index_,
 289                                 std::numeric_limits<int64_t>::min(),
 290                                 target_timestamp,
 291                                 std::numeric_limits<int64_t>::max(),
 292                                 0), print());
 293
 294                 auto flush_packet       = create_packet();
 295                 flush_packet->data      = nullptr;
 296                 flush_packet->size      = 0;
 297                 flush_packet->pos       = target;
 298
 299                 video_stream_.push(flush_packet);
 300
 301                 for (auto& audio_stream : audio_streams_)
 302                         audio_stream.push(flush_packet);
 303         }
 304
 305         void tick()
 306         {
 307                 if(seek_target_)
 308                 {
 309                         internal_seek(*seek_target_);
 310                         seek_target_.reset();
 311                 }
 312
 313                 auto packet = create_packet();
 314
 315                 auto ret = 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.
 316
 317                 if(is_eof(ret))
 318                 {
 319                         if (loop_)
 320                                 internal_seek(start_);
 321                         else
 322                         {
 323                                 eof_ = true;
 324                         }
 325                 }
 326                 else
 327                 {
 328                         THROW_ON_ERROR(ret, "av_read_frame", print());
 329
 330                         THROW_ON_ERROR2(av_dup_packet(packet.get()), print());
 331
 332                         // Make sure that the packet is correctly deallocated even if size and data is modified during decoding.
 333                         const auto size = packet->size;
 334                         const auto data = packet->data;
 335
 336                         packet = spl::shared_ptr<AVPacket>(packet.get(), [packet, size, data](AVPacket*)
 337                         {
 338                                 packet->size = size;
 339                                 packet->data = data;
 340                         });
 341
 342                         const auto stream_time_base = format_context_->streams[packet->stream_index]->time_base;
 343                         const auto packet_frame_number = static_cast<uint32_t>((static_cast<double>(packet->pts * stream_time_base.num)/stream_time_base.den)*fps_);
 344
 345                         if(packet->stream_index == default_stream_index_)
 346                                 frame_number_ = packet_frame_number;
 347
 348                         if(packet_frame_number >= start_ && packet_frame_number < length_)
 349                         {
 350                                 video_stream_.push(packet);
 351
 352                                 for (auto& audio_stream : audio_streams_)
 353                                         audio_stream.push(packet);
 354                         }
 355                 }
 356
 357                 if (video_stream_.is_available())
 358                         graph_->set_value("video-buffer", std::min(1.0, static_cast<double>(video_stream_.size())/MIN_FRAMES));
 359
 360                 for (int i = 0; i < audio_streams_.size(); ++i)
 361                         graph_->set_value(
 362                                         "audio-buffer" + boost::lexical_cast<std::string>(i + 1),
 363                                         std::min(1.0, static_cast<double>(audio_streams_[i].size())/MIN_FRAMES));
 364         }
 365
 366         bool full() const
 367         {
 368                 bool video_full = video_stream_.size() >= MIN_FRAMES;
 369
 370                 if (!video_full)
 371                         return false;
 372
 373                 for (auto& audio_stream : audio_streams_)
 374                         if (audio_stream.size() < MIN_FRAMES)
 375                                 return false;
 376
 377                 return true;
 378         }
 379
 380         void run(bool thumbnail_mode)
 381         {
 382                 ensure_gpf_handler_installed_for_thread(u8(print()).c_str());
 383                 auto quiet_logging = temporary_enable_quiet_logging_for_thread(thumbnail_mode);
 384
 385                 while(is_running_)
 386                 {
 387                         try
 388                         {
 389
 390                                 {
 391                                         boost::unique_lock<boost::mutex> lock(mutex_);
 392
 393                                         while((eof_ || full()) && !seek_target_ && is_running_)
 394                                                 cond_.wait(lock);
 395
 396                                         tick();
 397                                 }
 398                         }
 399                         catch(...)
 400                         {
 401                                 CASPAR_LOG_CURRENT_EXCEPTION();
 402                                 is_running_ = false;
 403                         }
 404                 }
 405         }
 406
 407         bool is_eof(int ret)
 408         {
 409                 #pragma warning (disable : 4146)
 410                 return ret == AVERROR_EOF || ret == AVERROR(EIO) || frame_number_ >= length_; // av_read_frame doesn't always correctly return AVERROR_EOF;
 411         }
 412 };
 413
 414 input::input(const spl::shared_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop, uint32_t start, uint32_t length, bool thumbnail_mode)
 415         : impl_(new impl(graph, filename, loop, start, length, thumbnail_mode)){}
 416 int input::get_actual_audio_stream_index(int audio_stream_index) const { return impl_->get_actual_audio_stream_index(audio_stream_index); };
 417 int input::num_audio_streams() const { return static_cast<int>(impl_->audio_streams_.size()); }
 418 bool input::try_pop_video(std::shared_ptr<AVPacket>& packet){return impl_->try_pop_video(packet);}
 419 bool input::try_pop_audio(std::shared_ptr<AVPacket>& packet, int audio_stream_index){return impl_->try_pop_audio(packet, audio_stream_index);}
 420 AVFormatContext& input::context(){return *impl_->format_context_;}
 421 void input::loop(bool value){impl_->loop_ = value;}
 422 bool input::loop() const{return impl_->loop_;}
 423 void input::seek(uint32_t target){impl_->seek(target);}
 424 void input::start(uint32_t value){impl_->start_ = value;}
 425 uint32_t input::start() const{return impl_->start_;}
 426 void input::length(uint32_t value){impl_->length_ = value;}
 427 uint32_t input::length() const{return impl_->length_;}
 428 bool input::eof() const { return impl_->eof_; }
 429 }}