orthanc: UnitTestsSources/MultiThreadingTests.cpp comparison

comparison UnitTestsSources/MultiThreadingTests.cpp @ 2565:0f06b4d5b3d0 jobs

JobsEngine

author	Sebastien Jodogne <s.jodogne@gmail.com>
date	Mon, 07 May 2018 13:20:29 +0200
parents	f8681f251caa
children	c09ce3c038fc

comparison

equal deleted inserted replaced

-:f8681f251caa
+:0f06b4d5b3d0
 JobState_Success,
 JobState_Failure,
 JobState_Paused,
 JobState_Retry
 };
+static const char* EnumerationToString(JobState state)
+{
+switch (state)
+{
+case JobState_Pending:
+return "Pending";
+case JobState_Running:
+return "Running";
+case JobState_Success:
+return "Success";
+case JobState_Failure:
+return "Failure";
+case JobState_Paused:
+return "Paused";
+case JobState_Retry:
+return "Retry";
+default:
+throw OrthancException(ErrorCode_ParameterOutOfRange);
+}
+}
 enum JobStepCode
 {
 JobStepCode_Success,
 JobStepCode_Failure,
 JobStatus(ErrorCode code,
 IJob& job) :
 errorCode_(code),
 progress_(job.GetProgress())
 {
-if (progress_ < 0 ||
+if (progress_ < 0)
-progress_ > 1)
+{
-{
+progress_ = 0;
-throw OrthancException(ErrorCode_ParameterOutOfRange);
+}
+if (progress_ > 1)
+{
+progress_ = 1;
 }
 job.GetDescription(description_);
 }
 class JobInfo
 {
 private:
 std::string                       id_;
 int                               priority_;
-ErrorCode                         errorCode_;
 JobState                          state_;
-boost::posix_time::ptime          infoTime_;
+boost::posix_time::ptime          timestamp_;
 boost::posix_time::ptime          creationTime_;
 boost::posix_time::time_duration  runtime_;
 boost::posix_time::ptime          eta_;
 JobStatus                         status_;
 const boost::posix_time::ptime& creationTime,
 const boost::posix_time::time_duration& runtime) :
 id_(id),
 priority_(priority),
 state_(state),
-infoTime_(boost::posix_time::microsec_clock::universal_time()),
+timestamp_(boost::posix_time::microsec_clock::universal_time()),
 creationTime_(creationTime),
 runtime_(runtime),
 status_(status)
 {
 float ms = static_cast<float>(runtime_.total_milliseconds());
 float remaining = boost::math::llround(1.0f - status_.GetProgress()) * ms;
-eta_ = infoTime_ + boost::posix_time::milliseconds(remaining);
+eta_ = timestamp_ + boost::posix_time::milliseconds(remaining);
+}
+JobInfo() :
+priority_(0),
+state_(JobState_Failure),
+timestamp_(boost::posix_time::microsec_clock::universal_time()),
+creationTime_(timestamp_),
+runtime_(boost::posix_time::milliseconds(0)),
+eta_(timestamp_)
+{
 }
 const std::string& GetIdentifier() const
 {
 return id_;
 int GetPriority() const
 {
 return priority_;
 }
-ErrorCode GetErrorCode() const
-{
-return errorCode_;
-}
 JobState GetState() const
 {
 return state_;
 }
 const boost::posix_time::ptime& GetInfoTime() const
 {
-return infoTime_;
+return timestamp_;
 }
 const boost::posix_time::ptime& GetCreationTime() const
 {
 return creationTime_;
 }
 JobStatus& GetStatus()
 {
 return status_;
+}
+void Format(Json::Value& target) const
+{
+target = Json::objectValue;
+target["ID"] = id_;
+target["Priority"] = priority_;
+target["ErrorCode"] = EnumerationToString(status_.GetErrorCode());
+target["State"] = EnumerationToString(state_);
+target["Timestamp"] = boost::posix_time::to_iso_string(timestamp_);
+target["CreationTime"] = boost::posix_time::to_iso_string(creationTime_);
+target["Runtime"] = static_cast<uint32_t>(runtime_.total_milliseconds());
+target["EstimatedTimeOfArrival"] = boost::posix_time::to_iso_string(eta_);
+target["Progress"] = boost::math::iround(status_.GetProgress() * 100.0f);
+target["Description"] = status_.GetDescription();
 }
 };
 class JobHandler : public boost::noncopyable
 }
 else
 {
 return retryTime_ <= now;
 }
+}
+const boost::posix_time::ptime& GetCreationTime() const
+{
+return creationTime_;
+}
+const boost::posix_time::time_duration& GetRuntime() const
+{
+return runtime_;
 }
 const JobStatus& GetLastStatus() const
 {
 return lastStatus_;
 target.insert(it->first);
 }
 }
