X-Git-Url: https://git.carlh.net/gitweb/?a=blobdiff_plain;f=src%2Flib%2Fencoder.cc;h=f252563792374bbac363be6af999e55f5d08f716;hb=7913cba90bccb9501b63a0518c58abbd5a6b330d;hp=c2416b87e754a0a50208aa4ced7c03e84e90890e;hpb=5e4ab7ebd9a6b62b64fbaf91d7aa1a2a9d4bdec4;p=dcpomatic.git diff --git a/src/lib/encoder.cc b/src/lib/encoder.cc index c2416b87e..f25256379 100644 --- a/src/lib/encoder.cc +++ b/src/lib/encoder.cc @@ -50,26 +50,23 @@ using namespace boost; int const Encoder::_history_size = 25; -/** @param f Film that we are encoding. - * @param o Options. - */ -Encoder::Encoder (shared_ptr f) +/** @param f Film that we are encoding */ +Encoder::Encoder (shared_ptr f) : _film (f) - , _just_skipped (false) , _video_frames_in (0) , _video_frames_out (0) #ifdef HAVE_SWRESAMPLE , _swr_context (0) #endif , _have_a_real_frame (false) - , _terminate_encoder (false) + , _terminate (false) { } Encoder::~Encoder () { - terminate_worker_threads (); + terminate_threads (); if (_writer) { _writer->finish (); } @@ -108,14 +105,14 @@ Encoder::process_begin () } for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) { - _worker_threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, (ServerDescription *) 0))); + _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, (ServerDescription *) 0))); } vector servers = Config::instance()->servers (); for (vector::iterator i = servers.begin(); i != servers.end(); ++i) { for (int j = 0; j < (*i)->threads (); ++j) { - _worker_threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, *i))); + _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, *i))); } } @@ -143,40 +140,40 @@ Encoder::process_end () } out->set_frames (frames); - _writer->write (out); + write_audio (out); } swr_free (&_swr_context); } #endif - boost::mutex::scoped_lock lock (_worker_mutex); + boost::mutex::scoped_lock lock (_mutex); - _film->log()->log ("Clearing queue of " + lexical_cast (_encode_queue.size ())); + _film->log()->log ("Clearing queue of " + lexical_cast (_queue.size ())); /* Keep waking workers until the queue is empty */ - while (!_encode_queue.empty ()) { - _film->log()->log ("Waking with " + lexical_cast (_encode_queue.size ()), Log::VERBOSE); - _worker_condition.notify_all (); - _worker_condition.wait (lock); + while (!_queue.empty ()) { + _film->log()->log ("Waking with " + lexical_cast (_queue.size ()), Log::VERBOSE); + _condition.notify_all (); + _condition.wait (lock); } lock.unlock (); - terminate_worker_threads (); + terminate_threads (); - _film->log()->log ("Mopping up " + lexical_cast (_encode_queue.size())); + _film->log()->log ("Mopping up " + lexical_cast (_queue.size())); /* The following sequence of events can occur in the above code: 1. a remote worker takes the last image off the queue 2. the loop above terminates 3. the remote worker fails to encode the image and puts it back on the queue - 4. the remote worker is then terminated by terminate_worker_threads + 4. the remote worker is then terminated by terminate_threads So just mop up anything left in the queue here. */ - for (list >::iterator i = _encode_queue.begin(); i != _encode_queue.end(); ++i) { + for (list >::iterator i = _queue.begin(); i != _queue.end(); ++i) { _film->log()->log (String::compose ("Encode left-over frame %1", (*i)->frame ())); try { _writer->write ((*i)->encode_locally(), (*i)->frame ()); @@ -207,14 +204,6 @@ Encoder::current_frames_per_second () const return _history_size / (seconds (now) - seconds (_time_history.back ())); } -/** @return true if the last frame to be processed was skipped as it already existed */ -bool -Encoder::skipping () const -{ - boost::mutex::scoped_lock (_history_mutex); - return _just_skipped; -} - /** @return Number of video frames that have been sent out */ int Encoder::video_frames_out () const @@ -230,7 +219,6 @@ void Encoder::frame_done () { boost::mutex::scoped_lock lock (_history_mutex); - _just_skipped = false; struct timeval tv; gettimeofday (&tv, 0); @@ -240,16 +228,6 @@ Encoder::frame_done () } } -/** Called by a subclass when it has just skipped the processing - of a frame because it has already been done. -*/ -void -Encoder::frame_skipped () -{ - boost::mutex::scoped_lock lock (_history_mutex); - _just_skipped = true; -} - void Encoder::process_video (shared_ptr image, bool same, boost::shared_ptr sub) { @@ -260,34 +238,36 @@ Encoder::process_video (shared_ptr image, bool same, boost::shared_ptr= _worker_threads.size() * 2 && !_terminate_encoder) { - TIMING ("decoder sleeps with queue of %1", _encode_queue.size()); - _worker_condition.wait (lock); - TIMING ("decoder wakes with queue of %1", _encode_queue.size()); + while (_queue.size() >= _threads.size() * 2 && !_terminate) { + TIMING ("decoder sleeps with queue of %1", _queue.size()); + _condition.wait (lock); + TIMING ("decoder wakes with queue of %1", _queue.size()); } - if (_terminate_encoder) { + if (_terminate) { return; } - /* Only do the processing if we don't already have a file for this frame */ - if (boost::filesystem::exists (_film->frame_out_path (_video_frames_out, false))) { - frame_skipped (); - return; + if (_writer->thrown ()) { + _writer->rethrow (); } - if (same && _have_a_real_frame) { + if (_writer->can_fake_write (_video_frames_out)) { + _writer->fake_write (_video_frames_out); + _have_a_real_frame = false; + frame_done (); + } else if (same && _have_a_real_frame) { /* Use the last frame that we encoded. */ _writer->repeat (_video_frames_out); frame_done (); } else { /* Queue this new frame for encoding */ pair const s = Filter::ffmpeg_strings (_film->filters()); - TIMING ("adding to queue of %1", _encode_queue.size ()); - _encode_queue.push_back (boost::shared_ptr ( + TIMING ("adding to queue of %1", _queue.size ()); + _queue.push_back (boost::shared_ptr ( new DCPVideoFrame ( image, sub, _film->format()->dcp_size(), _film->format()->dcp_padding (_film), _film->subtitle_offset(), _film->subtitle_scale(), @@ -297,7 +277,7 @@ Encoder::process_video (shared_ptr image, bool same, boost::shared_ptr data) } #endif - if (_film->audio_channels() == 1) { - /* We need to switch things around so that the mono channel is on - the centre channel of a 5.1 set (with other channels silent). - */ - - shared_ptr b (new AudioBuffers (6, data->frames ())); - b->make_silent (libdcp::LEFT); - b->make_silent (libdcp::RIGHT); - memcpy (b->data()[libdcp::CENTRE], data->data()[0], data->frames() * sizeof(float)); - b->make_silent (libdcp::LFE); - b->make_silent (libdcp::LS); - b->make_silent (libdcp::RS); - - data = b; - } - - _writer->write (data); + write_audio (data); } void -Encoder::terminate_worker_threads () +Encoder::terminate_threads () { - boost::mutex::scoped_lock lock (_worker_mutex); - _terminate_encoder = true; - _worker_condition.notify_all (); + boost::mutex::scoped_lock lock (_mutex); + _terminate = true; + _condition.notify_all (); lock.unlock (); - for (list::iterator i = _worker_threads.begin(); i != _worker_threads.end(); ++i) { + for (list::iterator i = _threads.begin(); i != _threads.end(); ++i) { (*i)->join (); delete *i; } @@ -384,19 +348,19 @@ Encoder::encoder_thread (ServerDescription* server) while (1) { TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id()); - boost::mutex::scoped_lock lock (_worker_mutex); - while (_encode_queue.empty () && !_terminate_encoder) { - _worker_condition.wait (lock); + boost::mutex::scoped_lock lock (_mutex); + while (_queue.empty () && !_terminate) { + _condition.wait (lock); } - if (_terminate_encoder) { + if (_terminate) { return; } - TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _encode_queue.size()); - boost::shared_ptr vf = _encode_queue.front (); + TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size()); + boost::shared_ptr vf = _queue.front (); _film->log()->log (String::compose ("Encoder thread %1 pops frame %2 from queue", boost::this_thread::get_id(), vf->frame()), Log::VERBOSE); - _encode_queue.pop_front (); + _queue.pop_front (); lock.unlock (); @@ -443,7 +407,7 @@ Encoder::encoder_thread (ServerDescription* server) _film->log()->log ( String::compose ("Encoder thread %1 pushes frame %2 back onto queue after failure", boost::this_thread::get_id(), vf->frame()) ); - _encode_queue.push_front (vf); + _queue.push_front (vf); lock.unlock (); } @@ -452,6 +416,28 @@ Encoder::encoder_thread (ServerDescription* server) } lock.lock (); - _worker_condition.notify_all (); + _condition.notify_all (); } } + +void +Encoder::write_audio (shared_ptr data) +{ + if (_film->audio_channels() == 1) { + /* We need to switch things around so that the mono channel is on + the centre channel of a 5.1 set (with other channels silent). + */ + + shared_ptr b (new AudioBuffers (6, data->frames ())); + b->make_silent (libdcp::LEFT); + b->make_silent (libdcp::RIGHT); + memcpy (b->data()[libdcp::CENTRE], data->data()[0], data->frames() * sizeof(float)); + b->make_silent (libdcp::LFE); + b->make_silent (libdcp::LS); + b->make_silent (libdcp::RS); + + data = b; + } + + _writer->write (data); +}