Merge branch 'grok2' into v2.17.xv2.17.8

This is the DoM support for Aaron Boxer's "grok" GPU J2K encoder,
with some cleanups and other assorted/related DoM changes.

1 files changed, 233 insertions, 154 deletions

diff --git a/src/lib/j2k_encoder.cc b/src/lib/j2k_encoder.cc
index d2e840f85..e68402483 100644
--- a/src/lib/j2k_encoder.cc
+++ b/src/lib/j2k_encoder.cc
@@ -32,6 +32,12 @@
 #include "encode_server_description.h"
 #include "encode_server_finder.h"
 #include "film.h"
+#include "cpu_j2k_encoder_thread.h"
+#ifdef DCPOMATIC_GROK
+#include "grok/context.h"
+#include "grok_j2k_encoder_thread.h"
+#endif
+#include "remote_j2k_encoder_thread.h"
 #include "j2k_encoder.h"
 #include "log.h"
 #include "player_video.h"
@@ -44,6 +50,7 @@
 
 
 using std::cout;
+using std::dynamic_pointer_cast;
 using std::exception;
 using std::list;
 using std::make_shared;
@@ -53,6 +60,33 @@ using boost::optional;
 using dcp::Data;
 using namespace dcpomatic;
 
+#ifdef DCPOMATIC_GROK
+
+namespace grk_plugin {
+
+IMessengerLogger* sLogger = nullptr;
+
+#if defined(__GNUC__) || defined(__clang__)
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wunused-function"
+#endif
+void setMessengerLogger(grk_plugin::IMessengerLogger* logger)
+{
+ 	delete sLogger;
+ 	sLogger = logger;
+}
+#if defined(__GNUC__) || defined(__clang__)
+#pragma GCC diagnostic pop
+#endif
+grk_plugin::IMessengerLogger* getMessengerLogger(void)
+{
+	return sLogger;
+}
+
+}
+
+#endif
+
 
 /** @param film Film that we are encoding.
  *  @param writer Writer that we are using.
@@ -62,6 +96,13 @@ J2KEncoder::J2KEncoder(shared_ptr<const Film> film, Writer& writer)
 	, _history (200)
 	, _writer (writer)
 {
+#ifdef DCPOMATIC_GROK
+	auto grok = Config::instance()->grok().get_value_or({});
+	_dcpomatic_context = new grk_plugin::DcpomaticContext(film, writer, _history, grok.binary_location);
+	if (grok.enable) {
+		_context = new grk_plugin::GrokContext(_dcpomatic_context);
+	}
+#endif
 	servers_list_changed ();
 }
 
@@ -70,8 +111,29 @@ J2KEncoder::~J2KEncoder ()
 {
 	_server_found_connection.disconnect();
 
-	boost::mutex::scoped_lock lm (_threads_mutex);
-	terminate_threads ();
+	terminate_threads();
+
+#ifdef DCPOMATIC_GROK
+	delete _context;
+	delete _dcpomatic_context;
+#endif
+}
+
+
+void
+J2KEncoder::servers_list_changed()
+{
+	auto config = Config::instance();
+#ifdef DCPOMATIC_GROK
+	auto const grok_enable = config->grok().get_value_or({}).enable;
+#else
+	auto const grok_enable = false;
+#endif
+
+	auto const cpu = (grok_enable || config->only_servers_encode()) ? 0 : config->master_encoding_threads();
+	auto const gpu = grok_enable ? config->master_encoding_threads() : 0;
+
+	remake_threads(cpu, gpu, EncodeServerFinder::instance()->servers());
 }
 
 
@@ -85,7 +147,40 @@ J2KEncoder::begin ()
 
 
 void
-J2KEncoder::end ()
+J2KEncoder::pause()
+{
+#ifdef DCPOMATIC_GROK
+	if (!Config::instance()->grok().get_value_or({}).enable) {
+		return;
+	}
+	return;
+
+	terminate_threads ();
+
+	/* Something might have been thrown during terminate_threads */
+	rethrow ();
+
+	delete _context;
+	_context = nullptr;
+#endif
+}
+
+
+void J2KEncoder::resume()
+{
+#ifdef DCPOMATIC_GROK
+	if (!Config::instance()->grok().get_value_or({}).enable) {
+		return;
+	}
+
+	_context = new grk_plugin::GrokContext(_dcpomatic_context);
+	servers_list_changed();
+#endif
+}
+
+
+void
+J2KEncoder::end()
 {
 	boost::mutex::scoped_lock lock (_queue_mutex);
 
@@ -94,18 +189,13 @@ J2KEncoder::end ()
 	/* 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"));
 
-	{
-		boost::mutex::scoped_lock lm (_threads_mutex);
-		terminate_threads ();
-	}
+	terminate_threads ();
 
 	/* Something might have been thrown during terminate_threads */
 	rethrow ();
@@ -120,20 +210,35 @@ J2KEncoder::end ()
 
 	     So just mop up anything left in the queue here.
 	*/
-
-	for (auto const& i: _queue) {
-		LOG_GENERAL(N_("Encode left-over frame %1"), i.index());
-		try {
-			_writer.write(
-				make_shared<dcp::ArrayData>(i.encode_locally()),
-				i.index(),
-				i.eyes()
-				);
-			frame_done ();
-		} catch (std::exception& e) {
-			LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
+	for (auto & i: _queue) {
+#ifdef DCPOMATIC_GROK
+		if (Config::instance()->grok().get_value_or({}).enable) {
+			if (!_context->scheduleCompress(i)){
+				LOG_GENERAL (N_("[%1] J2KEncoder thread pushes frame %2 back onto queue after failure"), thread_id(), i.index());
+				// handle error
+			}
+		} else {
+#else
+		{
+#endif
+			LOG_GENERAL(N_("Encode left-over frame %1"), i.index());
+			try {
+				_writer.write(
+					make_shared<dcp::ArrayData>(i.encode_locally()),
+					i.index(),
+					i.eyes()
+					);
+				frame_done ();
+			} catch (std::exception& e) {
+				LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
+			}
 		}
 	}
+
+#ifdef DCPOMATIC_GROK
+	delete _context;
+	_context = nullptr;
+#endif
 }
 
 
@@ -183,7 +288,7 @@ J2KEncoder::encode (shared_ptr<PlayerVideo> pv, DCPTime time)
 	size_t threads = 0;
 	{
 		boost::mutex::scoped_lock lm (_threads_mutex);
-		threads = _threads->size();
+		threads = _threads.size();
 	}
 
 	boost::mutex::scoped_lock queue_lock (_queue_mutex);
@@ -223,13 +328,14 @@ J2KEncoder::encode (shared_ptr<PlayerVideo> pv, DCPTime time)
 		LOG_DEBUG_ENCODE("Frame @ %1 ENCODE", to_string(time));
 		/* Queue this new frame for encoding */
 		LOG_TIMING ("add-frame-to-queue queue=%1", _queue.size ());
-		_queue.push_back (DCPVideo(
+		auto dcpv = DCPVideo(
 				pv,
 				position,
 				_film->video_frame_rate(),
 				_film->j2k_bandwidth(),
 				_film->resolution()
-				));
+				);
+		_queue.push_back (dcpv);
 
 		/* The queue might not be empty any more, so notify anything which is
 		   waiting on that.
@@ -242,170 +348,143 @@ J2KEncoder::encode (shared_ptr<PlayerVideo> pv, DCPTime time)
 }
 
 
-/** Caller must hold a lock on _threads_mutex */
 void
 J2KEncoder::terminate_threads ()
 {
+	boost::mutex::scoped_lock lm(_threads_mutex);
 	boost::this_thread::disable_interruption dis;
 
-	if (!_threads) {
-		return;
-	}
-
-	_threads->interrupt_all ();
-	try {
-		_threads->join_all ();
-	} catch (exception& e) {
-		LOG_ERROR ("join() threw an exception: %1", e.what());
-	} catch (...) {
-		LOG_ERROR_NC ("join() threw an exception");
+	for (auto& thread: _threads) {
+		thread->stop();
 	}
 
-	_threads.reset ();
+	_threads.clear();
+	_ending = true;
 }
 
 
 void
-J2KEncoder::encoder_thread (optional<EncodeServerDescription> server)
-try
+#ifdef DCPOMATIC_GROK
+J2KEncoder::remake_threads(int cpu, int gpu, list<EncodeServerDescription> servers)
+#else
+J2KEncoder::remake_threads(int cpu, int, list<EncodeServerDescription> servers)
+#endif
 {
-	start_of_thread ("J2KEncoder");
+	boost::mutex::scoped_lock lm (_threads_mutex);
+	if (_ending) {
+		return;
+	}
 
-	if (server) {
-		LOG_TIMING ("start-encoder-thread thread=%1 server=%2", thread_id (), server->host_name ());
-	} else {
-		LOG_TIMING ("start-encoder-thread thread=%1 server=localhost", thread_id ());
+	auto remove_threads = [this](int wanted, int current, std::function<bool (shared_ptr<J2KEncoderThread>)> predicate) {
+		for (auto i = wanted; i < current; ++i) {
+			auto iter = std::find_if(_threads.begin(), _threads.end(), predicate);
+			if (iter != _threads.end()) {
+				(*iter)->stop();
+				_threads.erase(iter);
+			}
+		}
+	};
+
+
+	/* CPU */
+
+	auto const is_cpu_thread = [](shared_ptr<J2KEncoderThread> thread) {
+		return static_cast<bool>(dynamic_pointer_cast<CPUJ2KEncoderThread>(thread));
+	};
+
+	auto const current_cpu_threads = std::count_if(_threads.begin(), _threads.end(), is_cpu_thread);
+
+	for (auto i = current_cpu_threads; i < cpu; ++i) {
+		auto thread = make_shared<CPUJ2KEncoderThread>(*this);
+		thread->start();
+		_threads.push_back(thread);
 	}
 
-	/* Number of seconds that we currently wait between attempts
-	   to connect to the server; not relevant for localhost
-	   encodings.
-	*/
-	int remote_backoff = 0;
+	remove_threads(cpu, current_cpu_threads, is_cpu_thread);
+
+#ifdef DCPOMATIC_GROK
+	/* GPU */
+
+	auto const is_grok_thread = [](shared_ptr<J2KEncoderThread> thread) {
+		return static_cast<bool>(dynamic_pointer_cast<GrokJ2KEncoderThread>(thread));
+	};
+
+	auto const current_gpu_threads = std::count_if(_threads.begin(), _threads.end(), is_grok_thread);
+
+	for (auto i = current_gpu_threads; i < gpu; ++i) {
+		auto thread = make_shared<GrokJ2KEncoderThread>(*this, _context);
+		thread->start();
+		_threads.push_back(thread);
+	}
+
+	remove_threads(gpu, current_gpu_threads, is_grok_thread);
+#endif
 
-	while (true) {
+	/* Remote */
 
-		LOG_TIMING ("encoder-sleep thread=%1", thread_id ());
-		boost::mutex::scoped_lock lock (_queue_mutex);
-		while (_queue.empty ()) {
-			_empty_condition.wait (lock);
+	for (auto const& server: servers) {
+		if (!server.current_link_version()) {
+			continue;
 		}
 
-		LOG_TIMING ("encoder-wake thread=%1 queue=%2", thread_id(), _queue.size());
-		auto vf = _queue.front ();
+		auto is_remote_thread = [server](shared_ptr<J2KEncoderThread> thread) {
+			auto remote = dynamic_pointer_cast<RemoteJ2KEncoderThread>(thread);
+			return remote && remote->server().host_name() == server.host_name();
+		};
 
-		/* We're about to commit to either encoding this frame or putting it back onto the queue,
-		   so we must not be interrupted until one or other of these things have happened.  This
-		   block has thread interruption disabled.
-		*/
-		{
-			boost::this_thread::disable_interruption dis;
-
-			LOG_TIMING ("encoder-pop thread=%1 frame=%2 eyes=%3", thread_id(), vf.index(), static_cast<int>(vf.eyes()));
-			_queue.pop_front ();
-
-			lock.unlock ();
-
-			shared_ptr<Data> encoded;
-
-			/* We need to encode this input */
-			if (server) {
-				try {
-					encoded = make_shared<dcp::ArrayData>(vf.encode_remotely(server.get()));
-
-					if (remote_backoff > 0) {
-						LOG_GENERAL ("%1 was lost, but now she is found; removing backoff", server->host_name ());
-					}
-
-					/* This job succeeded, so remove any backoff */
-					remote_backoff = 0;
-
-				} catch (std::exception& e) {
-					if (remote_backoff < 60) {
-						/* back off more */
-						remote_backoff += 10;
-					}
-					LOG_ERROR (
-						N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
-						vf.index(), server->host_name(), e.what(), remote_backoff
-						);
-				}
-
-			} else {
-				try {
-					LOG_TIMING ("start-local-encode thread=%1 frame=%2", thread_id(), vf.index());
-					encoded = make_shared<dcp::ArrayData>(vf.encode_locally());
-					LOG_TIMING ("finish-local-encode thread=%1 frame=%2", thread_id(), vf.index());
-				} catch (std::exception& e) {
-					/* This is very bad, so don't cope with it, just pass it on */
-					LOG_ERROR (N_("Local encode failed (%1)"), e.what ());
-					throw;
-				}
-			}
+		auto const current_threads = std::count_if(_threads.begin(), _threads.end(), is_remote_thread);
 
-			if (encoded) {
-				_writer.write(encoded, vf.index(), vf.eyes());
-				frame_done ();
-			} else {
-				lock.lock ();
-				LOG_GENERAL (N_("[%1] J2KEncoder thread pushes frame %2 back onto queue after failure"), thread_id(), vf.index());
-				_queue.push_front (vf);
-				lock.unlock ();
-			}
+		auto const wanted_threads = server.threads();
+
+		if (wanted_threads > current_threads) {
+			LOG_GENERAL(N_("Adding %1 worker threads for remote %2"), wanted_threads - current_threads, server.host_name());
+		} else if (wanted_threads < current_threads) {
+			LOG_GENERAL(N_("Removing %1 worker threads for remote %2"), current_threads - wanted_threads, server.host_name());
 		}
 
-		if (remote_backoff > 0) {
-			boost::this_thread::sleep (boost::posix_time::seconds (remote_backoff));
+		for (auto i = current_threads; i < wanted_threads; ++i) {
+			auto thread = make_shared<RemoteJ2KEncoderThread>(*this, server);
+			thread->start();
+			_threads.push_back(thread);
 		}
 
-		/* The queue might not be full any more, so notify anything that is waiting on that */
-		lock.lock ();
-		_full_condition.notify_all ();
+		remove_threads(wanted_threads, current_threads, is_remote_thread);
 	}
-}
-catch (boost::thread_interrupted& e) {
-	/* Ignore these and just stop the thread */
-	_full_condition.notify_all ();
-}
-catch (...)
-{
-	store_current ();
-	/* Wake anything waiting on _full_condition so it can see the exception */
-	_full_condition.notify_all ();
+
+	_writer.set_encoder_threads(_threads.size());
 }
 
 
-void
-J2KEncoder::servers_list_changed ()
+DCPVideo
+J2KEncoder::pop()
 {
-	boost::mutex::scoped_lock lm (_threads_mutex);
+	boost::mutex::scoped_lock lock(_queue_mutex);
+	while (_queue.empty()) {
+		_empty_condition.wait (lock);
+	}
 
-	terminate_threads ();
-	_threads = make_shared<boost::thread_group>();
+	LOG_TIMING("encoder-wake thread=%1 queue=%2", thread_id(), _queue.size());
 
-	/* XXX: could re-use threads */
+	auto vf = _queue.front();
+	_queue.pop_front();
 
-	if (!Config::instance()->only_servers_encode ()) {
-		for (int i = 0; i < Config::instance()->master_encoding_threads (); ++i) {
-#ifdef DCPOMATIC_LINUX
-			auto t = _threads->create_thread(boost::bind(&J2KEncoder::encoder_thread, this, optional<EncodeServerDescription>()));
-			pthread_setname_np (t->native_handle(), "encode-worker");
-#else
-			_threads->create_thread(boost::bind(&J2KEncoder::encoder_thread, this, optional<EncodeServerDescription>()));
-#endif
-		}
-	}
+	_full_condition.notify_all();
+	return vf;
+}
 
-	for (auto i: EncodeServerFinder::instance()->servers()) {
-		if (!i.current_link_version()) {
-			continue;
-		}
 
-		LOG_GENERAL (N_("Adding %1 worker threads for remote %2"), i.threads(), i.host_name ());
-		for (int j = 0; j < i.threads(); ++j) {
-			_threads->create_thread(boost::bind(&J2KEncoder::encoder_thread, this, i));
-		}
-	}
+void
+J2KEncoder::retry(DCPVideo video)
+{
+	boost::mutex::scoped_lock lock(_queue_mutex);
+	_queue.push_front(video);
+	_empty_condition.notify_all();
+}
 
-	_writer.set_encoder_threads(_threads->size());
+
+void
+J2KEncoder::write(shared_ptr<const dcp::Data> data, int index, Eyes eyes)
+{
+	_writer.write(data, index, eyes);
+	frame_done();
 }