--- /dev/null
+/*
+ Copyright (C) 2023 Carl Hetherington <cth@carlh.net>
+
+ This file is part of DCP-o-matic.
+
+ DCP-o-matic is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+
+ DCP-o-matic is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with DCP-o-matic. If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "cpu_j2k_encoder_thread.h"
+#include "cross.h"
+#include "dcpomatic_log.h"
+#include "dcp_video.h"
+#include "j2k_encoder.h"
+#include "util.h"
+
+#include "i18n.h"
+
+
+using std::make_shared;
+using std::shared_ptr;
+
+
+CPUJ2KEncoderThread::CPUJ2KEncoderThread(J2KEncoder& encoder)
+ : J2KSyncEncoderThread(encoder)
+{
+
+}
+
+
+void
+CPUJ2KEncoderThread::log_thread_start() const
+{
+ start_of_thread("CPUJ2KEncoder");
+ LOG_TIMING("start-encoder-thread thread=%1 server=localhost", thread_id());
+}
+
+
+shared_ptr<dcp::ArrayData>
+CPUJ2KEncoderThread::encode(DCPVideo const& frame)
+{
+ try {
+ return make_shared<dcp::ArrayData>(frame.encode_locally());
+ } catch (std::exception& e) {
+ LOG_ERROR(N_("Local encode failed (%1)"), e.what());
+ }
+
+ return {};
+}
+
--- /dev/null
+#include "j2k_sync_encoder_thread.h"
+#include <dcp/data.h>
+
+
+class DCPVideo;
+
+
+class CPUJ2KEncoderThread : public J2KSyncEncoderThread
+{
+public:
+ CPUJ2KEncoderThread(J2KEncoder& encoder);
+
+ void log_thread_start() const override;
+ std::shared_ptr<dcp::ArrayData> encode(DCPVideo const& frame) override;
+};
+
class Player;
class PlayerVideo;
+struct frames_not_lost_when_threads_disappear;
+
/** @class DCPEncoder */
class DCPEncoder : public Encoder
private:
+ friend struct ::frames_not_lost_when_threads_disappear;
+
void video (std::shared_ptr<PlayerVideo>, dcpomatic::DCPTime);
void audio (std::shared_ptr<AudioBuffers>, dcpomatic::DCPTime);
void text (PlayerText, TextType, boost::optional<DCPTextTrack>, dcpomatic::DCPTimePeriod);
#endif
, _verbose (verbose)
, _num_threads (num_threads)
+ , _frames_encoded(0)
{
}
throw;
}
+ ++_frames_encoded;
+
return dcp_video_frame.index ();
}
#include "exception_store.h"
#include "server.h"
#include <boost/asio.hpp>
+#include <boost/atomic.hpp>
#include <boost/thread.hpp>
#include <boost/thread/condition.hpp>
#include <string>
void run () override;
+ int frames_encoded() const {
+ return _frames_encoded;
+ }
+
private:
void handle (std::shared_ptr<Socket>) override;
void worker_thread ();
bool _verbose;
int _num_threads;
Waker _waker;
+ boost::atomic<int> _frames_encoded;
struct Broadcast {
--- /dev/null
+/*
+ Copyright (C) 2023 Carl Hetherington <cth@carlh.net>
+
+ This file is part of DCP-o-matic.
+
+ DCP-o-matic is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+
+ DCP-o-matic is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with DCP-o-matic. If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "config.h"
+#include "cross.h"
+#include "dcpomatic_log.h"
+#include "dcp_video.h"
+#include "grok_j2k_encoder_thread.h"
+#include "j2k_encoder.h"
+#include "util.h"
+#include <dcp/scope_guard.h>
+
+#include "i18n.h"
+
+
+using std::make_shared;
+using std::shared_ptr;
+
+
+GrokJ2KEncoderThread::GrokJ2KEncoderThread(J2KEncoder& encoder, grk_plugin::GrokContext* context)
+ : J2KEncoderThread(encoder)
+ , _context(context)
+{
+
+}
+
+
+void
+GrokJ2KEncoderThread::run()
+try
+{
+ while (true)
+ {
+ LOG_TIMING("encoder-sleep thread=%1", thread_id());
+ auto frame = _encoder.pop();
+
+<<<<<<< HEAD
+ ScopeGuard frame_guard([this, &frame]() {
+||||||| parent of 04d2316ac (fixup! Rearrange encoder threading.)
+ ScopeGuard frame_guard([this, &frame]() {
+ LOG_ERROR("Failed to schedule encode of %1 using grok", frame.index());
+=======
+ dcp::ScopeGuard frame_guard([this, &frame]() {
+ LOG_ERROR("Failed to schedule encode of %1 using grok", frame.index());
+>>>>>>> 04d2316ac (fixup! Rearrange encoder threading.)
+ _encoder.retry(frame);
+ });
+
+ LOG_TIMING("encoder-pop thread=%1 frame=%2 eyes=%3", thread_id(), frame.index(), static_cast<int>(frame.eyes()));
+
+ if (_context->launch(frame, Config::instance()->selected_gpu()) && _context->scheduleCompress(frame)) {
+ frame_guard.cancel();
+ }
+ }
+} catch (boost::thread_interrupted& e) {
+} catch (...) {
+ store_current();
+}
--- /dev/null
+/*
+ Copyright (C) 2023 Carl Hetherington <cth@carlh.net>
+
+ This file is part of DCP-o-matic.
+
+ DCP-o-matic is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+
+ DCP-o-matic is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with DCP-o-matic. If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "exception_store.h"
+#include "j2k_encoder_thread.h"
+
+
+namespace grk_plugin {
+ class GrokContext;
+}
+
+
+class GrokJ2KEncoderThread : public J2KEncoderThread, public ExceptionStore
+{
+public:
+ GrokJ2KEncoderThread(J2KEncoder& encoder, grk_plugin::GrokContext* context);
+
+ void run() override;
+
+private:
+ grk_plugin::GrokContext* _context;
+};
+
#include "encode_server_description.h"
#include "encode_server_finder.h"
#include "film.h"
+#include "cpu_j2k_encoder_thread.h"
+#include "grok_j2k_encoder_thread.h"
+#include "remote_j2k_encoder_thread.h"
#include "j2k_encoder.h"
#include "log.h"
#include "player_video.h"
using std::cout;
+using std::dynamic_pointer_cast;
using std::exception;
using std::list;
using std::make_shared;
using dcp::Data;
using namespace dcpomatic;
+
static grk_plugin::GrokInitializer grokInitializer;
/** @param film Film that we are encoding.
{
_server_found_connection.disconnect();
- {
- boost::mutex::scoped_lock lm(_threads_mutex);
- terminate_threads();
- }
+ terminate_threads();
delete _context;
}
+
+void
+J2KEncoder::servers_list_changed()
+{
+ auto config = Config::instance();
+
+ auto const cpu = (config->enable_gpu() || config->only_servers_encode()) ? 0 : config->master_encoding_threads();
+ auto const gpu = config->enable_gpu() ? config->master_encoding_threads() : 0;
+
+ remake_threads(cpu, gpu, EncodeServerFinder::instance()->servers());
+}
+
+
void
J2KEncoder::begin ()
{
return;
}
- {
- boost::mutex::scoped_lock lm (_threads_mutex);
- terminate_threads ();
- }
+ terminate_threads ();
/* Something might have been thrown during terminate_threads */
rethrow ();
/* 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 ();
LOG_GENERAL(N_("Encode left-over frame %1"), i.index());
try {
_writer.write(
- make_shared<dcp::ArrayData>(i.encode_locally()),
+ make_shared<dcp::ArrayData>(i.encode_locally()),
i.index(),
i.eyes()
);
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);
}
-/** 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 ();
+ boost::mutex::scoped_lock lm (_threads_mutex);
+ if (_ending) {
+ return;
+ }
- start_of_thread ("J2KEncoder");
+ 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);
+ }
+ }
+ };
- 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 ());
+
+ /* 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);
+
- while (true) {
+ /* GPU */
- LOG_TIMING ("encoder-sleep thread=%1", thread_id ());
- boost::mutex::scoped_lock lock (_queue_mutex);
- while (_queue.empty ()) {
- _empty_condition.wait (lock);
+ 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);
+
+
+ /* 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);
}
+
+ _writer.set_encoder_threads(_threads.size());
}
-catch (boost::thread_interrupted& e) {
- /* Ignore these and just stop the thread */
- _full_condition.notify_all ();
-}
-catch (...)
+
+
+DCPVideo
+J2KEncoder::pop()
{
- store_current ();
- /* Wake anything waiting on _full_condition so it can see the exception */
- _full_condition.notify_all ();
+ boost::mutex::scoped_lock lock(_queue_mutex);
+ while (_queue.empty()) {
+ _empty_condition.wait (lock);
+ }
+
+ LOG_TIMING("encoder-wake thread=%1 queue=%2", thread_id(), _queue.size());
+
+ auto vf = _queue.front();
+ _queue.pop_front();
+
+ _full_condition.notify_all();
+ return vf;
}
void
-J2KEncoder::servers_list_changed ()
+J2KEncoder::retry(DCPVideo video)
{
- boost::mutex::scoped_lock lm (_threads_mutex);
-
- terminate_threads ();
- _threads = make_shared<boost::thread_group>();
-
- /* XXX: could re-use threads */
-
- 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
- }
- }
-
- for (auto i: EncodeServerFinder::instance()->servers()) {
- if (!i.current_link_version()) {
- continue;
- }
+ boost::mutex::scoped_lock lock(_queue_mutex);
+ _queue.push_front(video);
+ _empty_condition.notify_all();
+}
- 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));
- }
- }
- _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();
}
#include "enum_indexed_vector.h"
#include "event_history.h"
#include "exception_store.h"
+#include "j2k_encoder_thread.h"
#include "writer.h"
#include <boost/optional.hpp>
#include <boost/signals2.hpp>
class Job;
class PlayerVideo;
+struct local_threads_created_and_destroyed;
+struct remote_threads_created_and_destroyed;
+struct frames_not_lost_when_threads_disappear;
+
/** @class J2KEncoder
* @brief Class to manage encoding to J2K.
boost::optional<float> current_encoding_rate () const;
int video_frames_enqueued () const;
- void servers_list_changed ();
+ DCPVideo pop();
+ void retry(DCPVideo frame);
+ void write(std::shared_ptr<const dcp::Data> data, int index, Eyes eyes);
private:
+ friend struct ::local_threads_created_and_destroyed;
+ friend struct ::remote_threads_created_and_destroyed;
+ friend struct ::frames_not_lost_when_threads_disappear;
void frame_done ();
-
- void encoder_thread (boost::optional<EncodeServerDescription>);
+ void servers_list_changed ();
+ void remake_threads(int cpu, int gpu, std::list<EncodeServerDescription> servers);
void terminate_threads ();
/** Film that we are encoding */
EventHistory _history;
boost::mutex _threads_mutex;
- std::shared_ptr<boost::thread_group> _threads;
+ std::vector<std::shared_ptr<J2KEncoderThread>> _threads;
mutable boost::mutex _queue_mutex;
std::list<DCPVideo> _queue;
grk_plugin::DcpomaticContext _dcpomatic_context;
grk_plugin::GrokContext *_context;
+
+ bool _ending = false;
};
--- /dev/null
+/*
+ Copyright (C) 2023 Carl Hetherington <cth@carlh.net>
+
+ This file is part of DCP-o-matic.
+
+ DCP-o-matic is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+
+ DCP-o-matic is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with DCP-o-matic. If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "dcp_video.h"
+#include "dcpomatic_log.h"
+#include "j2k_encoder.h"
+#include "j2k_encoder_thread.h"
+
+
+J2KEncoderThread::J2KEncoderThread(J2KEncoder& encoder)
+ : _encoder(encoder)
+{
+
+}
+
+
+void
+J2KEncoderThread::start()
+{
+ _thread = boost::thread(boost::bind(&J2KEncoderThread::run, this));
+#ifdef DCPOMATIC_LINUX
+ pthread_setname_np(_thread.native_handle(), "encode-worker");
+#endif
+}
+
+
+void
+J2KEncoderThread::stop()
+{
+ _thread.interrupt();
+ try {
+ _thread.join();
+ } catch (std::exception& e) {
+ LOG_ERROR("join() threw an exception: %1", e.what());
+ } catch (...) {
+ LOG_ERROR_NC("join() threw an exception");
+ }
+}
+
+
--- /dev/null
+/*
+ Copyright (C) 2023 Carl Hetherington <cth@carlh.net>
+
+ This file is part of DCP-o-matic.
+
+ DCP-o-matic is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+
+ DCP-o-matic is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with DCP-o-matic. If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#ifndef DCPOMATIC_J2K_ENCODER_THREAD
+#define DCPOMATIC_J2K_ENCODER_THREAD
+
+
+#include <boost/thread.hpp>
+
+
+class J2KEncoder;
+
+
+class J2KEncoderThread
+{
+public:
+ J2KEncoderThread(J2KEncoder& encoder);
+
+ J2KEncoderThread(J2KEncoderThread const&) = delete;
+ J2KEncoderThread& operator=(J2KEncoderThread const&) = delete;
+
+ void start();
+ void stop();
+
+ virtual void run() = 0;
+
+protected:
+ J2KEncoder& _encoder;
+
+private:
+ boost::thread _thread;
+};
+
+
+#endif
--- /dev/null
+/*
+ Copyright (C) 2023 Carl Hetherington <cth@carlh.net>
+
+ This file is part of DCP-o-matic.
+
+ DCP-o-matic is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+
+ DCP-o-matic is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with DCP-o-matic. If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "dcp_video.h"
+#include "dcpomatic_log.h"
+#include "j2k_encoder.h"
+#include "j2k_sync_encoder_thread.h"
+#include <dcp/scope_guard.h>
+
+
+J2KSyncEncoderThread::J2KSyncEncoderThread(J2KEncoder& encoder)
+ : J2KEncoderThread(encoder)
+{
+
+}
+
+
+void
+J2KSyncEncoderThread::run()
+try
+{
+ log_thread_start();
+
+ while (true) {
+ LOG_TIMING("encoder-sleep thread=%1", thread_id());
+ auto frame = _encoder.pop();
+
+ dcp::ScopeGuard frame_guard([this, &frame]() {
+ boost::this_thread::disable_interruption dis;
+ _encoder.retry(frame);
+ });
+
+ LOG_TIMING("encoder-pop thread=%1 frame=%2 eyes=%3", thread_id(), frame.index(), static_cast<int>(frame.eyes()));
+
+ auto encoded = encode(frame);
+
+ if (encoded) {
+ boost::this_thread::disable_interruption dis;
+ frame_guard.cancel();
+ _encoder.write(encoded, frame.index(), frame.eyes());
+ }
+ }
+} catch (boost::thread_interrupted& e) {
+} catch (...) {
+ store_current();
+}
+
--- /dev/null
+#ifndef DCPOMATIC_J2K_SYNC_ENCODER_THREAD_H
+#define DCPOMATIC_J2K_SYNC_ENCODER_THREAD_H
+
+
+#include "exception_store.h"
+#include "j2k_encoder_thread.h"
+#include <dcp/array_data.h>
+#include <boost/thread.hpp>
+
+
+class DCPVideo;
+class J2KEncoder;
+
+
+class J2KSyncEncoderThread : public J2KEncoderThread, public ExceptionStore
+{
+public:
+ J2KSyncEncoderThread(J2KEncoder& encoder);
+
+ J2KSyncEncoderThread(J2KSyncEncoderThread const&) = delete;
+ J2KSyncEncoderThread& operator=(J2KSyncEncoderThread const&) = delete;
+
+ virtual ~J2KSyncEncoderThread() {}
+
+ void run() override;
+
+ virtual void log_thread_start() const = 0;
+ virtual std::shared_ptr<dcp::ArrayData> encode(DCPVideo const& frame) = 0;
+};
+
+
+#endif
using std::string;
-/** Add suitable Jobs to the JobManager to create a DCP for a Film */
-void
+/** Add suitable Job to the JobManager to create a DCP for a Film */
+shared_ptr<TranscodeJob>
make_dcp (shared_ptr<Film> film, TranscodeJob::ChangedBehaviour behaviour)
{
if (film->dcp_name().find("/") != string::npos) {
auto tj = make_shared<DCPTranscodeJob>(film, behaviour);
tj->set_encoder (make_shared<DCPEncoder>(film, tj));
JobManager::instance()->add (tj);
+
+ return tj;
}
class Film;
-void make_dcp (std::shared_ptr<Film> film, TranscodeJob::ChangedBehaviour behaviour);
+std::shared_ptr<TranscodeJob> make_dcp(std::shared_ptr<Film> film, TranscodeJob::ChangedBehaviour behaviour);
--- /dev/null
+/*
+ Copyright (C) 2023 Carl Hetherington <cth@carlh.net>
+
+ This file is part of DCP-o-matic.
+
+ DCP-o-matic is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+
+ DCP-o-matic is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with DCP-o-matic. If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "dcp_video.h"
+#include "dcpomatic_log.h"
+#include "j2k_encoder.h"
+#include "remote_j2k_encoder_thread.h"
+#include "util.h"
+
+#include "i18n.h"
+
+
+using std::make_shared;
+using std::shared_ptr;
+
+
+RemoteJ2KEncoderThread::RemoteJ2KEncoderThread(J2KEncoder& encoder, EncodeServerDescription server)
+ : J2KSyncEncoderThread(encoder)
+ , _server(server)
+{
+
+}
+
+
+void
+RemoteJ2KEncoderThread::log_thread_start() const
+{
+ start_of_thread("RemoteJ2KEncoder");
+ LOG_TIMING("start-encoder-thread thread=%1 server=%2", thread_id(), _server.host_name());
+}
+
+
+shared_ptr<dcp::ArrayData>
+RemoteJ2KEncoderThread::encode(DCPVideo const& frame)
+{
+ shared_ptr<dcp::ArrayData> encoded;
+
+ try {
+ encoded = make_shared<dcp::ArrayData>(frame.encode_remotely(_server));
+ if (_remote_backoff > 0) {
+ LOG_GENERAL("%1 was lost, but now she is found; removing backoff", _server.host_name());
+ _remote_backoff = 0;
+ }
+ } catch (std::exception& e) {
+ LOG_ERROR(
+ N_("Remote encode of %1 on %2 failed (%3); thread sleeping for %4s"),
+ frame.index(), _server.host_name(), e.what(), _remote_backoff
+ );
+ } catch (...) {
+ LOG_ERROR(
+ N_("Remote encode of %1 on %2 failed; thread sleeping for %4s"),
+ frame.index(), _server.host_name(), _remote_backoff
+ );
+ }
+
+ if (!encoded) {
+ if (_remote_backoff < 60) {
+ /* back off more */
+ _remote_backoff += 10;
+ }
+ boost::this_thread::sleep(boost::posix_time::seconds(_remote_backoff));
+ }
+
+ return encoded;
+}
+
--- /dev/null
+#include "encode_server_description.h"
+#include "j2k_sync_encoder_thread.h"
+
+
+class RemoteJ2KEncoderThread : public J2KSyncEncoderThread
+{
+public:
+ RemoteJ2KEncoderThread(J2KEncoder& encoder, EncodeServerDescription server);
+
+ void log_thread_start() const override;
+ std::shared_ptr<dcp::ArrayData> encode(DCPVideo const& frame) override;
+
+ EncodeServerDescription server() const {
+ return _server;
+ }
+
+private:
+ EncodeServerDescription _server;
+ /** Number of seconds that we currently wait between attempts to connect to the server */
+ int _remote_backoff = 0;
+};
class Encoder;
+struct frames_not_lost_when_threads_disappear;
+
/** @class TranscodeJob
* @brief A job which transcodes a Film to another format.
void set_encoder (std::shared_ptr<Encoder> t);
private:
+ friend struct ::frames_not_lost_when_threads_disappear;
+
virtual void post_transcode () {}
float frames_per_second() const;
content_factory.cc
combine_dcp_job.cc
copy_dcp_details_to_film.cc
+ cpu_j2k_encoder_thread.cc
create_cli.cc
crop.cc
cross_common.cc
font_id_map.cc
frame_interval_checker.cc
frame_rate_change.cc
+ grok_j2k_encoder_thread.cc
guess_crop.cc
hints.cc
internet.cc
job.cc
job_manager.cc
j2k_encoder.cc
+ j2k_encoder_thread.cc
+ j2k_sync_encoder_thread.cc
json_server.cc
kdm_cli.cc
kdm_recipient.cc
referenced_reel_asset.cc
release_notes.cc
render_text.cc
+ remote_j2k_encoder_thread.cc
resampler.cc
resolution.cc
rgba.cc
/** @file test/client_server_test.cc
- * @brief Test the server class.
+ * @brief Test the remote encoding code.
* @ingroup feature
- *
- * Create a test image and then encode it using the standard mechanism
- * and also using a EncodeServer object running on localhost. Compare the resulting
- * encoded data to check that they are the same.
*/
+#include "lib/content_factory.h"
#include "lib/cross.h"
#include "lib/dcp_video.h"
#include "lib/dcpomatic_log.h"
#include "lib/encode_server.h"
#include "lib/encode_server_description.h"
+#include "lib/encode_server_finder.h"
#include "lib/file_log.h"
#include "lib/image.h"
#include "lib/j2k_image_proxy.h"
}
+BOOST_AUTO_TEST_CASE(real_encode_with_server)
+{
+ auto content = content_factory(TestPaths::private_data() / "dolby_aurora.vob");
+ auto film = new_test_film2("real_encode_with_server", content);
+
+ EncodeServerFinder::instance();
+
+ EncodeServer server(true, 4);
+ thread server_thread(boost::bind(&EncodeServer::run, &server));
+
+ make_and_verify_dcp(film);
+
+ server.stop();
+ server_thread.join();
+
+ BOOST_CHECK(server.frames_encoded() > 0);
+ EncodeServerFinder::drop();
+}
+
--- /dev/null
+/*
+ Copyright (C) 2023 Carl Hetherington <cth@carlh.net>
+
+ This file is part of DCP-o-matic.
+
+ DCP-o-matic is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 2 of the License, or
+ (at your option) any later version.
+
+ DCP-o-matic is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with DCP-o-matic. If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "lib/config.h"
+#include "lib/content_factory.h"
+#include "lib/dcp_encoder.h"
+#include "lib/dcp_transcode_job.h"
+#include "lib/encode_server_description.h"
+#include "lib/film.h"
+#include "lib/j2k_encoder.h"
+#include "lib/job_manager.h"
+#include "lib/make_dcp.h"
+#include "lib/transcode_job.h"
+#include "test.h"
+#include <dcp/cpl.h>
+#include <dcp/dcp.h>
+#include <dcp/reel.h>
+#include <dcp/reel_picture_asset.h>
+#include <boost/test/unit_test.hpp>
+
+
+using std::dynamic_pointer_cast;
+using std::list;
+
+
+BOOST_AUTO_TEST_CASE(local_threads_created_and_destroyed)
+{
+ auto film = new_test_film2("local_threads_created_and_destroyed", {});
+ Writer writer(film, {});
+ J2KEncoder encoder(film, writer);
+
+ encoder.remake_threads(32, 0, {});
+ BOOST_CHECK_EQUAL(encoder._threads.size(), 32U);
+
+ encoder.remake_threads(9, 0, {});
+ BOOST_CHECK_EQUAL(encoder._threads.size(), 9U);
+
+ encoder.end();
+ BOOST_CHECK_EQUAL(encoder._threads.size(), 0U);
+}
+
+
+BOOST_AUTO_TEST_CASE(remote_threads_created_and_destroyed)
+{
+ auto film = new_test_film2("remote_threads_created_and_destroyed", {});
+ Writer writer(film, {});
+ J2KEncoder encoder(film, writer);
+
+ list<EncodeServerDescription> servers = {
+ { "fred", 7, SERVER_LINK_VERSION },
+ { "jim", 2, SERVER_LINK_VERSION },
+ { "sheila", 14, SERVER_LINK_VERSION },
+ };
+
+ encoder.remake_threads(0, 0, servers);
+ BOOST_CHECK_EQUAL(encoder._threads.size(), 7U + 2U + 14U);
+
+ servers = {
+ { "fred", 7, SERVER_LINK_VERSION },
+ { "jim", 5, SERVER_LINK_VERSION },
+ { "sheila", 14, SERVER_LINK_VERSION },
+ };
+
+ encoder.remake_threads(0, 0, servers);
+ BOOST_CHECK_EQUAL(encoder._threads.size(), 7U + 5U + 14U);
+
+ servers = {
+ { "fred", 0, SERVER_LINK_VERSION },
+ { "jim", 0, SERVER_LINK_VERSION },
+ { "sheila", 11, SERVER_LINK_VERSION },
+ };
+
+ encoder.remake_threads(0, 0, servers);
+ BOOST_CHECK_EQUAL(encoder._threads.size(), 11U);
+}
+
+
+BOOST_AUTO_TEST_CASE(frames_not_lost_when_threads_disappear)
+{
+ auto content = content_factory(TestPaths::private_data() / "clapperboard.mp4");
+ auto film = new_test_film2("frames_not_lost", content);
+ film->write_metadata();
+ auto job = make_dcp(film, TranscodeJob::ChangedBehaviour::IGNORE);
+ auto& encoder = dynamic_pointer_cast<DCPEncoder>(job->_encoder)->_j2k_encoder;
+
+ while (JobManager::instance()->work_to_do()) {
+ encoder.remake_threads(rand() % 8, 0, {});
+ dcpomatic_sleep_seconds(1);
+ }
+
+ BOOST_CHECK(!JobManager::instance()->errors());
+
+ dcp::DCP dcp(film->dir(film->dcp_name()));
+ dcp.read();
+ BOOST_REQUIRE_EQUAL(dcp.cpls().size(), 1U);
+ BOOST_REQUIRE_EQUAL(dcp.cpls()[0]->reels().size(), 1U);
+ BOOST_REQUIRE_EQUAL(dcp.cpls()[0]->reels()[0]->main_picture()->intrinsic_duration(), 423U);
+}
+
interrupt_encoder_test.cc
isdcf_name_test.cc
j2k_bandwidth_test.cc
+ j2k_encode_threading_test.cc
job_manager_test.cc
kdm_cli_test.cc
kdm_naming_test.cc
projects / dcpomatic.git / commitdiff