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