src/lib/encoder.cc

   1 /*
   2     Copyright (C) 2012 Carl Hetherington <cth@carlh.net>
   3
   4     This program is free software; you can redistribute it and/or modify
   5     it under the terms of the GNU General Public License as published by
   6     the Free Software Foundation; either version 2 of the License, or
   7     (at your option) any later version.
   8
   9     This program is distributed in the hope that it will be useful,
  10     but WITHOUT ANY WARRANTY; without even the implied warranty of
  11     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12     GNU General Public License for more details.
  13
  14     You should have received a copy of the GNU General Public License
  15     along with this program; if not, write to the Free Software
  16     Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  17
  18 */
  19
  20 /** @file src/encoder.h
  21  *  @brief Parent class for classes which can encode video and audio frames.
  22  */
  23
  24 #include <iostream>
  25 #include <boost/filesystem.hpp>
  26 #include <boost/lexical_cast.hpp>
  27 #include "encoder.h"
  28 #include "util.h"
  29 #include "options.h"
  30 #include "film.h"
  31 #include "log.h"
  32 #include "exceptions.h"
  33 #include "filter.h"
  34 #include "config.h"
  35 #include "dcp_video_frame.h"
  36 #include "server.h"
  37 #include "format.h"
  38 #include "cross.h"
  39
  40 using std::pair;
  41 using std::string;
  42 using std::stringstream;
  43 using std::vector;
  44 using std::list;
  45 using std::cout;
  46 using std::make_pair;
  47 using namespace boost;
  48
  49 int const Encoder::_history_size = 25;
  50
  51 /** @param f Film that we are encoding.
  52  *  @param o Options.
  53  */
  54 Encoder::Encoder (shared_ptr<const Film> f)
  55         : _film (f)
  56         , _just_skipped (false)
  57         , _video_frame (0)
  58         , _audio_frame (0)
  59 #ifdef HAVE_SWRESAMPLE
  60         , _swr_context (0)
  61 #endif
  62         , _audio_frames_written (0)
  63         , _process_end (false)
  64 {
  65         if (_film->audio_stream()) {
  66                 /* Create sound output files with .tmp suffixes; we will rename
  67                    them if and when we complete.
  68                 */
  69                 for (int i = 0; i < dcp_audio_channels (_film->audio_channels()); ++i) {
  70                         SF_INFO sf_info;
  71                         sf_info.samplerate = dcp_audio_sample_rate (_film->audio_stream()->sample_rate());
  72                         /* We write mono files */
  73                         sf_info.channels = 1;
  74                         sf_info.format = SF_FORMAT_WAV | SF_FORMAT_PCM_24;
  75                         SNDFILE* f = sf_open (_film->multichannel_audio_out_path (i, true).c_str (), SFM_WRITE, &sf_info);
  76                         if (f == 0) {
  77                                 throw CreateFileError (_film->multichannel_audio_out_path (i, true));
  78                         }
  79                         _sound_files.push_back (f);
  80                 }
  81         }
  82 }
  83
  84 Encoder::~Encoder ()
  85 {
  86         close_sound_files ();
  87         terminate_worker_threads ();
  88 }
  89
  90 void
  91 Encoder::process_begin ()
  92 {
  93         if (_film->audio_stream() && _film->audio_stream()->sample_rate() != _film->target_audio_sample_rate()) {
  94 #ifdef HAVE_SWRESAMPLE
  95
  96                 stringstream s;
  97                 s << "Will resample audio from " << _film->audio_stream()->sample_rate() << " to " << _film->target_audio_sample_rate();
  98                 _film->log()->log (s.str ());
  99
 100                 /* We will be using planar float data when we call the resampler */
 101                 _swr_context = swr_alloc_set_opts (
 102                         0,
 103                         _film->audio_stream()->channel_layout(),
 104                         AV_SAMPLE_FMT_FLTP,
 105                         _film->target_audio_sample_rate(),
 106                         _film->audio_stream()->channel_layout(),
 107                         AV_SAMPLE_FMT_FLTP,
 108                         _film->audio_stream()->sample_rate(),
 109                         0, 0
 110                         );
 111
 112                 swr_init (_swr_context);
 113 #else
 114                 throw EncodeError ("Cannot resample audio as libswresample is not present");
 115 #endif
 116         } else {
 117 #ifdef HAVE_SWRESAMPLE
 118                 _swr_context = 0;
 119 #endif
 120         }
 121
 122         for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
 123                 _worker_threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, (ServerDescription *) 0)));
 124         }
 125
 126         vector<ServerDescription*> servers = Config::instance()->servers ();
 127
 128         for (vector<ServerDescription*>::iterator i = servers.begin(); i != servers.end(); ++i) {
 129                 for (int j = 0; j < (*i)->threads (); ++j) {
 130                         _worker_threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, *i)));
 131                 }
 132         }
 133 }
 134
 135
 136 void
 137 Encoder::process_end ()
 138 {
 139 #if HAVE_SWRESAMPLE
 140         if (_film->audio_stream() && _film->audio_stream()->channels() && _swr_context) {
 141
 142                 shared_ptr<AudioBuffers> out (new AudioBuffers (_film->audio_stream()->channels(), 256));
 143
 144                 while (1) {
 145                         int const frames = swr_convert (_swr_context, (uint8_t **) out->data(), 256, 0, 0);
 146
 147                         if (frames < 0) {
 148                                 throw EncodeError ("could not run sample-rate converter");
 149                         }
 150
 151                         if (frames == 0) {
 152                                 break;
 153                         }
 154
 155                         out->set_frames (frames);
 156                         write_audio (out);
 157                 }
 158
 159                 swr_free (&_swr_context);
 160         }
 161 #endif
 162
 163         if (_film->audio_stream()) {
 164                 close_sound_files ();
 165
 166                 /* Rename .wav.tmp files to .wav */
 167                 for (int i = 0; i < dcp_audio_channels (_film->audio_channels()); ++i) {
 168                         if (boost::filesystem::exists (_film->multichannel_audio_out_path (i, false))) {
 169                                 boost::filesystem::remove (_film->multichannel_audio_out_path (i, false));
 170                         }
 171                         boost::filesystem::rename (_film->multichannel_audio_out_path (i, true), _film->multichannel_audio_out_path (i, false));
 172                 }
 173         }
 174
 175         boost::mutex::scoped_lock lock (_worker_mutex);
 176
 177         _film->log()->log ("Clearing queue of " + lexical_cast<string> (_queue.size ()));
 178
 179         /* Keep waking workers until the queue is empty */
 180         while (!_queue.empty ()) {
 181                 _film->log()->log ("Waking with " + lexical_cast<string> (_queue.size ()), Log::VERBOSE);
 182                 _worker_condition.notify_all ();
 183                 _worker_condition.wait (lock);
 184         }
 185
 186         lock.unlock ();
 187
 188         terminate_worker_threads ();
 189
 190         _film->log()->log ("Mopping up " + lexical_cast<string> (_queue.size()));
 191
 192         /* The following sequence of events can occur in the above code:
 193              1. a remote worker takes the last image off the queue
 194              2. the loop above terminates
 195              3. the remote worker fails to encode the image and puts it back on the queue
 196              4. the remote worker is then terminated by terminate_worker_threads
 197
 198              So just mop up anything left in the queue here.
 199         */
 200
 201         for (list<shared_ptr<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
 202                 _film->log()->log (String::compose ("Encode left-over frame %1", (*i)->frame ()));
 203                 try {
 204                         shared_ptr<EncodedData> e = (*i)->encode_locally ();
 205                         e->write (_film, (*i)->frame ());
 206                         frame_done ();
 207                 } catch (std::exception& e) {
 208                         _film->log()->log (String::compose ("Local encode failed (%1)", e.what ()));
 209                 }
 210         }
 211
 212         /* Now do links (or copies on windows) to duplicate frames */
 213         for (list<pair<int, int> >::iterator i = _links_required.begin(); i != _links_required.end(); ++i) {
 214                 link (_film->frame_out_path (i->first, false), _film->frame_out_path (i->second, false));
 215                 link (_film->hash_out_path (i->first, false), _film->hash_out_path (i->second, false));
 216         }
 217 }
 218
 219 /** @return an estimate of the current number of frames we are encoding per second,
 220  *  or 0 if not known.
 221  */
 222 float
 223 Encoder::current_frames_per_second () const
 224 {
 225         boost::mutex::scoped_lock lock (_history_mutex);
 226         if (int (_time_history.size()) < _history_size) {
 227                 return 0;
 228         }
 229
 230         struct timeval now;
 231         gettimeofday (&now, 0);
 232
 233         return _history_size / (seconds (now) - seconds (_time_history.back ()));
 234 }
 235
 236 /** @return true if the last frame to be processed was skipped as it already existed */
 237 bool
 238 Encoder::skipping () const
 239 {
 240         boost::mutex::scoped_lock (_history_mutex);
 241         return _just_skipped;
 242 }
 243
 244 /** @return Number of video frames that have been received */
 245 SourceFrame
 246 Encoder::video_frame () const
 247 {
 248         boost::mutex::scoped_lock (_history_mutex);
 249         return _video_frame;
 250 }
 251
 252 /** Should be called when a frame has been encoded successfully.
 253  *  @param n Source frame index.
 254  */
 255 void
 256 Encoder::frame_done ()
 257 {
 258         boost::mutex::scoped_lock lock (_history_mutex);
 259         _just_skipped = false;
 260
 261         struct timeval tv;
 262         gettimeofday (&tv, 0);
 263         _time_history.push_front (tv);
 264         if (int (_time_history.size()) > _history_size) {
 265                 _time_history.pop_back ();
 266         }
 267 }
 268
 269 /** Called by a subclass when it has just skipped the processing
 270     of a frame because it has already been done.
 271 */
 272 void
 273 Encoder::frame_skipped ()
 274 {
 275         boost::mutex::scoped_lock lock (_history_mutex);
 276         _just_skipped = true;
 277 }
 278
 279 void
 280 Encoder::process_video (shared_ptr<Image> image, bool same, boost::shared_ptr<Subtitle> sub)
 281 {
 282         DCPFrameRate dfr (_film->frames_per_second ());
 283
 284         if (dfr.skip && (_video_frame % 2)) {
 285                 ++_video_frame;
 286                 return;
 287         }
 288
 289         if (_film->video_range ()) {
 290                 pair<SourceFrame, SourceFrame> const r = _film->video_range().get();
 291                 if (_video_frame < r.first || _video_frame >= r.second) {
 292                         ++_video_frame;
 293                         return;
 294                 }
 295         }
 296
 297         boost::mutex::scoped_lock lock (_worker_mutex);
 298
 299         /* Wait until the queue has gone down a bit */
 300         while (_queue.size() >= _worker_threads.size() * 2 && !_process_end) {
 301                 TIMING ("decoder sleeps with queue of %1", _queue.size());
 302                 _worker_condition.wait (lock);
 303                 TIMING ("decoder wakes with queue of %1", _queue.size());
 304         }
 305
 306         if (_process_end) {
 307                 return;
 308         }
 309
 310         /* Only do the processing if we don't already have a file for this frame */
 311         if (boost::filesystem::exists (_film->frame_out_path (_video_frame, false))) {
 312                 frame_skipped ();
 313                 return;
 314         }
 315
 316         if (same && _last_real_frame) {
 317                 /* Use the last frame that we encoded.  We need to postpone doing the actual link,
 318                    as on windows the link is really a copy and the reference frame might not have
 319                    finished encoding yet.
 320                 */
 321                 _links_required.push_back (make_pair (_last_real_frame.get(), _video_frame));
 322         } else {
 323                 /* Queue this new frame for encoding */
 324                 pair<string, string> const s = Filter::ffmpeg_strings (_film->filters());
 325                 TIMING ("adding to queue of %1", _queue.size ());
 326                 _queue.push_back (boost::shared_ptr<DCPVideoFrame> (
 327                                           new DCPVideoFrame (
 328                                                   image, sub, _film->format()->dcp_size(), _film->format()->dcp_padding (_film),
 329                                                   _film->subtitle_offset(), _film->subtitle_scale(),
 330                                                   _film->scaler(), _video_frame, _film->frames_per_second(), s.second,
 331                                                   _film->colour_lut(), _film->j2k_bandwidth(),
 332                                                   _film->log()
 333                                                   )
 334                                           ));
 335
 336                 _worker_condition.notify_all ();
 337                 _last_real_frame = _video_frame;
 338         }
 339
 340         ++_video_frame;
 341 }
 342
 343 void
 344 Encoder::process_audio (shared_ptr<AudioBuffers> data)
 345 {
 346         if (_film->audio_range ()) {
 347                 shared_ptr<AudioBuffers> trimmed (new AudioBuffers (*data.get ()));
 348
 349                 /* Range that we are encoding */
 350                 pair<int64_t, int64_t> required_range = _film->audio_range().get();
 351                 /* Range of this block of data */
 352                 pair<int64_t, int64_t> this_range (_audio_frame, _audio_frame + trimmed->frames());
 353
 354                 if (this_range.second < required_range.first || required_range.second < this_range.first) {
 355                         /* No part of this audio is within the required range */
 356                         return;
 357                 } else if (required_range.first >= this_range.first && required_range.first < this_range.second) {
 358                         /* Trim start */
 359                         int64_t const shift = required_range.first - this_range.first;
 360                         trimmed->move (shift, 0, trimmed->frames() - shift);
 361                         trimmed->set_frames (trimmed->frames() - shift);
 362                 } else if (required_range.second >= this_range.first && required_range.second < this_range.second) {
 363                         /* Trim end */
 364                         trimmed->set_frames (required_range.second - this_range.first);
 365                 }
 366
 367                 data = trimmed;
 368         }
 369
 370 #if HAVE_SWRESAMPLE
 371         /* Maybe sample-rate convert */
 372         if (_swr_context) {
 373
 374                 /* Compute the resampled frames count and add 32 for luck */
 375                 int const max_resampled_frames = ceil ((int64_t) data->frames() * _film->target_audio_sample_rate() / _film->audio_stream()->sample_rate()) + 32;
 376
 377                 shared_ptr<AudioBuffers> resampled (new AudioBuffers (_film->audio_stream()->channels(), max_resampled_frames));
 378
 379                 /* Resample audio */
 380                 int const resampled_frames = swr_convert (
 381                         _swr_context, (uint8_t **) resampled->data(), max_resampled_frames, (uint8_t const **) data->data(), data->frames()
 382                         );
 383
 384                 if (resampled_frames < 0) {
 385                         throw EncodeError ("could not run sample-rate converter");
 386                 }
 387
 388                 resampled->set_frames (resampled_frames);
 389
 390                 /* And point our variables at the resampled audio */
 391                 data = resampled;
 392         }
 393 #endif
 394
 395         if (_film->audio_channels() == 1) {
 396                 /* We need to switch things around so that the mono channel is on
 397                    the centre channel of a 5.1 set (with other channels silent).
 398                 */
 399
 400                 shared_ptr<AudioBuffers> b (new AudioBuffers (6, data->frames ()));
 401                 b->make_silent (libdcp::LEFT);
 402                 b->make_silent (libdcp::RIGHT);
 403                 memcpy (b->data()[libdcp::CENTRE], data->data()[0], data->frames() * sizeof(float));
 404                 b->make_silent (libdcp::LFE);
 405                 b->make_silent (libdcp::LS);
 406                 b->make_silent (libdcp::RS);
 407
 408                 data = b;
 409         }
 410
 411         write_audio (data);
 412
 413         _audio_frame += data->frames ();
 414 }
 415
 416 void
 417 Encoder::write_audio (shared_ptr<const AudioBuffers> audio)
 418 {
 419         for (int i = 0; i < audio->channels(); ++i) {
 420                 sf_write_float (_sound_files[i], audio->data(i), audio->frames());
 421         }
 422
 423         _audio_frames_written += audio->frames ();
 424 }
 425
 426 void
 427 Encoder::close_sound_files ()
 428 {
 429         for (vector<SNDFILE*>::iterator i = _sound_files.begin(); i != _sound_files.end(); ++i) {
 430                 sf_close (*i);
 431         }
 432
 433         _sound_files.clear ();
 434 }
 435
 436 void
 437 Encoder::terminate_worker_threads ()
 438 {
 439         boost::mutex::scoped_lock lock (_worker_mutex);
 440         _process_end = true;
 441         _worker_condition.notify_all ();
 442         lock.unlock ();
 443
 444         for (list<boost::thread *>::iterator i = _worker_threads.begin(); i != _worker_threads.end(); ++i) {
 445                 (*i)->join ();
 446                 delete *i;
 447         }
 448 }
 449
 450 void
 451 Encoder::encoder_thread (ServerDescription* server)
 452 {
 453         /* Number of seconds that we currently wait between attempts
 454            to connect to the server; not relevant for localhost
 455            encodings.
 456         */
 457         int remote_backoff = 0;
 458
 459         while (1) {
 460
 461                 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
 462                 boost::mutex::scoped_lock lock (_worker_mutex);
 463                 while (_queue.empty () && !_process_end) {
 464                         _worker_condition.wait (lock);
 465                 }
 466
 467                 if (_process_end) {
 468                         return;
 469                 }
 470
 471                 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 472                 boost::shared_ptr<DCPVideoFrame> vf = _queue.front ();
 473                 _film->log()->log (String::compose ("Encoder thread %1 pops frame %2 from queue", boost::this_thread::get_id(), vf->frame()), Log::VERBOSE);
 474                 _queue.pop_front ();
 475
 476                 lock.unlock ();
 477
 478                 shared_ptr<EncodedData> encoded;
 479
 480                 if (server) {
 481                         try {
 482                                 encoded = vf->encode_remotely (server);
 483
 484                                 if (remote_backoff > 0) {
 485                                         _film->log()->log (String::compose ("%1 was lost, but now she is found; removing backoff", server->host_name ()));
 486                                 }
 487
 488                                 /* This job succeeded, so remove any backoff */
 489                                 remote_backoff = 0;
 490
 491                         } catch (std::exception& e) {
 492                                 if (remote_backoff < 60) {
 493                                         /* back off more */
 494                                         remote_backoff += 10;
 495                                 }
 496                                 _film->log()->log (
 497                                         String::compose (
 498                                                 "Remote encode of %1 on %2 failed (%3); thread sleeping for %4s",
 499                                                 vf->frame(), server->host_name(), e.what(), remote_backoff)
 500                                         );
 501                         }
 502
 503                 } else {
 504                         try {
 505                                 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
 506                                 encoded = vf->encode_locally ();
 507                                 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
 508                         } catch (std::exception& e) {
 509                                 _film->log()->log (String::compose ("Local encode failed (%1)", e.what ()));
 510                         }
 511                 }
 512
 513                 if (encoded) {
 514                         encoded->write (_film, vf->frame ());
 515                         frame_done ();
 516                 } else {
 517                         lock.lock ();
 518                         _film->log()->log (
 519                                 String::compose ("Encoder thread %1 pushes frame %2 back onto queue after failure", boost::this_thread::get_id(), vf->frame())
 520                                 );
 521                         _queue.push_front (vf);
 522                         lock.unlock ();
 523                 }
 524
 525                 if (remote_backoff > 0) {
 526                         dvdomatic_sleep (remote_backoff);
 527                 }
 528
 529                 lock.lock ();
 530                 _worker_condition.notify_all ();
 531         }
 532 }
 533
 534 void
 535 Encoder::link (string a, string b) const
 536 {
 537 #ifdef DVDOMATIC_POSIX
 538         int const r = symlink (a.c_str(), b.c_str());
 539         if (r) {
 540                 throw EncodeError (String::compose ("could not create symlink from %1 to %2", a, b));
 541         }
 542 #endif
 543
 544 #ifdef DVDOMATIC_WINDOWS
 545         boost::filesystem::copy_file (a, b);
 546 #endif
 547 }