src/lib/j2k_encoder.cc

   1 /*
   2     Copyright (C) 2012-2019 Carl Hetherington <cth@carlh.net>
   3
   4     This file is part of DCP-o-matic.
   5
   6     DCP-o-matic is free software; you can redistribute it and/or modify
   7     it under the terms of the GNU General Public License as published by
   8     the Free Software Foundation; either version 2 of the License, or
   9     (at your option) any later version.
  10
  11     DCP-o-matic is distributed in the hope that it will be useful,
  12     but WITHOUT ANY WARRANTY; without even the implied warranty of
  13     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14     GNU General Public License for more details.
  15
  16     You should have received a copy of the GNU General Public License
  17     along with DCP-o-matic.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 */
  20
  21 /** @file src/j2k_encoder.cc
  22  *  @brief J2K encoder class.
  23  */
  24
  25 #include "j2k_encoder.h"
  26 #include "util.h"
  27 #include "film.h"
  28 #include "log.h"
  29 #include "dcpomatic_log.h"
  30 #include "config.h"
  31 #include "dcp_video.h"
  32 #include "cross.h"
  33 #include "writer.h"
  34 #include "encode_server_finder.h"
  35 #include "player.h"
  36 #include "player_video.h"
  37 #include "encode_server_description.h"
  38 #include "compose.hpp"
  39 #include <libcxml/cxml.h>
  40 #include <boost/foreach.hpp>
  41 #include <iostream>
  42
  43 #include "i18n.h"
  44
  45 using std::list;
  46 using std::cout;
  47 using std::exception;
  48 using boost::shared_ptr;
  49 using boost::weak_ptr;
  50 using boost::optional;
  51 using dcp::Data;
  52 using namespace dcpomatic;
  53
  54 /** @param film Film that we are encoding.
  55  *  @param writer Writer that we are using.
  56  */
  57 J2KEncoder::J2KEncoder (shared_ptr<const Film> film, shared_ptr<Writer> writer)
  58         : _film (film)
  59         , _history (200)
  60         , _writer (writer)
  61 {
  62         servers_list_changed ();
  63 }
  64
  65 J2KEncoder::~J2KEncoder ()
  66 {
  67         try {
  68                 terminate_threads ();
  69         } catch (...) {
  70                 /* Destructors must not throw exceptions; anything bad
  71                    happening now is too late to worry about anyway,
  72                    I think.
  73                 */
  74         }
  75 }
  76
  77 void
  78 J2KEncoder::begin ()
  79 {
  80         weak_ptr<J2KEncoder> wp = shared_from_this ();
  81         _server_found_connection = EncodeServerFinder::instance()->ServersListChanged.connect (
  82                 boost::bind (&J2KEncoder::call_servers_list_changed, wp)
  83                 );
  84 }
  85
  86 /* We don't want the servers-list-changed callback trying to do things
  87    during destruction of J2KEncoder, and I think this is the neatest way
  88    to achieve that.
  89 */
  90 void
  91 J2KEncoder::call_servers_list_changed (weak_ptr<J2KEncoder> encoder)
  92 {
  93         shared_ptr<J2KEncoder> e = encoder.lock ();
  94         if (e) {
  95                 e->servers_list_changed ();
  96         }
  97 }
  98
  99 void
 100 J2KEncoder::end ()
 101 {
 102         boost::mutex::scoped_lock lock (_queue_mutex);
 103
 104         LOG_GENERAL (N_("Clearing queue of %1"), _queue.size ());
 105
 106         /* Keep waking workers until the queue is empty */
 107         while (!_queue.empty ()) {
 108                 rethrow ();
 109                 _empty_condition.notify_all ();
 110                 _full_condition.wait (lock);
 111         }
 112
 113         lock.unlock ();
 114
 115         LOG_GENERAL_NC (N_("Terminating encoder threads"));
 116
 117         terminate_threads ();
 118
 119         /* Something might have been thrown during terminate_threads */
 120         rethrow ();
 121
 122         LOG_GENERAL (N_("Mopping up %1"), _queue.size());
 123
 124         /* The following sequence of events can occur in the above code:
 125              1. a remote worker takes the last image off the queue
 126              2. the loop above terminates
 127              3. the remote worker fails to encode the image and puts it back on the queue
 128              4. the remote worker is then terminated by terminate_threads
 129
 130              So just mop up anything left in the queue here.
 131         */
 132
 133         for (list<shared_ptr<DCPVideo> >::iterator i = _queue.begin(); i != _queue.end(); ++i) {
 134                 LOG_GENERAL (N_("Encode left-over frame %1"), (*i)->index ());
 135                 try {
 136                         _writer->write (
 137                                 (*i)->encode_locally(),
 138                                 (*i)->index(),
 139                                 (*i)->eyes()
 140                                 );
 141                         frame_done ();
 142                 } catch (std::exception& e) {
 143                         LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
 144                 }
 145         }
 146 }
 147
 148 /** @return an estimate of the current number of frames we are encoding per second,
 149  *  if known.
 150  */
 151 optional<float>
 152 J2KEncoder::current_encoding_rate () const
 153 {
 154         return _history.rate ();
 155 }
 156
 157 /** @return Number of video frames that have been queued for encoding */
 158 int
 159 J2KEncoder::video_frames_enqueued () const
 160 {
 161         if (!_last_player_video_time) {
 162                 return 0;
 163         }
 164
 165         return _last_player_video_time->frames_floor (_film->video_frame_rate ());
 166 }
 167
 168 /** Should be called when a frame has been encoded successfully */
 169 void
 170 J2KEncoder::frame_done ()
 171 {
 172         _history.event ();
 173 }
 174
 175 /** Called to request encoding of the next video frame in the DCP.  This is called in order,
 176  *  so each time the supplied frame is the one after the previous one.
 177  *  pv represents one video frame, and could be empty if there is nothing to encode
 178  *  for this DCP frame.
 179  *
 180  *  @param pv PlayerVideo to encode.
 181  *  @param time Time of \p pv within the DCP.
 182  */
 183 void
 184 J2KEncoder::encode (shared_ptr<PlayerVideo> pv, DCPTime time)
 185 {
 186         _waker.nudge ();
 187
 188         size_t threads = _threads->size();
 189
 190         boost::mutex::scoped_lock queue_lock (_queue_mutex);
 191
 192         /* Wait until the queue has gone down a bit.  Allow one thing in the queue even
 193            when there are no threads.
 194         */
 195         while (_queue.size() >= (threads * 2) + 1) {
 196                 LOG_TIMING ("decoder-sleep queue=%1 threads=%2", _queue.size(), threads);
 197                 _full_condition.wait (queue_lock);
 198                 LOG_TIMING ("decoder-wake queue=%1 threads=%2", _queue.size(), threads);
 199         }
 200
 201         _writer->rethrow ();
 202         /* Re-throw any exception raised by one of our threads.  If more
 203            than one has thrown an exception, only one will be rethrown, I think;
 204            but then, if that happens something has gone badly wrong.
 205         */
 206         rethrow ();
 207
 208         Frame const position = time.frames_floor(_film->video_frame_rate());
 209
 210         if (_writer->can_fake_write (position)) {
 211                 /* We can fake-write this frame */
 212                 LOG_DEBUG_ENCODE("Frame @ %1 FAKE", to_string(time));
 213                 _writer->fake_write (position, pv->eyes ());
 214                 frame_done ();
 215         } else if (pv->has_j2k() && !_film->reencode_j2k()) {
 216                 LOG_DEBUG_ENCODE("Frame @ %1 J2K", to_string(time));
 217                 /* This frame already has J2K data, so just write it */
 218                 _writer->write (pv->j2k(), position, pv->eyes ());
 219         } else if (_last_player_video[pv->eyes()] && _writer->can_repeat(position) && pv->same (_last_player_video[pv->eyes()])) {
 220                 LOG_DEBUG_ENCODE("Frame @ %1 REPEAT", to_string(time));
 221                 _writer->repeat (position, pv->eyes ());
 222         } else {
 223                 LOG_DEBUG_ENCODE("Frame @ %1 ENCODE", to_string(time));
 224                 /* Queue this new frame for encoding */
 225                 LOG_TIMING ("add-frame-to-queue queue=%1", _queue.size ());
 226                 _queue.push_back (shared_ptr<DCPVideo> (
 227                                           new DCPVideo (
 228                                                   pv,
 229                                                   position,
 230                                                   _film->video_frame_rate(),
 231                                                   _film->j2k_bandwidth(),
 232                                                   _film->resolution()
 233                                                   )
 234                                           ));
 235
 236                 /* The queue might not be empty any more, so notify anything which is
 237                    waiting on that.
 238                 */
 239                 _empty_condition.notify_all ();
 240         }
 241
 242         _last_player_video[pv->eyes()] = pv;
 243         _last_player_video_time = time;
 244 }
 245
 246 void
 247 J2KEncoder::terminate_threads ()
 248 {
 249         if (!_threads) {
 250                 return;
 251         }
 252
 253         _threads->interrupt_all ();
 254         try {
 255                 _threads->join_all ();
 256         } catch (exception& e) {
 257                 LOG_ERROR ("join() threw an exception: %1", e.what());
 258         } catch (...) {
 259                 LOG_ERROR_NC ("join() threw an exception");
 260         }
 261
 262         _threads.reset ();
 263 }
 264
 265 void
 266 J2KEncoder::encoder_thread (optional<EncodeServerDescription> server)
 267 try
 268 {
 269         if (server) {
 270                 LOG_TIMING ("start-encoder-thread thread=%1 server=%2", thread_id (), server->host_name ());
 271         } else {
 272                 LOG_TIMING ("start-encoder-thread thread=%1 server=localhost", thread_id ());
 273         }
 274
 275         /* Number of seconds that we currently wait between attempts
 276            to connect to the server; not relevant for localhost
 277            encodings.
 278         */
 279         int remote_backoff = 0;
 280
 281         while (true) {
 282
 283                 LOG_TIMING ("encoder-sleep thread=%1", thread_id ());
 284                 boost::mutex::scoped_lock lock (_queue_mutex);
 285                 while (_queue.empty ()) {
 286                         _empty_condition.wait (lock);
 287                 }
 288
 289                 LOG_TIMING ("encoder-wake thread=%1 queue=%2", thread_id(), _queue.size());
 290                 shared_ptr<DCPVideo> vf = _queue.front ();
 291
 292                 /* We're about to commit to either encoding this frame or putting it back onto the queue,
 293                    so we must not be interrupted until one or other of these things have happened.  This
 294                    block has thread interruption disabled.
 295                 */
 296                 {
 297                         boost::this_thread::disable_interruption dis;
 298
 299                         LOG_TIMING ("encoder-pop thread=%1 frame=%2 eyes=%3", thread_id(), vf->index(), (int) vf->eyes ());
 300                         _queue.pop_front ();
 301
 302                         lock.unlock ();
 303
 304                         optional<Data> encoded;
 305
 306                         /* We need to encode this input */
 307                         if (server) {
 308                                 try {
 309                                         encoded = vf->encode_remotely (server.get ());
 310
 311                                         if (remote_backoff > 0) {
 312                                                 LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
 313                                         }
 314
 315                                         /* This job succeeded, so remove any backoff */
 316                                         remote_backoff = 0;
 317
 318                                 } catch (std::exception& e) {
 319                                         if (remote_backoff < 60) {
 320                                                 /* back off more */
 321                                                 remote_backoff += 10;
 322                                         }
 323                                         LOG_ERROR (
 324                                                 N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
 325                                                 vf->index(), server->host_name(), e.what(), remote_backoff
 326                                                 );
 327                                 }
 328
 329                         } else {
 330                                 try {
 331                                         LOG_TIMING ("start-local-encode thread=%1 frame=%2", thread_id(), vf->index());
 332                                         encoded = vf->encode_locally ();
 333                                         LOG_TIMING ("finish-local-encode thread=%1 frame=%2", thread_id(), vf->index());
 334                                 } catch (std::exception& e) {
 335                                         /* This is very bad, so don't cope with it, just pass it on */
 336                                         LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
 337                                         throw;
 338                                 }
 339                         }
 340
 341                         if (encoded) {
 342                                 _writer->write (encoded.get(), vf->index (), vf->eyes ());
 343                                 frame_done ();
 344                         } else {
 345                                 lock.lock ();
 346                                 LOG_GENERAL (N_("[%1] J2KEncoder thread pushes frame %2 back onto queue after failure"), thread_id(), vf->index());
 347                                 _queue.push_front (vf);
 348                                 lock.unlock ();
 349                         }
 350                 }
 351
 352                 if (remote_backoff > 0) {
 353                         boost::this_thread::sleep (boost::posix_time::seconds (remote_backoff));
 354                 }
 355
 356                 /* The queue might not be full any more, so notify anything that is waiting on that */
 357                 lock.lock ();
 358                 _full_condition.notify_all ();
 359         }
 360 }
 361 catch (boost::thread_interrupted& e) {
 362         /* Ignore these and just stop the thread */
 363         _full_condition.notify_all ();
 364 }
 365 catch (...)
 366 {
 367         store_current ();
 368         /* Wake anything waiting on _full_condition so it can see the exception */
 369         _full_condition.notify_all ();
 370 }
 371
 372 void
 373 J2KEncoder::servers_list_changed ()
 374 {
 375         terminate_threads ();
 376         _threads.reset (new boost::thread_group());
 377
 378         /* XXX: could re-use threads */
 379
 380 #ifdef BOOST_THREAD_PLATFORM_WIN32
 381         OSVERSIONINFO info;
 382         info.dwOSVersionInfoSize = sizeof (OSVERSIONINFO);
 383         GetVersionEx (&info);
 384         bool const windows_xp = (info.dwMajorVersion == 5 && info.dwMinorVersion == 1);
 385         if (windows_xp) {
 386                 LOG_GENERAL_NC (N_("Setting thread affinity for Windows XP"));
 387         }
 388 #endif
 389
 390         if (!Config::instance()->only_servers_encode ()) {
 391                 for (int i = 0; i < Config::instance()->master_encoding_threads (); ++i) {
 392 #ifdef DCPOMATIC_LINUX
 393                         boost::thread* t = _threads->create_thread(boost::bind(&J2KEncoder::encoder_thread, this, optional<EncodeServerDescription>()));
 394                         pthread_setname_np (t->native_handle(), "encode-worker");
 395 #endif
 396 #ifdef DCPOMATIC_OSX
 397                         _threads->create_thread(boost::bind(&J2KEncoder::encoder_thread, this, optional<EncodeServerDescription>()));
 398 #endif
 399 #ifdef DCPOMATIC_WINDOWS
 400                         boost::thread* t = _threads->create_thread(boost::bind(&J2KEncoder::encoder_thread, this, optional<EncodeServerDescription>()));
 401                         if (windows_xp) {
 402                                 SetThreadAffinityMask (t->native_handle(), 1 << i);
 403                         }
 404 #endif
 405                 }
 406         }
 407
 408         BOOST_FOREACH (EncodeServerDescription i, EncodeServerFinder::instance()->servers()) {
 409                 if (!i.current_link_version()) {
 410                         continue;
 411                 }
 412
 413                 LOG_GENERAL (N_("Adding %1 worker threads for remote %2"), i.threads(), i.host_name ());
 414                 for (int j = 0; j < i.threads(); ++j) {
 415                         _threads->create_thread(boost::bind(&J2KEncoder::encoder_thread, this, i));
 416                 }
 417         }
 418
 419         _writer->set_encoder_threads (_threads->size());
 420 }