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 "encoder.h"
  25 #include "util.h"
  26 #include "film.h"
  27 #include "log.h"
  28 #include "config.h"
  29 #include "dcp_video.h"
  30 #include "server.h"
  31 #include "cross.h"
  32 #include "writer.h"
  33 #include "server_finder.h"
  34 #include "player.h"
  35 #include "player_video.h"
  36 #include <libcxml/cxml.h>
  37 #include <boost/lambda/lambda.hpp>
  38 #include <iostream>
  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 (
 131                                 (*i)->encode_locally (boost::bind (&Log::dcp_log, _film->log().get(), _1, _2)),
 132                                 (*i)->index (),
 133                                 (*i)->eyes ()
 134                                 );
 135                         frame_done ();
 136                 } catch (std::exception& e) {
 137                         LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
 138                 }
 139         }
 140 }
 141
 142 /** @return an estimate of the current number of frames we are encoding per second,
 143  *  or 0 if not known.
 144  */
 145 float
 146 Encoder::current_encoding_rate () const
 147 {
 148         boost::mutex::scoped_lock lock (_state_mutex);
 149         if (int (_time_history.size()) < _history_size) {
 150                 return 0;
 151         }
 152
 153         struct timeval now;
 154         gettimeofday (&now, 0);
 155
 156         return _history_size / (seconds (now) - seconds (_time_history.back ()));
 157 }
 158
 159 /** @return Number of video frames that have been sent out */
 160 int
 161 Encoder::video_frames_out () const
 162 {
 163         boost::mutex::scoped_lock (_state_mutex);
 164         return _video_frames_out;
 165 }
 166
 167 /** Should be called when a frame has been encoded successfully.
 168  *  @param n Source frame index.
 169  */
 170 void
 171 Encoder::frame_done ()
 172 {
 173         boost::mutex::scoped_lock lock (_state_mutex);
 174
 175         struct timeval tv;
 176         gettimeofday (&tv, 0);
 177         _time_history.push_front (tv);
 178         if (int (_time_history.size()) > _history_size) {
 179                 _time_history.pop_back ();
 180         }
 181 }
 182
 183 /** Called in order, so each time this is called the supplied frame is the one
 184  *  after the previous one.
 185  */
 186 void
 187 Encoder::enqueue (shared_ptr<PlayerVideo> pv)
 188 {
 189         _waker.nudge ();
 190
 191         boost::mutex::scoped_lock lock (_mutex);
 192
 193         /* XXX: discard 3D here if required */
 194
 195         /* Wait until the queue has gone down a bit */
 196         while (_queue.size() >= _threads.size() * 2 && !_terminate) {
 197                 LOG_TIMING ("decoder sleeps with queue of %1", _queue.size());
 198                 _full_condition.wait (lock);
 199                 LOG_TIMING ("decoder wakes with queue of %1", _queue.size());
 200         }
 201
 202         if (_terminate) {
 203                 return;
 204         }
 205
 206         _writer->rethrow ();
 207         /* Re-throw any exception raised by one of our threads.  If more
 208            than one has thrown an exception, only one will be rethrown, I think;
 209            but then, if that happens something has gone badly wrong.
 210         */
 211         rethrow ();
 212
 213         if (_writer->can_fake_write (_video_frames_out)) {
 214                 /* We can fake-write this frame */
 215                 _writer->fake_write (_video_frames_out, pv->eyes ());
 216                 frame_done ();
 217         } else if (pv->has_j2k ()) {
 218                 /* This frame already has JPEG2000 data, so just write it */
 219                 _writer->write (pv->j2k(), _video_frames_out, pv->eyes ());
 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<DCPVideo> (
 224                                           new DCPVideo (
 225                                                   pv,
 226                                                   _video_frames_out,
 227                                                   _film->video_frame_rate(),
 228                                                   _film->j2k_bandwidth(),
 229                                                   _film->resolution(),
 230                                                   _film->burn_subtitles(),
 231                                                   _film->log()
 232                                                   )
 233                                           ));
 234
 235                 /* The queue might not be empty any more, so notify anything which is
 236                    waiting on that.
 237                 */
 238                 _empty_condition.notify_all ();
 239         }
 240
 241         if (pv->eyes() != EYES_LEFT) {
 242                 ++_video_frames_out;
 243         }
 244 }
 245
 246 void
 247 Encoder::terminate_threads ()
 248 {
 249         {
 250                 boost::mutex::scoped_lock lock (_mutex);
 251                 _terminate = true;
 252                 _full_condition.notify_all ();
 253                 _empty_condition.notify_all ();
 254         }
 255
 256         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 257                 if ((*i)->joinable ()) {
 258                         (*i)->join ();
 259                 }
 260                 delete *i;
 261         }
 262
 263         _threads.clear ();
 264 }
 265
 266 void
 267 Encoder::encoder_thread (optional<ServerDescription> server)
 268 try
 269 {
 270         /* Number of seconds that we currently wait between attempts
 271            to connect to the server; not relevant for localhost
 272            encodings.
 273         */
 274         int remote_backoff = 0;
 275         shared_ptr<DCPVideo> last_dcp_video;
 276         shared_ptr<EncodedData> last_encoded;
 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<DCPVideo> 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 (last_dcp_video && vf->same (last_dcp_video)) {
 300                         /* We already have encoded data for the same input as this one, so take a short-cut */
 301                         encoded = last_encoded;
 302                 } else {
 303                         /* We need to encode this input */
 304                         if (server) {
 305                                 try {
 306                                         encoded = vf->encode_remotely (server.get ());
 307
 308                                         if (remote_backoff > 0) {
 309                                                 LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
 310                                         }
 311
 312                                         /* This job succeeded, so remove any backoff */
 313                                         remote_backoff = 0;
 314
 315                                 } catch (std::exception& e) {
 316                                         if (remote_backoff < 60) {
 317                                                 /* back off more */
 318                                                 remote_backoff += 10;
 319                                         }
 320                                         LOG_ERROR (
 321                                                 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 322                                                 vf->index(), server->host_name(), e.what(), remote_backoff
 323                                                 );
 324                                 }
 325
 326                         } else {
 327                                 try {
 328                                         LOG_TIMING ("[%1] encoder thread begins local encode of %2", boost::this_thread::get_id(), vf->index());
 329                                         encoded = vf->encode_locally (boost::bind (&Log::dcp_log, _film->log().get(), _1, _2));
 330                                         LOG_TIMING ("[%1] encoder thread finishes local encode of %2", boost::this_thread::get_id(), vf->index());
 331                                 } catch (std::exception& e) {
 332                                         LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
 333                                 }
 334                         }
 335                 }
 336
 337                 last_dcp_video = vf;
 338                 last_encoded = encoded;
 339
 340                 if (encoded) {
 341                         _writer->write (encoded, vf->index (), vf->eyes ());
 342                         frame_done ();
 343                 } else {
 344                         lock.lock ();
 345                         LOG_GENERAL (N_("[%1] Encoder thread pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->index());
 346                         _queue.push_front (vf);
 347                         lock.unlock ();
 348                 }
 349
 350                 if (remote_backoff > 0) {
 351                         dcpomatic_sleep (remote_backoff);
 352                 }
 353
 354                 /* The queue might not be full any more, so notify anything that is waiting on that */
 355                 lock.lock ();
 356                 _full_condition.notify_all ();
 357         }
 358 }
 359 catch (...)
 360 {
 361         store_current ();
 362 }
 363
 364 void
 365 Encoder::server_found (ServerDescription s)
 366 {
 367         add_worker_threads (s);
 368 }