Difficult to see how any unprotected write with multiple readers
[dcpomatic.git] / src / lib / writer.cc
index 10e4514fbcc134ef333d333d5086511b6b5660a3..699f220c4df31fdef6dfbae02086515119e60fd6 100644 (file)
@@ -1,5 +1,5 @@
 /*
-    Copyright (C) 2012-2017 Carl Hetherington <cth@carlh.net>
+    Copyright (C) 2012-2019 Carl Hetherington <cth@carlh.net>
 
     This file is part of DCP-o-matic.
 
@@ -23,6 +23,7 @@
 #include "film.h"
 #include "ratio.h"
 #include "log.h"
+#include "dcpomatic_log.h"
 #include "dcp_video.h"
 #include "dcp_content_type.h"
 #include "audio_mapping.h"
@@ -34,6 +35,7 @@
 #include "font.h"
 #include "util.h"
 #include "reel_writer.h"
+#include "text_content.h"
 #include <dcp/cpl.h>
 #include <dcp/locale_convert.h>
 #include <boost/foreach.hpp>
 
 #include "i18n.h"
 
-#define LOG_GENERAL(...) _film->log()->log (String::compose (__VA_ARGS__), LogEntry::TYPE_GENERAL);
-#define LOG_GENERAL_NC(...) _film->log()->log (__VA_ARGS__, LogEntry::TYPE_GENERAL);
-#define LOG_DEBUG_ENCODE(...) _film->log()->log (String::compose (__VA_ARGS__), LogEntry::TYPE_DEBUG_ENCODE);
-#define LOG_TIMING(...) _film->log()->log (String::compose (__VA_ARGS__), LogEntry::TYPE_TIMING);
-#define LOG_WARNING_NC(...) _film->log()->log (__VA_ARGS__, LogEntry::TYPE_WARNING);
-#define LOG_WARNING(...) _film->log()->log (String::compose (__VA_ARGS__), LogEntry::TYPE_WARNING);
-#define LOG_ERROR(...) _film->log()->log (String::compose (__VA_ARGS__), LogEntry::TYPE_ERROR);
-
 /* OS X strikes again */
 #undef set_key
 
@@ -63,18 +57,22 @@ using std::cout;
 using std::map;
 using std::min;
 using std::max;
+using std::vector;
 using boost::shared_ptr;
 using boost::weak_ptr;
 using boost::dynamic_pointer_cast;
+using boost::optional;
 using dcp::Data;
+using namespace dcpomatic;
 
 Writer::Writer (shared_ptr<const Film> film, weak_ptr<Job> j)
        : _film (film)
        , _job (j)
-       , _thread (0)
        , _finish (false)
        , _queued_full_in_memory (0)
-       , _maximum_frames_in_memory (0)
+       /* These will be reset to sensible values when J2KEncoder is created */
+       , _maximum_frames_in_memory (8)
+       , _maximum_queue_size (8)
        , _full_written (0)
        , _fake_written (0)
        , _repeat_written (0)
@@ -89,11 +87,16 @@ Writer::Writer (shared_ptr<const Film> film, weak_ptr<Job> j)
                _reels.push_back (ReelWriter (film, p, job, reel_index++, reels.size(), _film->content_summary(p)));
        }
 
-       /* We can keep track of the current audio and subtitle reels easily because audio
-          and subs arrive to the Writer in sequence.  This is not so for video.
+       _last_written.resize (reels.size());
+
+       /* We can keep track of the current audio, subtitle and closed caption reels easily because audio
+          and captions arrive to the Writer in sequence.  This is not so for video.
        */
        _audio_reel = _reels.begin ();
        _subtitle_reel = _reels.begin ();
+       BOOST_FOREACH (DCPTextTrack i, _film->closed_caption_tracks()) {
+               _caption_reels[i] = _reels.begin ();
+       }
 
        /* Check that the signer is OK if we need one */
        string reason;
