X-Git-Url: https://main.carlh.net/gitweb/?a=blobdiff_plain;f=src%2Flib%2Fjob_manager.cc;h=c40178f418a8960914193245f8fbebe557eead97;hb=83416808c0b1ca732e7a186d3811f1ec796fea08;hp=3992e685ed4fe76c9ee4c5912ae7cd41fcaac873;hpb=b3234e76d2f614c2b05034c0bdae8d5a4e9de9ea;p=dcpomatic.git diff --git a/src/lib/job_manager.cc b/src/lib/job_manager.cc index 3992e685e..c40178f41 100644 --- a/src/lib/job_manager.cc +++ b/src/lib/job_manager.cc @@ -1,5 +1,5 @@ /* - Copyright (C) 2012-2015 Carl Hetherington + Copyright (C) 2012-2020 Carl Hetherington This file is part of DCP-o-matic. @@ -26,6 +26,7 @@ #include "job.h" #include "cross.h" #include "analyse_audio_job.h" +#include "analyse_subtitles_job.h" #include "film.h" #include #include @@ -39,12 +40,13 @@ using boost::weak_ptr; using boost::function; using boost::dynamic_pointer_cast; using boost::optional; +using boost::bind; JobManager* JobManager::_instance = 0; JobManager::JobManager () : _terminate (false) - , _scheduler (0) + , _paused (false) { } @@ -52,26 +54,31 @@ JobManager::JobManager () 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_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 (); + if (_scheduler.joinable()) { + try { + _scheduler.join(); + } catch (...) { + } } - - delete _scheduler; } shared_ptr @@ -80,6 +87,23 @@ JobManager::add (shared_ptr 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; +} + +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); + DCPOMATIC_ASSERT (i != _jobs.end()); + _jobs.insert (i, j); + _empty_condition.notify_all (); } emit (boost::bind (boost::ref (JobAdded), weak_ptr (j))); @@ -110,8 +134,8 @@ 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; } } @@ -124,40 +148,54 @@ JobManager::scheduler () { while (true) { - optional active_job; - - { - boost::mutex::scoped_lock lm (_mutex); - if (_terminate) { - return; - } + boost::mutex::scoped_lock lm (_mutex); + while (true) { + bool have_new = false; + bool have_running = false; BOOST_FOREACH (shared_ptr i, _jobs) { - - if (!i->finished ()) { - active_job = i->json_name (); - } - - if (i->running ()) { - /* Something is already happening */ - break; + if (i->running()) { + have_running = true; } - if (i->is_new()) { - i->start (); - /* Only start one job at once */ - break; + have_new = true; } } + + if ((!have_running && have_new) || _terminate) { + break; + } + + _empty_condition.wait (lm); + } + + if (_terminate) { + break; } - if (active_job != _last_active_job) { - emit (boost::bind (boost::ref (ActiveJobsChanged), _last_active_job, active_job)); - _last_active_job = active_job; + 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 * @@ -182,6 +220,7 @@ void JobManager::analyse_audio ( shared_ptr film, shared_ptr playlist, + bool from_zero, boost::signals2::connection& connection, function ready ) @@ -191,7 +230,7 @@ JobManager::analyse_audio ( BOOST_FOREACH (shared_ptr i, _jobs) { shared_ptr a = dynamic_pointer_cast (i); - if (a && a->playlist () == playlist) { + if (a && a->path() == film->audio_analysis_path(playlist)) { i->when_finished (connection, ready); return; } @@ -203,14 +242,51 @@ JobManager::analyse_audio ( { boost::mutex::scoped_lock lm (_mutex); - job.reset (new AnalyseAudioJob (film, playlist)); + 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::analyse_subtitles ( + shared_ptr film, + shared_ptr content, + 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->subtitle_analysis_path(content)) { + i->when_finished (connection, ready); + return; + } + } + } + + shared_ptr job; + + { + boost::mutex::scoped_lock lm (_mutex); + + job.reset (new AnalyseSubtitlesJob(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::increase_priority (shared_ptr job) { @@ -265,13 +341,16 @@ JobManager::decrease_priority (shared_ptr job) { bool changed = false; - 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; + { + 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; + } } } @@ -279,3 +358,37 @@ JobManager::decrease_priority (shared_ptr job) 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; +}