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 && i->running()) {
 170                                 i->pause_by_priority();
 171                         } else if (!have_running && (i->is_new() || i->paused_by_priority())) {
 172                                 if (i->is_new()) {
 173                                         _connections.push_back (i->FinishedImmediate.connect(bind(&JobManager::job_finished, this)));
 174                                         i->start ();
 175                                 } else {
 176                                         i->resume ();
 177                                 }
 178                                 emit (boost::bind (boost::ref (ActiveJobsChanged), _last_active_job, i->json_name()));
 179                                 _last_active_job = i->json_name ();
 180                                 have_running = true;
 181                         } else if (!have_running && i->running()) {
 182                                 have_running = true;
 183                         }
 184                 }
 185
 186                 _schedule_condition.wait(lm);
 187         }
 188 }
 189
 190
 191 void
 192 JobManager::job_finished ()
 193 {
 194         {
 195                 boost::mutex::scoped_lock lm (_mutex);
 196                 emit (boost::bind(boost::ref (ActiveJobsChanged), _last_active_job, optional<string>()));
 197                 _last_active_job = optional<string>();
 198         }
 199
 200         _schedule_condition.notify_all();
 201 }
 202
 203
 204 JobManager *
 205 JobManager::instance ()
 206 {
 207         if (!_instance) {
 208                 _instance = new JobManager ();
 209                 _instance->start ();
 210         }
 211
 212         return _instance;
 213 }
 214
 215
 216 void
 217 JobManager::drop ()
 218 {
 219         delete _instance;
 220         _instance = nullptr;
 221 }
 222
 223
 224 void
 225 JobManager::analyse_audio (
 226         shared_ptr<const Film> film,
 227         shared_ptr<const Playlist> playlist,
 228         bool from_zero,
 229         boost::signals2::connection& connection,
 230         function<void()> ready
 231         )
 232 {
 233         {
 234                 boost::mutex::scoped_lock lm (_mutex);
 235
 236                 for (auto i: _jobs) {
 237                         auto a = dynamic_pointer_cast<AnalyseAudioJob> (i);
 238                         if (a && a->path() == film->audio_analysis_path(playlist) && !i->finished_cancelled()) {
 239                                 i->when_finished (connection, ready);
 240                                 return;
 241                         }
 242                 }
 243         }
 244
 245         shared_ptr<AnalyseAudioJob> job;
 246
 247         {
 248                 boost::mutex::scoped_lock lm (_mutex);
 249
 250                 job = make_shared<AnalyseAudioJob> (film, playlist, from_zero);
 251                 connection = job->Finished.connect (ready);
 252                 _jobs.push_back (job);
 253                 _schedule_condition.notify_all ();
 254         }
 255
 256         emit (boost::bind (boost::ref (JobAdded), weak_ptr<Job> (job)));
 257 }
 258
 259
 260 void
 261 JobManager::analyse_subtitles (
 262         shared_ptr<const Film> film,
 263         shared_ptr<Content> content,
 264         boost::signals2::connection& connection,
 265         function<void()> ready
 266         )
 267 {
 268         {
 269                 boost::mutex::scoped_lock lm (_mutex);
 270
 271                 for (auto i: _jobs) {
 272                         auto a = dynamic_pointer_cast<AnalyseSubtitlesJob> (i);
 273                         if (a && a->path() == film->subtitle_analysis_path(content)) {
 274                                 i->when_finished (connection, ready);
 275                                 return;
 276                         }
 277                 }
 278         }
 279
 280         shared_ptr<AnalyseSubtitlesJob> job;
 281
 282         {
 283                 boost::mutex::scoped_lock lm (_mutex);
 284
 285                 job = make_shared<AnalyseSubtitlesJob>(film, content);
 286                 connection = job->Finished.connect (ready);
 287                 _jobs.push_back (job);
 288                 _schedule_condition.notify_all ();
 289         }
 290
 291         emit (boost::bind(boost::ref(JobAdded), weak_ptr<Job>(job)));
 292 }
 293
 294
 295 void
 296 JobManager::increase_priority (shared_ptr<Job> job)
 297 {
 298         {
 299                 boost::mutex::scoped_lock lm (_mutex);
 300                 auto iter = std::find(_jobs.begin(), _jobs.end(), job);
 301                 if (iter == _jobs.begin() || iter == _jobs.end()) {
 302                         return;
 303                 }
 304                 swap(*iter, *std::prev(iter));
 305         }
 306
 307         _schedule_condition.notify_all();
 308         emit(boost::bind(boost::ref(JobsReordered)));
 309 }
 310
 311
 312 void
 313 JobManager::decrease_priority (shared_ptr<Job> job)
 314 {
 315         {
 316                 boost::mutex::scoped_lock lm (_mutex);
 317                 auto iter = std::find(_jobs.begin(), _jobs.end(), job);
 318                 if (iter == _jobs.end() || std::next(iter) == _jobs.end()) {
 319                         return;
 320                 }
 321                 swap(*iter, *std::next(iter));
 322         }
 323
 324         _schedule_condition.notify_all();
 325         emit(boost::bind(boost::ref(JobsReordered)));
 326 }
 327
 328
 329 void
 330 JobManager::pause ()
 331 {
 332         boost::mutex::scoped_lock lm (_mutex);
 333
 334         if (_paused_job) {
 335                 return;
 336         }
 337
 338         for (auto i: _jobs) {
 339                 if (i->pause_by_user()) {
 340                         _paused_job = i;
 341                 }
 342         }
 343 }
 344
 345
 346 void
 347 JobManager::resume ()
 348 {
 349         boost::mutex::scoped_lock lm (_mutex);
 350
 351         if (!_paused_job) {
 352                 return;
 353         }
 354
 355         _paused_job->resume();
 356         _paused_job.reset();
 357 }