src/lib/encoder.cc

   1 /*
   2     Copyright (C) 2012 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 "dcp_video.h"
  39
  40 #include "i18n.h"
  41
  42 using std::pair;
  43 using std::string;
  44 using std::stringstream;
  45 using std::vector;
  46 using std::list;
  47 using std::cout;
  48 using std::min;
  49 using std::make_pair;
  50 using boost::shared_ptr;
  51 using boost::weak_ptr;
  52 using boost::optional;
  53 using boost::scoped_array;
  54
  55 int const Encoder::_history_size = 25;
  56
  57 /** @param f Film that we are encoding */
  58 Encoder::Encoder (shared_ptr<const Film> f, weak_ptr<Job> j)
  59         : _film (f)
  60         , _job (j)
  61         , _video_frames_out (0)
  62         , _terminate (false)
  63 {
  64
  65 }
  66
  67 Encoder::~Encoder ()
  68 {
  69         terminate_threads ();
  70 }
  71
  72 /** Add a worker thread for a each thread on a remote server.  Caller must hold
  73  *  a lock on _mutex, or know that one is not currently required to
  74  *  safely modify _threads.
  75  */
  76 void
  77 Encoder::add_worker_threads (ServerDescription d)
  78 {
  79         for (int i = 0; i < d.threads(); ++i) {
  80                 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
  81         }
  82 }
  83
  84 void
  85 Encoder::process_begin ()
  86 {
  87         for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
  88                 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
  89         }
  90
  91         _writer.reset (new Writer (_film, _job));
  92         ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
  93 }
  94
  95 void
  96 Encoder::process_end ()
  97 {
  98         boost::mutex::scoped_lock lock (_mutex);
  99
 100         _film->log()->log (String::compose (N_("Clearing queue of %1"), _queue.size ()));
 101
 102         /* Keep waking workers until the queue is empty */
 103         while (!_queue.empty ()) {
 104                 _film->log()->log (String::compose (N_("Waking with %1"), _queue.size ()), Log::VERBOSE);
 105                 _condition.notify_all ();
 106                 _condition.wait (lock);
 107         }
 108
 109         lock.unlock ();
 110
 111         terminate_threads ();
 112
 113         _film->log()->log (String::compose (N_("Mopping up %1"), _queue.size()));
 114
 115         /* The following sequence of events can occur in the above code:
 116              1. a remote worker takes the last image off the queue
 117              2. the loop above terminates
 118              3. the remote worker fails to encode the image and puts it back on the queue
 119              4. the remote worker is then terminated by terminate_threads
 120
 121              So just mop up anything left in the queue here.
 122         */
 123
 124         for (list<shared_ptr<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
 125                 _film->log()->log (String::compose (N_("Encode left-over frame %1"), (*i)->frame ()));
 126                 try {
 127                         _writer->write ((*i)->encode_locally(), (*i)->frame (), (*i)->eyes ());
 128                         frame_done ();
 129                 } catch (std::exception& e) {
 130                         _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 131                 }
 132         }
 133
 134         _writer->finish ();
 135         _writer.reset ();
 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 void
 180 Encoder::process_video (shared_ptr<DCPVideo> frame)
 181 {
 182         _waker.nudge ();
 183
 184         boost::mutex::scoped_lock lock (_mutex);
 185
 186         /* XXX: discard 3D here if required */
 187
 188         /* Wait until the queue has gone down a bit */
 189         while (_queue.size() >= _threads.size() * 2 && !_terminate) {
 190                 TIMING ("decoder sleeps with queue of %1", _queue.size());
 191                 _condition.wait (lock);
 192                 TIMING ("decoder wakes with queue of %1", _queue.size());
 193         }
 194
 195         if (_terminate) {
 196                 return;
 197         }
 198
 199         _writer->rethrow ();
 200         /* Re-throw any exception raised by one of our threads.  If more
 201            than one has thrown an exception, only one will be rethrown, I think;
 202            but then, if that happens something has gone badly wrong.
 203         */
 204         rethrow ();
 205
 206         if (_writer->can_fake_write (_video_frames_out)) {
 207                 _writer->fake_write (_video_frames_out, frame->eyes ());
 208                 frame_done ();
 209         } else {
 210                 /* Queue this new frame for encoding */
 211                 TIMING ("adding to queue of %1", _queue.size ());
 212                 _queue.push_back (shared_ptr<DCPVideoFrame> (
 213                                           new DCPVideoFrame (
 214                                                   frame->image(PIX_FMT_RGB24, false),
 215                                                   _video_frames_out,
 216                                                   frame->eyes(),
 217                                                   frame->conversion(),
 218                                                   _film->video_frame_rate(),
 219                                                   _film->j2k_bandwidth(),
 220                                                   _film->resolution(),
 221                                                   _film->log()
 222                                                   )
 223                                           ));
 224
 225                 _condition.notify_all ();
 226         }
 227
 228         if (frame->eyes() != EYES_LEFT) {
 229                 ++_video_frames_out;
 230         }
 231 }
 232
 233 void
 234 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
 235 {
 236         _writer->write (data);
 237 }
 238
 239 void
 240 Encoder::terminate_threads ()
 241 {
 242         {
 243                 boost::mutex::scoped_lock lock (_mutex);
 244                 _terminate = true;
 245                 _condition.notify_all ();
 246         }
 247
 248         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 249                 if ((*i)->joinable ()) {
 250                         (*i)->join ();
 251                 }
 252                 delete *i;
 253         }
 254
 255         _threads.clear ();
 256 }
 257
 258 void
 259 Encoder::encoder_thread (optional<ServerDescription> server)
 260 try
 261 {
 262         /* Number of seconds that we currently wait between attempts
 263            to connect to the server; not relevant for localhost
 264            encodings.
 265         */
 266         int remote_backoff = 0;
 267
 268         while (1) {
 269
 270                 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
 271                 boost::mutex::scoped_lock lock (_mutex);
 272                 while (_queue.empty () && !_terminate) {
 273                         _condition.wait (lock);
 274                 }
 275
 276                 if (_terminate) {
 277                         return;
 278                 }
 279
 280                 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 281                 shared_ptr<DCPVideoFrame> vf = _queue.front ();
 282                 TIMING ("encoder thread %1 pops frame %2 (%3) from queue", boost::this_thread::get_id(), vf->frame(), vf->eyes ());
 283                 _queue.pop_front ();
 284
 285                 lock.unlock ();
 286
 287                 shared_ptr<EncodedData> encoded;
 288
 289                 if (server) {
 290                         try {
 291                                 encoded = vf->encode_remotely (server.get ());
 292
 293                                 if (remote_backoff > 0) {
 294                                         _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
 295                                 }
 296
 297                                 /* This job succeeded, so remove any backoff */
 298                                 remote_backoff = 0;
 299
 300                         } catch (std::exception& e) {
 301                                 if (remote_backoff < 60) {
 302                                         /* back off more */
 303                                         remote_backoff += 10;
 304                                 }
 305                                 _film->log()->log (
 306                                         String::compose (
 307                                                 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 308                                                 vf->frame(), server->host_name(), e.what(), remote_backoff)
 309                                         );
 310                         }
 311
 312                 } else {
 313                         try {
 314                                 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
 315                                 encoded = vf->encode_locally ();
 316                                 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
 317                         } catch (std::exception& e) {
 318                                 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 319                         }
 320                 }
 321
 322                 if (encoded) {
 323                         _writer->write (encoded, vf->frame (), vf->eyes ());
 324                         frame_done ();
 325                 } else {
 326                         lock.lock ();
 327                         _film->log()->log (
 328                                 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
 329                                 );
 330                         _queue.push_front (vf);
 331                         lock.unlock ();
 332                 }
 333
 334                 if (remote_backoff > 0) {
 335                         dcpomatic_sleep (remote_backoff);
 336                 }
 337
 338                 lock.lock ();
 339                 _condition.notify_all ();
 340         }
 341 }
 342 catch (...)
 343 {
 344         store_current ();
 345 }
 346
 347 void
 348 Encoder::server_found (ServerDescription s)
 349 {
 350         add_worker_threads (s);
 351 }