X-Git-Url: https://git.carlh.net/gitweb/?p=dcpomatic.git;a=blobdiff_plain;f=src%2Flib%2Fjob_manager.cc;h=9fcd86f2d38b7c034a332740a74457f672bd1b9b;hp=86e010c106b929d0cb6869026eda4a70baab86e5;hb=a6c4b4fa16d9c6597e362044b875f3d6df80753f;hpb=e975b14d2c962eab149f56a79c35b68b608226d4 diff --git a/src/lib/job_manager.cc b/src/lib/job_manager.cc index 86e010c10..9fcd86f2d 100644 --- a/src/lib/job_manager.cc +++ b/src/lib/job_manager.cc @@ -1,99 +1,134 @@ /* - Copyright (C) 2012-2015 Carl Hetherington + Copyright (C) 2012-2021 Carl Hetherington - This program is free software; you can redistribute it and/or modify + 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. - This program is distributed in the hope that it will be useful, + 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 this program; if not, write to the Free Software - Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. + along with DCP-o-matic. If not, see . */ + /** @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 "util.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 std::make_shared; +using std::shared_ptr; +using std::string; +using std::weak_ptr; +using boost::bind; using boost::optional; -JobManager* JobManager::_instance = 0; + +JobManager* JobManager::_instance = nullptr; + JobManager::JobManager () - : _terminate (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"); +#endif } + JobManager::~JobManager () { + boost::this_thread::disable_interruption dis; + + for (auto& i: _connections) { + i.disconnect (); + } + { boost::mutex::scoped_lock lm (_mutex); _terminate = true; + _schedule_condition.notify_all(); } - if (_scheduler) { - DCPOMATIC_ASSERT (_scheduler->joinable ()); - _scheduler->join (); - } - - delete _scheduler; + try { + _scheduler.join(); + } catch (...) {} } + shared_ptr JobManager::add (shared_ptr j) { { boost::mutex::scoped_lock lm (_mutex); _jobs.push_back (j); + _schedule_condition.notify_all(); + } + + 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); + auto i = find (_jobs.begin(), _jobs.end(), after); + DCPOMATIC_ASSERT (i != _jobs.end()); + _jobs.insert (i, j); + _schedule_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; } @@ -101,12 +136,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; } } @@ -114,51 +150,63 @@ JobManager::errors () const return false; } + void JobManager::scheduler () { - while (true) { - - optional active_job; + start_of_thread ("JobManager"); - { - boost::mutex::scoped_lock lm (_mutex); - if (_terminate) { - return; - } - - BOOST_FOREACH (shared_ptr i, _jobs) { + while (true) { - if (!i->finished ()) { - active_job = i->json_name (); - } + boost::mutex::scoped_lock lm (_mutex); - if (i->running ()) { - /* Something is already happening */ - break; - } + if (_terminate) { + break; + } + bool have_running = false; + for (auto i: _jobs) { + if ((have_running || _paused) && i->running()) { + /* We already have a running job, or are totally paused, so this job should not be running */ + i->pause_by_priority(); + } else if (!have_running && !_paused && (i->is_new() || i->paused_by_priority())) { + /* We don't have a running job, and we should have one, so start/resume this */ if (i->is_new()) { + _connections.push_back (i->FinishedImmediate.connect(bind(&JobManager::job_finished, this))); i->start (); - /* Only start one job at once */ - break; + } 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; } } - if (active_job != _last_active_job) { - emit (boost::bind (boost::ref (ActiveJobsChanged), _last_active_job, active_job)); - _last_active_job = active_job; - } + _schedule_condition.wait(lm); + } +} + - dcpomatic_sleep (1); +void +JobManager::job_finished () +{ + { + boost::mutex::scoped_lock lm (_mutex); + emit (boost::bind(boost::ref (ActiveJobsChanged), _last_active_job, optional())); + _last_active_job = optional(); } + + _schedule_condition.notify_all(); } + JobManager * JobManager::instance () { - if (_instance == 0) { + if (!_instance) { _instance = new JobManager (); _instance->start (); } @@ -166,27 +214,30 @@ JobManager::instance () return _instance; } + void JobManager::drop () { delete _instance; - _instance = 0; + _instance = nullptr; } + void JobManager::analyse_audio ( shared_ptr film, shared_ptr playlist, + bool from_zero, boost::signals2::connection& connection, - function ready + function ready ) { { 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; } @@ -198,10 +249,100 @@ JobManager::analyse_audio ( { boost::mutex::scoped_lock lm (_mutex); - job.reset (new AnalyseAudioJob (film, playlist)); + job = make_shared (film, playlist, from_zero); connection = job->Finished.connect (ready); _jobs.push_back (job); + _schedule_condition.notify_all (); } emit (boost::bind (boost::ref (JobAdded), weak_ptr (job))); } + + +void +JobManager::analyse_subtitles ( + shared_ptr film, + shared_ptr content, + boost::signals2::connection& connection, + function ready + ) +{ + { + boost::mutex::scoped_lock lm (_mutex); + + 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; + } + } + } + + shared_ptr job; + + { + boost::mutex::scoped_lock lm (_mutex); + + job = make_shared(film, content); + connection = job->Finished.connect (ready); + _jobs.push_back (job); + _schedule_condition.notify_all (); + } + + emit (boost::bind(boost::ref(JobAdded), weak_ptr(job))); +} + + +void +JobManager::increase_priority (shared_ptr job) +{ + { + boost::mutex::scoped_lock lm (_mutex); + auto iter = std::find(_jobs.begin(), _jobs.end(), job); + if (iter == _jobs.begin() || iter == _jobs.end()) { + return; + } + swap(*iter, *std::prev(iter)); + } + + _schedule_condition.notify_all(); + emit(boost::bind(boost::ref(JobsReordered))); +} + + +void +JobManager::decrease_priority (shared_ptr job) +{ + { + boost::mutex::scoped_lock lm (_mutex); + auto iter = std::find(_jobs.begin(), _jobs.end(), job); + if (iter == _jobs.end() || std::next(iter) == _jobs.end()) { + return; + } + swap(*iter, *std::next(iter)); + } + + _schedule_condition.notify_all(); + emit(boost::bind(boost::ref(JobsReordered))); +} + + +/** Pause all job processing */ +void +JobManager::pause () +{ + boost::mutex::scoped_lock lm (_mutex); + _paused = true; + _schedule_condition.notify_all(); +} + + +/** Resume processing jobs after a previous pause() */ +void +JobManager::resume () +{ + boost::mutex::scoped_lock lm (_mutex); + _paused = false; + _schedule_condition.notify_all(); +}