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/lambda/lambda.hpp>
  26 #include <libcxml/cxml.h>
  27 #include "encoder.h"
  28 #include "util.h"
  29 #include "film.h"
  30 #include "log.h"
  31 #include "config.h"
  32 #include "dcp_video_frame.h"
  33 #include "server.h"
  34 #include "cross.h"
  35 #include "writer.h"
  36 #include "server_finder.h"
  37 #include "player.h"
  38 #include "dcp_video.h"
  39
  40 #include "i18n.h"
  41
  42 using std::pair;
  43 using std::string;
  44 using std::stringstream;
  45 using std::vector;
  46 using std::list;
  47 using std::cout;
  48 using std::min;
  49 using std::make_pair;
  50 using boost::shared_ptr;
  51 using boost::weak_ptr;
  52 using boost::optional;
  53 using boost::scoped_array;
  54
  55 int const Encoder::_history_size = 25;
  56
  57 /** @param f Film that we are encoding */
  58 Encoder::Encoder (shared_ptr<const Film> f, weak_ptr<Job> j)
  59         : _film (f)
  60         , _job (j)
  61         , _video_frames_out (0)
  62         , _terminate (false)
  63 {
  64
  65 }
  66
  67 Encoder::~Encoder ()
  68 {
  69         terminate_threads ();
  70 }
  71
  72 /** Add a worker thread for a each thread on a remote server.  Caller must hold
  73  *  a lock on _mutex, or know that one is not currently required to
  74  *  safely modify _threads.
  75  */
  76 void
  77 Encoder::add_worker_threads (ServerDescription d)
  78 {
  79         _film->log()->log (String::compose (N_("Adding %1 worker threads for remote %2"), d.host_name ()));
  80         for (int i = 0; i < d.threads(); ++i) {
  81                 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
  82         }
  83 }
  84
  85 void
  86 Encoder::process_begin ()
  87 {
  88         for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
  89                 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
  90         }
  91
  92         _writer.reset (new Writer (_film, _job));
  93         ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
  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 (), (*i)->eyes ());
 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 (_state_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 (_state_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 (_state_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<DCPVideo> frame)
 182 {
 183         _waker.nudge ();
 184
 185         boost::mutex::scoped_lock lock (_mutex);
 186
 187         /* XXX: discard 3D here if required */
 188
 189         /* Wait until the queue has gone down a bit */
 190         while (_queue.size() >= _threads.size() * 2 && !_terminate) {
 191                 TIMING ("decoder sleeps with queue of %1", _queue.size());
 192                 _condition.wait (lock);
 193                 TIMING ("decoder wakes with queue of %1", _queue.size());
 194         }
 195
 196         if (_terminate) {
 197                 return;
 198         }
 199
 200         _writer->rethrow ();
 201         /* Re-throw any exception raised by one of our threads.  If more
 202            than one has thrown an exception, only one will be rethrown, I think;
 203            but then, if that happens something has gone badly wrong.
 204         */
 205         rethrow ();
 206
 207         if (_writer->can_fake_write (_video_frames_out)) {
 208                 _writer->fake_write (_video_frames_out, frame->eyes ());
 209                 frame_done ();
 210         } else {
 211                 /* Queue this new frame for encoding */
 212                 TIMING ("adding to queue of %1", _queue.size ());
 213                 _queue.push_back (shared_ptr<DCPVideoFrame> (
 214                                           new DCPVideoFrame (
 215                                                   frame->image(PIX_FMT_RGB24, false),
 216                                                   _video_frames_out,
 217                                                   frame->eyes(),
 218                                                   frame->conversion(),
 219                                                   _film->video_frame_rate(),
 220                                                   _film->j2k_bandwidth(),
 221                                                   _film->resolution(),
 222                                                   _film->log()
 223                                                   )
 224                                           ));
 225
 226                 _condition.notify_all ();
 227         }
 228
 229         if (frame->eyes() != EYES_LEFT) {
 230                 ++_video_frames_out;
 231         }
 232 }
 233
 234 void
 235 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
 236 {
 237         _writer->write (data);
 238 }
 239
 240 void
 241 Encoder::terminate_threads ()
 242 {
 243         {
 244                 boost::mutex::scoped_lock lock (_mutex);
 245                 _terminate = true;
 246                 _condition.notify_all ();
 247         }
 248
 249         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 250                 if ((*i)->joinable ()) {
 251                         (*i)->join ();
 252                 }
 253                 delete *i;
 254         }
 255
 256         _threads.clear ();
 257 }
 258
 259 void
 260 Encoder::encoder_thread (optional<ServerDescription> server)
 261 try
 262 {
 263         /* Number of seconds that we currently wait between attempts
 264            to connect to the server; not relevant for localhost
 265            encodings.
 266         */
 267         int remote_backoff = 0;
 268
 269         while (1) {
 270
 271                 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
 272                 boost::mutex::scoped_lock lock (_mutex);
 273                 while (_queue.empty () && !_terminate) {
 274                         _condition.wait (lock);
 275                 }
 276
 277                 if (_terminate) {
 278                         return;
 279                 }
 280
 281                 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 282                 shared_ptr<DCPVideoFrame> vf = _queue.front ();
 283                 TIMING ("encoder thread %1 pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->frame(), vf->eyes ());
 284                 _queue.pop_front ();
 285
 286                 lock.unlock ();
 287
 288                 shared_ptr<EncodedData> encoded;
 289
 290                 if (server) {
 291                         try {
 292                                 encoded = vf->encode_remotely (server.get ());
 293
 294                                 if (remote_backoff > 0) {
 295                                         _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
 296                                 }
 297
 298                                 /* This job succeeded, so remove any backoff */
 299                                 remote_backoff = 0;
 300
 301                         } catch (std::exception& e) {
 302                                 if (remote_backoff < 60) {
 303                                         /* back off more */
 304                                         remote_backoff += 10;
 305                                 }
 306                                 _film->log()->log (
 307                                         String::compose (
 308                                                 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 309                                                 vf->frame(), server->host_name(), e.what(), remote_backoff)
 310                                         );
 311                         }
 312
 313                 } else {
 314                         try {
 315                                 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
 316                                 encoded = vf->encode_locally ();
 317                                 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
 318                         } catch (std::exception& e) {
 319                                 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 320                         }
 321                 }
 322
 323                 if (encoded) {
 324                         _writer->write (encoded, vf->frame (), vf->eyes ());
 325                         frame_done ();
 326                 } else {
 327                         lock.lock ();
 328                         _film->log()->log (
 329                                 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
 330                                 );
 331                         _queue.push_front (vf);
 332                         lock.unlock ();
 333                 }
 334
 335                 if (remote_backoff > 0) {
 336                         dcpomatic_sleep (remote_backoff);
 337                 }
 338
 339                 lock.lock ();
 340                 _condition.notify_all ();
 341         }
 342 }
 343 catch (...)
 344 {
 345         store_current ();
 346 }
 347
 348 void
 349 Encoder::server_found (ServerDescription s)
 350 {
 351         add_worker_threads (s);
 352 }