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                 TIMING ("adding to queue of %1", _queue.size ());
 211                 /* XXX: padding */
 212                 _queue.push_back (shared_ptr<DCPVideoFrame> (
 213                                           new DCPVideoFrame (
 214                                                   image, sub, _film->container()->dcp_size(), 0,
 215                                                   _film->subtitle_offset(), _film->subtitle_scale(),
 216                                                   _film->scaler(), _video_frames_out, _film->dcp_video_frame_rate(),
 217                                                   _film->colour_lut(), _film->j2k_bandwidth(),
 218                                                   _film->log()
 219                                                   )
 220                                           ));
 221
 222                 _condition.notify_all ();
 223                 _have_a_real_frame = true;
 224         }
 225
 226         ++_video_frames_out;
 227 }
 228
 229 void
 230 Encoder::process_audio (shared_ptr<const AudioBuffers> data, Time)
 231 {
 232         _writer->write (data);
 233 }
 234
 235 void
 236 Encoder::terminate_threads ()
 237 {
 238         boost::mutex::scoped_lock lock (_mutex);
 239         _terminate = true;
 240         _condition.notify_all ();
 241         lock.unlock ();
 242
 243         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 244                 if ((*i)->joinable ()) {
 245                         (*i)->join ();
 246                 }
 247                 delete *i;
 248         }
 249 }
 250
 251 void
 252 Encoder::encoder_thread (ServerDescription* server)
 253 {
 254         /* Number of seconds that we currently wait between attempts
 255            to connect to the server; not relevant for localhost
 256            encodings.
 257         */
 258         int remote_backoff = 0;
 259
 260         while (1) {
 261
 262                 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
 263                 boost::mutex::scoped_lock lock (_mutex);
 264                 while (_queue.empty () && !_terminate) {
 265                         _condition.wait (lock);
 266                 }
 267
 268                 if (_terminate) {
 269                         return;
 270                 }
 271
 272                 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 273                 shared_ptr<DCPVideoFrame> vf = _queue.front ();
 274                 _film->log()->log (String::compose (N_("Encoder thread %1 pops frame %2 from queue"), boost::this_thread::get_id(), vf->frame()), Log::VERBOSE);
 275                 _queue.pop_front ();
 276
 277                 lock.unlock ();
 278
 279                 shared_ptr<EncodedData> encoded;
 280
 281                 if (server) {
 282                         try {
 283                                 encoded = vf->encode_remotely (server);
 284
 285                                 if (remote_backoff > 0) {
 286                                         _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
 287                                 }
 288
 289                                 /* This job succeeded, so remove any backoff */
 290                                 remote_backoff = 0;
 291
 292                         } catch (std::exception& e) {
 293                                 if (remote_backoff < 60) {
 294                                         /* back off more */
 295                                         remote_backoff += 10;
 296                                 }
 297                                 _film->log()->log (
 298                                         String::compose (
 299                                                 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 300                                                 vf->frame(), server->host_name(), e.what(), remote_backoff)
 301                                         );
 302                         }
 303
 304                 } else {
 305                         try {
 306                                 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
 307                                 encoded = vf->encode_locally ();
 308                                 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
 309                         } catch (std::exception& e) {
 310                                 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 311                         }
 312                 }
 313
 314                 if (encoded) {
 315                         _writer->write (encoded, vf->frame ());
 316                         frame_done ();
 317                 } else {
 318                         lock.lock ();
 319                         _film->log()->log (
 320                                 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
 321                                 );
 322                         _queue.push_front (vf);
 323                         lock.unlock ();
 324                 }
 325
 326                 if (remote_backoff > 0) {
 327                         dcpomatic_sleep (remote_backoff);
 328                 }
 329
 330                 lock.lock ();
 331                 _condition.notify_all ();
 332         }
 333 }