# HG changeset patch # User Sebastien Jodogne # Date 1526052799 -7200 # Node ID 8da2cffc23787e695c5d7fe750991c7127d6e4cd # Parent 055d7d4a823f2de5e7d9604de4064cea7c2ae9f6 JobsRegistry::Cancel() diff -r 055d7d4a823f -r 8da2cffc2378 Core/Enumerations.cpp --- a/Core/Enumerations.cpp Fri May 11 09:35:17 2018 +0200 +++ b/Core/Enumerations.cpp Fri May 11 17:33:19 2018 +0200 @@ -164,6 +164,9 @@ case ErrorCode_DatabaseUnavailable: return "The database is currently not available (probably a transient situation)"; + case ErrorCode_CanceledJob: + return "This job was canceled"; + case ErrorCode_SQLiteNotOpened: return "SQLite: The database is not opened"; diff -r 055d7d4a823f -r 8da2cffc2378 Core/Enumerations.h --- a/Core/Enumerations.h Fri May 11 09:35:17 2018 +0200 +++ b/Core/Enumerations.h Fri May 11 17:33:19 2018 +0200 @@ -96,6 +96,7 @@ ErrorCode_NotAcceptable = 34 /*!< Cannot send a response which is acceptable according to the Accept HTTP header */, ErrorCode_NullPointer = 35 /*!< Cannot handle a NULL pointer */, ErrorCode_DatabaseUnavailable = 36 /*!< The database is currently not available (probably a transient situation) */, + ErrorCode_CanceledJob = 37 /*!< This job was canceled */, ErrorCode_SQLiteNotOpened = 1000 /*!< SQLite: The database is not opened */, ErrorCode_SQLiteAlreadyOpened = 1001 /*!< SQLite: Connection is already open */, ErrorCode_SQLiteCannotOpen = 1002 /*!< SQLite: Unable to open the database */, diff -r 055d7d4a823f -r 8da2cffc2378 Core/JobsEngine/JobStatus.h --- a/Core/JobsEngine/JobStatus.h Fri May 11 09:35:17 2018 +0200 +++ b/Core/JobsEngine/JobStatus.h Fri May 11 17:33:19 2018 +0200 @@ -57,6 +57,11 @@ return errorCode_; } + void SetErrorCode(ErrorCode error) + { + errorCode_ = error; + } + float GetProgress() const { return progress_; diff -r 055d7d4a823f -r 8da2cffc2378 Core/JobsEngine/JobsEngine.cpp --- a/Core/JobsEngine/JobsEngine.cpp Fri May 11 09:35:17 2018 +0200 +++ b/Core/JobsEngine/JobsEngine.cpp Fri May 11 17:33:19 2018 +0200 @@ -60,6 +60,13 @@ return false; } + if (running.IsCancelScheduled()) + { + running.GetJob().ReleaseResources(); + running.MarkCanceled(); + return false; + } + std::auto_ptr result; { diff -r 055d7d4a823f -r 8da2cffc2378 Core/JobsEngine/JobsRegistry.cpp --- a/Core/JobsEngine/JobsRegistry.cpp Fri May 11 09:35:17 2018 +0200 +++ b/Core/JobsEngine/JobsRegistry.cpp Fri May 11 17:33:19 2018 +0200 @@ -52,6 +52,7 @@ boost::posix_time::time_duration runtime_; boost::posix_time::ptime retryTime_; bool pauseScheduled_; + bool cancelScheduled_; JobStatus lastStatus_; void Touch() @@ -70,6 +71,7 @@ { state_ = state; pauseScheduled_ = false; + cancelScheduled_ = false; Touch(); } @@ -84,7 +86,8 @@ lastStateChangeTime_(creationTime_), runtime_(boost::posix_time::milliseconds(0)), retryTime_(creationTime_), - pauseScheduled_(false) + pauseScheduled_(false), + cancelScheduled_(false) { if (job == NULL) { @@ -162,11 +165,29 @@ } } + void ScheduleCancel() + { + if (state_ == JobState_Running) + { + cancelScheduled_ = true; + } + else + { + // Only valid for running jobs + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + } + bool IsPauseScheduled() { return pauseScheduled_; } + bool IsCancelScheduled() + { + return cancelScheduled_; + } + bool IsRetryReady(const boost::posix_time::ptime& now) const { if (state_ != JobState_Retry) @@ -204,6 +225,11 @@ lastStatus_ = status; Touch(); } + + void SetLastErrorCode(ErrorCode code) + { + lastStatus_.SetErrorCode(code); + } }; @@ -335,6 +361,18 @@ } + void JobsRegistry::SetCompletedJob(JobHandler& job, + bool success) + { + job.SetState(success ? JobState_Success : JobState_Failure); + + completedJobs_.push_back(&job); + ForgetOldCompletedJobs(); + + someJobComplete_.notify_all(); + } + + void JobsRegistry::MarkRunningAsCompleted(JobHandler& job, bool success) { @@ -342,14 +380,9 @@ << ": " << job.GetId(); CheckInvariants(); + assert(job.GetState() == JobState_Running); - - job.SetState(success ? JobState_Success : JobState_Failure); - - completedJobs_.push_back(&job); - ForgetOldCompletedJobs(); - - someJobComplete_.notify_all(); + SetCompletedJob(job, success); CheckInvariants(); } @@ -501,8 +534,6 @@ std::string id; Submit(id, job, priority); - printf(">> %s\n", id.c_str()); fflush(stdout); - JobState state; { @@ -561,6 +592,34 @@ } + void JobsRegistry::RemovePendingJob(const std::string& id) + { + // If the job is pending, we need to reconstruct the priority + // queue to remove it + PendingJobs copy; + std::swap(copy, pendingJobs_); + + assert(pendingJobs_.empty()); + while (!copy.empty()) + { + if (copy.top()->GetId() != id) + { + pendingJobs_.push(copy.top()); + } + + copy.pop(); + } + } + + + void JobsRegistry::RemoveRetryJob(JobHandler* handler) + { + RetryJobs::iterator item = retryJobs_.find(handler); + assert(item != retryJobs_.end()); + retryJobs_.erase(item); + } + + bool JobsRegistry::Pause(const std::string& id) { LOG(INFO) << "Pausing job: " << id; @@ -580,38 +639,14 @@ switch (found->second->GetState()) { case JobState_Pending: - { - // If the job is pending, we need to reconstruct the - // priority queue to remove it - PendingJobs copy; - std::swap(copy, pendingJobs_); - - assert(pendingJobs_.empty()); - while (!copy.empty()) - { - if (copy.top()->GetId() != id) - { - pendingJobs_.push(copy.top()); - } - - copy.pop(); - } - + RemovePendingJob(id); found->second->SetState(JobState_Paused); - break; - } case JobState_Retry: - { - RetryJobs::iterator item = retryJobs_.find(found->second); - assert(item != retryJobs_.end()); - retryJobs_.erase(item); - + RemoveRetryJob(found->second); found->second->SetState(JobState_Paused); - break; - } case JobState_Paused: case JobState_Success: @@ -633,6 +668,60 @@ } + bool JobsRegistry::Cancel(const std::string& id) + { + LOG(INFO) << "Canceling job: " << id; + + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + JobsIndex::iterator found = jobsIndex_.find(id); + + if (found == jobsIndex_.end()) + { + LOG(WARNING) << "Unknown job: " << id; + return false; + } + else + { + switch (found->second->GetState()) + { + case JobState_Pending: + RemovePendingJob(id); + SetCompletedJob(*found->second, false); + found->second->SetLastErrorCode(ErrorCode_CanceledJob); + break; + + case JobState_Retry: + RemoveRetryJob(found->second); + SetCompletedJob(*found->second, false); + found->second->SetLastErrorCode(ErrorCode_CanceledJob); + break; + + case JobState_Paused: + SetCompletedJob(*found->second, false); + found->second->SetLastErrorCode(ErrorCode_CanceledJob); + break; + + case JobState_Success: + case JobState_Failure: + // Nothing to be done + break; + + case JobState_Running: + found->second->ScheduleCancel(); + break; + + default: + throw OrthancException(ErrorCode_InternalError); + } + + CheckInvariants(); + return true; + } + } + + bool JobsRegistry::Resume(const std::string& id) { LOG(INFO) << "Resuming job: " << id; @@ -751,7 +840,8 @@ registry_(registry), handler_(NULL), targetState_(JobState_Failure), - targetRetryTimeout_(0) + targetRetryTimeout_(0), + canceled_(false) { { boost::mutex::scoped_lock lock(registry_.mutex_); @@ -779,6 +869,7 @@ assert(handler_->GetState() == JobState_Pending); handler_->SetState(JobState_Running); + handler_->SetLastErrorCode(ErrorCode_Success); job_ = &handler_->GetJob(); id_ = handler_->GetId(); @@ -797,6 +888,12 @@ { case JobState_Failure: registry_.MarkRunningAsCompleted(*handler_, false); + + if (canceled_) + { + handler_->SetLastErrorCode(ErrorCode_CanceledJob); + } + break; case JobState_Success: @@ -881,6 +978,23 @@ } + bool JobsRegistry::RunningJob::IsCancelScheduled() + { + if (!IsValid()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + boost::mutex::scoped_lock lock(registry_.mutex_); + registry_.CheckInvariants(); + assert(handler_->GetState() == JobState_Running); + + return handler_->IsCancelScheduled(); + } + } + + void JobsRegistry::RunningJob::MarkSuccess() { if (!IsValid()) @@ -907,6 +1021,20 @@ } + void JobsRegistry::RunningJob::MarkCanceled() + { + if (!IsValid()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + targetState_ = JobState_Failure; + canceled_ = true; + } + } + + void JobsRegistry::RunningJob::MarkPause() { if (!IsValid()) diff -r 055d7d4a823f -r 8da2cffc2378 Core/JobsEngine/JobsRegistry.h --- a/Core/JobsEngine/JobsRegistry.h Fri May 11 09:35:17 2018 +0200 +++ b/Core/JobsEngine/JobsRegistry.h Fri May 11 17:33:19 2018 +0200 @@ -93,6 +93,9 @@ void ForgetOldCompletedJobs(); + void SetCompletedJob(JobHandler& job, + bool success); + void MarkRunningAsCompleted(JobHandler& job, bool success); @@ -104,6 +107,10 @@ bool GetStateInternal(JobState& state, const std::string& id); + void RemovePendingJob(const std::string& id); + + void RemoveRetryJob(JobHandler* handler); + public: JobsRegistry() : maxCompletedJobs_(10) @@ -138,6 +145,8 @@ bool Resume(const std::string& id); bool Resubmit(const std::string& id); + + bool Cancel(const std::string& id); void ScheduleRetries(); @@ -158,6 +167,7 @@ int priority_; JobState targetState_; unsigned int targetRetryTimeout_; + bool canceled_; public: RunningJob(JobsRegistry& registry, @@ -175,12 +185,16 @@ bool IsPauseScheduled(); + bool IsCancelScheduled(); + void MarkSuccess(); void MarkFailure(); void MarkPause(); + void MarkCanceled(); + void MarkRetry(unsigned int timeout); void UpdateStatus(ErrorCode code); diff -r 055d7d4a823f -r 8da2cffc2378 OrthancServer/main.cpp --- a/OrthancServer/main.cpp Fri May 11 09:35:17 2018 +0200 +++ b/OrthancServer/main.cpp Fri May 11 17:33:19 2018 +0200 @@ -574,6 +574,7 @@ PrintErrorCode(ErrorCode_NotAcceptable, "Cannot send a response which is acceptable according to the Accept HTTP header"); PrintErrorCode(ErrorCode_NullPointer, "Cannot handle a NULL pointer"); PrintErrorCode(ErrorCode_DatabaseUnavailable, "The database is currently not available (probably a transient situation)"); + PrintErrorCode(ErrorCode_CanceledJob, "This job was canceled"); PrintErrorCode(ErrorCode_SQLiteNotOpened, "SQLite: The database is not opened"); PrintErrorCode(ErrorCode_SQLiteAlreadyOpened, "SQLite: Connection is already open"); PrintErrorCode(ErrorCode_SQLiteCannotOpen, "SQLite: Unable to open the database"); diff -r 055d7d4a823f -r 8da2cffc2378 Plugins/Include/orthanc/OrthancCPlugin.h --- a/Plugins/Include/orthanc/OrthancCPlugin.h Fri May 11 09:35:17 2018 +0200 +++ b/Plugins/Include/orthanc/OrthancCPlugin.h Fri May 11 17:33:19 2018 +0200 @@ -235,6 +235,7 @@ OrthancPluginErrorCode_NotAcceptable = 34 /*!< Cannot send a response which is acceptable according to the Accept HTTP header */, OrthancPluginErrorCode_NullPointer = 35 /*!< Cannot handle a NULL pointer */, OrthancPluginErrorCode_DatabaseUnavailable = 36 /*!< The database is currently not available (probably a transient situation) */, + OrthancPluginErrorCode_CanceledJob = 37 /*!< This job was canceled */, OrthancPluginErrorCode_SQLiteNotOpened = 1000 /*!< SQLite: The database is not opened */, OrthancPluginErrorCode_SQLiteAlreadyOpened = 1001 /*!< SQLite: Connection is already open */, OrthancPluginErrorCode_SQLiteCannotOpen = 1002 /*!< SQLite: Unable to open the database */, diff -r 055d7d4a823f -r 8da2cffc2378 Resources/ErrorCodes.json --- a/Resources/ErrorCodes.json Fri May 11 09:35:17 2018 +0200 +++ b/Resources/ErrorCodes.json Fri May 11 17:33:19 2018 +0200 @@ -207,6 +207,11 @@ "Name": "DatabaseUnavailable", "Description": "The database is currently not available (probably a transient situation)" }, + { + "Code": 37, + "Name": "CanceledJob", + "Description": "This job was canceled" + }, diff -r 055d7d4a823f -r 8da2cffc2378 Resources/ImplementationNotes/JobsEngineStates.dot --- a/Resources/ImplementationNotes/JobsEngineStates.dot Fri May 11 09:35:17 2018 +0200 +++ b/Resources/ImplementationNotes/JobsEngineStates.dot Fri May 11 17:33:19 2018 +0200 @@ -20,4 +20,9 @@ pending -> paused [label="Pause()" fontcolor="red"]; retry -> paused [label="Pause()" fontcolor="red"]; running -> paused [label="Pause()" fontcolor="red"]; + + paused -> failure [label="Cancel()" fontcolor="red"]; + pending -> failure [label="Cancel()" fontcolor="red"]; + retry -> failure [label="Cancel()" fontcolor="red"]; + running -> failure [label="Cancel()" fontcolor="red"]; } diff -r 055d7d4a823f -r 8da2cffc2378 Resources/ImplementationNotes/JobsEngineStates.pdf Binary file Resources/ImplementationNotes/JobsEngineStates.pdf has changed diff -r 055d7d4a823f -r 8da2cffc2378 UnitTestsSources/MultiThreadingTests.cpp --- a/UnitTestsSources/MultiThreadingTests.cpp Fri May 11 09:35:17 2018 +0200 +++ b/UnitTestsSources/MultiThreadingTests.cpp Fri May 11 17:33:19 2018 +0200 @@ -323,6 +323,22 @@ } +static bool CheckErrorCode(Orthanc::JobsRegistry& registry, + const std::string& id, + Orthanc::ErrorCode code) +{ + Orthanc::JobInfo s; + if (registry.GetJobInfo(s, id)) + { + return code == s.GetStatus().GetErrorCode(); + } + else + { + return false; + } +} + + TEST(JobsRegistry, Priority) { JobsRegistry registry; @@ -611,6 +627,99 @@ } +TEST(JobsRegistry, Cancel) +{ + JobsRegistry registry; + + std::string id; + registry.Submit(id, new DummyJob(), 10); + + ASSERT_FALSE(registry.Cancel("nope")); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_Success)); + + ASSERT_TRUE(registry.Cancel(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Failure)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); + + ASSERT_TRUE(registry.Cancel(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Failure)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); + + ASSERT_TRUE(registry.Resubmit(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_Success)); + + job.MarkSuccess(); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Success)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_Success)); + + ASSERT_TRUE(registry.Cancel(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Success)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_Success)); + + registry.Submit(id, new DummyJob(), 10); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + ASSERT_EQ(id, job.GetId()); + + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_Success)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + + job.MarkCanceled(); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Failure)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); + + ASSERT_TRUE(registry.Resubmit(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); + + ASSERT_TRUE(registry.Pause(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Paused)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); + + ASSERT_TRUE(registry.Cancel(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Failure)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); + + ASSERT_TRUE(registry.Resubmit(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + ASSERT_EQ(id, job.GetId()); + + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_Success)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + + job.MarkRetry(500); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Retry)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_Success)); + + ASSERT_TRUE(registry.Cancel(id)); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Failure)); + ASSERT_TRUE(CheckErrorCode(registry, id, ErrorCode_CanceledJob)); +} + + TEST(JobsEngine, Basic)