X-Git-Url: https://main.carlh.net/gitweb/?p=dcpomatic.git;a=blobdiff_plain;f=src%2Flib%2Fjob_manager.cc;h=26e2d28970a71c0f5a9d90b7582ed2738fd2fca0;hp=f962754677c2c1c672092838a0eca23f73e5873f;hb=f0c10e92b849566e458bc323f8783a6fe83e52d2;hpb=afc495f722f89fea0bcc579046d1a5d362e36f69 diff --git a/src/lib/job_manager.cc b/src/lib/job_manager.cc index f96275467..26e2d2897 100644 --- a/src/lib/job_manager.cc +++ b/src/lib/job_manager.cc @@ -1,19 +1,20 @@ /* - Copyright (C) 2012 Carl Hetherington + Copyright (C) 2012-2018 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 . */ @@ -21,41 +22,96 @@ * @brief A simple scheduler for jobs. */ -#include -#include #include "job_manager.h" #include "job.h" #include "cross.h" -#include "ui_signaller.h" +#include "analyse_audio_job.h" +#include "film.h" +#include +#include +#include using std::string; 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 boost::bind; JobManager* JobManager::_instance = 0; JobManager::JobManager () - : _last_active_jobs (false) + : _terminate (false) + , _paused (false) + , _scheduler (0) +{ + +} + +void +JobManager::start () { - boost::thread (boost::bind (&JobManager::scheduler, this)); + _scheduler = new boost::thread (boost::bind (&JobManager::scheduler, this)); +#ifdef DCPOMATIC_LINUX + pthread_setname_np (_scheduler->native_handle(), "job-scheduler"); +#endif +} + +JobManager::~JobManager () +{ + BOOST_FOREACH (boost::signals2::connection& i, _connections) { + i.disconnect (); + } + + { + boost::mutex::scoped_lock lm (_mutex); + _terminate = true; + _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; } shared_ptr JobManager::add (shared_ptr j) { - boost::mutex::scoped_lock lm (_mutex); - _jobs.push_back (j); + { + boost::mutex::scoped_lock lm (_mutex); + _jobs.push_back (j); + _empty_condition.notify_all (); + } + + emit (boost::bind (boost::ref (JobAdded), weak_ptr (j))); + return j; } -void +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); - assert (i != _jobs.end ()); - ++i; - _jobs.insert (i, j); + { + boost::mutex::scoped_lock lm (_mutex); + list >::iterator 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))); + + return j; } list > @@ -81,53 +137,68 @@ 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 ()) { + BOOST_FOREACH (shared_ptr i, _jobs) { + if (i->finished_in_error ()) { return true; } } return false; -} +} void JobManager::scheduler () { - while (1) { + while (true) { - bool active_jobs = false; + boost::mutex::scoped_lock lm (_mutex); - { - boost::mutex::scoped_lock lm (_mutex); - for (list >::iterator i = _jobs.begin(); i != _jobs.end(); ++i) { - - if (!(*i)->finished ()) { - active_jobs = true; - } - - if ((*i)->running ()) { - /* Something is already happening */ - break; + while (true) { + bool have_new = false; + bool have_running = false; + BOOST_FOREACH (shared_ptr i, _jobs) { + if (i->running()) { + have_running = true; } - - if ((*i)->is_new()) { - (*i)->start (); - - /* Only start one job at once */ - break; + if (i->is_new()) { + have_new = true; } } + + if ((!have_running && have_new) || _terminate) { + break; + } + + _empty_condition.wait (lm); + } + + if (_terminate) { + break; } - if (active_jobs != _last_active_jobs) { - _last_active_jobs = active_jobs; - if (ui_signaller) { - ui_signaller->emit (boost::bind (boost::ref (ActiveJobsChanged), active_jobs)); + BOOST_FOREACH (shared_ptr i, _jobs) { + if (i->is_new()) { + _connections.push_back (i->FinishedImmediate.connect(bind(&JobManager::job_finished, this))); + i->start (); + emit (boost::bind (boost::ref (ActiveJobsChanged), _last_active_job, i->json_name())); + _last_active_job = i->json_name (); + /* Only start one job at once */ + break; } } + } +} - 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(); } + + _empty_condition.notify_all (); } JobManager * @@ -135,7 +206,156 @@ JobManager::instance () { if (_instance == 0) { _instance = new JobManager (); + _instance->start (); } return _instance; } + +void +JobManager::drop () +{ + delete _instance; + _instance = 0; +} + +void +JobManager::analyse_audio ( + shared_ptr film, + shared_ptr playlist, + bool from_zero, + boost::signals2::connection& connection, + function ready + ) +{ + { + boost::mutex::scoped_lock lm (_mutex); + + BOOST_FOREACH (shared_ptr i, _jobs) { + shared_ptr a = dynamic_pointer_cast (i); + if (a && a->path() == film->audio_analysis_path(playlist)) { + i->when_finished (connection, ready); + return; + } + } + } + + shared_ptr job; + + { + boost::mutex::scoped_lock lm (_mutex); + + job.reset (new AnalyseAudioJob (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) +{ + 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; + } + last = i; + } + } + + if (changed) { + priority_changed (); + } +} + +void +JobManager::priority_changed () +{ + { + 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 (); + } + } + } + } + + emit (boost::bind (boost::ref (JobsReordered))); +} + +void +JobManager::decrease_priority (shared_ptr job) +{ + bool changed = false; + + { + boost::mutex::scoped_lock lm (_mutex); + for (list >::iterator i = _jobs.begin(); i != _jobs.end(); ++i) { + list >::iterator next = i; + ++next; + if (*i == job && next != _jobs.end()) { + swap (*i, *next); + changed = true; + break; + } + } + } + + if (changed) { + priority_changed (); + } +} + +void +JobManager::pause () +{ + boost::mutex::scoped_lock lm (_mutex); + + if (_paused) { + return; + } + + BOOST_FOREACH (shared_ptr i, _jobs) { + if (i->pause_by_user()) { + _paused_job = i; + } + } + + _paused = true; +} + +void +JobManager::resume () +{ + boost::mutex::scoped_lock lm (_mutex); + if (!_paused) { + return; + } + + if (_paused_job) { + _paused_job->resume (); + } + + _paused_job.reset (); + _paused = false; +}