orthanc: OrthancServer/Sources/ServerJobs/ThreadedSetOfInstancesJob.cpp comparison

comparison OrthancServer/Sources/ServerJobs/ThreadedSetOfInstancesJob.cpp @ 5134:6aa41d86b948

fix ModificationJob state machine

author	Alain Mazy <am@osimis.io>
date	Tue, 10 Jan 2023 11:46:00 +0100
parents	f2dcdbe05884
children	252385892197

comparison

equal deleted inserted replaced

-:7dacd2c3c009
+:6aa41d86b948
 processedInstancesCount_(0),
 hasPostProcessing_(hasPostProcessing),
 started_(false),
 stopRequested_(false),
 permissive_(false),
-currentStep_(ThreadedJobStep_BeforeStart),
+currentStep_(ThreadedJobStep_ProcessingInstances),
 workersCount_(workersCount),
 context_(context),
-keepSource_(keepSource)
+keepSource_(keepSource),
+errorCode_(ErrorCode_Success)
 {
 }
 ThreadedSetOfInstancesJob::~ThreadedSetOfInstancesJob()
 {
 // no need to lock mutex here since we access variables used only by the "master" thread
 StopWorkers();
+WaitWorkersComplete();
 }
 void ThreadedSetOfInstancesJob::InitWorkers(size_t workersCount)
 {
 }
 void ThreadedSetOfInstancesJob::StopWorkers()
 {
-// no need to lock mutex here since we access variables used or set only by the "master" thread
+boost::recursive_mutex::scoped_lock lock(mutex_);
 stopRequested_ = true;
-WaitWorkersComplete();
 }
 void ThreadedSetOfInstancesJob::Stop(JobStopReason reason)
 {
 reason == JobStopReason_Failure ||
 reason == JobStopReason_Retry)
 {
 // deallocate resources
 StopWorkers();
+WaitWorkersComplete();
 }
 else if (reason == JobStopReason_Paused)
 {
 // keep resources allocated.
 // note that, right now, since all instances are queued from the start, this kind of jobs is not paused while in ProcessingInstances state
 return JobStepResult::Success();
 }
 try
 {
-if (currentStep_ == ThreadedJobStep_BeforeStart)
-{
-currentStep_ = ThreadedJobStep_ProcessingInstances;
-}
 if (currentStep_ == ThreadedJobStep_ProcessingInstances)
 {
+// create the workers and enqueue all instances
+for (std::set<std::string>::const_iterator it = instances_.begin(); it != instances_.end(); ++it)
+{
+instancesToProcess_.Enqueue(new SingleValueObject<std::string>(*it));
+}
+InitWorkers(workersCount_);
 // wait until all instances are processed by the workers
 WaitWorkersComplete();
+// check job has really completed !!! it might have been interrupted because of an error
+if ((processedInstancesCount_ != instances_.size())
+|| (!IsPermissive() && failedInstances_.size() > 0))
+{
+return JobStepResult::Failure(GetErrorCode(), NULL);
+}
 currentStep_ = ThreadedJobStep_PostProcessingInstances;
 return JobStepResult::Continue();
 }
 else if (currentStep_ == ThreadedJobStep_PostProcessingInstances)
 {
 || instanceId->GetValue() == EXIT_WORKER_MESSAGE)
 {
 return;
 }
-bool processed = that->HandleInstance(instanceId->GetValue());
+try
+{
-{
+bool processed = that->HandleInstance(instanceId->GetValue());
-boost::recursive_mutex::scoped_lock lock(that->mutex_);
+{
-that->processedInstancesCount_++;
+boost::recursive_mutex::scoped_lock lock(that->mutex_);
-if (!processed)
-{
+that->processedInstancesCount_++;
-that->failedInstances_.insert(instanceId->GetValue());
+if (!processed)
-}
+{
-}
+that->failedInstances_.insert(instanceId->GetValue());
+}
+}
+}
+catch (const Orthanc::OrthancException& e)
+{
+if (that->IsPermissive())
+{
+LOG(WARNING) << "Ignoring an error in a permissive job: " << e.What();
+}
+else
+{
+LOG(ERROR) << "Error in a non-permissive job: " << e.What();
+that->SetErrorCode(e.GetErrorCode());
+that->StopWorkers();
+}
+}
+catch (...)
+{
+LOG(ERROR) << "Native exception while executing a job";
+that->SetErrorCode(e.GetErrorCode());
+that->StopWorkers();
+}
 }
 }
 bool ThreadedSetOfInstancesJob::GetOutput(std::string &output,
 void ThreadedSetOfInstancesJob::Start()
 {
 boost::recursive_mutex::scoped_lock lock(mutex_);
 started_ = true;
-// create the workers and enqueue all instances
-for (std::set<std::string>::const_iterator it = instances_.begin(); it != instances_.end(); ++it)
-{
-instancesToProcess_.Enqueue(new SingleValueObject<std::string>(*it));
-}
-InitWorkers(workersCount_);
 }
 void ThreadedSetOfInstancesJob::Reset()
 {
 boost::recursive_mutex::scoped_lock lock(mutex_);
 if (started_)
 {
-currentStep_ = ThreadedJobStep_BeforeStart;
+currentStep_ = ThreadedJobStep_ProcessingInstances;
 stopRequested_ = false;
 processedInstancesCount_ = 0;
 failedInstances_.clear();
 instancesToProcess_.Clear();
 }
 processedInstancesCount_(0),
 hasPostProcessing_(hasPostProcessing),
 started_(false),
 stopRequested_(false),
 permissive_(false),
-currentStep_(ThreadedJobStep_BeforeStart),
+currentStep_(ThreadedJobStep_ProcessingInstances),
 workersCount_(1),
 context_(context),
-keepSource_(defaultKeepSource)
+keepSource_(defaultKeepSource),
+errorCode_(ErrorCode_Success)
 {
 SerializationToolbox::ReadSetOfStrings(failedInstances_, source, KEY_FAILED_INSTANCES);
 if (source.isMember(KEY_PARENT_RESOURCES))
 {
 }
 if (source.isMember(KEY_INSTANCES))
 {
 SerializationToolbox::ReadSetOfStrings(instances_, source, KEY_INSTANCES);
+}
+if (source.isMember(KEY_CURRENT_STEP))
+{
+currentStep_ = static_cast<ThreadedJobStep>(SerializationToolbox::ReadUnsignedInteger(source, KEY_CURRENT_STEP));
 }
 }
 void ThreadedSetOfInstancesJob::SetKeepSource(bool keep)
 boost::recursive_mutex::scoped_lock lock(mutex_);
 return description_;
 }
+void ThreadedSetOfInstancesJob::SetErrorCode(ErrorCode errorCode)
+{
+boost::recursive_mutex::scoped_lock lock(mutex_);
+errorCode_ = errorCode;
+}
+ErrorCode ThreadedSetOfInstancesJob::GetErrorCode() const
+{
+boost::recursive_mutex::scoped_lock lock(mutex_);
+return errorCode_;
+}
 bool ThreadedSetOfInstancesJob::IsPermissive() const
 {
 boost::recursive_mutex::scoped_lock lock(mutex_);

Mercurial > hg > orthanc

comparison OrthancServer/Sources/ServerJobs/ThreadedSetOfInstancesJob.cpp @ 5134:6aa41d86b948