src/lib/job_manager.cc

   1 /*
   2     Copyright (C) 2012-2021 Carl Hetherington <cth@carlh.net>
   3
   4     This file is part of DCP-o-matic.
   5
   6     DCP-o-matic is free software; you can redistribute it and/or modify
   7     it under the terms of the GNU General Public License as published by
   8     the Free Software Foundation; either version 2 of the License, or
   9     (at your option) any later version.
  10
  11     DCP-o-matic is distributed in the hope that it will be useful,
  12     but WITHOUT ANY WARRANTY; without even the implied warranty of
  13     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14     GNU General Public License for more details.
  15
  16     You should have received a copy of the GNU General Public License
  17     along with DCP-o-matic.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 */
  20
  21
  22 /** @file  src/job_manager.cc
  23  *  @brief A simple scheduler for jobs.
  24  */
  25
  26
  27 #include "analyse_audio_job.h"
  28 #include "analyse_subtitles_job.h"
  29 #include "cross.h"
  30 #include "film.h"
  31 #include "job.h"
  32 #include "job_manager.h"
  33 #include "util.h"
  34 #include <boost/thread.hpp>
  35
  36
  37 using std::dynamic_pointer_cast;
  38 using std::function;
  39 using std::list;
  40 using std::make_shared;
  41 using std::shared_ptr;
  42 using std::string;
  43 using std::weak_ptr;
  44 using boost::bind;
  45 using boost::optional;
  46
  47
  48 JobManager* JobManager::_instance = nullptr;
  49
  50
  51 JobManager::JobManager ()
  52 {
  53
  54 }
  55
  56
  57 void
  58 JobManager::start ()
  59 {
  60         _scheduler = boost::thread (boost::bind(&JobManager::scheduler, this));
  61 #ifdef DCPOMATIC_LINUX
  62         pthread_setname_np (_scheduler.native_handle(), "job-scheduler");
  63 #endif
  64 }
  65
  66
  67 JobManager::~JobManager ()
  68 {
  69         boost::this_thread::disable_interruption dis;
  70
  71         for (auto& i: _connections) {
  72                 i.disconnect ();
  73         }
  74
  75         {
  76                 boost::mutex::scoped_lock lm (_mutex);
  77                 _terminate = true;
  78                 _schedule_condition.notify_all();
  79         }
  80
  81         try {
  82                 _scheduler.join();
  83         } catch (...) {}
  84 }
  85
  86
  87 shared_ptr<Job>
  88 JobManager::add (shared_ptr<Job> j)
  89 {
  90         {
  91                 boost::mutex::scoped_lock lm (_mutex);
  92                 _jobs.push_back (j);
  93                 _schedule_condition.notify_all();
  94         }
  95
  96         emit (boost::bind(boost::ref(JobAdded), weak_ptr<Job>(j)));
  97
  98         return j;
  99 }
 100
 101
 102 shared_ptr<Job>
 103 JobManager::add_after (shared_ptr<Job> after, shared_ptr<Job> j)
 104 {
 105         {
 106                 boost::mutex::scoped_lock lm (_mutex);
 107                 auto i = find (_jobs.begin(), _jobs.end(), after);
 108                 DCPOMATIC_ASSERT (i != _jobs.end());
 109                 _jobs.insert (i, j);
 110                 _schedule_condition.notify_all();
 111         }
 112
 113         emit (boost::bind(boost::ref(JobAdded), weak_ptr<Job>(j)));
 114
 115         return j;
 116 }
 117
 118
 119 list<shared_ptr<Job>>
 120 JobManager::get () const
 121 {
 122         boost::mutex::scoped_lock lm (_mutex);
 123         return _jobs;
 124 }
 125
 126
 127 bool
 128 JobManager::work_to_do () const
 129 {
 130         boost::mutex::scoped_lock lm (_mutex);
 131         auto i = _jobs.begin();
 132         while (i != _jobs.end() && (*i)->finished()) {
 133                 ++i;
 134         }
 135
 136         return i != _jobs.end ();
 137 }
 138
 139
 140 bool
 141 JobManager::errors () const
 142 {
 143         boost::mutex::scoped_lock lm (_mutex);
 144         for (auto i: _jobs) {
 145                 if (i->finished_in_error()) {
 146                         return true;
 147                 }
 148         }
 149
 150         return false;
 151 }
 152
 153
 154 void
 155 JobManager::scheduler ()
 156 {
 157         start_of_thread ("JobManager");
 158
 159         while (true) {
 160
 161                 boost::mutex::scoped_lock lm (_mutex);
 162
 163                 if (_terminate) {
 164                         break;
 165                 }
 166
 167                 bool have_running = false;
 168                 for (auto i: _jobs) {
 169                         if ((have_running || _paused) && i->running()) {
 170                                 /* We already have a running job, or are totally paused, so this job should not be running */
 171                                 i->pause_by_priority();
 172                         } else if (!have_running && !_paused && (i->is_new() || i->paused_by_priority())) {
 173                                 /* We don't have a running job, and we should have one, so start/resume this */
 174                                 if (i->is_new()) {
 175                                         _connections.push_back (i->FinishedImmediate.connect(bind(&JobManager::job_finished, this)));
 176                                         i->start ();
 177                                 } else {
 178                                         i->resume ();
 179                                 }
 180                                 emit (boost::bind (boost::ref (ActiveJobsChanged), _last_active_job, i->json_name()));
 181                                 _last_active_job = i->json_name ();
 182                                 have_running = true;
 183                         } else if (!have_running && i->running()) {
 184                                 have_running = true;
 185                         }
 186                 }
 187
 188                 _schedule_condition.wait(lm);
 189         }
 190 }
 191
 192
 193 void
 194 JobManager::job_finished ()
 195 {
 196         {
 197                 boost::mutex::scoped_lock lm (_mutex);
 198                 emit (boost::bind(boost::ref (ActiveJobsChanged), _last_active_job, optional<string>()));
 199                 _last_active_job = optional<string>();
 200         }
 201
 202         _schedule_condition.notify_all();
 203 }
 204
 205
 206 JobManager *
 207 JobManager::instance ()
 208 {
 209         if (!_instance) {
 210                 _instance = new JobManager ();
 211                 _instance->start ();
 212         }
 213
 214         return _instance;
 215 }
 216
 217
 218 void
 219 JobManager::drop ()
 220 {
 221         delete _instance;
 222         _instance = nullptr;
 223 }
 224
 225
 226 void
 227 JobManager::analyse_audio (
 228         shared_ptr<const Film> film,
 229         shared_ptr<const Playlist> playlist,
 230         bool from_zero,
 231         boost::signals2::connection& connection,
 232         function<void (Job::Result)> ready
 233         )
 234 {
 235         {
 236                 boost::mutex::scoped_lock lm (_mutex);
 237
 238                 for (auto i: _jobs) {
 239                         auto a = dynamic_pointer_cast<AnalyseAudioJob> (i);
 240                         if (a && a->path() == film->audio_analysis_path(playlist) && !i->finished_cancelled()) {
 241                                 i->when_finished (connection, ready);
 242                                 return;
 243                         }
 244                 }
 245         }
 246
 247         shared_ptr<AnalyseAudioJob> job;
 248
 249         {
 250                 boost::mutex::scoped_lock lm (_mutex);
 251
 252                 job = make_shared<AnalyseAudioJob> (film, playlist, from_zero);
 253                 connection = job->Finished.connect (ready);
 254                 _jobs.push_back (job);
 255                 _schedule_condition.notify_all ();
 256         }
 257
 258         emit (boost::bind (boost::ref (JobAdded), weak_ptr<Job> (job)));
 259 }
 260
 261
 262 void
 263 JobManager::analyse_subtitles (
 264         shared_ptr<const Film> film,
 265         shared_ptr<Content> content,
 266         boost::signals2::connection& connection,
 267         function<void (Job::Result)> ready
 268         )
 269 {
 270         {
 271                 boost::mutex::scoped_lock lm (_mutex);
 272
 273                 for (auto i: _jobs) {
 274                         auto a = dynamic_pointer_cast<AnalyseSubtitlesJob> (i);
 275                         if (a && a->path() == film->subtitle_analysis_path(content) && !i->finished_cancelled()) {
 276                                 i->when_finished (connection, ready);
 277                                 return;
 278                         }
 279                 }
 280         }
 281
 282         shared_ptr<AnalyseSubtitlesJob> job;
 283
 284         {
 285                 boost::mutex::scoped_lock lm (_mutex);
 286
 287                 job = make_shared<AnalyseSubtitlesJob>(film, content);
 288                 connection = job->Finished.connect (ready);
 289                 _jobs.push_back (job);
 290                 _schedule_condition.notify_all ();
 291         }
 292
 293         emit (boost::bind(boost::ref(JobAdded), weak_ptr<Job>(job)));
 294 }
 295
 296
 297 void
 298 JobManager::increase_priority (shared_ptr<Job> job)
 299 {
 300         {
 301                 boost::mutex::scoped_lock lm (_mutex);
 302                 auto iter = std::find(_jobs.begin(), _jobs.end(), job);
 303                 if (iter == _jobs.begin() || iter == _jobs.end()) {
 304                         return;
 305                 }
 306                 swap(*iter, *std::prev(iter));
 307         }
 308
 309         _schedule_condition.notify_all();
 310         emit(boost::bind(boost::ref(JobsReordered)));
 311 }
 312
 313
 314 void
 315 JobManager::decrease_priority (shared_ptr<Job> job)
 316 {
 317         {
 318                 boost::mutex::scoped_lock lm (_mutex);
 319                 auto iter = std::find(_jobs.begin(), _jobs.end(), job);
 320                 if (iter == _jobs.end() || std::next(iter) == _jobs.end()) {
 321                         return;
 322                 }
 323                 swap(*iter, *std::next(iter));
 324         }
 325
 326         _schedule_condition.notify_all();
 327         emit(boost::bind(boost::ref(JobsReordered)));
 328 }
 329
 330
 331 /** Pause all job processing */
 332 void
 333 JobManager::pause ()
 334 {
 335         boost::mutex::scoped_lock lm (_mutex);
 336         _paused = true;
 337         _schedule_condition.notify_all();
 338 }
 339
 340
 341 /** Resume processing jobs after a previous pause() */
 342 void
 343 JobManager::resume ()
 344 {
 345         boost::mutex::scoped_lock lm (_mutex);
 346         _paused = false;
 347         _schedule_condition.notify_all();
 348 }