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