src/lib/job.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.cc
  23  *  @brief A parent class to represent long-running tasks which are run in their own thread.
  24  */
  25
  26
  27 #include "compose.hpp"
  28 #include "constants.h"
  29 #include "cross.h"
  30 #include "dcpomatic_log.h"
  31 #include "exceptions.h"
  32 #include "film.h"
  33 #include "job.h"
  34 #include "log.h"
  35 #include "util.h"
  36 #include "variant.h"
  37 #include <dcp/exceptions.h>
  38 #include <sub/exceptions.h>
  39 #include <boost/date_time/posix_time/posix_time.hpp>
  40 #include <boost/filesystem.hpp>
  41 #include <boost/thread.hpp>
  42 #include <time.h>
  43 #include <iostream>
  44
  45 #include "i18n.h"
  46
  47
  48 using std::cout;
  49 using std::function;
  50 using std::list;
  51 using std::shared_ptr;
  52 using std::string;
  53 using boost::optional;
  54 using namespace dcpomatic;
  55
  56
  57 /** @param film Associated film, or 0 */
  58 Job::Job (shared_ptr<const Film> film)
  59         : _film (film)
  60         , _state (NEW)
  61         , _sub_start_time (0)
  62         , _progress (0)
  63         , _rate_limit_progress(true)
  64 {
  65
  66 }
  67
  68
  69 Job::~Job ()
  70 {
  71 #ifdef DCPOMATIC_DEBUG
  72         /* Any subclass should have called stop_thread in its destructor */
  73         assert (!_thread.joinable());
  74 #endif
  75 }
  76
  77
  78 void
  79 Job::stop_thread ()
  80 {
  81         boost::this_thread::disable_interruption dis;
  82
  83         _thread.interrupt ();
  84         try {
  85                 _thread.join ();
  86         } catch (...) {}
  87 }
  88
  89
  90 /** Start the job in a separate thread, returning immediately */
  91 void
  92 Job::start ()
  93 {
  94         set_state (RUNNING);
  95         _start_time = time (0);
  96         _sub_start_time = time (0);
  97         _thread = boost::thread (boost::bind(&Job::run_wrapper, this));
  98 #ifdef DCPOMATIC_LINUX
  99         pthread_setname_np (_thread.native_handle(), "job-wrapper");
 100 #endif
 101 }
 102
 103
 104 /** A wrapper for the ::run() method to catch exceptions */
 105 void
 106 Job::run_wrapper ()
 107 {
 108         start_of_thread (String::compose("Job-%1", json_name()));
 109
 110         try {
 111
 112                 run ();
 113
 114         } catch (dcp::FileError& e) {
 115
 116                 string m = String::compose(_("An error occurred whilst handling the file %1."), e.filename().filename());
 117
 118                 try {
 119                         auto const s = dcp::filesystem::space(e.filename());
 120                         if (s.available < pow (1024, 3)) {
 121                                 m += N_("\n\n");
 122                                 m += _("The drive that the film is stored on is low in disc space.  Free some more space and try again.");
 123                         }
 124                 } catch (...) {
 125
 126                 }
 127
 128                 set_error (e.what(), m);
 129                 set_progress (1);
 130                 set_state (FINISHED_ERROR);
 131
 132         } catch (dcp::StartCompressionError& e) {
 133
 134                 bool done = false;
 135
 136 #ifdef DCPOMATIC_WINDOWS
 137 #if (__GNUC__ && !__x86_64__)
 138                 /* 32-bit */
 139                 set_error (
 140                         _("Failed to encode the DCP."),
 141                         String::compose(
 142                                 _("This error has probably occurred because you are running the 32-bit version of %1 and "
 143                                   "trying to use too many encoding threads.  Please reduce the 'number of threads %2 should "
 144                                   "use' in the General tab of Preferences and try again."),
 145                                 variant::dcpomatic(),
 146                                 variant::dcpomatic()
 147                                 )
 148                         );
 149                 done = true;
 150 #else
 151                 /* 64-bit */
 152                 if (running_32_on_64()) {
 153                         set_error (
 154                                 _("Failed to encode the DCP."),
 155                                 String::compose(
 156                                         _("This error has probably occurred because you are running the 32-bit version of %1.  "
 157                                           "Please re-install %2 with the 64-bit installer and try again."),
 158                                         variant::dcpomatic(),
 159                                         variant::dcpomatic()
 160                                         )
 161                                 );
 162                         done = true;
 163                 }
 164 #endif
 165 #endif
 166
 167                 if (!done) {
 168                         set_error (
 169                                 e.what (),
 170                                 string (_("It is not known what caused this error.")) + "  " + REPORT_PROBLEM
 171                                 );
 172                 }
 173
 174                 set_progress (1);
 175                 set_state (FINISHED_ERROR);
 176
 177         } catch (OpenFileError& e) {
 178
 179                 set_error (
 180                         String::compose (_("Could not open %1"), e.file().string()),
 181                         String::compose(_("%1 could not open the file %2 (%3).  Perhaps it does not exist or is in an unexpected format."),
 182                                 variant::dcpomatic(),
 183                                 dcp::filesystem::absolute(e.file()).string(),
 184                                 e.what()
 185                                 )
 186                         );
 187
 188                 set_progress (1);
 189                 set_state (FINISHED_ERROR);
 190
 191         } catch (boost::filesystem::filesystem_error& e) {
 192
 193                 if (e.code() == boost::system::errc::no_such_file_or_directory) {
 194                         set_error (
 195                                 String::compose (_("Could not open %1"), e.path1().string ()),
 196                                 String::compose(_("%1 could not open the file %2 (%3).  Perhaps it does not exist or is in an unexpected format."),
 197                                         variant::dcpomatic(),
 198                                         dcp::filesystem::absolute(e.path1()).string(),
 199                                         e.what()
 200                                         )
 201                                 );
 202                 } else {
 203                         set_error (
 204                                 e.what (),
 205                                 string (_("It is not known what caused this error.")) + "  " + REPORT_PROBLEM
 206                                 );
 207                 }
 208
 209                 set_progress (1);
 210                 set_state (FINISHED_ERROR);
 211
 212         } catch (boost::thread_interrupted &) {
 213                 /* The job was cancelled; there's nothing else we need to do here */
 214         } catch (sub::SubripError& e) {
 215
 216                 string extra = "Error is near:\n";
 217                 for (auto i: e.context()) {
 218                         extra += i + "\n";
 219                 }
 220
 221                 set_error (e.what (), extra);
 222                 set_progress (1);
 223                 set_state (FINISHED_ERROR);
 224
 225         } catch (std::bad_alloc& e) {
 226
 227                 set_error (_("Out of memory"), _("There was not enough memory to do this.  If you are running a 32-bit operating system try reducing the number of encoding threads in the General tab of Preferences."));
 228                 set_progress (1);
 229                 set_state (FINISHED_ERROR);
 230
 231         } catch (dcp::ReadError& e) {
 232
 233                 set_error (e.message(), e.detail().get_value_or(""));
 234                 set_progress (1);
 235                 set_state (FINISHED_ERROR);
 236
 237         } catch (KDMError& e) {
 238
 239                 set_error (e.summary(), e.detail());
 240                 set_progress (1);
 241                 set_state (FINISHED_ERROR);
 242
 243         } catch (FileError& e) {
 244
 245                 set_error (e.what(), e.what());
 246                 set_progress (1);
 247                 set_state (FINISHED_ERROR);
 248
 249         } catch (CPLNotFoundError& e) {
 250
 251                 set_error(e.what());
 252                 set_progress(1);
 253                 set_state(FINISHED_ERROR);
 254
 255         } catch (MissingConfigurationError& e) {
 256
 257                 set_error(e.what());
 258                 set_progress(1);
 259                 set_state(FINISHED_ERROR);
 260
 261         } catch (std::exception& e) {
 262
 263                 set_error (
 264                         e.what (),
 265                         string (_("It is not known what caused this error.")) + "  " + REPORT_PROBLEM
 266                         );
 267
 268                 set_progress (1);
 269                 set_state (FINISHED_ERROR);
 270
 271         } catch (...) {
 272
 273                 set_error (
 274                         _("Unknown error"),
 275                         string (_("It is not known what caused this error.")) + "  " + REPORT_PROBLEM
 276                         );
 277
 278                 set_progress (1);
 279                 set_state (FINISHED_ERROR);
 280         }
 281 }
 282
 283
 284 /** @return true if this job is new (ie has not started running) */
 285 bool
 286 Job::is_new () const
 287 {
 288         boost::mutex::scoped_lock lm (_state_mutex);
 289         return _state == NEW;
 290 }
 291
 292
 293 /** @return true if the job is running */
 294 bool
 295 Job::running () const
 296 {
 297         boost::mutex::scoped_lock lm (_state_mutex);
 298         return _state == RUNNING;
 299 }
 300
 301
 302 /** @return true if the job has finished (either successfully or unsuccessfully) */
 303 bool
 304 Job::finished () const
 305 {
 306         boost::mutex::scoped_lock lm (_state_mutex);
 307         return _state == FINISHED_OK || _state == FINISHED_ERROR || _state == FINISHED_CANCELLED;
 308 }
 309
 310
 311 /** @return true if the job has finished successfully */
 312 bool
 313 Job::finished_ok () const
 314 {
 315         boost::mutex::scoped_lock lm (_state_mutex);
 316         return _state == FINISHED_OK;
 317 }
 318
 319
 320 /** @return true if the job has finished unsuccessfully */
 321 bool
 322 Job::finished_in_error () const
 323 {
 324         boost::mutex::scoped_lock lm (_state_mutex);
 325         return _state == FINISHED_ERROR;
 326 }
 327
 328
 329 bool
 330 Job::finished_cancelled () const
 331 {
 332         boost::mutex::scoped_lock lm (_state_mutex);
 333         return _state == FINISHED_CANCELLED;
 334 }
 335
 336
 337 bool
 338 Job::paused_by_user () const
 339 {
 340         boost::mutex::scoped_lock lm (_state_mutex);
 341         return _state == PAUSED_BY_USER;
 342 }
 343
 344
 345 bool
 346 Job::paused_by_priority () const
 347 {
 348         boost::mutex::scoped_lock lm (_state_mutex);
 349         return _state == PAUSED_BY_PRIORITY;
 350 }
 351
 352
 353 /** Set the state of this job.
 354  *  @param s New state.
 355  */
 356 void
 357 Job::set_state (State s)
 358 {
 359         bool finished = false;
 360
 361         {
 362                 boost::mutex::scoped_lock lm (_state_mutex);
 363                 if (_state == s) {
 364                         return;
 365                 }
 366
 367                 _state = s;
 368
 369                 if (_state == FINISHED_OK || _state == FINISHED_ERROR || _state == FINISHED_CANCELLED) {
 370                         _finish_time = time(nullptr);
 371                         finished = true;
 372                         _sub_name.clear ();
 373                 }
 374         }
 375
 376         if (finished) {
 377                 auto const result = state_to_result(s);
 378                 emit(boost::bind(boost::ref(Finished), result));
 379                 FinishedImmediate(result);
 380         }
 381 }
 382
 383
 384 Job::Result
 385 Job::state_to_result(State state) const
 386 {
 387         switch (state) {
 388         case FINISHED_OK:
 389                 return Result::RESULT_OK;
 390         case FINISHED_ERROR:
 391                 return Result::RESULT_ERROR;
 392         case FINISHED_CANCELLED:
 393                 return Result::RESULT_CANCELLED;
 394         default:
 395                 DCPOMATIC_ASSERT(false);
 396         };
 397
 398         DCPOMATIC_ASSERT(false);
 399 }
 400
 401
 402 /** @return DCPTime (in seconds) that this sub-job has been running */
 403 int
 404 Job::elapsed_sub_time () const
 405 {
 406         if (_sub_start_time == 0) {
 407                 return 0;
 408         }
 409
 410         return time (0) - _sub_start_time;
 411 }
 412
 413
 414 /** Check to see if this job has been interrupted or paused */
 415 void
 416 Job::check_for_interruption_or_pause ()
 417 {
 418         boost::this_thread::interruption_point ();
 419
 420         boost::mutex::scoped_lock lm (_state_mutex);
 421         while (_state == PAUSED_BY_USER || _state == PAUSED_BY_PRIORITY) {
 422                 emit (boost::bind (boost::ref (Progress)));
 423                 _pause_changed.wait (lm);
 424         }
 425 }
 426
 427
 428 optional<float>
 429 Job::seconds_since_last_progress_update () const
 430 {
 431         boost::mutex::scoped_lock lm (_progress_mutex);
 432         if (!_last_progress_update) {
 433                 return {};
 434         }
 435
 436         struct timeval now;
 437         gettimeofday (&now, 0);
 438
 439         return seconds(now) - seconds(*_last_progress_update);
 440 }
 441
 442
 443 /** Set the progress of the current part of the job.
 444  *  @param p Progress (from 0 to 1)
 445  *  @param force Do not ignore this update, even if it hasn't been long since the last one.
 446  */
 447 void
 448 Job::set_progress (float p, bool force)
 449 {
 450         check_for_interruption_or_pause ();
 451
 452         if (!force && _rate_limit_progress) {
 453                 /* Check for excessively frequent progress reporting */
 454                 boost::mutex::scoped_lock lm (_progress_mutex);
 455                 struct timeval now;
 456                 gettimeofday (&now, 0);
 457                 if (_last_progress_update && _last_progress_update->tv_sec > 0) {
 458                         double const elapsed = seconds(now) - seconds(*_last_progress_update);
 459                         if (elapsed < 0.5) {
 460                                 return;
 461                         }
 462                 }
 463                 _last_progress_update = now;
 464         }
 465
 466         set_progress_common (p);
 467 }
 468
 469
 470 void
 471 Job::set_progress_common (optional<float> p)
 472 {
 473         {
 474                 boost::mutex::scoped_lock lm (_progress_mutex);
 475                 _progress = p;
 476         }
 477
 478         emit (boost::bind (boost::ref (Progress)));
 479 }
 480
 481
 482 /** @return fractional progress of the current sub-job, if known */
 483 optional<float>
 484 Job::progress () const
 485 {
 486         boost::mutex::scoped_lock lm (_progress_mutex);
 487         return _progress;
 488 }
 489
 490
 491 void
 492 Job::sub (string n)
 493 {
 494         {
 495                 boost::mutex::scoped_lock lm (_progress_mutex);
 496                 LOG_GENERAL ("Sub-job %1 starting", n);
 497                 _sub_name = n;
 498         }
 499
 500         set_progress (0, true);
 501         _sub_start_time = time (0);
 502 }
 503
 504
 505 string
 506 Job::error_details () const
 507 {
 508         boost::mutex::scoped_lock lm (_state_mutex);
 509         return _error_details;
 510 }
 511
 512
 513 /** @return A summary of any error that the job has generated */
 514 string
 515 Job::error_summary () const
 516 {
 517         boost::mutex::scoped_lock lm (_state_mutex);
 518         return _error_summary;
 519 }
 520
 521
 522 /** Set the current error string.
 523  *  @param s New error string.
 524  *  @param d New error detail string.
 525  */
 526 void
 527 Job::set_error (string s, string d)
 528 {
 529         if (_film) {
 530                 _film->log()->log (String::compose ("Error in job: %1 (%2)", s, d), LogEntry::TYPE_ERROR);
 531         }
 532
 533         boost::mutex::scoped_lock lm (_state_mutex);
 534         _error_summary = s;
 535         _error_details = d;
 536 }
 537
 538
 539 /** Say that this job's progress will be unknown until further notice */
 540 void
 541 Job::set_progress_unknown ()
 542 {
 543         check_for_interruption_or_pause ();
 544         set_progress_common (optional<float> ());
 545 }
 546
 547
 548 /** @return Human-readable status of this job */
 549 string
 550 Job::status () const
 551 {
 552         optional<float> p = progress ();
 553         int const t = elapsed_sub_time ();
 554         int const r = remaining_time ();
 555
 556         auto day_of_week_to_string = [](boost::gregorian::greg_weekday d) -> std::string {
 557                 switch (d.as_enum()) {
 558                 case boost::date_time::Sunday:
 559                         return _("Sunday");
 560                 case boost::date_time::Monday:
 561                         return _("Monday");
 562                 case boost::date_time::Tuesday:
 563                         return _("Tuesday");
 564                 case boost::date_time::Wednesday:
 565                         return _("Wednesday");
 566                 case boost::date_time::Thursday:
 567                         return _("Thursday");
 568                 case boost::date_time::Friday:
 569                         return _("Friday");
 570                 case boost::date_time::Saturday:
 571                         return _("Saturday");
 572                 }
 573
 574                 return d.as_long_string();
 575         };
 576
 577         string s;
 578         if (!finished () && p) {
 579                 int pc = lrintf (p.get() * 100);
 580                 if (pc == 100) {
 581                         /* 100% makes it sound like we've finished when we haven't */
 582                         pc = 99;
 583                 }
 584
 585                 char buffer[64];
 586                 snprintf (buffer, sizeof(buffer), "%d%%", pc);
 587                 s += buffer;
 588
 589                 if (t > 10 && r > 0) {
 590                         auto now = boost::posix_time::second_clock::local_time();
 591                         auto finish = now + boost::posix_time::seconds(r);
 592                         char finish_string[16];
 593                         snprintf (finish_string, sizeof(finish_string), "%02d:%02d", int(finish.time_of_day().hours()), int(finish.time_of_day().minutes()));
 594                         string day;
 595                         if (now.date() != finish.date()) {
 596                                 /// TRANSLATORS: the %1 in this string will be filled in with a day of the week
 597                                 /// to say what day a job will finish.
 598                                 day = String::compose (_(" on %1"), day_of_week_to_string(finish.date().day_of_week()));
 599                         }
 600                         /// TRANSLATORS: "remaining; finishing at" here follows an amount of time that is remaining
 601                         /// on an operation; after it is an estimated wall-clock completion time.
 602                         s += String::compose(
 603                                 _("; %1 remaining; finishing at %2%3"),
 604                                 seconds_to_approximate_hms(r), finish_string, day
 605                                 );
 606                 }
 607         } else if (finished_ok ()) {
 608                 auto time_string = [](time_t time) {
 609                         auto tm = localtime(&time);
 610                         char buffer[8];
 611                         snprintf(buffer, sizeof(buffer), "%02d:%02d", tm->tm_hour, tm->tm_min);
 612                         return string(buffer);
 613                 };
 614                 auto const duration = _finish_time - _start_time;
 615                 if (duration < 10) {
 616                         /* It took less than 10 seconds; it doesn't seem worth saying how long it took */
 617                         s = _("OK");
 618                 } else if (duration < 600) {
 619                         /* It took less than 10 minutes; it doesn't seem worth saying when it started and finished */
 620                         s = String::compose(_("OK (ran for %1)"), seconds_to_hms(duration));
 621                 } else {
 622                         s = String::compose(_("OK (ran for %1 from %2 to %3)"),  seconds_to_hms(duration), time_string(_start_time), time_string(_finish_time));
 623                 }
 624         } else if (finished_in_error ()) {
 625                 s = String::compose (_("Error: %1"), error_summary ());
 626         } else if (finished_cancelled ()) {
 627                 s = _("Cancelled");
 628         }
 629
 630         return s;
 631 }
 632
 633
 634 string
 635 Job::json_status () const
 636 {
 637         boost::mutex::scoped_lock lm (_state_mutex);
 638
 639         switch (_state) {
 640         case NEW:
 641                 return N_("new");
 642         case RUNNING:
 643                 return N_("running");
 644         case PAUSED_BY_USER:
 645         case PAUSED_BY_PRIORITY:
 646                 return N_("paused");
 647         case FINISHED_OK:
 648                 return N_("finished_ok");
 649         case FINISHED_ERROR:
 650                 return N_("finished_error");
 651         case FINISHED_CANCELLED:
 652                 return N_("finished_cancelled");
 653         }
 654
 655         return "";
 656 }
 657
 658
 659 /** @return An estimate of the remaining time for this sub-job, in seconds */
 660 int
 661 Job::remaining_time () const
 662 {
 663         if (progress().get_value_or(0) == 0) {
 664                 return elapsed_sub_time ();
 665         }
 666
 667         return elapsed_sub_time() / progress().get() - elapsed_sub_time();
 668 }
 669
 670
 671 void
 672 Job::cancel ()
 673 {
 674         if (_thread.joinable()) {
 675                 Job::resume();
 676
 677                 _thread.interrupt ();
 678                 _thread.join ();
 679         }
 680
 681         set_state (FINISHED_CANCELLED);
 682 }
 683
 684
 685 /** @return true if the job was paused, false if it was not running */
 686 bool
 687 Job::pause_by_user ()
 688 {
 689         bool paused = false;
 690         {
 691                 boost::mutex::scoped_lock lm (_state_mutex);
 692                 /* We can set _state here directly because we have a lock and we aren't
 693                    setting the job to FINISHED_*
 694                 */
 695                 if (_state == RUNNING) {
 696                         paused = true;
 697                         _state = PAUSED_BY_USER;
 698                 }
 699         }
 700
 701         if (paused) {
 702                 pause();
 703                 _pause_changed.notify_all ();
 704         }
 705
 706         return paused;
 707 }
 708
 709
 710 void
 711 Job::pause_by_priority ()
 712 {
 713         if (running ()) {
 714                 set_state (PAUSED_BY_PRIORITY);
 715                 pause();
 716                 _pause_changed.notify_all ();
 717         }
 718 }
 719
 720
 721 void
 722 Job::resume ()
 723 {
 724         if (paused_by_user() || paused_by_priority()) {
 725                 set_state (RUNNING);
 726                 _pause_changed.notify_all ();
 727         }
 728 }
 729
 730
 731 void
 732 Job::when_finished(boost::signals2::connection& connection, function<void(Result)> finished)
 733 {
 734         boost::mutex::scoped_lock lm (_state_mutex);
 735         if (_state == FINISHED_OK || _state == FINISHED_ERROR || _state == FINISHED_CANCELLED) {
 736                 finished(state_to_result(_state));
 737         } else {
 738                 connection = Finished.connect (finished);
 739         }
 740 }
 741
 742
 743 optional<string>
 744 Job::message () const
 745 {
 746         boost::mutex::scoped_lock lm (_state_mutex);
 747         return _message;
 748 }
 749
 750
 751 void
 752 Job::set_message (string m)
 753 {
 754         boost::mutex::scoped_lock lm (_state_mutex);
 755         _message = m;
 756 }
 757
 758
 759 void
 760 Job::set_rate_limit_progress(bool rate_limit)
 761 {
 762         _rate_limit_progress = rate_limit;
 763 }
 764