orthanc: Core/JobsEngine/JobsRegistry.cpp comparison

comparison Core/JobsEngine/JobsRegistry.cpp @ 4024:1d2b31fc782f more-changes

new 'changes': JobSubmitted, JobSuccess, JobFailure

author	Alain Mazy <alain@mazy.be>
date	Tue, 09 Jun 2020 12:20:42 +0200
parents	a2e4edc7b9aa
children

comparison

equal deleted inserted replaced

-:cbdf62468d77
+:1d2b31fc782f
 * modify file(s) with this exception, you may extend this exception to
 * your version of the file(s), but you are not obligated to do so. If
 * you do not wish to do so, delete this exception statement from your
 * version. If you delete this exception statement from all source files
 * in the program, then also delete it here.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * General Public License for more details.
 *
 static const char* JOBS = "Jobs";
 static const char* JOBS_REGISTRY = "JobsRegistry";
 static const char* CREATION_TIME = "CreationTime";
 static const char* LAST_CHANGE_TIME = "LastChangeTime";
 static const char* RUNTIME = "Runtime";
 class JobsRegistry::JobHandler : public boost::noncopyable
 {
 private:
 std::string                       id_;
 JobState                          state_;
 std::string                       jobType_;
 std::unique_ptr<IJob>             job_;
 }
 lastStateChangeTime_ = now;
 }
 void SetStateInternal(JobState state)
 {
 state_ = state;
 pauseScheduled_ = false;
 cancelScheduled_ = false;
 Touch();
 JobState GetState() const
 {
 return state_;
 }
 void SetState(JobState state)
 {
 if (state == JobState_Retry)
 {
 // Use "SetRetryState()"
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 void SetRetryState(unsigned int timeout)
 {
 if (state_ == JobState_Running)
 {
 SetStateInternal(JobState_Retry);
 retryTime_ = (boost::posix_time::microsec_clock::universal_time() +
 boost::posix_time::milliseconds(timeout));
 }
 else
 {
 // Only valid for running jobs
 else
 {
 ok = false;
 }
 }
 else
 {
 ok = job_->Serialize(target[JOB]);
 }
 if (ok)
 bool JobsRegistry::PriorityComparator::operator() (JobHandler*& a,
 JobHandler*& b) const
 {
 return a->GetPriority() < b->GetPriority();
 }
 #if defined(NDEBUG)
 void JobsRegistry::CheckInvariants() const
 {
 }
 #else
 bool JobsRegistry::IsPendingJob(const JobHandler& job) const
 {
 PendingJobs copy = pendingJobs_;
 while (!copy.empty())
 switch (job.GetState())
 {
 case JobState_Pending:
 assert(!IsRetryJob(job) && IsPendingJob(job) && !IsCompletedJob(job));
 break;
 case JobState_Success:
 case JobState_Failure:
 assert(!IsRetryJob(job) && !IsPendingJob(job) && IsCompletedJob(job));
 break;
 case JobState_Retry:
 assert(IsRetryJob(job) && !IsPendingJob(job) && !IsCompletedJob(job));
 break;
 case JobState_Running:
 case JobState_Paused:
 assert(!IsRetryJob(job) && !IsPendingJob(job) && !IsCompletedJob(job));
 break;
 break;
 default:
 throw OrthancException(ErrorCode_InternalError);
 }
 LOG(INFO) << "Job has completed with " << tmp << ": " << job.GetId();
 CheckInvariants();
 assert(job.GetState() == JobState_Running);
 if (reason == CompletedReason_Canceled)
 {
 job.SetLastErrorCode(ErrorCode_CanceledJob);
 }
-{
+if (observer_ != NULL)
-boost::shared_lock<boost::shared_mutex> lock(observersMutex_);
+{
+if (reason == CompletedReason_Success)
-for (Observers::iterator it = observers_.begin(); it != observers_.end(); ++it)
+{
-{
+observer_->SignalJobSuccess(job.GetId());
-if (reason == CompletedReason_Success)
+}
-{
+else
-(*it)->SignalJobSuccess(job.GetId());
+{
-}
+observer_->SignalJobFailure(job.GetId());
-else
-{
-(*it)->SignalJobFailure(job.GetId());
-}
 }
 }
 // WARNING: The following call might make "job" invalid if the job
 // history size is empty
 state = it->second->GetState();
 return true;
 }
 }
 JobsRegistry::~JobsRegistry()
 {
 for (JobsIndex::iterator it = jobsIndex_.begin(); it != jobsIndex_.end(); ++it)
 {
 assert(it->second != NULL);
 {
 if (handler == NULL)
 {
 throw OrthancException(ErrorCode_NullPointer);
 }
 std::unique_ptr<JobHandler>  protection(handler);
 {
 boost::mutex::scoped_lock lock(mutex_);
 CheckInvariants();
 id = handler->GetId();
 int priority = handler->GetPriority();
 jobsIndex_.insert(std::make_pair(id, protection.release()));
 case JobState_Running:
 handler->SetState(JobState_Pending);
 pendingJobs_.push(handler);
 pendingJobAvailable_.notify_one();
 break;
 case JobState_Success:
 SetCompletedJob(*handler, true);
 break;
 case JobState_Failure:
 SetCompletedJob(*handler, false);
 break;
 case JobState_Paused:
 break;
 default:
 {
 std::string details = ("A job should not be loaded from state: " +
 std::string(EnumerationToString(handler->GetState())));
 throw OrthancException(ErrorCode_InternalError, details);
 }
 }
 LOG(INFO) << "New job submitted with priority " << priority << ": " << id;
-{
+if (observer_ != NULL)
-boost::shared_lock<boost::shared_mutex> lock(observersMutex_);
+{
+observer_->SignalJobSubmitted(id);
-for (Observers::iterator it = observers_.begin(); it != observers_.end(); ++it)
-{
-(*it)->SignalJobSubmitted(id);
-}
 }
 // WARNING: The following call might make "handler" invalid if
 // the job history size is empty
 ForgetOldCompletedJobs();
 else
 {
 const JobStatus& status = it->second->GetLastStatus();
 successContent = status.GetPublicContent();
 }
 return;
 }
 else
 {
 // This job has not finished yet, wait for new completion
 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)
 case JobState_Paused:
 SetCompletedJob(*found->second, false);
 found->second->SetLastErrorCode(ErrorCode_CanceledJob);
 break;
 case JobState_Success:
 case JobState_Failure:
 // Nothing to be done
 break;
 {
 found->second->SetState(JobState_Pending);
 pendingJobs_.push(found->second);
 pendingJobAvailable_.notify_one();
 CheckInvariants();
 return true;
 }
 }
 bool JobsRegistry::Resubmit(const std::string& id)
 return false;
 }
 else
 {
 found->second->GetJob().Reset();
 bool ok = false;
 for (CompletedJobs::iterator it = completedJobs_.begin();
 it != completedJobs_.end(); ++it)
 {
 if (*it == found->second)
 {
 ok = true;
 boost::mutex::scoped_lock lock(mutex_);
 return GetStateInternal(state, id);
 }
-void JobsRegistry::AddObserver(JobsRegistry::IObserver& observer)
+void JobsRegistry::SetObserver(JobsRegistry::IObserver& observer)
 {
-boost::unique_lock<boost::shared_mutex> lock(observersMutex_);
+boost::mutex::scoped_lock lock(mutex_);
-observers_.insert(&observer);
+observer_ = &observer;
 }
-void JobsRegistry::ResetObserver(JobsRegistry::IObserver& observer)
+void JobsRegistry::ResetObserver()
 {
-boost::unique_lock<boost::shared_mutex> lock(observersMutex_);
+boost::mutex::scoped_lock lock(mutex_);
-Observers::iterator it = observers_.find(&observer);
+observer_ = NULL;
-if (it != observers_.end())
+}
-{
-observers_.erase(it);
-}
-}
 JobsRegistry::RunningJob::RunningJob(JobsRegistry& registry,
 unsigned int timeout) :
 registry_(registry),
 handler_(NULL),
 targetState_(JobState_Failure),
 id_ = handler_->GetId();
 priority_ = handler_->GetPriority();
 }
 }
 JobsRegistry::RunningJob::~RunningJob()
 {
 if (IsValid())
 {
 boost::mutex::scoped_lock lock(registry_.mutex_);
 registry_.MarkRunningAsCompleted(*handler_, CompletedReason_Success);
 break;
 case JobState_Paused:
 registry_.MarkRunningAsPaused(*handler_);
 break;
 case JobState_Retry:
 registry_.MarkRunningAsRetry(*handler_, targetRetryTimeout_);
 break;
 default:
 assert(0);
 }
 }
 }
 bool JobsRegistry::RunningJob::IsValid() const
 {
 return (handler_ != NULL &&
 job_ != NULL);
 }
 const std::string& JobsRegistry::RunningJob::GetId() const
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 {
 return id_;
 }
 }
 int JobsRegistry::RunningJob::GetPriority() const
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 else
 {
 return priority_;
 }
 }
 IJob& JobsRegistry::RunningJob::GetJob()
 {
 if (!IsValid())
 {
 {
 return *job_;
 }
 }
 bool JobsRegistry::RunningJob::IsPauseScheduled()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 else
 {
 boost::mutex::scoped_lock lock(registry_.mutex_);
 registry_.CheckInvariants();
 assert(handler_->GetState() == JobState_Running);
 return handler_->IsPauseScheduled();
 }
 }
 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())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 {
 targetState_ = JobState_Success;
 }
 }
 void JobsRegistry::RunningJob::MarkFailure()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 {
 targetState_ = JobState_Failure;
 }
 }
 void JobsRegistry::RunningJob::MarkCanceled()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 targetState_ = JobState_Failure;
 canceled_ = true;
 }
 }
 void JobsRegistry::RunningJob::MarkPause()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 {
 targetState_ = JobState_Paused;
 }
 }
 void JobsRegistry::RunningJob::MarkRetry(unsigned int timeout)
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 {
 targetState_ = JobState_Retry;
 targetRetryTimeout_ = timeout;
 }
 }
 void JobsRegistry::RunningJob::UpdateStatus(ErrorCode code,
 const std::string& details)
 {
 if (!IsValid())
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
 else
 {
 JobStatus status(code, details, *job_);
 boost::mutex::scoped_lock lock(registry_.mutex_);
 registry_.CheckInvariants();
 assert(handler_->GetState() == JobState_Running);
 handler_->SetLastStatus(status);
 }
 }
 CheckInvariants();
 target = Json::objectValue;
 target[TYPE] = JOBS_REGISTRY;
 target[JOBS] = Json::objectValue;
 for (JobsIndex::const_iterator it = jobsIndex_.begin();
 it != jobsIndex_.end(); ++it)
 {
 Json::Value v;
 if (it->second->Serialize(v))
 {
 JobsRegistry::JobsRegistry(IJobUnserializer& unserializer,
 const Json::Value& s,
 size_t maxCompletedJobs) :
-maxCompletedJobs_(maxCompletedJobs)
+maxCompletedJobs_(maxCompletedJobs),
+observer_(NULL)
 {
 if (SerializationToolbox::ReadString(s, TYPE) != JOBS_REGISTRY ||
 !s.isMember(JOBS) ||
 s[JOBS].type() != Json::objectValue)
 {
 pending = 0;
 running = 0;
 success = 0;
 failed = 0;
 for (JobsIndex::const_iterator it = jobsIndex_.begin();
 it != jobsIndex_.end(); ++it)
 {
 JobHandler& job = *it->second;
 case JobState_Paused:
 case JobState_Running:
 running ++;
 break;
 case JobState_Success:
 success ++;
 break;
 case JobState_Failure:
 break;
 default:
 throw OrthancException(ErrorCode_InternalError);
 }
 }
 }
 }

Mercurial > hg > orthanc

comparison Core/JobsEngine/JobsRegistry.cpp @ 4024:1d2b31fc782f more-changes