orthanc: UnitTestsSources/MultiThreadingTests.cpp comparison

comparison UnitTestsSources/MultiThreadingTests.cpp @ 2562:1e66fe3ddf9f jobs

refactoring

author	Sebastien Jodogne <s.jodogne@gmail.com>
date	Fri, 04 May 2018 17:28:47 +0200
parents	9b7680dee75d
children	98dfc1948d00

comparison

equal deleted inserted replaced

-:7d4a3eca96af
+:1e66fe3ddf9f
 JobState_Failure,
 JobState_Paused,
 JobState_Retry
 };
-enum JobStepStatus
+enum JobStepCode
 {
-JobStepStatus_Success,
+JobStepCode_Success,
-JobStepStatus_Failure,
+JobStepCode_Failure,
-JobStepStatus_Continue,
+JobStepCode_Continue,
-JobStepStatus_Retry
+JobStepCode_Retry
 };
 class JobStepResult
 {
 private:
-JobStepStatus status_;
+JobStepCode status_;
 public:
-explicit JobStepResult(JobStepStatus status) :
+explicit JobStepResult(JobStepCode status) :
 status_(status)
 {
 }
 virtual ~JobStepResult()
 {
 }
-JobStepStatus GetStatus() const
+JobStepCode GetCode() const
 {
 return status_;
 }
 };
 private:
 unsigned int  timeout_;   // Retry after "timeout_" milliseconds
 public:
 RetryResult(unsigned int timeout) :
-JobStepResult(JobStepStatus_Retry),
+JobStepResult(JobStepCode_Retry),
 timeout_(timeout)
 {
 }
 unsigned int  GetTimeout() const
 virtual void ReleaseResources() = 0;   // For pausing jobs
 virtual float GetProgress() = 0;
 virtual void FormatStatus(Json::Value& value) = 0;
