X-Git-Url: https://git.carlh.net/gitweb/?a=blobdiff_plain;f=src%2Flib%2Fencoder.cc;h=c1d1041ae539f9cdab1f087291eb2d8d7104abbb;hb=f861018389acd9d277fe34d7621182b9b54f977f;hp=0d5fc1c6efbca6dd8a843a284b7dfde905d3959e;hpb=a93bbd88eaa67a1e24471789335a2790b6cfbc78;p=dcpomatic.git diff --git a/src/lib/encoder.cc b/src/lib/encoder.cc index 0d5fc1c6e..c1d1041ae 100644 --- a/src/lib/encoder.cc +++ b/src/lib/encoder.cc @@ -27,7 +27,6 @@ #include #include "encoder.h" #include "util.h" -#include "options.h" #include "film.h" #include "log.h" #include "exceptions.h" @@ -37,6 +36,11 @@ #include "server.h" #include "format.h" #include "cross.h" +#include "writer.h" +#include "player.h" +#include "audio_mapping.h" + +#include "i18n.h" using std::pair; using std::string; @@ -45,81 +49,65 @@ using std::vector; using std::list; using std::cout; using std::make_pair; -using namespace boost; +using boost::shared_ptr; +using boost::optional; int const Encoder::_history_size = 25; -unsigned int const Encoder::_maximum_frames_in_memory = 8; -/** @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) - , _audio_frames_in (0) , _video_frames_out (0) - , _audio_frames_out (0) #ifdef HAVE_SWRESAMPLE , _swr_context (0) #endif , _have_a_real_frame (false) - , _terminate_encoder (false) - , _writer_thread (0) - , _finish_writer (false) - , _last_written_frame (-1) + , _terminate (false) { - if (_film->audio_stream()) { - /* Create sound output files with .tmp suffixes; we will rename - them if and when we complete. - */ - for (int i = 0; i < dcp_audio_channels (_film->audio_channels()); ++i) { - SF_INFO sf_info; - sf_info.samplerate = dcp_audio_sample_rate (_film->audio_stream()->sample_rate()); - /* We write mono files */ - sf_info.channels = 1; - sf_info.format = SF_FORMAT_WAV | SF_FORMAT_PCM_24; - SNDFILE* f = sf_open (_film->multichannel_audio_out_path (i, true).c_str (), SFM_WRITE, &sf_info); - if (f == 0) { - throw CreateFileError (_film->multichannel_audio_out_path (i, true)); - } - _sound_files.push_back (f); - } - } + } Encoder::~Encoder () { - close_sound_files (); - terminate_worker_threads (); - finish_writer_thread (); + terminate_threads (); + if (_writer) { + _writer->finish (); + } } void Encoder::process_begin () { - if (_film->audio_stream() && _film->audio_stream()->sample_rate() != _film->target_audio_sample_rate()) { + if (_film->has_audio() && _film->audio_frame_rate() != _film->target_audio_sample_rate()) { #ifdef HAVE_SWRESAMPLE stringstream s; - s << "Will resample audio from " << _film->audio_stream()->sample_rate() << " to " << _film->target_audio_sample_rate(); + s << String::compose (N_("Will resample audio from %1 to %2"), _film->audio_frame_rate(), _film->target_audio_sample_rate()); _film->log()->log (s.str ()); - /* We will be using planar float data when we call the resampler */ + /* We will be using planar float data when we call the + resampler. As far as I can see, the audio channel + layout is not necessary for our purposes; it seems + only to be used get the number of channels and + decide if rematrixing is needed. It won't be, since + input and output layouts are the same. + */ + _swr_context = swr_alloc_set_opts ( 0, - _film->audio_stream()->channel_layout(), + av_get_default_channel_layout (_film->audio_mapping().dcp_channels ()), AV_SAMPLE_FMT_FLTP, _film->target_audio_sample_rate(), - _film->audio_stream()->channel_layout(), + av_get_default_channel_layout (_film->audio_mapping().dcp_channels ()), AV_SAMPLE_FMT_FLTP, - _film->audio_stream()->sample_rate(), + _film->audio_frame_rate(), 0, 0 ); swr_init (_swr_context); #else - throw EncodeError ("Cannot resample audio as libswresample is not present"); + throw EncodeError (_("Cannot resample audio as libswresample is not present")); #endif } else { #ifdef HAVE_SWRESAMPLE @@ -128,29 +116,18 @@ 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))); } } - /* XXX! */ - _picture_asset.reset ( - new libdcp::MonoPictureAsset ( - _film->dir (_film->dcp_name()), - String::compose ("video_%1.mxf", 0), - DCPFrameRate (_film->frames_per_second()).frames_per_second, - _film->format()->dcp_size() - ) - ); - - _picture_asset_writer = _picture_asset->start_write (); - _writer_thread = new boost::thread (boost::bind (&Encoder::writer_thread, this)); + _writer.reset (new Writer (_film)); } @@ -158,15 +135,15 @@ void Encoder::process_end () { #if HAVE_SWRESAMPLE - if (_film->audio_stream() && _film->audio_stream()->channels() && _swr_context) { + if (_film->has_audio() && _swr_context) { - shared_ptr out (new AudioBuffers (_film->audio_stream()->channels(), 256)); + shared_ptr out (new AudioBuffers (_film->audio_mapping().dcp_channels(), 256)); while (1) { int const frames = swr_convert (_swr_context, (uint8_t **) out->data(), 256, 0, 0); if (frames < 0) { - throw EncodeError ("could not run sample-rate converter"); + throw EncodeError (_("could not run sample-rate converter")); } if (frames == 0) { @@ -174,75 +151,58 @@ Encoder::process_end () } out->set_frames (frames); - write_audio (out); + _writer->write (out); } swr_free (&_swr_context); } #endif - if (_film->audio_stream()) { - close_sound_files (); - - /* Rename .wav.tmp files to .wav */ - for (int i = 0; i < dcp_audio_channels (_film->audio_channels()); ++i) { - if (boost::filesystem::exists (_film->multichannel_audio_out_path (i, false))) { - boost::filesystem::remove (_film->multichannel_audio_out_path (i, false)); - } - boost::filesystem::rename (_film->multichannel_audio_out_path (i, true), _film->multichannel_audio_out_path (i, false)); - } - } - - 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 (String::compose (N_("Clearing queue of %1"), _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 (String::compose (N_("Waking with %1"), _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 (String::compose (N_("Mopping up %1"), _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) { - _film->log()->log (String::compose ("Encode left-over frame %1", (*i)->frame ())); + for (list >::iterator i = _queue.begin(); i != _queue.end(); ++i) { + _film->log()->log (String::compose (N_("Encode left-over frame %1"), (*i)->frame ())); try { - shared_ptr e = (*i)->encode_locally (); - { - boost::mutex::scoped_lock lock2 (_writer_mutex); - _write_queue.push_back (make_pair (e, (*i)->frame ())); - _writer_condition.notify_all (); - } + _writer->write ((*i)->encode_locally(), (*i)->frame ()); frame_done (); } catch (std::exception& e) { - _film->log()->log (String::compose ("Local encode failed (%1)", e.what ())); + _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ())); } } - finish_writer_thread (); - _picture_asset_writer->finalize (); + _writer->finish (); + _writer.reset (); } /** @return an estimate of the current number of frames we are encoding per second, * or 0 if not known. */ float -Encoder::current_frames_per_second () const +Encoder::current_encoding_rate () const { boost::mutex::scoped_lock lock (_history_mutex); if (int (_time_history.size()) < _history_size) { @@ -255,14 +215,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 @@ -278,7 +230,6 @@ void Encoder::frame_done () { boost::mutex::scoped_lock lock (_history_mutex); - _just_skipped = false; struct timeval tv; gettimeofday (&tv, 0); @@ -288,91 +239,80 @@ 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 () +Encoder::process_video (shared_ptr image, bool same, shared_ptr sub) { - boost::mutex::scoped_lock lock (_history_mutex); - _just_skipped = true; -} - -void -Encoder::process_video (shared_ptr image, bool same, boost::shared_ptr sub) -{ - DCPFrameRate dfr (_film->frames_per_second ()); + FrameRateConversion frc (_film->video_frame_rate(), _film->dcp_frame_rate()); - if (dfr.skip && (_video_frames_in % 2)) { + if (frc.skip && (_video_frames_in % 2)) { ++_video_frames_in; return; } - boost::mutex::scoped_lock lock (_worker_mutex); + boost::mutex::scoped_lock lock (_mutex); /* Wait until the queue has gone down a bit */ - while (_encode_queue.size() >= _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) { - /* Use the last frame that we encoded. We do this by putting a null encoded - frame straight onto the writer's queue. It will know to duplicate the previous frame - in this case. - */ - boost::mutex::scoped_lock lock2 (_writer_mutex); - _write_queue.push_back (make_pair (shared_ptr (), _video_frames_out)); + 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 (shared_ptr ( new DCPVideoFrame ( image, sub, _film->format()->dcp_size(), _film->format()->dcp_padding (_film), _film->subtitle_offset(), _film->subtitle_scale(), - _film->scaler(), _video_frames_out, _film->frames_per_second(), s.second, + _film->scaler(), _video_frames_out, _film->dcp_frame_rate(), s.second, _film->colour_lut(), _film->j2k_bandwidth(), _film->log() ) )); - _worker_condition.notify_all (); + _condition.notify_all (); _have_a_real_frame = true; } ++_video_frames_in; ++_video_frames_out; - if (dfr.repeat) { - boost::mutex::scoped_lock lock2 (_writer_mutex); - _write_queue.push_back (make_pair (shared_ptr (), _video_frames_out)); + if (frc.repeat) { + _writer->repeat (_video_frames_out); ++_video_frames_out; + frame_done (); } } void -Encoder::process_audio (shared_ptr data) +Encoder::process_audio (shared_ptr data) { #if HAVE_SWRESAMPLE /* Maybe sample-rate convert */ if (_swr_context) { /* Compute the resampled frames count and add 32 for luck */ - int const max_resampled_frames = ceil ((int64_t) data->frames() * _film->target_audio_sample_rate() / _film->audio_stream()->sample_rate()) + 32; + int const max_resampled_frames = ceil ((int64_t) data->frames() * _film->target_audio_sample_rate() / _film->audio_frame_rate()) + 32; - shared_ptr resampled (new AudioBuffers (_film->audio_stream()->channels(), max_resampled_frames)); + shared_ptr resampled (new AudioBuffers (_film->audio_mapping().dcp_channels(), max_resampled_frames)); /* Resample audio */ int const resampled_frames = swr_convert ( @@ -380,7 +320,7 @@ Encoder::process_audio (shared_ptr data) ); if (resampled_frames < 0) { - throw EncodeError ("could not run sample-rate converter"); + throw EncodeError (_("could not run sample-rate converter")); } resampled->set_frames (resampled_frames); @@ -390,78 +330,25 @@ Encoder::process_audio (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; - } - - write_audio (data); - - _audio_frames_in += data->frames (); + _writer->write (data); } void -Encoder::write_audio (shared_ptr audio) +Encoder::terminate_threads () { - for (int i = 0; i < audio->channels(); ++i) { - sf_write_float (_sound_files[i], audio->data(i), audio->frames()); - } - - _audio_frames_out += audio->frames (); -} - -void -Encoder::close_sound_files () -{ - for (vector::iterator i = _sound_files.begin(); i != _sound_files.end(); ++i) { - sf_close (*i); - } - - _sound_files.clear (); -} - -void -Encoder::terminate_worker_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) { - (*i)->join (); + for (list::iterator i = _threads.begin(); i != _threads.end(); ++i) { + if ((*i)->joinable ()) { + (*i)->join (); + } delete *i; } } -void -Encoder::finish_writer_thread () -{ - if (!_writer_thread) { - return; - } - - boost::mutex::scoped_lock lock (_writer_mutex); - _finish_writer = true; - _writer_condition.notify_all (); - lock.unlock (); - - _writer_thread->join (); - delete _writer_thread; - _writer_thread = 0; -} - void Encoder::encoder_thread (ServerDescription* server) { @@ -474,19 +361,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 (); - _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 (); + TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size()); + shared_ptr vf = _queue.front (); + _film->log()->log (String::compose (N_("Encoder thread %1 pops frame %2 from queue"), boost::this_thread::get_id(), vf->frame()), Log::VERBOSE); + _queue.pop_front (); lock.unlock (); @@ -497,7 +384,7 @@ Encoder::encoder_thread (ServerDescription* server) encoded = vf->encode_remotely (server); if (remote_backoff > 0) { - _film->log()->log (String::compose ("%1 was lost, but now she is found; removing backoff", server->host_name ())); + _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ())); } /* This job succeeded, so remove any backoff */ @@ -510,7 +397,7 @@ Encoder::encoder_thread (ServerDescription* server) } _film->log()->log ( String::compose ( - "Remote encode of %1 on %2 failed (%3); thread sleeping for %4s", + N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"), vf->frame(), server->host_name(), e.what(), remote_backoff) ); } @@ -521,135 +408,27 @@ Encoder::encoder_thread (ServerDescription* server) encoded = vf->encode_locally (); TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame()); } catch (std::exception& e) { - _film->log()->log (String::compose ("Local encode failed (%1)", e.what ())); + _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ())); } } if (encoded) { - boost::mutex::scoped_lock lock2 (_writer_mutex); - _write_queue.push_back (make_pair (encoded, vf->frame ())); - _writer_condition.notify_all (); + _writer->write (encoded, vf->frame ()); + frame_done (); } else { lock.lock (); _film->log()->log ( - String::compose ("Encoder thread %1 pushes frame %2 back onto queue after failure", boost::this_thread::get_id(), vf->frame()) + String::compose (N_("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 (); } if (remote_backoff > 0) { - dvdomatic_sleep (remote_backoff); + dcpomatic_sleep (remote_backoff); } lock.lock (); - _worker_condition.notify_all (); - } -} - -void -Encoder::link (string a, string b) const -{ -#ifdef DVDOMATIC_POSIX - int const r = symlink (a.c_str(), b.c_str()); - if (r) { - throw EncodeError (String::compose ("could not create symlink from %1 to %2", a, b)); - } -#endif - -#ifdef DVDOMATIC_WINDOWS - boost::filesystem::copy_file (a, b); -#endif -} - -struct WriteQueueSorter -{ - bool operator() (pair, int> const & a, pair, int> const & b) { - return a.second < b.second; - } -}; - -void -Encoder::writer_thread () -{ - while (1) - { - boost::mutex::scoped_lock lock (_writer_mutex); - - while (1) { - if (_finish_writer || - _write_queue.size() > _maximum_frames_in_memory || - (!_write_queue.empty() && _write_queue.front().second == (_last_written_frame + 1))) { - - break; - } - - TIMING ("writer sleeps with a queue of %1; %2 pending", _write_queue.size(), _pending.size()); - _writer_condition.wait (lock); - TIMING ("writer wakes with a queue of %1", _write_queue.size()); - - _write_queue.sort (WriteQueueSorter ()); - } - - if (_finish_writer && _write_queue.empty() && _pending.empty()) { - return; - } - - /* Write any frames that we can write; i.e. those that are in sequence */ - while (!_write_queue.empty() && _write_queue.front().second == (_last_written_frame + 1)) { - pair, int> encoded = _write_queue.front (); - _write_queue.pop_front (); - - lock.unlock (); - _film->log()->log (String::compose ("Writer writes %1 to MXF", encoded.second)); - if (encoded.first) { - _picture_asset_writer->write (encoded.first->data(), encoded.first->size()); - _last_written = encoded.first; - } else { - _picture_asset_writer->write (_last_written->data(), _last_written->size()); - } - lock.lock (); - - ++_last_written_frame; - } - - while (_write_queue.size() > _maximum_frames_in_memory) { - /* Too many frames in memory which can't yet be written to the stream. - Put some to disk. - */ - - pair, int> encoded = _write_queue.back (); - _write_queue.pop_back (); - if (!encoded.first) { - /* This is a `repeat-last' frame, so no need to write it to disk */ - continue; - } - - lock.unlock (); - _film->log()->log (String::compose ("Writer full (awaiting %1); pushes %2 to disk", _last_written_frame + 1, encoded.second)); - encoded.first->write (_film, encoded.second); - lock.lock (); - - _pending.push_back (encoded.second); - } - - while (_write_queue.size() < _maximum_frames_in_memory && !_pending.empty()) { - /* We have some space in memory. Fetch some frames back off disk. */ - - _pending.sort (); - int const fetch = _pending.front (); - - lock.unlock (); - _film->log()->log (String::compose ("Writer pulls %1 back from disk", fetch)); - shared_ptr encoded; - if (boost::filesystem::exists (_film->frame_out_path (fetch, false))) { - /* It's an actual frame (not a repeat-last); load it in */ - encoded.reset (new EncodedData (_film->frame_out_path (fetch, false))); - } - lock.lock (); - - _write_queue.push_back (make_pair (encoded, fetch)); - _pending.remove (fetch); - } + _condition.notify_all (); } }