Speculative fix for hangs during Encoder::end().

[dcpomatic.git] / src / lib / encoder.cc

diff --git a/src/lib/encoder.cc b/src/lib/encoder.cc
index 195b10f443bf65f66508dd885bb74a17747de4c8..ef58ca09e6e71b85fb7c0c6e29d192fb3dfb468f 100644 (file)
--- a/src/lib/encoder.cc
+++ b/src/lib/encoder.cc
@@ -34,6 +34,7 @@
 #include "player.h"
 #include "player_video.h"
 #include "data.h"
+#include "server_description.h"
 #include "compose.hpp"
 #include <libcxml/cxml.h>
 #include <boost/foreach.hpp>
@@ -42,6 +43,7 @@
 #include "i18n.h"
 
 #define LOG_GENERAL(...) _film->log()->log (String::compose (__VA_ARGS__), Log::TYPE_GENERAL);
+#define LOG_GENERAL_NC(...) _film->log()->log (__VA_ARGS__, Log::TYPE_GENERAL);
 #define LOG_ERROR(...) _film->log()->log (String::compose (__VA_ARGS__), Log::TYPE_ERROR);
 #define LOG_TIMING(...) _film->log()->microsecond_log (String::compose (__VA_ARGS__), Log::TYPE_TIMING);
 
@@ -60,7 +62,8 @@ Encoder::Encoder (shared_ptr<const Film> film, weak_ptr<Job> j, shared_ptr<Write
        , _video_frames_enqueued (0)
        , _left_done (false)
        , _right_done (false)
-       , _terminate (false)
+       , _terminate_enqueue (false)
+       , _terminate_encoding (false)
        , _writer (writer)
 {
        servers_list_changed ();
@@ -69,6 +72,11 @@ Encoder::Encoder (shared_ptr<const Film> film, weak_ptr<Job> j, shared_ptr<Write
 Encoder::~Encoder ()
 {
        terminate_threads ();
+
+       boost::mutex::scoped_lock lm (_queue_mutex);
+       _terminate_enqueue = true;
+       _full_condition.notify_all ();
+       _empty_condition.notify_all ();
 }
 
 void
@@ -82,18 +90,21 @@ Encoder::begin ()
 void
 Encoder::end ()
 {
-       boost::mutex::scoped_lock lock (_mutex);
+       boost::mutex::scoped_lock lock (_queue_mutex);
 
        LOG_GENERAL (N_("Clearing queue of %1"), _queue.size ());
 
        /* Keep waking workers until the queue is empty */
        while (!_queue.empty ()) {
+               rethrow ();
                _empty_condition.notify_all ();
                _full_condition.wait (lock);
        }
 
        lock.unlock ();
 
+       LOG_GENERAL_NC (N_("Terminating encoder threads"));
+
        terminate_threads ();
 
        LOG_GENERAL (N_("Mopping up %1"), _queue.size());
@@ -171,18 +182,24 @@ Encoder::enqueue (shared_ptr<PlayerVideo> pv)
 {
        _waker.nudge ();
 
-       boost::mutex::scoped_lock lock (_mutex);
+       size_t threads = 0;
+       {
+               boost::mutex::scoped_lock threads_lock (_threads_mutex);
+               threads = _threads.size ();
+       }
+
+       boost::mutex::scoped_lock queue_lock (_queue_mutex);
 
        /* XXX: discard 3D here if required */
 
        /* Wait until the queue has gone down a bit */
-       while (_queue.size() >= _threads.size() * 2 && !_terminate) {
+       while (_queue.size() >= threads * 2 && !_terminate_enqueue) {
                LOG_TIMING ("decoder-sleep queue=%1", _queue.size());
-               _full_condition.wait (lock);
+               _full_condition.wait (queue_lock);
                LOG_TIMING ("decoder-wake queue=%1", _queue.size());
        }
 
-       if (_terminate) {
+       if (_terminate_enqueue) {
                return;
        }
 
@@ -248,12 +265,14 @@ void
 Encoder::terminate_threads ()
 {
        {
-               boost::mutex::scoped_lock lock (_mutex);
-               _terminate = true;
+               boost::mutex::scoped_lock queue_lock (_queue_mutex);
+               _terminate_encoding = true;
                _full_condition.notify_all ();
                _empty_condition.notify_all ();
        }
 
+       boost::mutex::scoped_lock threads_lock (_threads_mutex);
+
        for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
                if ((*i)->joinable ()) {
                        (*i)->join ();
@@ -262,7 +281,7 @@ Encoder::terminate_threads ()
        }
 
        _threads.clear ();
-       _terminate = false;
+       _terminate_encoding = false;
 }
 
 void
@@ -284,12 +303,12 @@ try
        while (true) {
 
                LOG_TIMING ("encoder-sleep thread=%1", boost::this_thread::get_id());
-               boost::mutex::scoped_lock lock (_mutex);
-               while (_queue.empty () && !_terminate) {
+               boost::mutex::scoped_lock lock (_queue_mutex);
+               while (_queue.empty () && !_terminate_encoding) {
                        _empty_condition.wait (lock);
                }
 
-               if (_terminate) {
+               if (_terminate_encoding) {
                        return;
                }
 
@@ -332,6 +351,7 @@ try
                                LOG_TIMING ("finish-local-encode thread=%1 frame=%2", boost::this_thread::get_id(), vf->index());
                        } catch (std::exception& e) {
                                LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
+                               throw;
                        }
                }
 
@@ -357,6 +377,8 @@ try
 catch (...)
 {
        store_current ();
+       /* Wake anything waiting on _full_condition so it can see the exception */
+       _full_condition.notify_all ();
 }
 
 void
@@ -366,6 +388,8 @@ Encoder::servers_list_changed ()
 
        /* XXX: could re-use threads */
 
+       boost::mutex::scoped_lock lm (_threads_mutex);
+
        if (!Config::instance()->only_servers_encode ()) {
                for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
                        _threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));