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 <libdcp/picture_asset.h>
  28 #include "encoder.h"
  29 #include "util.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 #include "writer.h"
  40 #include "player.h"
  41 #include "audio_mapping.h"
  42
  43 #include "i18n.h"
  44
  45 using std::pair;
  46 using std::string;
  47 using std::stringstream;
  48 using std::vector;
  49 using std::list;
  50 using std::cout;
  51 using std::make_pair;
  52 using boost::shared_ptr;
  53 using boost::optional;
  54
  55 int const Encoder::_history_size = 25;
  56
  57 /** @param f Film that we are encoding */
  58 Encoder::Encoder (shared_ptr<Film> f, shared_ptr<Job> j)
  59         : _film (f)
  60         , _job (j)
  61         , _video_frames_in (0)
  62         , _video_frames_out (0)
  63         , _have_a_real_frame (false)
  64         , _terminate (false)
  65 {
  66
  67 }
  68
  69 Encoder::~Encoder ()
  70 {
  71         terminate_threads ();
  72         if (_writer) {
  73                 _writer->finish ();
  74         }
  75 }
  76
  77 void
  78 Encoder::process_begin ()
  79 {
  80         for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
  81                 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, (ServerDescription *) 0)));
  82         }
  83
  84         vector<ServerDescription*> servers = Config::instance()->servers ();
  85
  86         for (vector<ServerDescription*>::iterator i = servers.begin(); i != servers.end(); ++i) {
  87                 for (int j = 0; j < (*i)->threads (); ++j) {
  88                         _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, *i)));
  89                 }
  90         }
  91
  92         _writer.reset (new Writer (_film, _job));
  93 }
  94
  95
  96 void
  97 Encoder::process_end ()
  98 {
  99         boost::mutex::scoped_lock lock (_mutex);
 100
 101         _film->log()->log (String::compose (N_("Clearing queue of %1"), _queue.size ()));
 102
 103         /* Keep waking workers until the queue is empty */
 104         while (!_queue.empty ()) {
 105                 _film->log()->log (String::compose (N_("Waking with %1"), _queue.size ()), Log::VERBOSE);
 106                 _condition.notify_all ();
 107                 _condition.wait (lock);
 108         }
 109
 110         lock.unlock ();
 111
 112         terminate_threads ();
 113
 114         _film->log()->log (String::compose (N_("Mopping up %1"), _queue.size()));
 115
 116         /* The following sequence of events can occur in the above code:
 117              1. a remote worker takes the last image off the queue
 118              2. the loop above terminates
 119              3. the remote worker fails to encode the image and puts it back on the queue
 120              4. the remote worker is then terminated by terminate_threads
 121
 122              So just mop up anything left in the queue here.
 123         */
 124
 125         for (list<shared_ptr<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
 126                 _film->log()->log (String::compose (N_("Encode left-over frame %1"), (*i)->frame ()));
 127                 try {
 128                         _writer->write ((*i)->encode_locally(), (*i)->frame ());
 129                         frame_done ();
 130                 } catch (std::exception& e) {
 131                         _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 132                 }
 133         }
 134
 135         _writer->finish ();
 136         _writer.reset ();
 137 }
 138
 139 /** @return an estimate of the current number of frames we are encoding per second,
 140  *  or 0 if not known.
 141  */
 142 float
 143 Encoder::current_encoding_rate () const
 144 {
 145         boost::mutex::scoped_lock lock (_history_mutex);
 146         if (int (_time_history.size()) < _history_size) {
 147                 return 0;
 148         }
 149
 150         struct timeval now;
 151         gettimeofday (&now, 0);
 152
 153         return _history_size / (seconds (now) - seconds (_time_history.back ()));
 154 }
 155
 156 /** @return Number of video frames that have been sent out */
 157 int
 158 Encoder::video_frames_out () const
 159 {
 160         boost::mutex::scoped_lock (_history_mutex);
 161         return _video_frames_out;
 162 }
 163
 164 /** Should be called when a frame has been encoded successfully.
 165  *  @param n Source frame index.
 166  */
 167 void
 168 Encoder::frame_done ()
 169 {
 170         boost::mutex::scoped_lock lock (_history_mutex);
 171
 172         struct timeval tv;
 173         gettimeofday (&tv, 0);
 174         _time_history.push_front (tv);
 175         if (int (_time_history.size()) > _history_size) {
 176                 _time_history.pop_back ();
 177         }
 178 }
 179
 180 void
 181 Encoder::process_video (shared_ptr<const Image> image, bool same, shared_ptr<Subtitle> sub, Time)
 182 {
 183         FrameRateConversion frc (_film->video_frame_rate(), _film->dcp_frame_rate());
 184
 185         if (frc.skip && (_video_frames_in % 2)) {
 186                 ++_video_frames_in;
 187                 return;
 188         }
 189
 190         boost::mutex::scoped_lock lock (_mutex);
 191
 192         /* Wait until the queue has gone down a bit */
 193         while (_queue.size() >= _threads.size() * 2 && !_terminate) {
 194                 TIMING ("decoder sleeps with queue of %1", _queue.size());
 195                 _condition.wait (lock);
 196                 TIMING ("decoder wakes with queue of %1", _queue.size());
 197         }
 198
 199         if (_terminate) {
 200                 return;
 201         }
 202
 203         if (_writer->thrown ()) {
 204                 _writer->rethrow ();
 205         }
 206
 207         if (_writer->can_fake_write (_video_frames_out)) {
 208                 _writer->fake_write (_video_frames_out);
 209                 _have_a_real_frame = false;
 210                 frame_done ();
 211         } else if (same && _have_a_real_frame) {
 212                 /* Use the last frame that we encoded. */
 213                 _writer->repeat (_video_frames_out);
 214                 frame_done ();
 215         } else {
 216                 /* Queue this new frame for encoding */
 217                 pair<string, string> const s = Filter::ffmpeg_strings (_film->filters());
 218                 TIMING ("adding to queue of %1", _queue.size ());
 219                 _queue.push_back (shared_ptr<DCPVideoFrame> (
 220                                           new DCPVideoFrame (
 221                                                   image, sub, _film->format()->dcp_size(), _film->format()->dcp_padding (_film),
 222                                                   _film->subtitle_offset(), _film->subtitle_scale(),
 223                                                   _film->scaler(), _video_frames_out, _film->dcp_frame_rate(), s.second,
 224                                                   _film->colour_lut(), _film->j2k_bandwidth(),
 225                                                   _film->log()
 226                                                   )
 227                                           ));
 228
 229                 _condition.notify_all ();
 230                 _have_a_real_frame = true;
 231         }
 232
 233         ++_video_frames_in;
 234         ++_video_frames_out;
 235
 236         if (frc.repeat) {
 237                 _writer->repeat (_video_frames_out);
 238                 ++_video_frames_out;
 239                 frame_done ();
 240         }
 241 }
 242
 243 void
 244 Encoder::process_audio (shared_ptr<const AudioBuffers> data, Time)
 245 {
 246         _writer->write (data);
 247 }
 248
 249 void
 250 Encoder::terminate_threads ()
 251 {
 252         boost::mutex::scoped_lock lock (_mutex);
 253         _terminate = true;
 254         _condition.notify_all ();
 255         lock.unlock ();
 256
 257         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 258                 if ((*i)->joinable ()) {
 259                         (*i)->join ();
 260                 }
 261                 delete *i;
 262         }
 263 }
 264
 265 void
 266 Encoder::encoder_thread (ServerDescription* server)
 267 {
 268         /* Number of seconds that we currently wait between attempts
 269            to connect to the server; not relevant for localhost
 270            encodings.
 271         */
 272         int remote_backoff = 0;
 273
 274         while (1) {
 275
 276                 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
 277                 boost::mutex::scoped_lock lock (_mutex);
 278                 while (_queue.empty () && !_terminate) {
 279                         _condition.wait (lock);
 280                 }
 281
 282                 if (_terminate) {
 283                         return;
 284                 }
 285
 286                 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 287                 shared_ptr<DCPVideoFrame> vf = _queue.front ();
 288                 _film->log()->log (String::compose (N_("Encoder thread %1 pops frame %2 from queue"), boost::this_thread::get_id(), vf->frame()), Log::VERBOSE);
 289                 _queue.pop_front ();
 290
 291                 lock.unlock ();
 292
 293                 shared_ptr<EncodedData> encoded;
 294
 295                 if (server) {
 296                         try {
 297                                 encoded = vf->encode_remotely (server);
 298
 299                                 if (remote_backoff > 0) {
 300                                         _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
 301                                 }
 302
 303                                 /* This job succeeded, so remove any backoff */
 304                                 remote_backoff = 0;
 305
 306                         } catch (std::exception& e) {
 307                                 if (remote_backoff < 60) {
 308                                         /* back off more */
 309                                         remote_backoff += 10;
 310                                 }
 311                                 _film->log()->log (
 312                                         String::compose (
 313                                                 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 314                                                 vf->frame(), server->host_name(), e.what(), remote_backoff)
 315                                         );
 316                         }
 317
 318                 } else {
 319                         try {
 320                                 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
 321                                 encoded = vf->encode_locally ();
 322                                 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
 323                         } catch (std::exception& e) {
 324                                 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 325                         }
 326                 }
 327
 328                 if (encoded) {
 329                         _writer->write (encoded, vf->frame ());
 330                         frame_done ();
 331                 } else {
 332                         lock.lock ();
 333                         _film->log()->log (
 334                                 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
 335                                 );
 336                         _queue.push_front (vf);
 337                         lock.unlock ();
 338                 }
 339
 340                 if (remote_backoff > 0) {
 341                         dcpomatic_sleep (remote_backoff);
 342                 }
 343
 344                 lock.lock ();
 345                 _condition.notify_all ();
 346         }
 347 }