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 #include "container.h"
  43
  44 #include "i18n.h"
  45
  46 using std::pair;
  47 using std::string;
  48 using std::stringstream;
  49 using std::vector;
  50 using std::list;
  51 using std::cout;
  52 using std::make_pair;
  53 using boost::shared_ptr;
  54 using boost::optional;
  55
  56 int const Encoder::_history_size = 25;
  57
  58 /** @param f Film that we are encoding */
  59 Encoder::Encoder (shared_ptr<Film> f, shared_ptr<Job> j)
  60         : _film (f)
  61         , _job (j)
  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         boost::mutex::scoped_lock lock (_mutex);
 184
 185         /* Wait until the queue has gone down a bit */
 186         while (_queue.size() >= _threads.size() * 2 && !_terminate) {
 187                 TIMING ("decoder sleeps with queue of %1", _queue.size());
 188                 _condition.wait (lock);
 189                 TIMING ("decoder wakes with queue of %1", _queue.size());
 190         }
 191
 192         if (_terminate) {
 193                 return;
 194         }
 195
 196         if (_writer->thrown ()) {
 197                 _writer->rethrow ();
 198         }
 199
 200         if (_writer->can_fake_write (_video_frames_out)) {
 201                 _writer->fake_write (_video_frames_out);
 202                 _have_a_real_frame = false;
 203                 frame_done ();
 204         } else if (same && _have_a_real_frame) {
 205                 /* Use the last frame that we encoded. */
 206                 _writer->repeat (_video_frames_out);
 207                 frame_done ();
 208         } else {
 209                 /* Queue this new frame for encoding */
 210                 pair<string, string> const s = Filter::ffmpeg_strings (_film->filters());
 211                 TIMING ("adding to queue of %1", _queue.size ());
 212                 /* XXX: padding */
 213                 _queue.push_back (shared_ptr<DCPVideoFrame> (
 214                                           new DCPVideoFrame (
 215                                                   image, sub, _film->container()->dcp_size(), 0,
 216                                                   _film->subtitle_offset(), _film->subtitle_scale(),
 217                                                   _film->scaler(), _video_frames_out, _film->dcp_video_frame_rate(), s.second,
 218                                                   _film->colour_lut(), _film->j2k_bandwidth(),
 219                                                   _film->log()
 220                                                   )
 221                                           ));
 222
 223                 _condition.notify_all ();
 224                 _have_a_real_frame = true;
 225         }
 226
 227         ++_video_frames_out;
 228 }
 229
 230 void
 231 Encoder::process_audio (shared_ptr<const AudioBuffers> data, Time)
 232 {
 233         _writer->write (data);
 234 }
 235
 236 void
 237 Encoder::terminate_threads ()
 238 {
 239         boost::mutex::scoped_lock lock (_mutex);
 240         _terminate = true;
 241         _condition.notify_all ();
 242         lock.unlock ();
 243
 244         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 245                 if ((*i)->joinable ()) {
 246                         (*i)->join ();
 247                 }
 248                 delete *i;
 249         }
 250 }
 251
 252 void
 253 Encoder::encoder_thread (ServerDescription* server)
 254 {
 255         /* Number of seconds that we currently wait between attempts
 256            to connect to the server; not relevant for localhost
 257            encodings.
 258         */
 259         int remote_backoff = 0;
 260
 261         while (1) {
 262
 263                 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
 264                 boost::mutex::scoped_lock lock (_mutex);
 265                 while (_queue.empty () && !_terminate) {
 266                         _condition.wait (lock);
 267                 }
 268
 269                 if (_terminate) {
 270                         return;
 271                 }
 272
 273                 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 274                 shared_ptr<DCPVideoFrame> vf = _queue.front ();
 275                 _film->log()->log (String::compose (N_("Encoder thread %1 pops frame %2 from queue"), boost::this_thread::get_id(), vf->frame()), Log::VERBOSE);
 276                 _queue.pop_front ();
 277
 278                 lock.unlock ();
 279
 280                 shared_ptr<EncodedData> encoded;
 281
 282                 if (server) {
 283                         try {
 284                                 encoded = vf->encode_remotely (server);
 285
 286                                 if (remote_backoff > 0) {
 287                                         _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
 288                                 }
 289
 290                                 /* This job succeeded, so remove any backoff */
 291                                 remote_backoff = 0;
 292
 293                         } catch (std::exception& e) {
 294                                 if (remote_backoff < 60) {
 295                                         /* back off more */
 296                                         remote_backoff += 10;
 297                                 }
 298                                 _film->log()->log (
 299                                         String::compose (
 300                                                 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 301                                                 vf->frame(), server->host_name(), e.what(), remote_backoff)
 302                                         );
 303                         }
 304
 305                 } else {
 306                         try {
 307                                 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
 308                                 encoded = vf->encode_locally ();
 309                                 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
 310                         } catch (std::exception& e) {
 311                                 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 312                         }
 313                 }
 314
 315                 if (encoded) {
 316                         _writer->write (encoded, vf->frame ());
 317                         frame_done ();
 318                 } else {
 319                         lock.lock ();
 320                         _film->log()->log (
 321                                 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
 322                                 );
 323                         _queue.push_front (vf);
 324                         lock.unlock ();
 325                 }
 326
 327                 if (remote_backoff > 0) {
 328                         dcpomatic_sleep (remote_backoff);
 329                 }
 330
 331                 lock.lock ();
 332                 _condition.notify_all ();
 333         }
 334 }