From: Carl Hetherington Date: Thu, 17 Jan 2013 21:42:43 +0000 (+0000) Subject: Merge writer-thread with original which was time-cleanup. X-Git-Tag: v2.0.48~1337^2~743 X-Git-Url: https://git.carlh.net/gitweb/?a=commitdiff_plain;h=039ac5848730d154e0b3f3dd841a55e45ba922d8;hp=d42989d447bc9199b670d8e99fbb8b93f06ddb71;p=dcpomatic.git Merge writer-thread with original which was time-cleanup. --- diff --git a/src/lib/dcp_video_frame.cc b/src/lib/dcp_video_frame.cc index 427d447ef..1c408270e 100644 --- a/src/lib/dcp_video_frame.cc +++ b/src/lib/dcp_video_frame.cc @@ -59,6 +59,7 @@ using std::string; using std::stringstream; using std::ofstream; +using std::cout; using boost::shared_ptr; using libdcp::Size; @@ -372,6 +373,18 @@ DCPVideoFrame::encode_remotely (ServerDescription const * serv) return e; } +EncodedData::EncodedData (int s) + : _data (new uint8_t[s]) + , _size (s) +{ + +} + +EncodedData::~EncodedData () +{ + delete[] _data; +} + /** Write this data to a J2K file. * @param opt Options. * @param frame DCP Frame index. @@ -414,14 +427,15 @@ EncodedData::send (shared_ptr socket) socket->write (_data, _size, 30); } -/** @param s Size of data in bytes */ -RemotelyEncodedData::RemotelyEncodedData (int s) - : EncodedData (new uint8_t[s], s) +LocallyEncodedData::LocallyEncodedData (uint8_t* d, int s) + : EncodedData (s) { - + memcpy (_data, d, s); } -RemotelyEncodedData::~RemotelyEncodedData () +/** @param s Size of data in bytes */ +RemotelyEncodedData::RemotelyEncodedData (int s) + : EncodedData (s) { - delete[] _data; + } diff --git a/src/lib/dcp_video_frame.h b/src/lib/dcp_video_frame.h index 4271ebb28..e311724d8 100644 --- a/src/lib/dcp_video_frame.h +++ b/src/lib/dcp_video_frame.h @@ -39,15 +39,11 @@ class Subtitle; class EncodedData { public: - /** @param d Data (will not be freed by this class, but may be by subclasses) - * @param s Size of data, in bytes. + /** @param s Size of data, in bytes. */ - EncodedData (uint8_t* d, int s) - : _data (d) - , _size (s) - {} + EncodedData (int s); - virtual ~EncodedData () {} + virtual ~EncodedData (); void send (boost::shared_ptr socket); void write (boost::shared_ptr, int); @@ -65,6 +61,10 @@ public: protected: uint8_t* _data; ///< data int _size; ///< data size in bytes + +private: + /* No copy construction */ + EncodedData (EncodedData const &); }; /** @class LocallyEncodedData @@ -75,12 +75,10 @@ protected: class LocallyEncodedData : public EncodedData { public: - /** @param d Data (which will not be freed by this class) + /** @param d Data (which will be copied by this class) * @param s Size of data, in bytes. */ - LocallyEncodedData (uint8_t* d, int s) - : EncodedData (d, s) - {} + LocallyEncodedData (uint8_t* d, int s); }; /** @class RemotelyEncodedData @@ -91,7 +89,6 @@ class RemotelyEncodedData : public EncodedData { public: RemotelyEncodedData (int s); - ~RemotelyEncodedData (); }; /** @class DCPVideoFrame diff --git a/src/lib/encoder.cc b/src/lib/encoder.cc index 6df7c85a1..0f48310ef 100644 --- a/src/lib/encoder.cc +++ b/src/lib/encoder.cc @@ -61,7 +61,9 @@ Encoder::Encoder (shared_ptr f) #ifdef HAVE_SWRESAMPLE , _swr_context (0) #endif - , _process_end (false) + , _terminate_encoder (false) + , _writer_thread (0) + , _terminate_writer (false) { if (_film->audio_stream()) { /* Create sound output files with .tmp suffixes; we will rename @@ -86,6 +88,7 @@ Encoder::~Encoder () { close_sound_files (); terminate_worker_threads (); + terminate_writer_thread (); } void @@ -131,6 +134,8 @@ Encoder::process_begin () _worker_threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, *i))); } } + + _writer_thread = new boost::thread (boost::bind (&Encoder::writer_thread, this)); } @@ -175,11 +180,11 @@ Encoder::process_end () boost::mutex::scoped_lock lock (_worker_mutex); - _film->log()->log ("Clearing queue of " + lexical_cast (_queue.size ())); + _film->log()->log ("Clearing queue of " + lexical_cast (_encode_queue.size ())); /* Keep waking workers until the queue is empty */ - while (!_queue.empty ()) { - _film->log()->log ("Waking with " + lexical_cast (_queue.size ()), Log::VERBOSE); + while (!_encode_queue.empty ()) { + _film->log()->log ("Waking with " + lexical_cast (_encode_queue.size ()), Log::VERBOSE); _worker_condition.notify_all (); _worker_condition.wait (lock); } @@ -187,8 +192,9 @@ Encoder::process_end () lock.unlock (); terminate_worker_threads (); + terminate_writer_thread (); - _film->log()->log ("Mopping up " + lexical_cast (_queue.size())); + _film->log()->log ("Mopping up " + lexical_cast (_encode_queue.size())); /* The following sequence of events can occur in the above code: 1. a remote worker takes the last image off the queue @@ -199,7 +205,7 @@ Encoder::process_end () So just mop up anything left in the queue here. */ - for (list >::iterator i = _queue.begin(); i != _queue.end(); ++i) { + for (list >::iterator i = _encode_queue.begin(); i != _encode_queue.end(); ++i) { _film->log()->log (String::compose ("Encode left-over frame %1", (*i)->frame ())); try { shared_ptr e = (*i)->encode_locally (); @@ -210,6 +216,11 @@ Encoder::process_end () } } + /* Mop up any unwritten things in the writer's queue */ + for (list, int> >::iterator i = _write_queue.begin(); i != _write_queue.end(); ++i) { + i->first->write (_opt, i->second); + } + /* Now do links (or copies on windows) to duplicate frames */ for (list >::iterator i = _links_required.begin(); i != _links_required.end(); ++i) { link (_film->frame_out_path (i->first, false), _film->frame_out_path (i->second, false)); @@ -290,13 +301,13 @@ Encoder::process_video (shared_ptr image, bool same, boost::shared_ptr= _worker_threads.size() * 2 && !_process_end) { - TIMING ("decoder sleeps with queue of %1", _queue.size()); + while (_encode_queue.size() >= _worker_threads.size() * 2 && !_terminate_encoder) { + TIMING ("decoder sleeps with queue of %1", _encode_queue.size()); _worker_condition.wait (lock); - TIMING ("decoder wakes with queue of %1", _queue.size()); + TIMING ("decoder wakes with queue of %1", _encode_queue.size()); } - if (_process_end) { + if (_terminate_encoder) { return; } @@ -315,8 +326,8 @@ Encoder::process_video (shared_ptr image, bool same, boost::shared_ptr const s = Filter::ffmpeg_strings (_film->filters()); - TIMING ("adding to queue of %1", _queue.size ()); - _queue.push_back (boost::shared_ptr ( + TIMING ("adding to queue of %1", _encode_queue.size ()); + _encode_queue.push_back (boost::shared_ptr ( new DCPVideoFrame ( image, sub, _film->format()->dcp_size(), _film->format()->dcp_padding (_film), _film->subtitle_offset(), _film->subtitle_scale(), @@ -412,7 +423,7 @@ void Encoder::terminate_worker_threads () { boost::mutex::scoped_lock lock (_worker_mutex); - _process_end = true; + _terminate_encoder = true; _worker_condition.notify_all (); lock.unlock (); @@ -422,6 +433,23 @@ Encoder::terminate_worker_threads () } } +void +Encoder::terminate_writer_thread () +{ + if (!_writer_thread) { + return; + } + + boost::mutex::scoped_lock lock (_writer_mutex); + _terminate_writer = true; + _writer_condition.notify_all (); + lock.unlock (); + + _writer_thread->join (); + delete _writer_thread; + _writer_thread = 0; +} + void Encoder::encoder_thread (ServerDescription* server) { @@ -435,18 +463,18 @@ Encoder::encoder_thread (ServerDescription* server) TIMING ("encoder thread %1 sleeps", boost::this_thread::get_id()); boost::mutex::scoped_lock lock (_worker_mutex); - while (_queue.empty () && !_process_end) { + while (_encode_queue.empty () && !_terminate_encoder) { _worker_condition.wait (lock); } - if (_process_end) { + if (_terminate_encoder) { return; } - TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _queue.size()); - boost::shared_ptr vf = _queue.front (); + TIMING ("encoder thread %1 wakes with queue of %2", boost::this_thread::get_id(), _encode_queue.size()); + boost::shared_ptr vf = _encode_queue.front (); _film->log()->log (String::compose ("Encoder thread %1 pops frame %2 from queue", boost::this_thread::get_id(), vf->frame()), Log::VERBOSE); - _queue.pop_front (); + _encode_queue.pop_front (); lock.unlock (); @@ -486,14 +514,15 @@ Encoder::encoder_thread (ServerDescription* server) } if (encoded) { - encoded->write (_film, vf->frame ()); - frame_done (); + boost::mutex::scoped_lock lock2 (_writer_mutex); + _write_queue.push_back (make_pair (encoded, vf->frame ())); + _writer_condition.notify_all (); } else { lock.lock (); _film->log()->log ( String::compose ("Encoder thread %1 pushes frame %2 back onto queue after failure", boost::this_thread::get_id(), vf->frame()) ); - _queue.push_front (vf); + _encode_queue.push_front (vf); lock.unlock (); } @@ -520,3 +549,28 @@ Encoder::link (string a, string b) const boost::filesystem::copy_file (a, b); #endif } + +void +Encoder::writer_thread () +{ + while (1) + { + boost::mutex::scoped_lock lock (_writer_mutex); + TIMING ("writer sleeps with a queue of %1", _write_queue.size()); + while (_write_queue.empty() && !_terminate_writer) { + _writer_condition.wait (lock); + } + TIMING ("writer wakes with a queue of %1", _write_queue.size()); + + if (_terminate_writer) { + return; + } + + pair, int> encoded = _write_queue.front (); + _write_queue.pop_front (); + + lock.unlock (); + encoded.first->write (_opt, encoded.second); + lock.lock (); + } +} diff --git a/src/lib/encoder.h b/src/lib/encoder.h index 20255cca9..96e7a1d25 100644 --- a/src/lib/encoder.h +++ b/src/lib/encoder.h @@ -50,6 +50,7 @@ class AudioBuffers; class Film; class ServerDescription; class DCPVideoFrame; +class EncodedData; /** @class Encoder * @brief Encoder to J2K and WAV for DCP. @@ -119,6 +120,9 @@ private: /** Number of audio frames written for the DCP so far */ int64_t _audio_frames_out; + void writer_thread (); + void terminate_writer_thread (); + #if HAVE_SWRESAMPLE SwrContext* _swr_context; #endif @@ -133,11 +137,17 @@ private: std::vector _sound_files; boost::optional _last_real_frame; - bool _process_end; - std::list > _queue; + bool _terminate_encoder; + std::list > _encode_queue; std::list _worker_threads; mutable boost::mutex _worker_mutex; boost::condition _worker_condition; + + boost::thread* _writer_thread; + bool _terminate_writer; + std::list, int> > _write_queue; + mutable boost::mutex _writer_mutex; + boost::condition _writer_condition; }; #endif