@@ -105,7 +108,10 @@ Writer::Writer (shared_ptr<const Film> film, weak_ptr<Job> j)
 void
 Writer::start ()
 {
-       _thread = new boost::thread (boost::bind (&Writer::thread, this));
+       _thread = boost::thread (boost::bind(&Writer::thread, this));
+#ifdef DCPOMATIC_LINUX
+       pthread_setname_np (_thread.native_handle(), "writer");
+#endif
 }
 
 Writer::~Writer ()
@@ -125,7 +131,9 @@ Writer::write (Data encoded, Frame frame, Eyes eyes)
        boost::mutex::scoped_lock lock (_state_mutex);
 
        while (_queued_full_in_memory > _maximum_frames_in_memory) {
-               /* The queue is too big; wait until that is sorted out */
+               /* There are too many full frames in memory; wake the main writer thread and
+                  wait until it sorts everything out */
+               _empty_condition.notify_all ();
                _full_condition.wait (lock);
        }
 
@@ -168,8 +176,11 @@ Writer::repeat (Frame frame, Eyes eyes)
 {
        boost::mutex::scoped_lock lock (_state_mutex);
 
-       while (_queued_full_in_memory > _maximum_frames_in_memory) {
-               /* The queue is too big; wait until that is sorted out */
+       while (_queue.size() > _maximum_queue_size && have_sequenced_image_at_queue_head()) {
+               /* The queue is too big, and the main writer thread can run and fix it, so
+                  wake it and wait until it has done.
+               */
+               _empty_condition.notify_all ();
                _full_condition.wait (lock);
        }
 
@@ -196,26 +207,27 @@ Writer::fake_write (Frame frame, Eyes eyes)
 {
        boost::mutex::scoped_lock lock (_state_mutex);
 
-       while (_queued_full_in_memory > _maximum_frames_in_memory) {
-               /* The queue is too big; wait until that is sorted out */
+       while (_queue.size() > _maximum_queue_size && have_sequenced_image_at_queue_head()) {
+               /* The queue is too big, and the main writer thread can run and fix it, so
+                  wake it and wait until it has done.
+               */
+               _empty_condition.notify_all ();
                _full_condition.wait (lock);
        }
 
        size_t const reel = video_reel (frame);
-       Frame const reel_frame = frame - _reels[reel].start ();
-
-       FILE* file = fopen_boost (_film->info_file(_reels[reel].period()), "rb");
-       if (!file) {
-               throw ReadFileError (_film->info_file(_reels[reel].period()));
-       }
-       dcp::FrameInfo info = _reels[reel].read_frame_info (file, reel_frame, eyes);
-       fclose (file);
+       Frame const frame_in_reel = frame - _reels[reel].start ();
 
        QueueItem qi;
        qi.type = QueueItem::FAKE;
-       qi.size = info.size;
+
+       {
+               shared_ptr<InfoFileHandle> info_file = _film->info_file_handle(_reels[reel].period(), true);
+               qi.size = _reels[reel].read_frame_info(info_file, frame_in_reel, eyes).size;
+       }
+
        qi.reel = reel;
-       qi.frame = reel_frame;
+       qi.frame = frame_in_reel;
        if (_film->three_d() && eyes == EYES_BOTH) {
                qi.eyes = EYES_LEFT;
                _queue.push_back (qi);
@@ -258,8 +270,11 @@ Writer::write (shared_ptr<const AudioBuffers> audio, DCPTime const time)
                        /* Easy case: we can write all the audio to this reel */
                        _audio_reel->write (audio);
                        t = end;
+               } else if (_audio_reel->period().to <= t) {
+                       /* This reel is entirely before the start of our audio; just skip the reel */
+                       ++_audio_reel;
                } else {
-                       /* Split the audio into two and write the first part */
+                       /* This audio is over a reel boundary; split the audio into two and write the first part */
                        DCPTime part_lengths[2] = {
                                _audio_reel->period().to - t,
                                end - _audio_reel->period().to
@@ -271,15 +286,12 @@ Writer::write (shared_ptr<const AudioBuffers> audio, DCPTime const time)
                        };
 
                        if (part_frames[0]) {
-                               shared_ptr<AudioBuffers> part (new AudioBuffers (audio->channels(), part_frames[0]));
-                               part->copy_from (audio.get(), part_frames[0], 0, 0);
+                               shared_ptr<AudioBuffers> part (new AudioBuffers(audio, part_frames[0], 0));
                                _audio_reel->write (part);
                        }
 
                        if (part_frames[1]) {
-                               shared_ptr<AudioBuffers> part (new AudioBuffers (audio->channels(), part_frames[1]));
-                               part->copy_from (audio.get(), part_frames[1], part_frames[0], 0);
-                               audio = part;
+                               audio.reset (new AudioBuffers(audio, part_frames[1], part_frames[0]));
                        } else {
                                audio.reset ();
                        }
@@ -290,7 +302,8 @@ Writer::write (shared_ptr<const AudioBuffers> audio, DCPTime const time)
        }
 }
 
-/** This must be called from Writer::thread() with an appropriate lock held */
+
+/** Caller must hold a lock on _state_mutex */
 bool
 Writer::have_sequenced_image_at_queue_head ()
 {
@@ -299,30 +312,41 @@ Writer::have_sequenced_image_at_queue_head ()
        }
 
        _queue.sort ();
-
        QueueItem const & f = _queue.front();
-       ReelWriter const & reel = _reels[f.reel];
+       return _last_written[f.reel].next(f);
+}
 
-       /* The queue should contain only EYES_LEFT/EYES_RIGHT pairs or EYES_BOTH */
 
-       if (f.eyes == EYES_BOTH) {
+bool
+Writer::LastWritten::next (QueueItem qi) const
+{
+       if (qi.eyes == EYES_BOTH) {
                /* 2D */
-               return f.frame == (reel.last_written_video_frame() + 1);
+               return qi.frame == (_frame + 1);
        }
 
        /* 3D */
 
-       if (reel.last_written_eyes() == EYES_LEFT && f.frame == reel.last_written_video_frame() && f.eyes == EYES_RIGHT) {
+       if (_eyes == EYES_LEFT && qi.frame == _frame && qi.eyes == EYES_RIGHT) {
                return true;
        }
 
-       if (reel.last_written_eyes() == EYES_RIGHT && f.frame == (reel.last_written_video_frame() + 1) && f.eyes == EYES_LEFT) {
+       if (_eyes == EYES_RIGHT && qi.frame == (_frame + 1) && qi.eyes == EYES_LEFT) {
                return true;
        }
 
        return false;
 }
 
+
+void
+Writer::LastWritten::update (QueueItem qi)
+{
+       _frame = qi.frame;
+       _eyes = qi.eyes;
+}
+
+
 void
 Writer::thread ()
 try
@@ -357,11 +381,11 @@ try
                        /* (Hopefully temporarily) log anything that was not written */
                        if (!_queue.empty() && !have_sequenced_image_at_queue_head()) {
                                LOG_WARNING (N_("Finishing writer with a left-over queue of %1:"), _queue.size());
-                               for (list<QueueItem>::const_iterator i = _queue.begin(); i != _queue.end(); ++i) {
-                                       if (i->type == QueueItem::FULL) {
-                                               LOG_WARNING (N_("- type FULL, frame %1, eyes %2"), i->frame, (int) i->eyes);
+                               BOOST_FOREACH (QueueItem const& i, _queue) {
+                                       if (i.type == QueueItem::FULL) {
+                                               LOG_WARNING (N_("- type FULL, frame %1, eyes %2"), i.frame, (int) i.eyes);
                                        } else {
-                                               LOG_WARNING (N_("- type FAKE, size %1, frame %2, eyes %3"), i->size, i->frame, (int) i->eyes);
+                                               LOG_WARNING (N_("- type FAKE, size %1, frame %2, eyes %3"), i.size, i.frame, (int) i.eyes);
                                        }
                                }
                        }
@@ -371,6 +395,7 @@ try
                /* Write any frames that we can write; i.e. those that are in sequence. */
                while (have_sequenced_image_at_queue_head ()) {
                        QueueItem qi = _queue.front ();
+                       _last_written[qi.reel].update (qi);
                        _queue.pop_front ();
                        if (qi.type == QueueItem::FULL && qi.encoded) {
                                --_queued_full_in_memory;
@@ -391,7 +416,7 @@ try
                                break;
                        case QueueItem::FAKE:
                                LOG_DEBUG_ENCODE (N_("Writer FAKE-writes %1"), qi.frame);
-                               reel.fake_write (qi.frame, qi.eyes, qi.size);
+                               reel.fake_write (qi.size);
                                ++_fake_written;
                                break;
                        case QueueItem::REPEAT:
@@ -402,6 +427,7 @@ try
                        }
 
                        lock.lock ();
+                       _full_condition.notify_all ();
                }
 
                while (_queued_full_in_memory > _maximum_frames_in_memory) {
@@ -419,7 +445,7 @@ try
                        DCPOMATIC_ASSERT (i != _queue.rend());
                        ++_pushed_to_disk;
                        /* For the log message below */
-                       int const awaiting = _reels[_queue.front().reel].last_written_video_frame();
+                       int const awaiting = _last_written[_queue.front().reel].frame() + 1;
                        lock.unlock ();
 
                        /* i is valid here, even though we don't hold a lock on the mutex,
@@ -437,10 +463,8 @@ try
                        lock.lock ();
                        i->encoded.reset ();
                        --_queued_full_in_memory;
+                       _full_condition.notify_all ();
                }
-
-               /* The queue has probably just gone down a bit; notify anything wait()ing on _full_condition */
-               _full_condition.notify_all ();
        }
 }
 catch (...)
@@ -452,7 +476,7 @@ void
 Writer::terminate_thread (bool can_throw)
 {
        boost::mutex::scoped_lock lock (_state_mutex);
-       if (_thread == 0) {
+       if (!_thread.joinable()) {
                return;
        }
 
@@ -461,22 +485,17 @@ Writer::terminate_thread (bool can_throw)
        _full_condition.notify_all ();
        lock.unlock ();
 
-       if (_thread->joinable ()) {
-               _thread->join ();
-       }
+       _thread.join ();
 
        if (can_throw) {
                rethrow ();
        }
-
-       delete _thread;
-       _thread = 0;
 }
 
 void
 Writer::finish ()
 {
-       if (!_thread) {
+       if (!_thread.joinable()) {
                return;
        }
 
@@ -547,6 +566,8 @@ Writer::finish ()
        meta.set_issue_date_now ();
 
        cpl->set_metadata (meta);
+       cpl->set_ratings (vector_to_list(_film->ratings()));
+       cpl->set_content_version_label_text (_film->content_version());
 
        shared_ptr<const dcp::CertificateChain> signer;
        if (_film->is_signed ()) {
@@ -573,7 +594,7 @@ Writer::write_cover_sheet ()
        boost::filesystem::path const cover = _film->file ("COVER_SHEET.txt");
        FILE* f = fopen_boost (cover, "w");
        if (!f) {
-               throw OpenFileError (cover, errno, false);
+               throw OpenFileError (cover, errno, OpenFileError::WRITE);
        }
 
        string text = Config::instance()->cover_sheet ();
@@ -581,7 +602,16 @@ Writer::write_cover_sheet ()
        boost::algorithm::replace_all (text, "$TYPE", _film->dcp_content_type()->pretty_name());
        boost::algorithm::replace_all (text, "$CONTAINER", _film->container()->container_nickname());
        boost::algorithm::replace_all (text, "$AUDIO_LANGUAGE", _film->isdcf_metadata().audio_language);
-       boost::algorithm::replace_all (text, "$SUBTITLE_LANGUAGE", _film->isdcf_metadata().subtitle_language);
+
+       optional<string> subtitle_language;
+       BOOST_FOREACH (shared_ptr<Content> i, _film->content()) {
+               BOOST_FOREACH (shared_ptr<TextContent> j, i->text) {
+                       if (j->type() == TEXT_OPEN_SUBTITLE && j->use()) {
+                               subtitle_language = j->language ();
+                       }
+               }
+       }
+       boost::algorithm::replace_all (text, "$SUBTITLE_LANGUAGE", subtitle_language.get_value_or("None"));
 
        boost::uintmax_t size = 0;
        for (
@@ -624,7 +654,7 @@ Writer::write_cover_sheet ()
 
        boost::algorithm::replace_all (text, "$LENGTH", length);
 
-       fwrite (text.c_str(), 1, text.length(), f);
+       checked_fwrite (text.c_str(), text.length(), f, cover);
        fclose (f);
 }
 
@@ -634,6 +664,11 @@ Writer::write_cover_sheet ()
 bool
 Writer::can_fake_write (Frame frame) const
 {
+       if (_film->encrypted()) {
+               /* We need to re-write the frame because the asset ID is embedded in the HMAC... I think... */
+               return false;
+       }
+
        /* We have to do a proper write of the first frame so that we can set up the JPEG2000
           parameters in the asset writer.
        */
@@ -645,18 +680,32 @@ Writer::can_fake_write (Frame frame) const
        return (frame != 0 && frame < reel.first_nonexistant_frame());
 }
 
+/** @param track Closed caption track if type == TEXT_CLOSED_CAPTION */
 void
-Writer::write (PlayerSubtitles subs, DCPTimePeriod period)
+Writer::write (PlayerText text, TextType type, optional<DCPTextTrack> track, DCPTimePeriod period)
 {
-       if (subs.text.empty ()) {
-               return;
+       vector<ReelWriter>::iterator* reel = 0;
+
+       switch (type) {
+       case TEXT_OPEN_SUBTITLE:
+               reel = &_subtitle_reel;
+               break;
+       case TEXT_CLOSED_CAPTION:
+               DCPOMATIC_ASSERT (track);
+               DCPOMATIC_ASSERT (_caption_reels.find(*track) != _caption_reels.end());
+               reel = &_caption_reels[*track];
+               break;
+       default:
+               DCPOMATIC_ASSERT (false);
        }
 
-       if (_subtitle_reel->period().to <= period.from) {
-               ++_subtitle_reel;
+       DCPOMATIC_ASSERT (*reel != _reels.end());
+       while ((*reel)->period().to <= period.from) {
+               ++(*reel);
+               DCPOMATIC_ASSERT (*reel != _reels.end());
        }
 
-       _subtitle_reel->write (subs);
+       (*reel)->write (text, type, track, period);
 }
 
 void
@@ -701,7 +750,9 @@ operator== (QueueItem const & a, QueueItem const & b)
 void
 Writer::set_encoder_threads (int threads)
 {
+       boost::mutex::scoped_lock lm (_state_mutex);
        _maximum_frames_in_memory = lrint (threads * Config::instance()->frames_in_memory_multiplier());
+       _maximum_queue_size = threads * 16;
 }
 
 void
@@ -726,14 +777,16 @@ Writer::video_reel (int frame) const
 void
 Writer::set_digest_progress (Job* job, float progress)
 {
-       /* I believe this is thread-safe */
-       _digest_progresses[boost::this_thread::get_id()] = progress;
-
        boost::mutex::scoped_lock lm (_digest_progresses_mutex);
+
+       _digest_progresses[boost::this_thread::get_id()] = progress;
        float min_progress = FLT_MAX;
        for (map<boost::thread::id, float>::const_iterator i = _digest_progresses.begin(); i != _digest_progresses.end(); ++i) {
                min_progress = min (min_progress, i->second);
        }
 
        job->set_progress (min_progress);
+
+       Waker waker;
+       waker.nudge ();
 }