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