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