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.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.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, shared_ptr<Writer> writer)
  63         : _film (f)
  64         , _job (j)
  65         , _video_frames_out (0)
  66         , _terminate (false)
  67         , _writer (writer)
  68 {
  69
  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::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         ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
  98 }
  99
 100 void
 101 Encoder::end ()
 102 {
 103         boost::mutex::scoped_lock lock (_mutex);
 104
 105         LOG_GENERAL (N_("Clearing queue of %1"), _queue.size ());
 106
 107         /* Keep waking workers until the queue is empty */
 108         while (!_queue.empty ()) {
 109                 _empty_condition.notify_all ();
 110                 _full_condition.wait (lock);
 111         }
 112
 113         lock.unlock ();
 114
 115         terminate_threads ();
 116
 117         LOG_GENERAL (N_("Mopping up %1"), _queue.size());
 118
 119         /* The following sequence of events can occur in the above code:
 120              1. a remote worker takes the last image off the queue
 121              2. the loop above terminates
 122              3. the remote worker fails to encode the image and puts it back on the queue
 123              4. the remote worker is then terminated by terminate_threads
 124
 125              So just mop up anything left in the queue here.
 126         */
 127
 128         for (list<shared_ptr<DCPVideo> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
 129                 LOG_GENERAL (N_("Encode left-over frame %1"), (*i)->index ());
 130                 try {
 131                         _writer->write ((*i)->encode_locally(), (*i)->index (), (*i)->eyes ());
 132                         frame_done ();
 133                 } catch (std::exception& e) {
 134                         LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
 135                 }
 136         }
 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::enqueue (shared_ptr<PlayerVideo> 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                 LOG_TIMING ("decoder sleeps with queue of %1", _queue.size());
 192                 _full_condition.wait (lock);
 193                 LOG_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 if (pvf->has_j2k ()) {
 211                 _writer->write (pvf->j2k(), _video_frames_out, pvf->eyes ());
 212         } else {
 213                 /* Queue this new frame for encoding */
 214                 LOG_TIMING ("adding to queue of %1", _queue.size ());
 215                 _queue.push_back (shared_ptr<DCPVideo> (
 216                                           new DCPVideo (
 217                                                   pvf,
 218                                                   _video_frames_out,
 219                                                   _film->video_frame_rate(),
 220                                                   _film->j2k_bandwidth(),
 221                                                   _film->resolution(),
 222                                                   _film->burn_subtitles(),
 223                                                   _film->log()
 224                                                   )
 225                                           ));
 226
 227                 /* The queue might not be empty any more, so notify anything which is
 228                    waiting on that.
 229                 */
 230                 _empty_condition.notify_all ();
 231         }
 232
 233         if (pvf->eyes() != EYES_LEFT) {
 234                 ++_video_frames_out;
 235         }
 236 }
 237
 238 void
 239 Encoder::terminate_threads ()
 240 {
 241         {
 242                 boost::mutex::scoped_lock lock (_mutex);
 243                 _terminate = true;
 244                 _full_condition.notify_all ();
 245                 _empty_condition.notify_all ();
 246         }
 247
 248         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 249                 if ((*i)->joinable ()) {
 250                         (*i)->join ();
 251                 }
 252                 delete *i;
 253         }
 254
 255         _threads.clear ();
 256 }
 257
 258 void
 259 Encoder::encoder_thread (optional<ServerDescription> server)
 260 try
 261 {
 262         /* Number of seconds that we currently wait between attempts
 263            to connect to the server; not relevant for localhost
 264            encodings.
 265         */
 266         int remote_backoff = 0;
 267
 268         while (true) {
 269
 270                 LOG_TIMING ("[%1] encoder thread sleeps", boost::this_thread::get_id());
 271                 boost::mutex::scoped_lock lock (_mutex);
 272                 while (_queue.empty () && !_terminate) {
 273                         _empty_condition.wait (lock);
 274                 }
 275
 276                 if (_terminate) {
 277                         return;
 278                 }
 279
 280                 LOG_TIMING ("[%1] encoder thread wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 281                 shared_ptr<DCPVideo> vf = _queue.front ();
 282                 LOG_TIMING ("[%1] encoder thread pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->index(), vf->eyes ());
 283                 _queue.pop_front ();
 284
 285                 lock.unlock ();
 286
 287                 shared_ptr<EncodedData> encoded;
 288
 289                 if (server) {
 290                         try {
 291                                 encoded = vf->encode_remotely (server.get ());
 292
 293                                 if (remote_backoff > 0) {
 294                                         LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
 295                                 }
 296
 297                                 /* This job succeeded, so remove any backoff */
 298                                 remote_backoff = 0;
 299
 300                         } catch (std::exception& e) {
 301                                 if (remote_backoff < 60) {
 302                                         /* back off more */
 303                                         remote_backoff += 10;
 304                                 }
 305                                 LOG_ERROR (
 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                                 LOG_TIMING ("[%1] encoder thread begins local encode of %2", boost::this_thread::get_id(), vf->index());
 314                                 encoded = vf->encode_locally ();
 315                                 LOG_TIMING ("[%1] encoder thread finishes local encode of %2", boost::this_thread::get_id(), vf->index());
 316                         } catch (std::exception& e) {
 317                                 LOG_ERROR (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                         LOG_GENERAL (N_("[%1] Encoder thread pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->index());
 327                         _queue.push_front (vf);
 328                         lock.unlock ();
 329                 }
 330
 331                 if (remote_backoff > 0) {
 332                         dcpomatic_sleep (remote_backoff);
 333                 }
 334
 335                 /* The queue might not be full any more, so notify anything that is waiting on that */
 336                 lock.lock ();
 337                 _full_condition.notify_all ();
 338         }
 339 }
 340 catch (...)
 341 {
 342         store_current ();
 343 }
 344
 345 void
 346 Encoder::server_found (ServerDescription s)
 347 {
 348         add_worker_threads (s);
 349 }