+void GetJobsInfo(std::map<std::string, JobInfo>& target)
+{
+boost::mutex::scoped_lock lock(mutex_);
+CheckInvariants();
+for (JobsIndex::const_iterator it = jobsIndex_.begin();
+it != jobsIndex_.end(); ++it)
+{
+const JobHandler& handler = *it->second;
+target[it->first] = JobInfo(handler.GetId(),
+handler.GetPriority(),
+handler.GetState(),
+handler.GetLastStatus(),
+handler.GetCreationTime(),
+handler.GetRuntime());
+}
+}
 void Submit(std::string& id,
 IJob* job,        // Takes ownership
 int priority)
 {
 std::auto_ptr<JobHandler>  handler(new JobHandler(job, priority));
 pendingJobs_.push(handler.get());
 pendingJobAvailable_.notify_one();
 jobsIndex_.insert(std::make_pair(id, handler.release()));
-LOG(INFO) << "New job submitted: " << id;
+LOG(INFO) << "New job submitted with priority " << priority << ": " << id;
 CheckInvariants();
 }
 {
 return priority_;
 }
 }
+IJob& GetJob()
+{
+if (!IsValid())
+{
+throw OrthancException(ErrorCode_BadSequenceOfCalls);
+}
+else
+{
+return *job_;
+}
+}
 bool IsPauseScheduled()
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
 else
 {
-job_->ReleaseResources();
 targetState_ = JobState_Paused;
 }
 }
 void MarkRetry(unsigned int timeout)
 targetState_ = JobState_Retry;
 targetRetryTimeout_ = timeout;
 }
 }
-void UpdateStatus(const JobStatus& status)
+void UpdateStatus(ErrorCode code)
 {
 if (!IsValid())
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
 else
 {
+JobStatus status(code, *job_);
 boost::mutex::scoped_lock lock(registry_.mutex_);
 registry_.CheckInvariants();
 assert(handler_->GetState() == JobState_Running);
 handler_->SetLastStatus(status);
 }
 }
