src/lib/j2k_encoder.cc

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