X-Git-Url: https://git.carlh.net/gitweb/?a=blobdiff_plain;f=src%2Flib%2Fj2k_encoder.cc;h=de229113b3a93234f9af47d91c17c1133ae8575f;hb=1fc8c0c6d045404732497ba70bd2eccfbe4cc6f6;hp=e2fca09b11112e37b80a1e0cdfb4fa3bf4f4db20;hpb=c80eca531e01860e6714f526b2efb07a58c2e26e;p=dcpomatic.git

diff --git a/src/lib/j2k_encoder.cc b/src/lib/j2k_encoder.cc
index e2fca09b1..de229113b 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,7 +60,33 @@ using boost::optional;
 using dcp::Data;
 using namespace dcpomatic;
 
-static grk_plugin::GrokInitializer grokInitializer;
+#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.
@@ -61,10 +94,15 @@ static grk_plugin::GrokInitializer grokInitializer;
 J2KEncoder::J2KEncoder(shared_ptr<const Film> film, Writer& writer)
 	: _film (film)
 	, _history (200)
-	, _writer (writer) ,
-	dcpomaticContext_(film,writer,_history, Config::instance()->gpu_binary_location ()),
-	context_(Config::instance()->enable_gpu () ? new grk_plugin::GrokContext(dcpomaticContext_) : nullptr)
+	, _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
 }
 
 
@@ -72,14 +110,32 @@ J2KEncoder::~J2KEncoder ()
 {
 	_server_found_connection.disconnect();
 
-	{
-	boost::mutex::scoped_lock lm (_threads_mutex);
-	terminate_threads ();
-	}
+	terminate_threads();
 
-	delete context_;
+#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());
+}
+
+
 void
 J2KEncoder::begin ()
 {
@@ -89,41 +145,57 @@ J2KEncoder::begin ()
 	servers_list_changed ();
 }
 
-void J2KEncoder::pause(void){
-	if (Config::instance()->enable_gpu ())
-		end(false);
+
+void
+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(void){
-	if (Config::instance()->enable_gpu ()) {
-		context_ = new grk_plugin::GrokContext(dcpomaticContext_);
-		servers_list_changed ();
+
+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 (bool isFinal)
+J2KEncoder::end()
 {
-	if (isFinal) {
-		boost::mutex::scoped_lock lock (_queue_mutex);
+	boost::mutex::scoped_lock lock (_queue_mutex);
 
-		LOG_GENERAL (N_("Clearing queue of %1"), _queue.size ());
+	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 ();
+	/* Keep waking workers until the queue is empty */
+	while (!_queue.empty ()) {
+		rethrow ();
+		_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 ();
@@ -131,38 +203,42 @@ J2KEncoder::end (bool isFinal)
 	LOG_GENERAL (N_("Mopping up %1"), _queue.size());
 
 	/* The following sequence of events can occur in the above code:
-		 1. a remote worker takes the last image off the queue
-		 2. the loop above terminates
-		 3. the remote worker fails to encode the image and puts it back on the queue
-		 4. the remote worker is then terminated by terminate_threads
+	     1. a remote worker takes the last image off the queue
+	     2. the loop above terminates
+	     3. the remote worker fails to encode the image and puts it back on the queue
+	     4. the remote worker is then terminated by terminate_threads
 
-		 So just mop up anything left in the queue here.
+	     So just mop up anything left in the queue here.
 	*/
-	if (isFinal) {
-		for (auto & i: _queue) {
-			if (Config::instance()->enable_gpu ()) {
-				if (!context_->scheduleCompress(i)){
-					LOG_GENERAL (N_("[%1] J2KEncoder thread pushes frame %2 back onto queue after failure"), thread_id(), i.index());
-					// handle error
-				}
+	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 {
-				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 ());
-				}
+		} 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 ());
 			}
 		}
 	}
-	delete context_;
-	context_ = nullptr;
+
+#ifdef DCPOMATIC_GROK
+	delete _context;
+	_context = nullptr;
+#endif
 }
 
 
@@ -212,10 +288,7 @@ J2KEncoder::encode (shared_ptr<PlayerVideo> pv, DCPTime time)
 	size_t threads = 0;
 	{
 		boost::mutex::scoped_lock lm (_threads_mutex);
-		if (_threads)
-			threads = _threads->size();
-		else
-			threads = std::thread::hardware_concurrency();
+		threads = _threads.size();
 	}
 
 	boost::mutex::scoped_lock queue_lock (_queue_mutex);
@@ -275,182 +348,141 @@ 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
+J2KEncoder::remake_threads(int cpu, int gpu, list<EncodeServerDescription> servers)
 {
-	auto config = Config::instance ();
+	LOG_GENERAL("Making threads: CPU=%1, GPU=%2, Remote=%3", cpu, gpu, servers.size());
 
-	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));
+	};
 
-	while (true) {
+	auto const current_gpu_threads = std::count_if(_threads.begin(), _threads.end(), is_grok_thread);
 
-		LOG_TIMING ("encoder-sleep thread=%1", thread_id ());
-		boost::mutex::scoped_lock lock (_queue_mutex);
-		while (_queue.empty ()) {
-			_empty_condition.wait (lock);
+	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
+
+	/* Remote */
+
+	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 {
-				if (context_) {
-					if (!context_->launch(vf, config->selected_gpu()) || !context_->scheduleCompress(vf)) {
-						LOG_GENERAL (N_("[%1] J2KEncoder thread pushes frame %2 back onto queue after failure"), thread_id(), vf.index());
-						_queue.push_front (vf);
-					}
-
-				} 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 {
-				if (!Config::instance()->enable_gpu ()) {
-					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();
 }