src/lib/encoder.cc

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