]> git.sesse.net Git - casparcg/blob - modules/ffmpeg/producer/input/input.cpp
boost::this_thread::sleep is deprecated
[casparcg] / 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 MIN_FRAMES = 25;
64
65 class stream
66 {
67         stream(const stream&);
68         stream& operator=(const stream&);
69
70         typedef tbb::concurrent_bounded_queue<std::shared_ptr<AVPacket>>::size_type size_type;
71
72         int                                                                                                              index_;
73         tbb::concurrent_bounded_queue<std::shared_ptr<AVPacket>> packets_;
74 public:
75
76         stream(int index) 
77                 : index_(index)
78         {
79         }
80
81         bool is_available() const
82         {
83                 return index_ >= 0;
84         }
85         
86         void push(const std::shared_ptr<AVPacket>& packet)
87         {
88                 if(packet && packet->data && packet->stream_index != index_)
89                         return;
90
91                 packets_.push(packet);
92         }
93
94         bool try_pop(std::shared_ptr<AVPacket>& packet)
95         {
96                 return packets_.try_pop(packet);
97         }
98
99         void clear()
100         {
101                 std::shared_ptr<AVPacket> packet;
102                 while(packets_.try_pop(packet));
103         }
104                 
105         size_type size() const
106         {
107                 return is_available() ? packets_.size() : std::numeric_limits<size_type>::max();
108         }
109 };
110                 
111 struct input::impl : boost::noncopyable
112 {               
113         const spl::shared_ptr<diagnostics::graph>       graph_;
114
115         const std::wstring                                                      filename_;
116         const spl::shared_ptr<AVFormatContext>          format_context_                 = open_input(filename_); // Destroy this last
117         const int                                                                       default_stream_index_   = av_find_default_stream_index(format_context_.get());
118
119         tbb::atomic<uint32_t>                                           start_;         
120         tbb::atomic<uint32_t>                                           length_;
121         tbb::atomic<bool>                                                       loop_;
122         tbb::atomic<bool>                                                       eof_;
123         bool                                                                            thumbnail_mode_;
124         double                                                                          fps_                                    = read_fps(*format_context_, 0.0);
125         uint32_t                                                                        frame_number_                   = 0;
126
127         stream                                                                          video_stream_                   {                                                       av_find_best_stream(format_context_.get(), AVMEDIA_TYPE_VIDEO, -1, -1, 0, 0) };
128         stream                                                                          audio_stream_                   { thumbnail_mode_ ? -1 :        av_find_best_stream(format_context_.get(), AVMEDIA_TYPE_AUDIO, -1, -1, 0, 0) };
129
130         boost::optional<uint32_t>                                       seek_target_;
131
132         tbb::atomic<bool>                                                       is_running_;
133         boost::mutex                                                            mutex_;
134         boost::condition_variable                                       cond_;
135         boost::thread                                                           thread_;
136         
137         impl(
138                         const spl::shared_ptr<diagnostics::graph> graph,
139                         const std::wstring& filename,
140                         const bool loop,
141                         const uint32_t start,
142                         const uint32_t length,
143                         bool thumbnail_mode)
144                 : graph_(graph)
145                 , filename_(filename)
146                 , thumbnail_mode_(thumbnail_mode)
147         {
148                 start_                  = start;
149                 length_                 = length;
150                 loop_                   = loop;
151                 eof_                    = false;
152                 is_running_             = true;
153
154                 if(start_ != 0)
155                         seek_target_ = start_;
156                                                                                                                 
157                 graph_->set_color("seek", diagnostics::color(1.0f, 0.5f, 0.0f));
158
159                 if (audio_stream_.is_available())
160                         graph_->set_color("audio-buffer", diagnostics::color(0.7f, 0.4f, 0.4f));
161
162                 if (video_stream_.is_available())
163                         graph_->set_color("video-buffer", diagnostics::color(1.0f, 1.0f, 0.0f));
164                 
165                 for(int n = 0; n < 8; ++n)
166                         tick();
167
168                 if (!thumbnail_mode)
169                         thread_ = boost::thread([this]{run();});
170         }
171
172         ~impl()
173         {
174                 is_running_ = false;
175                 cond_.notify_one();
176
177                 if (!thumbnail_mode_)
178                         thread_.join();
179         }
180         
181         bool try_pop_video(std::shared_ptr<AVPacket>& packet)
182         {
183                 if (!video_stream_.is_available())
184                         return false;
185
186                 if (thumbnail_mode_)
187                 {
188                         int ticks = 0;
189                         while (!video_stream_.try_pop(packet))
190                         {
191                                 tick();
192                                 if (++ticks > 32) // Infinite loop should not be possible
193                                         return false;
194
195                                 // Play nice
196                                 boost::this_thread::sleep_for(boost::chrono::milliseconds(5));
197                         }
198
199                         return true;
200                 }
201
202                 bool result = video_stream_.try_pop(packet);
203
204                 if(result)
205                         cond_.notify_one();
206                 
207                 graph_->set_value("video-buffer", std::min(1.0, static_cast<double>(video_stream_.size())/MIN_FRAMES));
208                                 
209                 return result;
210         }
211         
212         bool try_pop_audio(std::shared_ptr<AVPacket>& packet)
213         {
214                 if (!audio_stream_.is_available())
215                         return false;
216
217                 bool result = audio_stream_.try_pop(packet);
218                 if(result)
219                         cond_.notify_one();
220                                 
221                 graph_->set_value("audio-buffer", std::min(1.0, static_cast<double>(audio_stream_.size())/MIN_FRAMES));
222
223                 return result;
224         }
225
226         void seek(uint32_t target)
227         {
228                 {
229                         boost::lock_guard<boost::mutex> lock(mutex_);
230
231                         seek_target_ = target;
232                         video_stream_.clear();
233                         audio_stream_.clear();
234                 }
235
236                 cond_.notify_one();
237         }
238                 
239         std::wstring print() const
240         {
241                 return L"ffmpeg_input[" + filename_ + L")]";
242         }
243
244 private:
245         void internal_seek(uint32_t target)
246         {
247                 eof_ = false;
248                 graph_->set_tag(diagnostics::tag_severity::INFO, "seek");
249
250                 if (is_logging_quiet_for_thread())
251                         CASPAR_LOG(trace) << print() << " Seeking: " << target;
252                 else
253                         CASPAR_LOG(debug) << print() << " Seeking: " << target;
254
255                 int flags = AVSEEK_FLAG_FRAME;
256                 if(target == 0)
257                 {
258                         // Fix VP6 seeking
259                         int vid_stream_index = av_find_best_stream(format_context_.get(), AVMEDIA_TYPE_VIDEO, -1, -1, 0, 0);
260                         if(vid_stream_index >= 0)
261                         {
262                                 auto codec_id = format_context_->streams[vid_stream_index]->codec->codec_id;
263                                 if(codec_id == CODEC_ID_VP6A || codec_id == CODEC_ID_VP6F || codec_id == CODEC_ID_VP6)
264                                         flags = AVSEEK_FLAG_BYTE;
265                         }
266                 }
267                 
268                 auto stream                             = format_context_->streams[default_stream_index_];
269                 auto fps                                = read_fps(*format_context_, 0.0);
270                 auto target_timestamp = static_cast<int64_t>((target / fps * stream->time_base.den) / stream->time_base.num);
271                 
272                 THROW_ON_ERROR2(avformat_seek_file(
273                                 format_context_.get(),
274                                 default_stream_index_,
275                                 std::numeric_limits<int64_t>::min(),
276                                 target_timestamp,
277                                 std::numeric_limits<int64_t>::max(),
278                                 0), print());
279                 
280                 video_stream_.push(nullptr);
281                 audio_stream_.push(nullptr);
282         }
283
284         void tick()
285         {
286                 if(seek_target_)                                
287                 {
288                         internal_seek(*seek_target_);
289                         seek_target_.reset();
290                 }
291
292                 auto packet = create_packet();
293                 
294                 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.  
295                 
296                 if(is_eof(ret))                                                                                                              
297                 {
298                         if (loop_)
299                                 internal_seek(start_);
300                         else
301                         {
302                                 eof_ = true;
303                         }
304                 }
305                 else
306                 {               
307                         THROW_ON_ERROR(ret, "av_read_frame", print());
308                                         
309                         THROW_ON_ERROR2(av_dup_packet(packet.get()), print());
310                                 
311                         // Make sure that the packet is correctly deallocated even if size and data is modified during decoding.
312                         const auto size = packet->size;
313                         const auto data = packet->data;
314                         
315                         packet = spl::shared_ptr<AVPacket>(packet.get(), [packet, size, data](AVPacket*)
316                         {
317                                 packet->size = size;
318                                 packet->data = data;                            
319                         });
320                                         
321                         const auto stream_time_base = format_context_->streams[packet->stream_index]->time_base;
322                         const auto packet_frame_number = static_cast<uint32_t>((static_cast<double>(packet->pts * stream_time_base.num)/stream_time_base.den)*fps_);
323
324                         if(packet->stream_index == default_stream_index_)
325                                 frame_number_ = packet_frame_number;
326                                         
327                         if(packet_frame_number >= start_ && packet_frame_number < length_)
328                         {
329                                 video_stream_.push(packet);
330                                 audio_stream_.push(packet);
331                         }
332                 }       
333
334                 if (video_stream_.is_available())
335                         graph_->set_value("video-buffer", std::min(1.0, static_cast<double>(video_stream_.size())/MIN_FRAMES));
336
337                 if (audio_stream_.is_available())
338                         graph_->set_value("audio-buffer", std::min(1.0, static_cast<double>(audio_stream_.size())/MIN_FRAMES));
339         }
340                         
341         bool full() const
342         {
343                 return video_stream_.size() >= MIN_FRAMES && audio_stream_.size() >= MIN_FRAMES;
344         }
345
346         void run()
347         {
348                 ensure_gpf_handler_installed_for_thread(u8(print()).c_str());
349
350                 while(is_running_)
351                 {
352                         try
353                         {
354                                 
355                                 {
356                                         boost::unique_lock<boost::mutex> lock(mutex_);
357
358                                         while(full() && !seek_target_ && is_running_)
359                                                 cond_.wait(lock);
360                                         
361                                         tick();
362                                 }
363                         }
364                         catch(...)
365                         {
366                                 CASPAR_LOG_CURRENT_EXCEPTION();
367                                 is_running_ = false;
368                         }
369                 }
370         }
371                         
372         bool is_eof(int ret)
373         {
374                 #pragma warning (disable : 4146)
375                 return ret == AVERROR_EOF || ret == AVERROR(EIO) || frame_number_ >= length_; // av_read_frame doesn't always correctly return AVERROR_EOF;
376         }
377 };
378
379 input::input(const spl::shared_ptr<diagnostics::graph>& graph, const std::wstring& filename, bool loop, uint32_t start, uint32_t length, bool thumbnail_mode)
380         : impl_(new impl(graph, filename, loop, start, length, thumbnail_mode)){}
381 bool input::try_pop_video(std::shared_ptr<AVPacket>& packet){return impl_->try_pop_video(packet);}
382 bool input::try_pop_audio(std::shared_ptr<AVPacket>& packet){return impl_->try_pop_audio(packet);}
383 AVFormatContext& input::context(){return *impl_->format_context_;}
384 void input::loop(bool value){impl_->loop_ = value;}
385 bool input::loop() const{return impl_->loop_;}
386 void input::seek(uint32_t target){impl_->seek(target);}
387 void input::start(uint32_t value){impl_->start_ = value;}
388 uint32_t input::start() const{return impl_->start_;}
389 void input::length(uint32_t value){impl_->length_ = value;}
390 uint32_t input::length() const{return impl_->length_;}
391 bool input::eof() const { return impl_->eof_; }
392 }}