From d0babb26ab341026fe227de810ed30c6c9f65cf2 Mon Sep 17 00:00:00 2001 From: Carl Hetherington Date: Fri, 11 Jan 2013 18:50:50 +0000 Subject: [PATCH] Try a separate writer thread. --- src/lib/encoder.cc | 94 +++++++++++++++++++++++++++++++++++----------- src/lib/encoder.h | 14 ++++++- 2 files changed, 85 insertions(+), 23 deletions(-) diff --git a/src/lib/encoder.cc b/src/lib/encoder.cc index 693bd5bc8..93a364fed 100644 --- a/src/lib/encoder.cc +++ b/src/lib/encoder.cc @@ -60,7 +60,9 @@ Encoder::Encoder (shared_ptr f, shared_ptr o) , _swr_context (0) #endif , _audio_frames_written (0) - , _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 @@ -85,6 +87,7 @@ Encoder::~Encoder () { close_sound_files (); terminate_worker_threads (); + terminate_writer_thread (); } void @@ -130,6 +133,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)); } @@ -174,11 +179,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); } @@ -186,8 +191,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 @@ -198,7 +204,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 (); @@ -209,6 +215,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 (_opt->frame_out_path (i->first, false), _opt->frame_out_path (i->second, false)); @@ -295,13 +306,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; } @@ -320,8 +331,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, _opt->out_size, _opt->padding, _film->subtitle_offset(), _film->subtitle_scale(), _film->scaler(), _video_frame, _film->frames_per_second(), s.second, @@ -434,7 +445,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 (); @@ -444,6 +455,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) { @@ -457,18 +485,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 (); @@ -508,14 +536,15 @@ Encoder::encoder_thread (ServerDescription* server) } if (encoded) { - encoded->write (_opt, vf->frame ()); - frame_done (); + boost::mutex::scoped_lock lock (_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 (); } @@ -542,3 +571,26 @@ 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); + while (_write_queue.empty() && !_terminate_writer) { + _writer_condition.wait (lock); + } + + 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 52ccfc166..fff264963 100644 --- a/src/lib/encoder.h +++ b/src/lib/encoder.h @@ -51,6 +51,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: void terminate_worker_threads (); void link (std::string, std::string) const; + void writer_thread (); + void terminate_writer_thread (); + #if HAVE_SWRESAMPLE SwrContext* _swr_context; #endif @@ -133,11 +137,17 @@ private: int64_t _audio_frames_written; 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 -- 2.30.2