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