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

comparison OrthancServer/Sources/ServerJobs/ThreadedSetOfInstancesJob.cpp @ 5136:e71b22a43c0b

Threaded modifications continued: call ReconstructInstance at the end of the modification to update the DB model

author	Alain Mazy <am@osimis.io>
date	Tue, 17 Jan 2023 17:54:38 +0100
parents	252385892197
children	15109c3f0f7d

comparison

equal deleted inserted replaced

-:252385892197
+:e71b22a43c0b
 ThreadedSetOfInstancesJob::ThreadedSetOfInstancesJob(ServerContext& context,
 bool hasPostProcessing,
 bool keepSource,
 size_t workersCount) :
-processedInstancesCount_(0),
 hasPostProcessing_(hasPostProcessing),
 started_(false),
 stopRequested_(false),
 permissive_(false),
 currentStep_(ThreadedJobStep_ProcessingInstances),
 // no need to lock mutex here since we access variables used only by the "master" thread
 // send a dummy "exit" message to all workers such that they stop waiting for messages on the queue
 for (size_t i = 0; i < instancesWorkers_.size(); i++)
 {
-instancesToProcess_.Enqueue(new SingleValueObject<std::string>(EXIT_WORKER_MESSAGE));
+instancesToProcessQueue_.Enqueue(new SingleValueObject<std::string>(EXIT_WORKER_MESSAGE));
 }
 for (size_t i = 0; i < instancesWorkers_.size(); i++)
 {
 if (instancesWorkers_[i]->joinable())
 try
 {
 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)
+for (std::set<std::string>::const_iterator it = instancesToProcess_.begin(); it != instancesToProcess_.end(); ++it)
 {
-instancesToProcess_.Enqueue(new SingleValueObject<std::string>(*it));
+instancesToProcessQueue_.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())
+if ((processedInstances_.size() != instancesToProcess_.size())
 || (!IsPermissive() && failedInstances_.size() > 0))
 {
 return JobStepResult::Failure(GetErrorCode(), NULL);
 }
 else if (currentStep_ == ThreadedJobStep_Cleanup)
 {
 // clean after the post processing step
 if (HasCleanupStep())
 {
-for (std::set<std::string>::const_iterator it = instances_.begin(); it != instances_.end(); ++it)
+for (std::set<std::string>::const_iterator it = instancesToProcess_.begin(); it != instancesToProcess_.end(); ++it)
 {
 Json::Value tmp;
 context_.DeleteResource(tmp, *it, ResourceType_Instance);
 }
 }
 boost::recursive_mutex::scoped_lock lock(mutex_);
 return hasPostProcessing_;
 }
+void ThreadedSetOfInstancesJob::PostProcessInstances()
+{
+if (HasPostProcessingStep())
+{
+throw OrthancException(ErrorCode_InternalError, "Job with post-processing should override PostProcessInstances");
+}
+}
 bool ThreadedSetOfInstancesJob::HasCleanupStep() const
 {
 boost::recursive_mutex::scoped_lock lock(mutex_);
 void ThreadedSetOfInstancesJob::InstanceWorkerThread(ThreadedSetOfInstancesJob* that)
 {
 while (true)
 {
-std::unique_ptr<SingleValueObject<std::string> > instanceId(dynamic_cast<SingleValueObject<std::string>*>(that->instancesToProcess_.Dequeue(0)));
+std::unique_ptr<SingleValueObject<std::string> > instanceId(dynamic_cast<SingleValueObject<std::string>*>(that->instancesToProcessQueue_.Dequeue(0)));
 if (that->stopRequested_                // no lock(mutex) to access this variable, this is safe since it's just reading a boolean
 || instanceId->GetValue() == EXIT_WORKER_MESSAGE)
 {
 return;
 }
+bool processed = false;
 try
 {
-bool processed = that->HandleInstance(instanceId->GetValue());
+processed = that->HandleInstance(instanceId->GetValue());
-{
-boost::recursive_mutex::scoped_lock lock(that->mutex_);
-that->processedInstancesCount_++;
-if (!processed)
-{
-that->failedInstances_.insert(instanceId->GetValue());
-}
-}
 }
 catch (const Orthanc::OrthancException& e)
 {
 if (that->IsPermissive())
 {
 else
 {
 LOG(ERROR) << "Error in a non-permissive job: " << e.What();
 that->SetErrorCode(e.GetErrorCode());
 that->StopWorkers();
+return;
 }
 }
 catch (...)
 {
 LOG(ERROR) << "Native exception while executing a job";
 that->SetErrorCode(ErrorCode_InternalError);
 that->StopWorkers();
-}
+return;
+}
+{
+boost::recursive_mutex::scoped_lock lock(that->mutex_);
+that->processedInstances_.insert(instanceId->GetValue());
+if (!processed)
+{
+that->failedInstances_.insert(instanceId->GetValue());
+}
+}
 }
 }
 bool ThreadedSetOfInstancesJob::GetOutput(std::string &output,
 size_t ThreadedSetOfInstancesJob::GetInstancesCount() const
 {
 boost::recursive_mutex::scoped_lock lock(mutex_);
-return instances_.size();
+return instancesToProcess_.size();
 }
 void ThreadedSetOfInstancesJob::GetFailedInstances(std::set<std::string>& target) const
 {
 void ThreadedSetOfInstancesJob::GetInstances(std::set<std::string>& target) const
 {
 boost::recursive_mutex::scoped_lock lock(mutex_);
-target = instances_;
+target = instancesToProcess_;
 }
 bool ThreadedSetOfInstancesJob::IsFailedInstance(const std::string &instance) const
 {
 started_ = true;
 }
+// Reset is called when resubmitting a failed job
 void ThreadedSetOfInstancesJob::Reset()
 {
 boost::recursive_mutex::scoped_lock lock(mutex_);
 if (started_)
 {
+// TODO: cleanup the instances that have been generated during the previous run
 currentStep_ = ThreadedJobStep_ProcessingInstances;
 stopRequested_ = false;
-processedInstancesCount_ = 0;
+processedInstances_.clear();
 failedInstances_.clear();
-instancesToProcess_.Clear();
+instancesToProcessQueue_.Clear();
 }
 else
 {
 throw OrthancException(ErrorCode_BadSequenceOfCalls);
 }
 target[KEY_CURRENT_STEP] = static_cast<unsigned int>(currentStep_);
 target[KEY_DESCRIPTION] = description_;
 target[KEY_KEEP_SOURCE] = keepSource_;
 target[KEY_WORKERS_COUNT] = static_cast<unsigned int>(workersCount_);
-SerializationToolbox::WriteSetOfStrings(target, instances_, KEY_INSTANCES);
+SerializationToolbox::WriteSetOfStrings(target, instancesToProcess_, KEY_INSTANCES);
 SerializationToolbox::WriteSetOfStrings(target, failedInstances_, KEY_FAILED_INSTANCES);
 SerializationToolbox::WriteSetOfStrings(target, parentResources_, KEY_PARENT_RESOURCES);
 return true;
 }
 ThreadedSetOfInstancesJob::ThreadedSetOfInstancesJob(ServerContext& context,
 const Json::Value& source,
 bool hasPostProcessing,
 bool defaultKeepSource) :
-processedInstancesCount_(0),
 hasPostProcessing_(hasPostProcessing),
 started_(false),
 stopRequested_(false),
 permissive_(false),
 currentStep_(ThreadedJobStep_ProcessingInstances),
 workersCount_ = SerializationToolbox::ReadUnsignedInteger(source, KEY_WORKERS_COUNT);
 }
 if (source.isMember(KEY_INSTANCES))
 {
-SerializationToolbox::ReadSetOfStrings(instances_, source, KEY_INSTANCES);
+SerializationToolbox::ReadSetOfStrings(instancesToProcess_, source, KEY_INSTANCES);
 }
 if (source.isMember(KEY_CURRENT_STEP))
 {
 currentStep_ = static_cast<ThreadedJobStep>(SerializationToolbox::ReadUnsignedInteger(source, KEY_CURRENT_STEP));
 return 1;
 }
 else
 {
 size_t totalProgress = GetInstancesCount();
-size_t currentProgress = processedInstancesCount_;
+size_t currentProgress = processedInstances_.size();
 if (HasPostProcessingStep())
 {
 ++totalProgress;
 if (currentStep_ > ThreadedJobStep_PostProcessingInstances)
 boost::recursive_mutex::scoped_lock lock(mutex_);
 for (std::list<std::string>::const_iterator
 it = instances.begin(); it != instances.end(); ++it)
 {
-instances_.insert(*it);
+instancesToProcess_.insert(*it);
 }
 }
 void ThreadedSetOfInstancesJob::AddParentResource(const std::string &resource)

Mercurial > hg > orthanc

comparison OrthancServer/Sources/ServerJobs/ThreadedSetOfInstancesJob.cpp @ 5136:e71b22a43c0b