+};
-bool ExecuteStep()
+};
-{
-if (!IsValid())
-{
-throw OrthancException(ErrorCode_BadSequenceOfCalls);
+class JobsEngine
-}
+{
+private:
-if (IsPauseScheduled())
+enum State
 {
-targetState_ = JobState_Paused;
+State_Setup,
-return false;
+State_Running,
-}
+State_Stopping,
+State_Done
-std::auto_ptr<JobStepResult> result;
+};
-ErrorCode code;
+boost::mutex                stateMutex_;
-{
+State                       state_;
-bool ok = false;
+JobsRegistry                registry_;
+boost::thread               retryHandler_;
-try
+std::vector<boost::thread>  workers_;
+bool ExecuteStep(JobsRegistry::RunningJob& running,
+size_t workerIndex)
+{
+assert(running.IsValid());
+LOG(INFO) << "Executing job with priority " << running.GetPriority()
+<< " in worker thread " << workerIndex << ": " << running.GetId();
+if (running.IsPauseScheduled())
+{
+running.GetJob().ReleaseResources();
+running.MarkPause();
+return false;
+}
+std::auto_ptr<JobStepResult> result;
+{
+try
+{
+result.reset(running.GetJob().ExecuteStep());
+if (result->GetCode() == JobStepCode_Failure)
 {
-result.reset(job_->ExecuteStep());
+running.UpdateStatus(ErrorCode_InternalError);
-ok = true;
-if (result->GetCode() == JobStepCode_Failure)
-{
-code = ErrorCode_InternalError;
-}
-}
-catch (OrthancException& e)
-{
-code = e.GetErrorCode();
-}
-catch (boost::bad_lexical_cast&)
-{
-code = ErrorCode_BadFileFormat;
-}
-catch (...)
-{
-code = ErrorCode_InternalError;
-}
-if (ok)
-{
-code = ErrorCode_Success;
 }
 else
 {
-result.reset(new JobStepResult(JobStepCode_Failure));
+running.UpdateStatus(ErrorCode_Success);
 }
 }
+catch (OrthancException& e)
 {
-JobStatus status(code, *job_);
+running.UpdateStatus(e.GetErrorCode());
-UpdateStatus(status);
+}
-}
+catch (boost::bad_lexical_cast&)
+{
-switch (result->GetCode())
+running.UpdateStatus(ErrorCode_BadFileFormat);
-{
+}
-case JobStepCode_Success:
+catch (...)
-targetState_ = JobState_Success;
+{
-return false;
+running.UpdateStatus(ErrorCode_InternalError);
+}
-case JobStepCode_Failure:
-targetState_ = JobState_Failure;
+if (result.get() == NULL)
-return false;
+{
+result.reset(new JobStepResult(JobStepCode_Failure));
-case JobStepCode_Retry:
+}
-targetState_ = JobState_Retry;
+}
-targetRetryTimeout_ = dynamic_cast<RetryResult&>(*result).GetTimeout();
-return false;
+switch (result->GetCode())
+{
-case JobStepCode_Continue:
+case JobStepCode_Success:
-return true;
+running.MarkSuccess();
+return false;
+case JobStepCode_Failure:
+running.MarkFailure();
+return false;
+case JobStepCode_Retry:
+running.MarkRetry(dynamic_cast<RetryResult&>(*result).GetTimeout());
+return false;
+case JobStepCode_Continue:
+return true;
 default:
 throw OrthancException(ErrorCode_InternalError);
 }
 }
-};
+static void RetryHandler(JobsEngine* engine)
+{
+assert(engine != NULL);
+for (;;)
+{
+boost::this_thread::sleep(boost::posix_time::milliseconds(200));
+{
+boost::mutex::scoped_lock lock(engine->stateMutex_);
+if (engine->state_ != State_Running)
+{
+return;
+}
+}
+engine->GetRegistry().ScheduleRetries();
+}
+}
+static void Worker(JobsEngine* engine,
+size_t workerIndex)
+{
+assert(engine != NULL);
+LOG(INFO) << "Worker thread " << workerIndex << " has started";
+for (;;)
+{
+{
+boost::mutex::scoped_lock lock(engine->stateMutex_);
+if (engine->state_ != State_Running)
+{
+return;
+}
+}
+JobsRegistry::RunningJob running(engine->GetRegistry(), 100);
+if (running.IsValid())
+{
+for (;;)
+{
+if (!engine->ExecuteStep(running, workerIndex))
+{
+break;
+}
+}
+}
+}
+}
+public:
+JobsEngine() :
+state_(State_Setup),
+workers_(1)
+{
+}
+~JobsEngine()
+{
+if (state_ != State_Setup &&
+state_ != State_Done)
+{
+LOG(ERROR) << "INTERNAL ERROR: JobsEngine::Stop() should be invoked manually to avoid mess in the destruction order!";
+Stop();
+}
+}
+void SetWorkersCount(size_t count)
+{
+if (count == 0)
+{
+throw OrthancException(ErrorCode_ParameterOutOfRange);
+}
+boost::mutex::scoped_lock lock(stateMutex_);
+if (state_ != State_Setup)
+{
+// Can only be invoked before calling "Start()"
+throw OrthancException(ErrorCode_BadSequenceOfCalls);
+}
+workers_.resize(count);
+}
+JobsRegistry& GetRegistry()
+{
+return registry_;
+}
+void Start()
+{
+boost::mutex::scoped_lock lock(stateMutex_);
+if (state_ != State_Setup)
+{
+throw OrthancException(ErrorCode_BadSequenceOfCalls);
+}
+retryHandler_ = boost::thread(RetryHandler, this);
+for (size_t i = 0; i < workers_.size(); i++)
+{
+workers_[i] = boost::thread(Worker, this, i);
+}
+state_ = State_Running;
+LOG(WARNING) << "The jobs engine has started";
+}
+void Stop()
+{
+{
+boost::mutex::scoped_lock lock(stateMutex_);
+if (state_ != State_Running)
+{
+return;
+}
+state_ = State_Stopping;
+}
+LOG(INFO) << "Stopping the jobs engine";
+if (retryHandler_.joinable())
+{
+retryHandler_.join();
+}
+for (size_t i = 0; i < workers_.size(); i++)
+{
+if (workers_[i].joinable())
+{
+workers_[i].join();
+}
+}
+{
+boost::mutex::scoped_lock lock(stateMutex_);
+state_ = State_Done;
+}
+LOG(WARNING) << "The jobs engine has stopped";
+}
 };
 }
 class DummyJob : public Orthanc::IJob
 {
 private:
 JobStepResult  result_;
+unsigned int count_;
+unsigned int steps_;
 public:
 DummyJob() :
-result_(Orthanc::JobStepCode_Success)
+result_(Orthanc::JobStepCode_Success),
+count_(0),
+steps_(4)
 {
 }
 explicit DummyJob(JobStepResult result) :
-result_(result)
+result_(result),
+count_(0),
+steps_(4)
 {
 }
 virtual JobStepResult* ExecuteStep()
 {
-return new JobStepResult(result_);
+boost::this_thread::sleep(boost::posix_time::milliseconds(50));
+if (count_ == steps_ - 1)
+{
+return new JobStepResult(result_);
+}
+else
+{
+count_++;
+return new JobStepResult(JobStepCode_Continue);
+}
 }
 virtual void ReleaseResources()
 {
 }
 virtual float GetProgress()
 {
-return 0;
+return static_cast<float>(count_) / static_cast<float>(steps_ - 1);
 }
 virtual void GetDescription(Json::Value& value)
 {
+value["hello"] = "world";
 }
 };
 static bool CheckState(Orthanc::JobsRegistry& registry,
 ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running));
 }
 ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Success));
 }
+TEST(JobsEngine, Basic)
+{
+JobsEngine engine;
+std::string s;
+for (size_t i = 0; i < 20; i++)
+engine.GetRegistry().Submit(s, new DummyJob(), rand() % 10);
+engine.SetWorkersCount(3);
+engine.Start();
+boost::this_thread::sleep(boost::posix_time::milliseconds(200));
+engine.Stop();
+typedef std::map<std::string, JobInfo> Jobs;
+Jobs jobs;
+engine.GetRegistry().GetJobsInfo(jobs);
+Json::Value v;
+for (Jobs::const_iterator it = jobs.begin(); it != jobs.end(); ++it)
+{
+Json::Value vv;
+it->second.Format(vv);
+v[it->first] = vv;
+}
+std::cout << v << std::endl;
+}

Mercurial > hg > orthanc

comparison UnitTestsSources/MultiThreadingTests.cpp @ 2565:0f06b4d5b3d0 jobs