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)
  63         : _film (f)
  64         , _job (j)
  65         , _video_frames_out (0)
  66         , _terminate (false)
  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.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::process_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         _writer.reset (new Writer (_film, _job));
  97         ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
  98 }
  99
 100 void
 101 Encoder::process_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         _writer->finish ();
 139         _writer.reset ();
 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 void
 184 Encoder::process_video (shared_ptr<PlayerVideo> pvf)
 185 {
 186         _waker.nudge ();
 187
 188         boost::mutex::scoped_lock lock (_mutex);
 189
 190         /* XXX: discard 3D here if required */
 191
 192         /* Wait until the queue has gone down a bit */
 193         while (_queue.size() >= _threads.size() * 2 && !_terminate) {
 194                 LOG_TIMING ("decoder sleeps with queue of %1", _queue.size());
 195                 _full_condition.wait (lock);
 196                 LOG_TIMING ("decoder wakes with queue of %1", _queue.size());
 197         }
 198
 199         if (_terminate) {
 200                 return;
 201         }
 202
 203         _writer->rethrow ();
 204         /* Re-throw any exception raised by one of our threads.  If more
 205            than one has thrown an exception, only one will be rethrown, I think;
 206            but then, if that happens something has gone badly wrong.
 207         */
 208         rethrow ();
 209
 210         if (_writer->can_fake_write (_video_frames_out)) {
 211                 _writer->fake_write (_video_frames_out, pvf->eyes ());
 212                 frame_done ();
 213         } else {
 214                 /* Queue this new frame for encoding */
 215                 LOG_TIMING ("adding to queue of %1", _queue.size ());
 216                 _queue.push_back (shared_ptr<DCPVideo> (
 217                                           new DCPVideo (
 218                                                   pvf,
 219                                                   _video_frames_out,
 220                                                   _film->video_frame_rate(),
 221                                                   _film->j2k_bandwidth(),
 222                                                   _film->resolution(),
 223                                                   _film->burn_subtitles(),
 224                                                   _film->log()
 225                                                   )
 226                                           ));
 227
 228                 /* The queue might not be empty any more, so notify anything which is
 229                    waiting on that.
 230                 */
 231                 _empty_condition.notify_all ();
 232         }
 233
 234         if (pvf->eyes() != EYES_LEFT) {
 235                 ++_video_frames_out;
 236         }
 237 }
 238
 239 void
 240 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
 241 {
 242         _writer->write (data);
 243 }
 244
 245 void
 246 Encoder::terminate_threads ()
 247 {
 248         {
 249                 boost::mutex::scoped_lock lock (_mutex);
 250                 _terminate = true;
 251                 _full_condition.notify_all ();
 252                 _empty_condition.notify_all ();
 253         }
 254
 255         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 256                 if ((*i)->joinable ()) {
 257                         (*i)->join ();
 258                 }
 259                 delete *i;
 260         }
 261
 262         _threads.clear ();
 263 }
 264
 265 void
 266 Encoder::encoder_thread (optional<ServerDescription> server)
 267 try
 268 {
 269         /* Number of seconds that we currently wait between attempts
 270            to connect to the server; not relevant for localhost
 271            encodings.
 272         */
 273         int remote_backoff = 0;
 274
 275         while (true) {
 276
 277                 LOG_TIMING ("[%1] encoder thread sleeps", boost::this_thread::get_id());
 278                 boost::mutex::scoped_lock lock (_mutex);
 279                 while (_queue.empty () && !_terminate) {
 280                         _empty_condition.wait (lock);
 281                 }
 282
 283                 if (_terminate) {
 284                         return;
 285                 }
 286
 287                 LOG_TIMING ("[%1] encoder thread wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 288                 shared_ptr<DCPVideo> vf = _queue.front ();
 289                 LOG_TIMING ("[%1] encoder thread pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->index(), vf->eyes ());
 290                 _queue.pop_front ();
 291
 292                 lock.unlock ();
 293
 294                 shared_ptr<EncodedData> encoded;
 295
 296                 if (server) {
 297                         try {
 298                                 encoded = vf->encode_remotely (server.get ());
 299
 300                                 if (remote_backoff > 0) {
 301                                         LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
 302                                 }
 303
 304                                 /* This job succeeded, so remove any backoff */
 305                                 remote_backoff = 0;
 306
 307                         } catch (std::exception& e) {
 308                                 if (remote_backoff < 60) {
 309                                         /* back off more */
 310                                         remote_backoff += 10;
 311                                 }
 312                                 LOG_ERROR (
 313                                         N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 314                                         vf->index(), server->host_name(), e.what(), remote_backoff
 315                                         );
 316                         }
 317
 318                 } else {
 319                         try {
 320                                 LOG_TIMING ("[%1] encoder thread begins local encode of %2", boost::this_thread::get_id(), vf->index());
 321                                 encoded = vf->encode_locally ();
 322                                 LOG_TIMING ("[%1] encoder thread finishes local encode of %2", boost::this_thread::get_id(), vf->index());
 323                         } catch (std::exception& e) {
 324                                 LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
 325                         }
 326                 }
 327
 328                 if (encoded) {
 329                         _writer->write (encoded, vf->index (), vf->eyes ());
 330                         frame_done ();
 331                 } else {
 332                         lock.lock ();
 333                         LOG_GENERAL (N_("[%1] Encoder thread pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->index());
 334                         _queue.push_front (vf);
 335                         lock.unlock ();
 336                 }
 337
 338                 if (remote_backoff > 0) {
 339                         dcpomatic_sleep (remote_backoff);
 340                 }
 341
 342                 /* The queue might not be full any more, so notify anything that is waiting on that */
 343                 lock.lock ();
 344                 _full_condition.notify_all ();
 345         }
 346 }
 347 catch (...)
 348 {
 349         store_current ();
 350 }
 351
 352 void
 353 Encoder::server_found (ServerDescription s)
 354 {
 355         add_worker_threads (s);
 356 }