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