X-Git-Url: https://git.carlh.net/gitweb/?p=dcpomatic.git;a=blobdiff_plain;f=src%2Flib%2Fjob_manager.cc;h=608df7ef0af9063729824f7c3967d6d76b7a02fe;hp=5fcb7086cf4bf586170ed2a74e8b2cff4f1213e7;hb=182b9d2e2feb6545592868606aaf0f0146095481;hpb=5892dc2cefde50516c27a3f630ee9fb483d07d20 diff --git a/src/lib/job_manager.cc b/src/lib/job_manager.cc index 5fcb7086c..608df7ef0 100644 --- a/src/lib/job_manager.cc +++ b/src/lib/job_manager.cc @@ -1,5 +1,5 @@ /* - Copyright (C) 2012-2018 Carl Hetherington + Copyright (C) 2012-2021 Carl Hetherington This file is part of DCP-o-matic. @@ -18,51 +18,56 @@ */ + /** @file src/job_manager.cc * @brief A simple scheduler for jobs. */ -#include "job_manager.h" -#include "job.h" -#include "cross.h" + #include "analyse_audio_job.h" +#include "analyse_subtitles_job.h" +#include "cross.h" #include "film.h" +#include "job.h" +#include "job_manager.h" #include -#include -#include -using std::string; + +using std::dynamic_pointer_cast; +using std::function; using std::list; -using std::cout; -using boost::shared_ptr; -using boost::weak_ptr; -using boost::function; -using boost::dynamic_pointer_cast; -using boost::optional; +using std::make_shared; +using std::shared_ptr; +using std::string; +using std::weak_ptr; using boost::bind; +using boost::optional; + + +JobManager* JobManager::_instance = nullptr; -JobManager* JobManager::_instance = 0; JobManager::JobManager () - : _terminate (false) - , _paused (false) - , _scheduler (0) { } + void JobManager::start () { - _scheduler = new boost::thread (boost::bind (&JobManager::scheduler, this)); + _scheduler = boost::thread (boost::bind(&JobManager::scheduler, this)); #ifdef DCPOMATIC_LINUX - pthread_setname_np (_scheduler->native_handle(), "job-scheduler"); + pthread_setname_np (_scheduler.native_handle(), "job-scheduler"); #endif } + JobManager::~JobManager () { - BOOST_FOREACH (boost::signals2::connection& i, _connections) { + boost::this_thread::disable_interruption dis; + + for (auto& i: _connections) { i.disconnect (); } @@ -72,18 +77,12 @@ JobManager::~JobManager () _empty_condition.notify_all (); } - if (_scheduler) { - /* Ideally this would be a DCPOMATIC_ASSERT(_scheduler->joinable()) but we - can't throw exceptions from a destructor. - */ - if (_scheduler->joinable ()) { - _scheduler->join (); - } - } - - delete _scheduler; + try { + _scheduler.join(); + } catch (...) {} } + shared_ptr JobManager::add (shared_ptr j) { @@ -93,39 +92,42 @@ JobManager::add (shared_ptr j) _empty_condition.notify_all (); } - emit (boost::bind (boost::ref (JobAdded), weak_ptr (j))); + emit (boost::bind(boost::ref(JobAdded), weak_ptr(j))); return j; } + shared_ptr JobManager::add_after (shared_ptr after, shared_ptr j) { { boost::mutex::scoped_lock lm (_mutex); - list >::iterator i = find (_jobs.begin(), _jobs.end(), after); + auto i = find (_jobs.begin(), _jobs.end(), after); DCPOMATIC_ASSERT (i != _jobs.end()); _jobs.insert (i, j); _empty_condition.notify_all (); } - emit (boost::bind (boost::ref (JobAdded), weak_ptr (j))); + emit (boost::bind(boost::ref(JobAdded), weak_ptr(j))); return j; } -list > + +list> JobManager::get () const { boost::mutex::scoped_lock lm (_mutex); return _jobs; } + bool JobManager::work_to_do () const { boost::mutex::scoped_lock lm (_mutex); - list >::const_iterator i = _jobs.begin(); + auto i = _jobs.begin(); while (i != _jobs.end() && (*i)->finished()) { ++i; } @@ -133,12 +135,13 @@ JobManager::work_to_do () const return i != _jobs.end (); } + bool JobManager::errors () const { boost::mutex::scoped_lock lm (_mutex); - for (list >::const_iterator i = _jobs.begin(); i != _jobs.end(); ++i) { - if ((*i)->finished_in_error ()) { + for (auto i: _jobs) { + if (i->finished_in_error()) { return true; } } @@ -146,67 +149,61 @@ JobManager::errors () const return false; } + void JobManager::scheduler () { + start_of_thread ("JobManager"); + while (true) { boost::mutex::scoped_lock lm (_mutex); - optional active_job; - - while (true) { - bool have_new = false; - bool have_running = false; - BOOST_FOREACH (shared_ptr i, _jobs) { - if (i->running()) { - have_running = true; - active_job = i->json_name(); - } - if (i->is_new()) { - have_new = true; - } - } - - if ((!have_running && have_new) || _terminate) { - break; - } - - _empty_condition.wait (lm); - } - if (_terminate) { break; } - BOOST_FOREACH (shared_ptr i, _jobs) { - if (i->is_new()) { - _connections.push_back (i->FinishedImmediate.connect(bind(&JobManager::job_finished, this))); - i->start (); - /* Only start one job at once */ - break; + bool have_running = false; + for (auto i: _jobs) { + if (have_running && i->running()) { + i->pause_by_priority(); + } else if (!have_running && (i->is_new() || i->paused_by_priority())) { + if (i->is_new()) { + _connections.push_back (i->FinishedImmediate.connect(bind(&JobManager::job_finished, this))); + i->start (); + } else { + i->resume (); + } + emit (boost::bind (boost::ref (ActiveJobsChanged), _last_active_job, i->json_name())); + _last_active_job = i->json_name (); + have_running = true; + } else if (!have_running && i->running()) { + have_running = true; } } - lm.unlock (); - - if (active_job != _last_active_job) { - emit (boost::bind (boost::ref (ActiveJobsChanged), _last_active_job, active_job)); - _last_active_job = active_job; - } + _empty_condition.wait (lm); } } + void JobManager::job_finished () { + { + boost::mutex::scoped_lock lm (_mutex); + emit (boost::bind(boost::ref (ActiveJobsChanged), _last_active_job, optional())); + _last_active_job = optional(); + } + _empty_condition.notify_all (); } + JobManager * JobManager::instance () { - if (_instance == 0) { + if (!_instance) { _instance = new JobManager (); _instance->start (); } @@ -214,13 +211,15 @@ JobManager::instance () return _instance; } + void JobManager::drop () { delete _instance; - _instance = 0; + _instance = nullptr; } + void JobManager::analyse_audio ( shared_ptr film, @@ -233,9 +232,9 @@ JobManager::analyse_audio ( { boost::mutex::scoped_lock lm (_mutex); - BOOST_FOREACH (shared_ptr i, _jobs) { - shared_ptr a = dynamic_pointer_cast (i); - if (a && a->playlist () == playlist) { + for (auto i: _jobs) { + auto a = dynamic_pointer_cast (i); + if (a && a->path() == film->audio_analysis_path(playlist) && !i->finished_cancelled()) { i->when_finished (connection, ready); return; } @@ -247,63 +246,76 @@ JobManager::analyse_audio ( { boost::mutex::scoped_lock lm (_mutex); - job.reset (new AnalyseAudioJob (film, playlist, from_zero)); + job = make_shared (film, playlist, from_zero); connection = job->Finished.connect (ready); _jobs.push_back (job); + _empty_condition.notify_all (); } emit (boost::bind (boost::ref (JobAdded), weak_ptr (job))); } + void -JobManager::increase_priority (shared_ptr job) +JobManager::analyse_subtitles ( + shared_ptr film, + shared_ptr content, + boost::signals2::connection& connection, + function ready + ) { - bool changed = false; - { boost::mutex::scoped_lock lm (_mutex); - list >::iterator last = _jobs.end (); - for (list >::iterator i = _jobs.begin(); i != _jobs.end(); ++i) { - if (*i == job && last != _jobs.end()) { - swap (*i, *last); - changed = true; - break; + + for (auto i: _jobs) { + auto a = dynamic_pointer_cast (i); + if (a && a->path() == film->subtitle_analysis_path(content)) { + i->when_finished (connection, ready); + return; } - last = i; } } - if (changed) { - priority_changed (); + shared_ptr job; + + { + boost::mutex::scoped_lock lm (_mutex); + + job = make_shared(film, content); + connection = job->Finished.connect (ready); + _jobs.push_back (job); + _empty_condition.notify_all (); } + + emit (boost::bind(boost::ref(JobAdded), weak_ptr(job))); } + void -JobManager::priority_changed () +JobManager::increase_priority (shared_ptr job) { + bool changed = false; + { boost::mutex::scoped_lock lm (_mutex); - - bool first = true; - BOOST_FOREACH (shared_ptr i, _jobs) { - if (first) { - if (i->is_new ()) { - i->start (); - } else if (i->paused_by_priority ()) { - i->resume (); - } - first = false; - } else { - if (i->running ()) { - i->pause_by_priority (); - } + auto last = _jobs.end (); + for (auto i = _jobs.begin(); i != _jobs.end(); ++i) { + if (*i == job && last != _jobs.end()) { + swap (*i, *last); + changed = true; + break; } + last = i; } } - emit (boost::bind (boost::ref (JobsReordered))); + if (changed) { + _empty_condition.notify_all (); + emit (boost::bind(boost::ref(JobsReordered))); + } } + void JobManager::decrease_priority (shared_ptr job) { @@ -311,8 +323,8 @@ JobManager::decrease_priority (shared_ptr job) { boost::mutex::scoped_lock lm (_mutex); - for (list >::iterator i = _jobs.begin(); i != _jobs.end(); ++i) { - list >::iterator next = i; + for (auto i = _jobs.begin(); i != _jobs.end(); ++i) { + auto next = i; ++next; if (*i == job && next != _jobs.end()) { swap (*i, *next); @@ -323,10 +335,12 @@ JobManager::decrease_priority (shared_ptr job) } if (changed) { - priority_changed (); + _empty_condition.notify_all (); + emit (boost::bind(boost::ref(JobsReordered))); } } + void JobManager::pause () { @@ -336,7 +350,7 @@ JobManager::pause () return; } - BOOST_FOREACH (shared_ptr i, _jobs) { + for (auto i: _jobs) { if (i->pause_by_user()) { _paused_job = i; } @@ -345,6 +359,7 @@ JobManager::pause () _paused = true; } + void JobManager::resume () {