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
  38 #include "i18n.h"
  39
  40 using std::pair;
  41 using std::string;
  42 using std::stringstream;
  43 using std::vector;
  44 using std::list;
  45 using std::cout;
  46 using std::min;
  47 using std::make_pair;
  48 using boost::shared_ptr;
  49 using boost::optional;
  50 using boost::scoped_array;
  51
  52 int const Encoder::_history_size = 25;
  53
  54 /** @param f Film that we are encoding */
  55 Encoder::Encoder (shared_ptr<const Film> f, shared_ptr<Job> j)
  56         : _film (f)
  57         , _job (j)
  58         , _video_frames_out (0)
  59         , _terminate (false)
  60 {
  61         _have_a_real_frame[EYES_BOTH] = false;
  62         _have_a_real_frame[EYES_LEFT] = false;
  63         _have_a_real_frame[EYES_RIGHT] = false;
  64 }
  65
  66 Encoder::~Encoder ()
  67 {
  68         terminate_threads ();
  69         if (_writer) {
  70                 _writer->finish ();
  71         }
  72 }
  73
  74 /** Add a worker thread for a each thread on a remote server.  Caller must hold
  75  *  a lock on _mutex, or know that one is not currently required to
  76  *  safely modify _threads.
  77  */
  78 void
  79 Encoder::add_worker_threads (ServerDescription d)
  80 {
  81         for (int i = 0; i < d.threads(); ++i) {
  82                 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)));
  83         }
  84 }
  85
  86 void
  87 Encoder::process_begin ()
  88 {
  89         for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
  90                 _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
  91         }
  92
  93         _writer.reset (new Writer (_film, _job));
  94         ServerFinder::instance()->connect (boost::bind (&Encoder::server_found, this, _1));
  95 }
  96
  97
  98 void
  99 Encoder::process_end ()
 100 {
 101         boost::mutex::scoped_lock lock (_mutex);
 102
 103         _film->log()->log (String::compose (N_("Clearing queue of %1"), _queue.size ()));
 104
 105         /* Keep waking workers until the queue is empty */
 106         while (!_queue.empty ()) {
 107                 _film->log()->log (String::compose (N_("Waking with %1"), _queue.size ()), Log::VERBOSE);
 108                 _condition.notify_all ();
 109                 _condition.wait (lock);
 110         }
 111
 112         lock.unlock ();
 113
 114         terminate_threads ();
 115
 116         _film->log()->log (String::compose (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<DCPVideoFrame> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
 128                 _film->log()->log (String::compose (N_("Encode left-over frame %1"), (*i)->frame ()));
 129                 try {
 130                         _writer->write ((*i)->encode_locally(), (*i)->frame (), (*i)->eyes ());
 131                         frame_done ();
 132                 } catch (std::exception& e) {
 133                         _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 134                 }
 135         }
 136
 137         _writer->finish ();
 138         _writer.reset ();
 139 }
 140
 141 /** @return an estimate of the current number of frames we are encoding per second,
 142  *  or 0 if not known.
 143  */
 144 float
 145 Encoder::current_encoding_rate () const
 146 {
 147         boost::mutex::scoped_lock lock (_state_mutex);
 148         if (int (_time_history.size()) < _history_size) {
 149                 return 0;
 150         }
 151
 152         struct timeval now;
 153         gettimeofday (&now, 0);
 154
 155         return _history_size / (seconds (now) - seconds (_time_history.back ()));
 156 }
 157
 158 /** @return Number of video frames that have been sent out */
 159 int
 160 Encoder::video_frames_out () const
 161 {
 162         boost::mutex::scoped_lock (_state_mutex);
 163         return _video_frames_out;
 164 }
 165
 166 /** Should be called when a frame has been encoded successfully.
 167  *  @param n Source frame index.
 168  */
 169 void
 170 Encoder::frame_done ()
 171 {
 172         boost::mutex::scoped_lock lock (_state_mutex);
 173
 174         struct timeval tv;
 175         gettimeofday (&tv, 0);
 176         _time_history.push_front (tv);
 177         if (int (_time_history.size()) > _history_size) {
 178                 _time_history.pop_back ();
 179         }
 180 }
 181
 182 void
 183 Encoder::process_video (shared_ptr<const Image> image, Eyes eyes, ColourConversion conversion, bool same)
 184 {
 185         boost::mutex::scoped_lock lock (_mutex);
 186
 187         /* XXX: discard 3D here if required */
 188
 189         /* Wait until the queue has gone down a bit */
 190         while (_queue.size() >= _threads.size() * 2 && !_terminate) {
 191                 TIMING ("decoder sleeps with queue of %1", _queue.size());
 192                 _condition.wait (lock);
 193                 TIMING ("decoder wakes with queue of %1", _queue.size());
 194         }
 195
 196         if (_terminate) {
 197                 return;
 198         }
 199
 200         if (_writer->thrown ()) {
 201                 _writer->rethrow ();
 202         }
 203
 204         if (_writer->can_fake_write (_video_frames_out)) {
 205                 _writer->fake_write (_video_frames_out, eyes);
 206                 _have_a_real_frame[eyes] = false;
 207                 frame_done ();
 208         } else if (same && _have_a_real_frame[eyes]) {
 209                 /* Use the last frame that we encoded. */
 210                 _writer->repeat (_video_frames_out, eyes);
 211                 frame_done ();
 212         } else {
 213                 /* Queue this new frame for encoding */
 214                 TIMING ("adding to queue of %1", _queue.size ());
 215                 _queue.push_back (shared_ptr<DCPVideoFrame> (
 216                                           new DCPVideoFrame (
 217                                                   image, _video_frames_out, eyes, conversion, _film->video_frame_rate(),
 218                                                   _film->j2k_bandwidth(), _film->log()
 219                                                   )
 220                                           ));
 221
 222                 _condition.notify_all ();
 223                 _have_a_real_frame[eyes] = true;
 224         }
 225
 226         if (eyes != EYES_LEFT) {
 227                 ++_video_frames_out;
 228         }
 229 }
 230
 231 void
 232 Encoder::process_audio (shared_ptr<const AudioBuffers> data)
 233 {
 234         _writer->write (data);
 235 }
 236
 237 void
 238 Encoder::terminate_threads ()
 239 {
 240         {
 241                 boost::mutex::scoped_lock lock (_mutex);
 242                 _terminate = true;
 243                 _condition.notify_all ();
 244         }
 245
 246         for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
 247                 if ((*i)->joinable ()) {
 248                         (*i)->join ();
 249                 }
 250                 delete *i;
 251         }
 252
 253         _threads.clear ();
 254 }
 255
 256 void
 257 Encoder::encoder_thread (optional<ServerDescription> server)
 258 {
 259         /* Number of seconds that we currently wait between attempts
 260            to connect to the server; not relevant for localhost
 261            encodings.
 262         */
 263         int remote_backoff = 0;
 264
 265         while (1) {
 266
 267                 TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id());
 268                 boost::mutex::scoped_lock lock (_mutex);
 269                 while (_queue.empty () && !_terminate) {
 270                         _condition.wait (lock);
 271                 }
 272
 273                 if (_terminate) {
 274                         return;
 275                 }
 276
 277                 TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size());
 278                 shared_ptr<DCPVideoFrame> vf = _queue.front ();
 279                 _film->log()->log (String::compose (N_("Encoder thread %1 pops frame %2 (%3) from queue"), boost::this_thread::get_id(), vf->frame(), vf->eyes ()));
 280                 _queue.pop_front ();
 281
 282                 lock.unlock ();
 283
 284                 shared_ptr<EncodedData> encoded;
 285
 286                 if (server) {
 287                         try {
 288                                 encoded = vf->encode_remotely (server.get ());
 289
 290                                 if (remote_backoff > 0) {
 291                                         _film->log()->log (String::compose (N_("%1 was lost, but now she is found; removing backoff"), server->host_name ()));
 292                                 }
 293
 294                                 /* This job succeeded, so remove any backoff */
 295                                 remote_backoff = 0;
 296
 297                         } catch (std::exception& e) {
 298                                 if (remote_backoff < 60) {
 299                                         /* back off more */
 300                                         remote_backoff += 10;
 301                                 }
 302                                 _film->log()->log (
 303                                         String::compose (
 304                                                 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 305                                                 vf->frame(), server->host_name(), e.what(), remote_backoff)
 306                                         );
 307                         }
 308
 309                 } else {
 310                         try {
 311                                 TIMING ("encoder thread %1 begins local encode of %2", boost::this_thread::get_id(), vf->frame());
 312                                 encoded = vf->encode_locally ();
 313                                 TIMING ("encoder thread %1 finishes local encode of %2", boost::this_thread::get_id(), vf->frame());
 314                         } catch (std::exception& e) {
 315                                 _film->log()->log (String::compose (N_("Local encode failed (%1)"), e.what ()));
 316                         }
 317                 }
 318
 319                 if (encoded) {
 320                         _writer->write (encoded, vf->frame (), vf->eyes ());
 321                         frame_done ();
 322                 } else {
 323                         lock.lock ();
 324                         _film->log()->log (
 325                                 String::compose (N_("Encoder thread %1 pushes frame %2 back onto queue after failure"), boost::this_thread::get_id(), vf->frame())
 326                                 );
 327                         _queue.push_front (vf);
 328                         lock.unlock ();
 329                 }
 330
 331                 if (remote_backoff > 0) {
 332                         dcpomatic_sleep (remote_backoff);
 333                 }
 334
 335                 lock.lock ();
 336                 _condition.notify_all ();
 337         }
 338 }
 339
 340 void
 341 Encoder::server_found (ServerDescription s)
 342 {
 343         add_worker_threads (s);
 344 }