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