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