+};
+struct JobStatus
+{
+ErrorCode      errorCode_;
+float          progress_;
+Json::Value    description_;
+JobStatus() :
+errorCode_(ErrorCode_Success),
+progress_(0),
+description_(Json::objectValue)
+{
+}
+JobStatus(ErrorCode code,
+float progress) :
+errorCode_(code),
+progress_(progress),
+description_(Json::objectValue)
+{
+if (progress < 0 ||
+progress > 1)
+{
+throw OrthancException(ErrorCode_ParameterOutOfRange);
+}
+}
 };
 class JobInfo
 {
 private:
 std::string                       id_;
 int                               priority_;
 ErrorCode                         errorCode_;
 JobState                          state_;
-float                             progress_;
 boost::posix_time::ptime          infoTime_;
 boost::posix_time::ptime          creationTime_;
 boost::posix_time::time_duration  runtime_;
 boost::posix_time::ptime          eta_;
-Json::Value                       status_;
+JobStatus                         status_;
 public:
 JobInfo(const std::string& id,
 int priority,
-ErrorCode errorCode,
 JobState state,
-float progress,
+const JobStatus& status,
 const boost::posix_time::ptime& creationTime,
 const boost::posix_time::time_duration& runtime) :
 id_(id),
 priority_(priority),
-errorCode_(errorCode),
 state_(state),
-progress_(progress),
 infoTime_(boost::posix_time::microsec_clock::universal_time()),
 creationTime_(creationTime),
-runtime_(runtime)
+runtime_(runtime),
-{
+status_(status)
-if (progress < 0 ||
+{
-progress > 1)
+float ms = static_cast<float>(runtime_.total_milliseconds());
-{
+float remaining = boost::math::llround(1.0f - status_.progress_) * ms;
-throw OrthancException(ErrorCode_ParameterOutOfRange);
+eta_ = infoTime_ + boost::posix_time::milliseconds(remaining);
-}
-float r = static_cast<float>(runtime_.total_milliseconds());
-eta_ = infoTime_ + boost::posix_time::milliseconds(boost::math::llround(1.0f - progress) * r);
 }
 const std::string& GetIdentifier() const
 {
 return id_;
 JobState GetState() const
 {
 return state_;
 }
-float GetProgress() const
-{
-return progress_;
-}
 const boost::posix_time::ptime& GetInfoTime() const
 {
 return infoTime_;
 }
 const boost::posix_time::ptime& GetEstimatedTimeOfArrival() const
 {
 return eta_;
 }
-const Json::Value& GetStatus() const
+const JobStatus& GetStatus() const
 {
 return status_;
 }
-Json::Value& GetStatus()
+JobStatus& GetStatus()
 {
 return status_;
 }
 };
 class JobHandler : public boost::noncopyable
 {
 private:
 std::string                       id_;
 JobState                          state_;
-boost::mutex                      jobMutex_;
 std::auto_ptr<IJob>               job_;
 int                               priority_;  // "+inf()" means highest priority
 boost::posix_time::ptime          creationTime_;
-boost::posix_time::ptime          lastStateChangeTime;
+boost::posix_time::ptime          lastStateChangeTime_;
+boost::posix_time::time_duration  runtime_;
 boost::posix_time::ptime          retryTime_;
-boost::posix_time::time_duration  runtime_;
 bool                              pauseScheduled_;
-ErrorCode                         lastErrorCode_;
+JobStatus                         lastStatus_;
-float                             lastProgress_;
-Json::Value                       lastStatus_;
 void SetStateInternal(JobState state)
 {
 const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time();
 if (state_ == JobState_Running)
 {
-runtime_ += (now - lastStateChangeTime);
+runtime_ += (now - lastStateChangeTime_);
 }
 state_ = state;
-lastStateChangeTime = now;
+lastStateChangeTime_ = now;
 pauseScheduled_ = false;
 }
 public:
 JobHandler(IJob* job,
 id_(Toolbox::GenerateUuid()),
 state_(JobState_Pending),
 job_(job),
 priority_(priority),
 creationTime_(boost::posix_time::microsec_clock::universal_time()),
-lastStateChangeTime(creationTime_),
+lastStateChangeTime_(creationTime_),
 runtime_(boost::posix_time::milliseconds(0)),
-pauseScheduled_(false),
+retryTime_(creationTime_),
-lastErrorCode_(ErrorCode_Success),
+pauseScheduled_(false)
-lastProgress_(0)
 {
 if (job == NULL)
 {
 throw OrthancException(ErrorCode_NullPointer);
 }
 }
 const std::string& GetId() const
 {
 return id_;
+}
+IJob& GetJob() const
+{
+assert(job_.get() != NULL);
+return *job_;
 }
 void SetPriority(int priority)
 {
 priority_ = priority;
 {
 return retryTime_ <= now;
 }
 }
-class JobLock
+JobStatus& GetLastStatus()
 {
-private:
+return lastStatus_;
-boost::mutex::scoped_lock  lock_;
+}
-JobHandler&                handler_;
+const JobStatus& GetLastStatus() const
-public:
+{
-JobLock(JobHandler& handler) :
+return lastStatus_;
-lock_(handler.jobMutex_),
+}
-handler_(handler)
-{
-}
-IJob& GetJob()
-{
-return *handler_.job_;
-}
-void UpdateStatus()
-{
-handler_.lastProgress_ = handler_.job_->GetProgress();
-handler_.job_->FormatStatus(handler_.lastStatus_);
-}
-void SetLastErrorCode(ErrorCode code)
-{
-handler_.lastErrorCode_ = code;
-}
-};
 };
 class JobsRegistry : public boost::noncopyable
 {
 bool success)
 {
 LOG(INFO) << "Job has completed with " << (success ? "success" : "failure")
 << ": " << job.GetId();
-boost::mutex::scoped_lock lock(mutex_);
 CheckInvariants();
 assert(job.GetState() == JobState_Running);
 job.SetState(success ? JobState_Success : JobState_Failure);
 void MarkRunningAsRetry(JobHandler& job,
 unsigned int timeout)
 {
 LOG(INFO) << "Job scheduled for retry in " << timeout << "ms: " << job.GetId();
-boost::mutex::scoped_lock lock(mutex_);
 CheckInvariants();
 assert(job.GetState() == JobState_Running &&
 retryJobs_.find(&job) == retryJobs_.end());
 void MarkRunningAsPaused(JobHandler& job)
 {
 LOG(INFO) << "Job paused: " << job.GetId();
-boost::mutex::scoped_lock lock(mutex_);
 CheckInvariants();
 assert(job.GetState() == JobState_Running);
 job.SetState(JobState_Paused);
 CheckInvariants();
-}
-JobHandler* WaitPendingJob(unsigned int timeout)
-{
-boost::mutex::scoped_lock lock(mutex_);
-while (pendingJobs_.empty())
-{
-if (timeout == 0)
-{
-pendingJobAvailable_.wait(lock);
-}
-else
-{
-bool success = pendingJobAvailable_.timed_wait
-(lock, boost::posix_time::milliseconds(timeout));
-if (!success)
-{
-return NULL;
-}
-}
-}
-JobHandler* job = pendingJobs_.top();
-pendingJobs_.pop();
-job->SetState(JobState_Running);
-return job;
 }
 public:
 JobsRegistry() :
 class RunningJob : public boost::noncopyable
 {
 private:
-JobsRegistry&  that_;
+JobsRegistry&  registry_;
-JobHandler*    handler_;
+JobHandler*    handler_;  // Can only be accessed if the registry
+// mutex is locked!
+IJob*          job_;  // Will by design be in mutual exclusion,
+// because only one RunningJob can be
+// executed at a time on a JobHandler
+std::string    id_;
+int            priority_;
 JobState       targetState_;
-unsigned int   retryTimeout_;
+unsigned int   targetRetryTimeout_;
 public:
-RunningJob(JobsRegistry& that,
+RunningJob(JobsRegistry& registry,
 unsigned int timeout) :
-that_(that),
+registry_(registry),
 handler_(NULL),
 targetState_(JobState_Failure),
-retryTimeout_(0)
+targetRetryTimeout_(0)
 {
-handler_ = that_.WaitPendingJob(timeout);
+{
+boost::mutex::scoped_lock lock(registry_.mutex_);
+while (registry_.pendingJobs_.empty())
+{
+if (timeout == 0)
+{
+registry_.pendingJobAvailable_.wait(lock);
+}
+else
+{
+bool success = registry_.pendingJobAvailable_.timed_wait
+(lock, boost::posix_time::milliseconds(timeout));
+if (!success)
+{
+// No pending job
+return;
+}
+}
+}
+handler_ = registry_.pendingJobs_.top();
+registry_.pendingJobs_.pop();
+assert(handler_->GetState() == JobState_Pending);
+handler_->SetState(JobState_Running);
+job_ = &handler_->GetJob();
+id_ = handler_->GetId();
+priority_ = handler_->GetPriority();
+}
 }
 ~RunningJob()
 {
 if (IsValid())
 {
+boost::mutex::scoped_lock lock(registry_.mutex_);
 switch (targetState_)
 {
 case JobState_Failure:
-that_.MarkRunningAsCompleted(*handler_, false);
+registry_.MarkRunningAsCompleted(*handler_, false);
 break;
 case JobState_Success:
-that_.MarkRunningAsCompleted(*handler_, true);
+registry_.MarkRunningAsCompleted(*handler_, true);
 break;
 case JobState_Paused:
-that_.MarkRunningAsPaused(*handler_);
+registry_.MarkRunningAsPaused(*handler_);
 break;
 case JobState_Retry:
-that_.MarkRunningAsRetry(*handler_, retryTimeout_);
+registry_.MarkRunningAsRetry(*handler_, targetRetryTimeout_);
 break;
 default:
 assert(0);
 }
 }
 }
 bool IsValid() const
 {
-return handler_ != NULL;
+return (handler_ != NULL &&
+job_ != NULL);
 }
 const std::string& GetId() const
 {
-if (IsValid())
+if (!IsValid())
 {
-return handler_->GetId();
+throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
 else
 {
+return id_;
+}
+}
+int GetPriority() const
+{
+if (!IsValid())
+{
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
-}
-}
-int GetPriority() const
-{
-if (IsValid())
-{
-return handler_->GetPriority();
 }
 else
 {
+return priority_;
+}
+}
+bool IsPauseScheduled()
+{
+if (!IsValid())
+{
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
-}
+else
+{
-bool IsPauseScheduled()
+boost::mutex::scoped_lock lock(registry_.mutex_);
+registry_.CheckInvariants();
+assert(handler_->GetState() == JobState_Running);
+return handler_->IsPauseScheduled();
+}
+}
+void MarkSuccess()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
+else
-boost::mutex::scoped_lock lock(that_.mutex_);
+{
-that_.CheckInvariants();
+targetState_ = JobState_Success;
-assert(handler_->GetState() == JobState_Running);
+}
+}
-return handler_->IsPauseScheduled();
-}
+void MarkFailure()
-void MarkSuccess()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
+else
-targetState_ = JobState_Success;
+{
-}
+targetState_ = JobState_Failure;
+}
-void MarkFailure()
+}
+void MarkPause()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
+else
-targetState_ = JobState_Failure;
+{
-}
+targetState_ = JobState_Paused;
+}
-void SchedulePause()
+}
+void MarkRetry(unsigned int timeout)
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
+else
-targetState_ = JobState_Paused;
+{
-}
+targetState_ = JobState_Retry;
+targetRetryTimeout_ = timeout;
-void MarkRetry(unsigned int timeout)
+}
+}
+/*void ExecuteStep()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
-targetState_ = JobState_Retry;
+if (IsPauseScheduled())
-retryTimeout_ = timeout;
-}
-void ExecuteStep()
-{
-if (!IsValid())
-{
-throw OrthancException(ErrorCode_BadSequenceOfCalls);
-}
-if (handler_->IsPauseScheduled())
 {
 targetState_ = JobState_Paused;
 return;
 }
 std::auto_ptr<JobStepResult> result;
+ErrorCode code;
-{
-JobHandler::JobLock lock(*handler_);
+{
 bool ok = false;
 try
 {
-result.reset(lock.GetJob().ExecuteStep());
+result.reset(job_->ExecuteStep());
-lock.UpdateStatus();
 ok = true;
+if (result->GetCode() == JobStepCode_Failure)
+{
+code = ErrorCode_InternalError;
+}
 }
 catch (OrthancException& e)
 {
-lock.SetLastErrorCode(e.GetErrorCode());
+code = e.GetErrorCode();
 }
 catch (boost::bad_lexical_cast&)
 {
-lock.SetLastErrorCode(ErrorCode_BadFileFormat);
+code = ErrorCode_BadFileFormat;
 }
 catch (...)
 {
-lock.SetLastErrorCode(ErrorCode_InternalError);
+code = ErrorCode_InternalError;
 }
 if (ok)
 {
-lock.SetLastErrorCode(ErrorCode_Success);
+code = ErrorCode_Success;
 }
 else
 {
-result.reset(new JobStepResult(JobStepStatus_Failure));
+result.reset(new JobStepResult(JobStepCode_Failure));
 }
 }
-switch (result->GetStatus())
+switch (result->GetCode())
 {
-case JobStepStatus_Success:
+case JobStepCode_Success:
 targetState_ = JobState_Success;
 break;
-case JobStepStatus_Failure:
+case JobStepCode_Failure:
 targetState_ = JobState_Failure;
 break;
-case JobStepStatus_Continue:
+case JobStepCode_Continue:
 targetState_ = JobState_Running;
 break;
-case JobStepStatus_Retry:
+case JobStepCode_Retry:
 targetState_ = JobState_Retry;
-retryTimeout_ = dynamic_cast<RetryResult&>(*result).GetTimeout();
+targetRetryTimeout_ = dynamic_cast<RetryResult&>(*result).GetTimeout();
 break;
 default:
 throw OrthancException(ErrorCode_InternalError);
 }
-}
+}*/
 };
 };
 }
 private:
 JobStepResult  result_;
 public:
 DummyJob() :
-result_(Orthanc::JobStepStatus_Success)
+result_(Orthanc::JobStepCode_Success)
 {
 }
 explicit DummyJob(JobStepResult result) :
 result_(result)
 {
 JobsRegistry::RunningJob job(registry, 0);
 ASSERT_TRUE(job.IsValid());
 registry.Resubmit(id);
-job.SchedulePause();
+job.MarkPause();
 ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running));
 }
 ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Paused));

Mercurial > hg > orthanc

comparison UnitTestsSources/MultiThreadingTests.cpp @ 2562:1e66fe3ddf9f jobs