X-Git-Url: https://main.carlh.net/gitweb/?p=dcpomatic.git;a=blobdiff_plain;f=src%2Flib%2Fbutler.cc;h=fe331fc79dca1609f9ba5ddacd23a20595bf0f94;hp=ea195b5d55b2fe2ed4581fe60687687f4cbea0ba;hb=f0c10e92b849566e458bc323f8783a6fe83e52d2;hpb=4073923b2c5ccf127c61435cd575542351421f05 diff --git a/src/lib/butler.cc b/src/lib/butler.cc index ea195b5d5..fe331fc79 100644 --- a/src/lib/butler.cc +++ b/src/lib/butler.cc @@ -1,5 +1,5 @@ /* - Copyright (C) 2016-2017 Carl Hetherington + Copyright (C) 2016-2018 Carl Hetherington This file is part of DCP-o-matic. @@ -20,40 +20,95 @@ #include "butler.h" #include "player.h" +#include "util.h" +#include "log.h" +#include "dcpomatic_log.h" +#include "cross.h" +#include "compose.hpp" +#include "exceptions.h" #include #include using std::cout; using std::pair; using std::make_pair; +using std::string; using boost::weak_ptr; using boost::shared_ptr; using boost::bind; using boost::optional; +using boost::function; -/** Video readahead in frames */ -#define VIDEO_READAHEAD 10 -/** Audio readahead in frames */ -#define AUDIO_READAHEAD 48000 +/** Minimum video readahead in frames */ +#define MINIMUM_VIDEO_READAHEAD 10 +/** Maximum video readahead in frames; should never be exceeded (by much) unless there are bugs in Player */ +#define MAXIMUM_VIDEO_READAHEAD 48 +/** Minimum audio readahead in frames */ +#define MINIMUM_AUDIO_READAHEAD (48000 * MINIMUM_VIDEO_READAHEAD / 24) +/** Maximum audio readahead in frames; should never be exceeded (by much) unless there are bugs in Player */ +#define MAXIMUM_AUDIO_READAHEAD (48000 * MAXIMUM_VIDEO_READAHEAD / 24) -Butler::Butler (weak_ptr film, shared_ptr player, AudioMapping audio_mapping, int audio_channels) - : _film (film) - , _player (player) +/** @param pixel_format Pixel format functor that will be used when calling ::image on PlayerVideos coming out of this + * butler. This will be used (where possible) to prepare the PlayerVideos so that calling image() on them is quick. + * @param aligned Same as above for the `aligned' flag. + * @param fast Same as above for the `fast' flag. + */ +Butler::Butler ( + shared_ptr player, + AudioMapping audio_mapping, + int audio_channels, + function pixel_format, + bool aligned, + bool fast + ) + : _player (player) + , _prepare_work (new boost::asio::io_service::work (_prepare_service)) , _pending_seek_accurate (false) + , _suspended (0) , _finished (false) + , _died (false) + , _stop_thread (false) , _audio_mapping (audio_mapping) , _audio_channels (audio_channels) - , _stop_thread (false) + , _disable_audio (false) + , _pixel_format (pixel_format) + , _aligned (aligned) + , _fast (fast) { _player_video_connection = _player->Video.connect (bind (&Butler::video, this, _1, _2)); - _player_audio_connection = _player->Audio.connect (bind (&Butler::audio, this, _1, _2)); - _player_changed_connection = _player->Changed.connect (bind (&Butler::player_changed, this)); + _player_audio_connection = _player->Audio.connect (bind (&Butler::audio, this, _1, _2, _3)); + _player_text_connection = _player->Text.connect (bind (&Butler::text, this, _1, _2, _3, _4)); + /* The butler must hear about things first, otherwise it might not sort out suspensions in time for + get_video() to be called in response to this signal. + */ + _player_change_connection = _player->Change.connect (bind (&Butler::player_change, this, _1, _3), boost::signals2::at_front); _thread = new boost::thread (bind (&Butler::thread, this)); +#ifdef DCPOMATIC_LINUX + pthread_setname_np (_thread->native_handle(), "butler"); +#endif + + /* Create some threads to do work on the PlayerVideos we are creating; at present this is used to + multi-thread JPEG2000 decoding. + */ + + LOG_TIMING("start-prepare-threads %1", boost::thread::hardware_concurrency() * 2); + + for (size_t i = 0; i < boost::thread::hardware_concurrency() * 2; ++i) { + _prepare_pool.create_thread (bind (&boost::asio::io_service::run, &_prepare_service)); + } } Butler::~Butler () { - _stop_thread = true; + { + boost::mutex::scoped_lock lm (_mutex); + _stop_thread = true; + } + + _prepare_work.reset (); + _prepare_pool.join_all (); + _prepare_service.stop (); + _thread->interrupt (); try { _thread->join (); @@ -63,6 +118,56 @@ Butler::~Butler () delete _thread; } +/** Caller must hold a lock on _mutex */ +bool +Butler::should_run () const +{ + if (_video.size() >= MAXIMUM_VIDEO_READAHEAD * 10) { + /* This is way too big */ + optional pos = _audio.peek(); + if (pos) { + throw ProgrammingError + (__FILE__, __LINE__, String::compose ("Butler video buffers reached %1 frames (audio is %2 at %3)", _video.size(), _audio.size(), pos->get())); + } else { + throw ProgrammingError + (__FILE__, __LINE__, String::compose ("Butler video buffers reached %1 frames (audio is %2)", _video.size(), _audio.size())); + } + } + + if (_audio.size() >= MAXIMUM_AUDIO_READAHEAD * 10) { + /* This is way too big */ + optional pos = _audio.peek(); + if (pos) { + throw ProgrammingError + (__FILE__, __LINE__, String::compose ("Butler audio buffers reached %1 frames at %2 (video is %3)", _audio.size(), pos->get(), _video.size())); + } else { + throw ProgrammingError + (__FILE__, __LINE__, String::compose ("Butler audio buffers reached %1 frames (video is %3)", _audio.size(), _video.size())); + } + } + + if (_video.size() >= MAXIMUM_VIDEO_READAHEAD * 2) { + LOG_WARNING ("Butler video buffers reached %1 frames (audio is %2)", _video.size(), _audio.size()); + } + + if (_audio.size() >= MAXIMUM_AUDIO_READAHEAD * 2) { + LOG_WARNING ("Butler audio buffers reached %1 frames (video is %2)", _audio.size(), _video.size()); + } + + if (_stop_thread || _finished || _died || _suspended) { + /* Definitely do not run */ + return false; + } + + if (_video.size() < MINIMUM_VIDEO_READAHEAD || (!_disable_audio && _audio.size() < MINIMUM_AUDIO_READAHEAD)) { + /* Definitely do run: we need data */ + return true; + } + + /* Run if we aren't full of video or audio */ + return (_video.size() < MAXIMUM_VIDEO_READAHEAD) && (_audio.size() < MAXIMUM_AUDIO_READAHEAD); +} + void Butler::thread () try @@ -71,12 +176,13 @@ try boost::mutex::scoped_lock lm (_mutex); /* Wait until we have something to do */ - while ((_video.size() >= VIDEO_READAHEAD && _audio.size() >= AUDIO_READAHEAD && !_pending_seek_position) || _stop_thread) { + while (!should_run() && !_pending_seek_position) { _summon.wait (lm); } /* Do any seek that has been requested */ if (_pending_seek_position) { + _finished = false; _player->seek (*_pending_seek_position, _pending_seek_accurate); _pending_seek_position = optional (); } @@ -85,33 +191,52 @@ try while lm is unlocked, as in that state nothing will be added to _video/_audio. */ - while ((_video.size() < VIDEO_READAHEAD || _audio.size() < AUDIO_READAHEAD) && !_pending_seek_position && !_stop_thread) { + while (should_run() && !_pending_seek_position) { lm.unlock (); - if (_player->pass ()) { + bool const r = _player->pass (); + lm.lock (); + if (r) { _finished = true; _arrived.notify_all (); break; } - lm.lock (); _arrived.notify_all (); } } } catch (boost::thread_interrupted) { /* The butler thread is being terminated */ + boost::mutex::scoped_lock lm (_mutex); + _finished = true; + _arrived.notify_all (); +} catch (...) { + store_current (); + boost::mutex::scoped_lock lm (_mutex); + _died = true; + _arrived.notify_all (); } pair, DCPTime> -Butler::get_video () +Butler::get_video (Error* e) { boost::mutex::scoped_lock lm (_mutex); + if (_suspended) { + if (e) { + *e = AGAIN; + } + return make_pair(shared_ptr(), DCPTime()); + } + /* Wait for data if we have none */ - while (_video.empty() && !_finished) { + while (_video.empty() && !_finished && !_died) { _arrived.wait (lm); } - if (_finished) { - return make_pair (shared_ptr(), DCPTime()); + if (_video.empty()) { + if (e) { + *e = NONE; + } + return make_pair(shared_ptr(), DCPTime()); } pair, DCPTime> const r = _video.get (); @@ -119,63 +244,163 @@ Butler::get_video () return r; } +optional +Butler::get_closed_caption () +{ + boost::mutex::scoped_lock lm (_mutex); + return _closed_caption.get (); +} + void Butler::seek (DCPTime position, bool accurate) { boost::mutex::scoped_lock lm (_mutex); - _video.clear (); + _awaiting = optional(); + seek_unlocked (position, accurate); +} + +void +Butler::seek_unlocked (DCPTime position, bool accurate) +{ + if (_died) { + return; + } + _finished = false; _pending_seek_position = position; _pending_seek_accurate = accurate; + + { + boost::mutex::scoped_lock lm (_buffers_mutex); + _video.clear (); + _audio.clear (); + _closed_caption.clear (); + } + _summon.notify_all (); } +void +Butler::prepare (weak_ptr weak_video) +try +{ + shared_ptr video = weak_video.lock (); + /* If the weak_ptr cannot be locked the video obviously no longer requires any work */ + if (video) { + LOG_TIMING("start-prepare in %1", thread_id()); + video->prepare (_pixel_format, _aligned, _fast); + LOG_TIMING("finish-prepare in %1", thread_id()); + } +} +catch (...) +{ + store_current (); + boost::mutex::scoped_lock lm (_mutex); + _died = true; +} + void Butler::video (shared_ptr video, DCPTime time) { - { - boost::mutex::scoped_lock lm (_mutex); - if (_pending_seek_position) { - /* Don't store any video while a seek is pending */ - return; - } + boost::mutex::scoped_lock lm (_mutex); + + if (_pending_seek_position) { + /* Don't store any video in this case */ + return; } + _prepare_service.post (bind (&Butler::prepare, this, weak_ptr(video))); + + boost::mutex::scoped_lock lm2 (_buffers_mutex); _video.put (video, time); } void -Butler::audio (shared_ptr audio, DCPTime time) +Butler::audio (shared_ptr audio, DCPTime time, int frame_rate) { { boost::mutex::scoped_lock lm (_mutex); - if (_pending_seek_position) { - /* Don't store any audio while a seek is pending */ + if (_pending_seek_position || _disable_audio) { + /* Don't store any audio in these cases */ return; } } - _audio.put (audio, time); + boost::mutex::scoped_lock lm2 (_buffers_mutex); + _audio.put (remap (audio, _audio_channels, _audio_mapping), time, frame_rate); +} + +/** Try to get `frames' frames of audio and copy it into `out'. Silence + * will be filled if no audio is available. + * @return time of this audio, or unset if there was a buffer underrun. + */ +optional +Butler::get_audio (float* out, Frame frames) +{ + optional t = _audio.get (out, _audio_channels, frames); + _summon.notify_all (); + return t; +} + +void +Butler::disable_audio () +{ + boost::mutex::scoped_lock lm (_mutex); + _disable_audio = true; +} + +pair +Butler::memory_used () const +{ + /* XXX: should also look at _audio.memory_used() */ + return _video.memory_used(); } void -Butler::player_changed () +Butler::player_change (ChangeType type, bool frequent) { - optional t; + boost::mutex::scoped_lock lm (_mutex); - { - boost::mutex::scoped_lock lm (_mutex); - t = _video.earliest (); - } + if (type == CHANGE_TYPE_PENDING) { + ++_suspended; + } else if (type == CHANGE_TYPE_DONE) { + --_suspended; + if (_died || _pending_seek_position || frequent) { + lm.unlock (); + _summon.notify_all (); + return; + } - if (t) { - seek (*t, true); + DCPTime seek_to; + DCPTime next = _video.get().second; + if (_awaiting && _awaiting > next) { + /* We have recently done a player_changed seek and our buffers haven't been refilled yet, + so assume that we're seeking to the same place as last time. + */ + seek_to = *_awaiting; + } else { + seek_to = next; + } + + seek_unlocked (seek_to, true); + _awaiting = seek_to; + } else if (type == CHANGE_TYPE_CANCELLED) { + --_suspended; } + + lm.unlock (); + _summon.notify_all (); } void -Butler::get_audio (float* out, Frame frames) +Butler::text (PlayerText pt, TextType type, optional track, DCPTimePeriod period) { - _audio.get (reinterpret_cast (out), _audio_channels, frames); - _summon.notify_all (); + if (type != TEXT_CLOSED_CAPTION) { + return; + } + + DCPOMATIC_ASSERT (track); + + boost::mutex::scoped_lock lm2 (_buffers_mutex); + _closed_caption.put (pt, *track, period); }