Mercurial > hg > orthanc
changeset 2574:84cbc5abf3cc jobs
merge
| author | Sebastien Jodogne <s.jodogne@gmail.com> |
|---|---|
| date | Wed, 09 May 2018 17:56:26 +0200 |
| parents | 3372c5255333 (diff) fcf447be9e97 (current diff) |
| children | a874bdbbfa11 |
| files | |
| diffstat | 39 files changed, 4139 insertions(+), 914 deletions(-) [+] |
line wrap: on
line diff
--- a/Core/Enumerations.cpp Wed May 09 15:57:44 2018 +0200 +++ b/Core/Enumerations.cpp Wed May 09 17:56:26 2018 +0200 @@ -984,6 +984,34 @@ } + 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); + } + } + + Encoding StringToEncoding(const char* encoding) { std::string s(encoding);
--- a/Core/Enumerations.h Wed May 09 15:57:44 2018 +0200 +++ b/Core/Enumerations.h Wed May 09 17:56:26 2018 +0200 @@ -542,6 +542,24 @@ TransferSyntax_Rle }; + enum JobState + { + JobState_Pending, + JobState_Running, + JobState_Success, + JobState_Failure, + JobState_Paused, + JobState_Retry + }; + + enum JobStepCode + { + JobStepCode_Success, + JobStepCode_Failure, + JobStepCode_Continue, + JobStepCode_Retry + }; + /** * WARNING: Do not change the explicit values in the enumerations @@ -622,6 +640,8 @@ const char* EnumerationToString(ValueRepresentation vr); + const char* EnumerationToString(JobState state); + Encoding StringToEncoding(const char* encoding); ResourceType StringToResourceType(const char* type);
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/IJob.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,65 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "JobStepResult.h" + +#include <boost/noncopyable.hpp> +#include <json/value.h> + +namespace Orthanc +{ + class IJob : public boost::noncopyable + { + public: + virtual ~IJob() + { + } + + // Method called once the job enters the jobs engine + virtual void Start() = 0; + + virtual JobStepResult* ExecuteStep() = 0; + + virtual void ReleaseResources() = 0; // For pausing jobs + + virtual float GetProgress() = 0; + + virtual void GetJobType(std::string& target) = 0; + + virtual void GetPublicContent(Json::Value& value) = 0; + + virtual void GetInternalContent(Json::Value& value) = 0; + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobInfo.cpp Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,146 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#include "../PrecompiledHeaders.h" +#include "JobInfo.h" + +#include "../OrthancException.h" + +namespace Orthanc +{ + JobInfo::JobInfo(const std::string& id, + int priority, + JobState state, + const JobStatus& status, + const boost::posix_time::ptime& creationTime, + const boost::posix_time::ptime& lastStateChangeTime, + const boost::posix_time::time_duration& runtime) : + id_(id), + priority_(priority), + state_(state), + timestamp_(boost::posix_time::microsec_clock::universal_time()), + creationTime_(creationTime), + lastStateChangeTime_(lastStateChangeTime), + runtime_(runtime), + hasEta_(false), + status_(status) + { + if (state_ == JobState_Running) + { + float ms = static_cast<float>(runtime_.total_milliseconds()); + + if (status_.GetProgress() > 0.01f && + ms > 0.01f) + { + float ratio = static_cast<float>(1.0 - status_.GetProgress()); + long long remaining = boost::math::llround(ratio * ms); + eta_ = timestamp_ + boost::posix_time::milliseconds(remaining); + hasEta_ = true; + } + } + } + + + JobInfo::JobInfo() : + priority_(0), + state_(JobState_Failure), + timestamp_(boost::posix_time::microsec_clock::universal_time()), + creationTime_(timestamp_), + lastStateChangeTime_(timestamp_), + runtime_(boost::posix_time::milliseconds(0)), + hasEta_(false) + { + } + + + bool JobInfo::HasCompletionTime() const + { + return (state_ == JobState_Success || + state_ == JobState_Failure); + } + + + const boost::posix_time::ptime& JobInfo::GetEstimatedTimeOfArrival() const + { + if (hasEta_) + { + return eta_; + } + else + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + } + + + const boost::posix_time::ptime& JobInfo::GetCompletionTime() const + { + if (HasCompletionTime()) + { + return lastStateChangeTime_; + } + else + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + } + + + void JobInfo::Serialize(Json::Value& target) const + { + target = Json::objectValue; + target["ID"] = id_; + target["Priority"] = priority_; + target["ErrorCode"] = static_cast<int>(status_.GetErrorCode()); + target["ErrorDescription"] = 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["EffectiveRuntime"] = static_cast<double>(runtime_.total_milliseconds()) / 1000.0; + target["Progress"] = boost::math::iround(status_.GetProgress() * 100.0f); + + target["Type"] = status_.GetJobType(); + target["PublicContent"] = status_.GetPublicContent(); + target["InternalContent"] = status_.GetInternalContent(); + + if (HasEstimatedTimeOfArrival()) + { + target["EstimatedTimeOfArrival"] = boost::posix_time::to_iso_string(GetEstimatedTimeOfArrival()); + } + + if (HasCompletionTime()) + { + target["CompletionTime"] = boost::posix_time::to_iso_string(GetCompletionTime()); + } + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobInfo.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,120 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "JobStatus.h" + +#include <boost/date_time/posix_time/posix_time.hpp> + +namespace Orthanc +{ + class JobInfo + { + private: + std::string id_; + int priority_; + JobState state_; + boost::posix_time::ptime timestamp_; + boost::posix_time::ptime creationTime_; + boost::posix_time::ptime lastStateChangeTime_; + boost::posix_time::time_duration runtime_; + bool hasEta_; + boost::posix_time::ptime eta_; + JobStatus status_; + + public: + JobInfo(const std::string& id, + int priority, + JobState state, + const JobStatus& status, + const boost::posix_time::ptime& creationTime, + const boost::posix_time::ptime& lastStateChangeTime, + const boost::posix_time::time_duration& runtime); + + JobInfo(); + + const std::string& GetIdentifier() const + { + return id_; + } + + int GetPriority() const + { + return priority_; + } + + JobState GetState() const + { + return state_; + } + + const boost::posix_time::ptime& GetInfoTime() const + { + return timestamp_; + } + + const boost::posix_time::ptime& GetCreationTime() const + { + return creationTime_; + } + + const boost::posix_time::time_duration& GetRuntime() const + { + return runtime_; + } + + bool HasEstimatedTimeOfArrival() const + { + return hasEta_; + } + + bool HasCompletionTime() const; + + const boost::posix_time::ptime& GetEstimatedTimeOfArrival() const; + + const boost::posix_time::ptime& GetCompletionTime() const; + + const JobStatus& GetStatus() const + { + return status_; + } + + JobStatus& GetStatus() + { + return status_; + } + + void Serialize(Json::Value& target) const; + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobStatus.cpp Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,70 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#include "../PrecompiledHeaders.h" +#include "JobStatus.h" + +namespace Orthanc +{ + JobStatus::JobStatus() : + errorCode_(ErrorCode_InternalError), + progress_(0), + jobType_("Invalid"), + publicContent_(Json::objectValue), + internalContent_(Json::objectValue) + { + } + + + JobStatus::JobStatus(ErrorCode code, + IJob& job) : + errorCode_(code), + progress_(job.GetProgress()), + publicContent_(Json::objectValue), + internalContent_(Json::objectValue) + { + if (progress_ < 0) + { + progress_ = 0; + } + + if (progress_ > 1) + { + progress_ = 1; + } + + job.GetJobType(jobType_); + job.GetPublicContent(publicContent_); + job.GetInternalContent(internalContent_); + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobStatus.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,80 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "IJob.h" + +namespace Orthanc +{ + class JobStatus + { + private: + ErrorCode errorCode_; + float progress_; + std::string jobType_; + Json::Value publicContent_; + Json::Value internalContent_; + + public: + JobStatus(); + + JobStatus(ErrorCode code, + IJob& job); + + ErrorCode GetErrorCode() const + { + return errorCode_; + } + + float GetProgress() const + { + return progress_; + } + + const std::string& GetJobType() const + { + return jobType_; + } + + const Json::Value& GetPublicContent() const + { + return publicContent_; + } + + const Json::Value& GetInternalContent() const + { + return internalContent_; + } + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobStepResult.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,60 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "../Enumerations.h" + +namespace Orthanc +{ + class JobStepResult + { + private: + JobStepCode code_; + + public: + explicit JobStepResult(JobStepCode code) : + code_(code) + { + } + + virtual ~JobStepResult() + { + } + + JobStepCode GetCode() const + { + return code_; + } + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobStepRetry.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,57 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "JobStepResult.h" + +namespace Orthanc +{ + class JobStepRetry : public JobStepResult + { + private: + unsigned int timeout_; // Retry after "timeout_" milliseconds + + public: + JobStepRetry(unsigned int timeout) : + JobStepResult(JobStepCode_Retry), + timeout_(timeout) + { + } + + unsigned int GetTimeout() const + { + return timeout_; + } + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobsEngine.cpp Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,266 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#include "../PrecompiledHeaders.h" +#include "JobsEngine.h" + +#include "JobStepRetry.h" + +#include "../Logging.h" +#include "../OrthancException.h" + +namespace Orthanc +{ + bool JobsEngine::IsRunning() + { + boost::mutex::scoped_lock lock(stateMutex_); + return (state_ == State_Running); + } + + + bool JobsEngine::ExecuteStep(JobsRegistry::RunningJob& running, + size_t workerIndex) + { + assert(running.IsValid()); + + 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) + { + running.UpdateStatus(ErrorCode_InternalError); + } + else + { + running.UpdateStatus(ErrorCode_Success); + } + } + catch (OrthancException& e) + { + running.UpdateStatus(e.GetErrorCode()); + } + catch (boost::bad_lexical_cast&) + { + running.UpdateStatus(ErrorCode_BadFileFormat); + } + catch (...) + { + running.UpdateStatus(ErrorCode_InternalError); + } + + if (result.get() == NULL) + { + result.reset(new JobStepResult(JobStepCode_Failure)); + } + } + + switch (result->GetCode()) + { + case JobStepCode_Success: + running.GetJob().ReleaseResources(); + running.MarkSuccess(); + return false; + + case JobStepCode_Failure: + running.GetJob().ReleaseResources(); + running.MarkFailure(); + return false; + + case JobStepCode_Retry: + running.GetJob().ReleaseResources(); + running.MarkRetry(dynamic_cast<JobStepRetry&>(*result).GetTimeout()); + return false; + + case JobStepCode_Continue: + return true; + + default: + throw OrthancException(ErrorCode_InternalError); + } + } + + + void JobsEngine::RetryHandler(JobsEngine* engine) + { + assert(engine != NULL); + + while (engine->IsRunning()) + { + boost::this_thread::sleep(boost::posix_time::milliseconds(200)); + engine->GetRegistry().ScheduleRetries(); + } + } + + + void JobsEngine::Worker(JobsEngine* engine, + size_t workerIndex) + { + assert(engine != NULL); + + LOG(INFO) << "Worker thread " << workerIndex << " has started"; + + while (engine->IsRunning()) + { + JobsRegistry::RunningJob running(engine->GetRegistry(), 100); + + if (running.IsValid()) + { + LOG(INFO) << "Executing job with priority " << running.GetPriority() + << " in worker thread " << workerIndex << ": " << running.GetId(); + + while (engine->IsRunning()) + { + if (!engine->ExecuteStep(running, workerIndex)) + { + break; + } + } + } + } + } + + + JobsEngine::JobsEngine() : + state_(State_Setup), + workers_(1) + { + } + + + JobsEngine::~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 JobsEngine::SetWorkersCount(size_t count) + { + boost::mutex::scoped_lock lock(stateMutex_); + + if (state_ != State_Setup) + { + // Can only be invoked before calling "Start()" + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + + workers_.resize(count); + } + + + void JobsEngine::Start() + { + boost::mutex::scoped_lock lock(stateMutex_); + + if (state_ != State_Setup) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + + retryHandler_ = boost::thread(RetryHandler, this); + + if (workers_.size() == 0) + { + // Use all the available CPUs + size_t n = boost::thread::hardware_concurrency(); + + if (n == 0) + { + n = 1; + } + + workers_.resize(n); + } + + 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 with " << workers_.size() << " threads"; + } + + + void JobsEngine::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"; + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobsEngine.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,85 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "JobsRegistry.h" + +#include <boost/thread.hpp> + +namespace Orthanc +{ + class JobsEngine + { + private: + enum State + { + State_Setup, + State_Running, + State_Stopping, + State_Done + }; + + boost::mutex stateMutex_; + State state_; + JobsRegistry registry_; + boost::thread retryHandler_; + std::vector<boost::thread> workers_; + + bool IsRunning(); + + bool ExecuteStep(JobsRegistry::RunningJob& running, + size_t workerIndex); + + static void RetryHandler(JobsEngine* engine); + + static void Worker(JobsEngine* engine, + size_t workerIndex); + + public: + JobsEngine(); + + ~JobsEngine(); + + void SetWorkersCount(size_t count); + + JobsRegistry& GetRegistry() + { + return registry_; + } + + void Start(); + + void Stop(); + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobsRegistry.cpp Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,954 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#include "../PrecompiledHeaders.h" +#include "JobsRegistry.h" + +#include "../Logging.h" +#include "../OrthancException.h" +#include "../Toolbox.h" + +namespace Orthanc +{ + class JobsRegistry::JobHandler : public boost::noncopyable + { + private: + std::string id_; + JobState state_; + std::auto_ptr<IJob> job_; + int priority_; // "+inf()" means highest priority + boost::posix_time::ptime creationTime_; + boost::posix_time::ptime lastStateChangeTime_; + boost::posix_time::time_duration runtime_; + boost::posix_time::ptime retryTime_; + bool pauseScheduled_; + JobStatus lastStatus_; + + void Touch() + { + const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time(); + + if (state_ == JobState_Running) + { + runtime_ += (now - lastStateChangeTime_); + } + + lastStateChangeTime_ = now; + } + + void SetStateInternal(JobState state) + { + state_ = state; + pauseScheduled_ = false; + Touch(); + } + + public: + JobHandler(IJob* job, + int priority) : + id_(Toolbox::GenerateUuid()), + state_(JobState_Pending), + job_(job), + priority_(priority), + creationTime_(boost::posix_time::microsec_clock::universal_time()), + lastStateChangeTime_(creationTime_), + runtime_(boost::posix_time::milliseconds(0)), + retryTime_(creationTime_), + pauseScheduled_(false) + { + if (job == NULL) + { + throw OrthancException(ErrorCode_NullPointer); + } + + lastStatus_ = JobStatus(ErrorCode_Success, *job); + job->Start(); + } + + const std::string& GetId() const + { + return id_; + } + + IJob& GetJob() const + { + assert(job_.get() != NULL); + return *job_; + } + + void SetPriority(int priority) + { + priority_ = priority; + } + + int GetPriority() const + { + return priority_; + } + + JobState GetState() const + { + return state_; + } + + void SetState(JobState state) + { + if (state == JobState_Retry) + { + // Use "SetRetryState()" + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + SetStateInternal(state); + } + } + + 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 + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + } + + void SchedulePause() + { + if (state_ == JobState_Running) + { + pauseScheduled_ = true; + } + else + { + // Only valid for running jobs + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + } + + bool IsPauseScheduled() + { + return pauseScheduled_; + } + + bool IsRetryReady(const boost::posix_time::ptime& now) const + { + if (state_ != JobState_Retry) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + return retryTime_ <= now; + } + } + + const boost::posix_time::ptime& GetCreationTime() const + { + return creationTime_; + } + + const boost::posix_time::ptime& GetLastStateChangeTime() const + { + return lastStateChangeTime_; + } + + const boost::posix_time::time_duration& GetRuntime() const + { + return runtime_; + } + + const JobStatus& GetLastStatus() const + { + return lastStatus_; + } + + void SetLastStatus(const JobStatus& status) + { + lastStatus_ = status; + Touch(); + } + }; + + + 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()) + { + if (copy.top() == &job) + { + return true; + } + + copy.pop(); + } + + return false; + } + + bool JobsRegistry::IsCompletedJob(JobHandler& job) const + { + for (CompletedJobs::const_iterator it = completedJobs_.begin(); + it != completedJobs_.end(); ++it) + { + if (*it == &job) + { + return true; + } + } + + return false; + } + + bool JobsRegistry::IsRetryJob(JobHandler& job) const + { + return retryJobs_.find(&job) != retryJobs_.end(); + } + + void JobsRegistry::CheckInvariants() const + { + { + PendingJobs copy = pendingJobs_; + while (!copy.empty()) + { + assert(copy.top()->GetState() == JobState_Pending); + copy.pop(); + } + } + + assert(completedJobs_.size() <= maxCompletedJobs_); + + for (CompletedJobs::const_iterator it = completedJobs_.begin(); + it != completedJobs_.end(); ++it) + { + assert((*it)->GetState() == JobState_Success || + (*it)->GetState() == JobState_Failure); + } + + for (RetryJobs::const_iterator it = retryJobs_.begin(); + it != retryJobs_.end(); ++it) + { + assert((*it)->GetState() == JobState_Retry); + } + + for (JobsIndex::const_iterator it = jobsIndex_.begin(); + it != jobsIndex_.end(); ++it) + { + JobHandler& job = *it->second; + + assert(job.GetId() == it->first); + + 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; + + default: + throw OrthancException(ErrorCode_InternalError); + } + } + } +#endif + + + void JobsRegistry::ForgetOldCompletedJobs() + { + if (maxCompletedJobs_ != 0) + { + while (completedJobs_.size() > maxCompletedJobs_) + { + assert(completedJobs_.front() != NULL); + + std::string id = completedJobs_.front()->GetId(); + assert(jobsIndex_.find(id) != jobsIndex_.end()); + + jobsIndex_.erase(id); + delete(completedJobs_.front()); + completedJobs_.pop_front(); + } + } + } + + + void JobsRegistry::MarkRunningAsCompleted(JobHandler& job, + bool success) + { + LOG(INFO) << "Job has completed with " << (success ? "success" : "failure") + << ": " << job.GetId(); + + CheckInvariants(); + assert(job.GetState() == JobState_Running); + + job.SetState(success ? JobState_Success : JobState_Failure); + + completedJobs_.push_back(&job); + ForgetOldCompletedJobs(); + + someJobComplete_.notify_all(); + + CheckInvariants(); + } + + + void JobsRegistry::MarkRunningAsRetry(JobHandler& job, + unsigned int timeout) + { + LOG(INFO) << "Job scheduled for retry in " << timeout << "ms: " << job.GetId(); + + CheckInvariants(); + + assert(job.GetState() == JobState_Running && + retryJobs_.find(&job) == retryJobs_.end()); + + retryJobs_.insert(&job); + job.SetRetryState(timeout); + + CheckInvariants(); + } + + + void JobsRegistry::MarkRunningAsPaused(JobHandler& job) + { + LOG(INFO) << "Job paused: " << job.GetId(); + + CheckInvariants(); + assert(job.GetState() == JobState_Running); + + job.SetState(JobState_Paused); + + CheckInvariants(); + } + + + bool JobsRegistry::GetStateInternal(JobState& state, + const std::string& id) + { + CheckInvariants(); + + JobsIndex::const_iterator it = jobsIndex_.find(id); + if (it == jobsIndex_.end()) + { + return false; + } + else + { + state = it->second->GetState(); + return true; + } + } + + + JobsRegistry::~JobsRegistry() + { + for (JobsIndex::iterator it = jobsIndex_.begin(); it != jobsIndex_.end(); ++it) + { + assert(it->second != NULL); + delete it->second; + } + } + + + void JobsRegistry::SetMaxCompletedJobs(size_t i) + { + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + maxCompletedJobs_ = i; + ForgetOldCompletedJobs(); + + CheckInvariants(); + } + + + void JobsRegistry::ListJobs(std::set<std::string>& target) + { + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + for (JobsIndex::const_iterator it = jobsIndex_.begin(); + it != jobsIndex_.end(); ++it) + { + target.insert(it->first); + } + } + + + bool JobsRegistry::GetJobInfo(JobInfo& target, + const std::string& id) + { + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + JobsIndex::const_iterator found = jobsIndex_.find(id); + + if (found == jobsIndex_.end()) + { + return false; + } + else + { + const JobHandler& handler = *found->second; + target = JobInfo(handler.GetId(), + handler.GetPriority(), + handler.GetState(), + handler.GetLastStatus(), + handler.GetCreationTime(), + handler.GetLastStateChangeTime(), + handler.GetRuntime()); + return true; + } + } + + + void JobsRegistry::Submit(std::string& id, + IJob* job, // Takes ownership + int priority) + { + std::auto_ptr<JobHandler> handler(new JobHandler(job, priority)); + + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + id = handler->GetId(); + + pendingJobs_.push(handler.get()); + pendingJobAvailable_.notify_one(); + + jobsIndex_.insert(std::make_pair(id, handler.release())); + + LOG(INFO) << "New job submitted with priority " << priority << ": " << id; + + CheckInvariants(); + } + + + void JobsRegistry::Submit(IJob* job, // Takes ownership + int priority) + { + std::string id; + Submit(id, job, priority); + } + + + bool JobsRegistry::SubmitAndWait(IJob* job, // Takes ownership + int priority) + { + std::string id; + Submit(id, job, priority); + + printf(">> %s\n", id.c_str()); fflush(stdout); + + JobState state; + + { + boost::mutex::scoped_lock lock(mutex_); + + while (GetStateInternal(state, id) && + state != JobState_Success && + state != JobState_Failure) + { + someJobComplete_.wait(lock); + } + } + + return (state == JobState_Success); + } + + + bool JobsRegistry::SetPriority(const std::string& id, + int priority) + { + LOG(INFO) << "Changing priority to " << priority << " for job: " << id; + + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + JobsIndex::iterator found = jobsIndex_.find(id); + + if (found == jobsIndex_.end()) + { + LOG(WARNING) << "Unknown job: " << id; + return false; + } + else + { + found->second->SetPriority(priority); + + if (found->second->GetState() == JobState_Pending) + { + // If the job is pending, we need to reconstruct the + // priority queue, as the heap condition has changed + + PendingJobs copy; + std::swap(copy, pendingJobs_); + + assert(pendingJobs_.empty()); + while (!copy.empty()) + { + pendingJobs_.push(copy.top()); + copy.pop(); + } + } + + CheckInvariants(); + return true; + } + } + + + bool JobsRegistry::Pause(const std::string& id) + { + LOG(INFO) << "Pausing job: " << id; + + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + JobsIndex::iterator found = jobsIndex_.find(id); + + if (found == jobsIndex_.end()) + { + LOG(WARNING) << "Unknown job: " << id; + return false; + } + else + { + switch (found->second->GetState()) + { + case JobState_Pending: + { + // If the job is pending, we need to reconstruct the + // priority queue to remove it + PendingJobs copy; + std::swap(copy, pendingJobs_); + + assert(pendingJobs_.empty()); + while (!copy.empty()) + { + if (copy.top()->GetId() != id) + { + pendingJobs_.push(copy.top()); + } + + copy.pop(); + } + + found->second->SetState(JobState_Paused); + + break; + } + + case JobState_Retry: + { + RetryJobs::iterator item = retryJobs_.find(found->second); + assert(item != retryJobs_.end()); + retryJobs_.erase(item); + + found->second->SetState(JobState_Paused); + + break; + } + + case JobState_Paused: + case JobState_Success: + case JobState_Failure: + // Nothing to be done + break; + + case JobState_Running: + found->second->SchedulePause(); + break; + + default: + throw OrthancException(ErrorCode_InternalError); + } + + CheckInvariants(); + return true; + } + } + + + bool JobsRegistry::Resume(const std::string& id) + { + LOG(INFO) << "Resuming job: " << id; + + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + JobsIndex::iterator found = jobsIndex_.find(id); + + if (found == jobsIndex_.end()) + { + LOG(WARNING) << "Unknown job: " << id; + return false; + } + else if (found->second->GetState() != JobState_Paused) + { + LOG(WARNING) << "Cannot resume a job that is not paused: " << id; + return false; + } + else + { + found->second->SetState(JobState_Pending); + pendingJobs_.push(found->second); + pendingJobAvailable_.notify_one(); + CheckInvariants(); + return true; + } + } + + + bool JobsRegistry::Resubmit(const std::string& id) + { + LOG(INFO) << "Resubmitting failed job: " << id; + + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + JobsIndex::iterator found = jobsIndex_.find(id); + + if (found == jobsIndex_.end()) + { + LOG(WARNING) << "Unknown job: " << id; + return false; + } + else if (found->second->GetState() != JobState_Failure) + { + LOG(WARNING) << "Cannot resubmit a job that has not failed: " << id; + return false; + } + else + { + bool ok = false; + for (CompletedJobs::iterator it = completedJobs_.begin(); + it != completedJobs_.end(); ++it) + { + if (*it == found->second) + { + ok = true; + completedJobs_.erase(it); + break; + } + } + + assert(ok); + + found->second->SetState(JobState_Pending); + pendingJobs_.push(found->second); + pendingJobAvailable_.notify_one(); + + CheckInvariants(); + return true; + } + } + + + void JobsRegistry::ScheduleRetries() + { + boost::mutex::scoped_lock lock(mutex_); + CheckInvariants(); + + RetryJobs copy; + std::swap(copy, retryJobs_); + + const boost::posix_time::ptime now = boost::posix_time::microsec_clock::universal_time(); + + assert(retryJobs_.empty()); + for (RetryJobs::iterator it = copy.begin(); it != copy.end(); ++it) + { + if ((*it)->IsRetryReady(now)) + { + LOG(INFO) << "Retrying job: " << (*it)->GetId(); + (*it)->SetState(JobState_Pending); + pendingJobs_.push(*it); + pendingJobAvailable_.notify_one(); + } + else + { + retryJobs_.insert(*it); + } + } + + CheckInvariants(); + } + + + bool JobsRegistry::GetState(JobState& state, + const std::string& id) + { + boost::mutex::scoped_lock lock(mutex_); + return GetStateInternal(state, id); + } + + + JobsRegistry::RunningJob::RunningJob(JobsRegistry& registry, + unsigned int timeout) : + registry_(registry), + handler_(NULL), + targetState_(JobState_Failure), + targetRetryTimeout_(0) + { + { + 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(); + } + } + + + JobsRegistry::RunningJob::~RunningJob() + { + if (IsValid()) + { + boost::mutex::scoped_lock lock(registry_.mutex_); + + switch (targetState_) + { + case JobState_Failure: + registry_.MarkRunningAsCompleted(*handler_, false); + break; + + case JobState_Success: + registry_.MarkRunningAsCompleted(*handler_, true); + 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); + } + else + { + return id_; + } + } + + + int JobsRegistry::RunningJob::GetPriority() const + { + if (!IsValid()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + return priority_; + } + } + + + IJob& JobsRegistry::RunningJob::GetJob() + { + if (!IsValid()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + 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(); + } + } + + + void JobsRegistry::RunningJob::MarkSuccess() + { + if (!IsValid()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + targetState_ = JobState_Success; + } + } + + + void JobsRegistry::RunningJob::MarkFailure() + { + if (!IsValid()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + targetState_ = JobState_Failure; + } + } + + + void JobsRegistry::RunningJob::MarkPause() + { + if (!IsValid()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + targetState_ = JobState_Paused; + } + } + + + void JobsRegistry::RunningJob::MarkRetry(unsigned int timeout) + { + if (!IsValid()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + targetState_ = JobState_Retry; + targetRetryTimeout_ = timeout; + } + } + + + void JobsRegistry::RunningJob::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); + } + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/JobsEngine/JobsRegistry.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,189 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#if !defined(ORTHANC_SANDBOXED) +# error The macro ORTHANC_SANDBOXED must be defined +#endif + +#if ORTHANC_SANDBOXED == 1 +# error The job engine cannot be used in sandboxed environments +#endif + +#include "JobInfo.h" + +#include <list> +#include <set> +#include <queue> +#include <boost/thread/mutex.hpp> +#include <boost/thread/condition_variable.hpp> + +namespace Orthanc +{ + // This class handles the state machine of the jobs engine + class JobsRegistry : public boost::noncopyable + { + private: + class JobHandler; + + struct PriorityComparator + { + bool operator() (JobHandler*& a, + JobHandler*& b) const; + }; + + typedef std::map<std::string, JobHandler*> JobsIndex; + typedef std::list<JobHandler*> CompletedJobs; + typedef std::set<JobHandler*> RetryJobs; + typedef std::priority_queue<JobHandler*, + std::vector<JobHandler*>, // Could be a "std::deque" + PriorityComparator> PendingJobs; + + boost::mutex mutex_; + JobsIndex jobsIndex_; + PendingJobs pendingJobs_; + CompletedJobs completedJobs_; + RetryJobs retryJobs_; + + boost::condition_variable pendingJobAvailable_; + boost::condition_variable someJobComplete_; + size_t maxCompletedJobs_; + + +#ifndef NDEBUG + bool IsPendingJob(const JobHandler& job) const; + + bool IsCompletedJob(JobHandler& job) const; + + bool IsRetryJob(JobHandler& job) const; +#endif + + void CheckInvariants() const; + + void ForgetOldCompletedJobs(); + + void MarkRunningAsCompleted(JobHandler& job, + bool success); + + void MarkRunningAsRetry(JobHandler& job, + unsigned int timeout); + + void MarkRunningAsPaused(JobHandler& job); + + bool GetStateInternal(JobState& state, + const std::string& id); + + public: + JobsRegistry() : + maxCompletedJobs_(10) + { + } + + + ~JobsRegistry(); + + void SetMaxCompletedJobs(size_t i); + + void ListJobs(std::set<std::string>& target); + + bool GetJobInfo(JobInfo& target, + const std::string& id); + + void Submit(std::string& id, + IJob* job, // Takes ownership + int priority); + + void Submit(IJob* job, // Takes ownership + int priority); + + bool SubmitAndWait(IJob* job, // Takes ownership + int priority); + + bool SetPriority(const std::string& id, + int priority); + + bool Pause(const std::string& id); + + bool Resume(const std::string& id); + + bool Resubmit(const std::string& id); + + void ScheduleRetries(); + + bool GetState(JobState& state, + const std::string& id); + + class RunningJob : public boost::noncopyable + { + private: + JobsRegistry& registry_; + 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 targetRetryTimeout_; + + public: + RunningJob(JobsRegistry& registry, + unsigned int timeout); + + ~RunningJob(); + + bool IsValid() const; + + const std::string& GetId() const; + + int GetPriority() const; + + IJob& GetJob(); + + bool IsPauseScheduled(); + + void MarkSuccess(); + + void MarkFailure(); + + void MarkPause(); + + void MarkRetry(unsigned int timeout); + + void UpdateStatus(ErrorCode code); + }; + }; +}
--- a/Core/MultiThreading/BagOfTasks.h Wed May 09 15:57:44 2018 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,84 +0,0 @@ -/** - * Orthanc - A Lightweight, RESTful DICOM Store - * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics - * Department, University Hospital of Liege, Belgium - * Copyright (C) 2017-2018 Osimis S.A., Belgium - * - * This program is free software: you can redistribute it and/or - * modify it under the terms of the GNU General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * In addition, as a special exception, the copyright holders of this - * program give permission to link the code of its release with the - * OpenSSL project's "OpenSSL" library (or with modified versions of it - * that use the same license as the "OpenSSL" library), and distribute - * the linked executables. You must obey the GNU General Public License - * in all respects for all of the code used other than "OpenSSL". If you - * 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. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see <http://www.gnu.org/licenses/>. - **/ - - -#pragma once - -#include "../ICommand.h" - -#include <list> -#include <cstddef> - -namespace Orthanc -{ - class BagOfTasks : public boost::noncopyable - { - private: - typedef std::list<ICommand*> Tasks; - - Tasks tasks_; - - public: - ~BagOfTasks() - { - for (Tasks::iterator it = tasks_.begin(); it != tasks_.end(); ++it) - { - delete *it; - } - } - - ICommand* Pop() - { - ICommand* task = tasks_.front(); - tasks_.pop_front(); - return task; - } - - void Push(ICommand* task) // Takes ownership - { - if (task != NULL) - { - tasks_.push_back(task); - } - } - - size_t GetSize() const - { - return tasks_.size(); - } - - bool IsEmpty() const - { - return tasks_.empty(); - } - }; -}
--- a/Core/MultiThreading/BagOfTasksProcessor.cpp Wed May 09 15:57:44 2018 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,277 +0,0 @@ -/** - * Orthanc - A Lightweight, RESTful DICOM Store - * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics - * Department, University Hospital of Liege, Belgium - * Copyright (C) 2017-2018 Osimis S.A., Belgium - * - * This program is free software: you can redistribute it and/or - * modify it under the terms of the GNU General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * In addition, as a special exception, the copyright holders of this - * program give permission to link the code of its release with the - * OpenSSL project's "OpenSSL" library (or with modified versions of it - * that use the same license as the "OpenSSL" library), and distribute - * the linked executables. You must obey the GNU General Public License - * in all respects for all of the code used other than "OpenSSL". If you - * 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. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see <http://www.gnu.org/licenses/>. - **/ - - -#include "../PrecompiledHeaders.h" -#include "BagOfTasksProcessor.h" - -#include "../Logging.h" -#include "../OrthancException.h" - -#include <stdio.h> - -namespace Orthanc -{ - class BagOfTasksProcessor::Task : public IDynamicObject - { - private: - uint64_t bag_; - std::auto_ptr<ICommand> command_; - - public: - Task(uint64_t bag, - ICommand* command) : - bag_(bag), - command_(command) - { - } - - bool Execute() - { - try - { - return command_->Execute(); - } - catch (OrthancException& e) - { - LOG(ERROR) << "Exception while processing a bag of tasks: " << e.What(); - return false; - } - catch (std::runtime_error& e) - { - LOG(ERROR) << "Runtime exception while processing a bag of tasks: " << e.what(); - return false; - } - catch (...) - { - LOG(ERROR) << "Native exception while processing a bag of tasks"; - return false; - } - } - - uint64_t GetBag() - { - return bag_; - } - }; - - - void BagOfTasksProcessor::SignalProgress(Task& task, - Bag& bag) - { - assert(bag.done_ < bag.size_); - - bag.done_ += 1; - - if (bag.done_ == bag.size_) - { - exitStatus_[task.GetBag()] = (bag.status_ == BagStatus_Running); - bagFinished_.notify_all(); - } - } - - void BagOfTasksProcessor::Worker(BagOfTasksProcessor* that) - { - while (that->continue_) - { - std::auto_ptr<IDynamicObject> obj(that->queue_.Dequeue(100)); - if (obj.get() != NULL) - { - Task& task = *dynamic_cast<Task*>(obj.get()); - - { - boost::mutex::scoped_lock lock(that->mutex_); - - Bags::iterator bag = that->bags_.find(task.GetBag()); - assert(bag != that->bags_.end()); - assert(bag->second.done_ < bag->second.size_); - - if (bag->second.status_ != BagStatus_Running) - { - // Do not execute this task, as its parent bag of tasks - // has failed or is tagged as canceled - that->SignalProgress(task, bag->second); - continue; - } - } - - bool success = task.Execute(); - - { - boost::mutex::scoped_lock lock(that->mutex_); - - Bags::iterator bag = that->bags_.find(task.GetBag()); - assert(bag != that->bags_.end()); - - if (!success) - { - bag->second.status_ = BagStatus_Failed; - } - - that->SignalProgress(task, bag->second); - } - } - } - } - - - void BagOfTasksProcessor::Cancel(int64_t bag) - { - boost::mutex::scoped_lock lock(mutex_); - - Bags::iterator it = bags_.find(bag); - if (it != bags_.end()) - { - it->second.status_ = BagStatus_Canceled; - } - } - - - bool BagOfTasksProcessor::Join(int64_t bag) - { - boost::mutex::scoped_lock lock(mutex_); - - while (continue_) - { - ExitStatus::iterator it = exitStatus_.find(bag); - if (it == exitStatus_.end()) // The bag is still running - { - bagFinished_.wait(lock); - } - else - { - bool status = it->second; - exitStatus_.erase(it); - return status; - } - } - - return false; // The processor is stopping - } - - - float BagOfTasksProcessor::GetProgress(int64_t bag) - { - boost::mutex::scoped_lock lock(mutex_); - - Bags::const_iterator it = bags_.find(bag); - if (it == bags_.end()) - { - // The bag of tasks has finished - return 1.0f; - } - else - { - return (static_cast<float>(it->second.done_) / - static_cast<float>(it->second.size_)); - } - } - - - bool BagOfTasksProcessor::Handle::Join() - { - if (hasJoined_) - { - return status_; - } - else - { - status_ = that_.Join(bag_); - hasJoined_ = true; - return status_; - } - } - - - BagOfTasksProcessor::BagOfTasksProcessor(size_t countThreads) : - countBags_(0), - continue_(true) - { - if (countThreads == 0) - { - throw OrthancException(ErrorCode_ParameterOutOfRange); - } - - threads_.resize(countThreads); - - for (size_t i = 0; i < threads_.size(); i++) - { - threads_[i] = new boost::thread(Worker, this); - } - } - - - BagOfTasksProcessor::~BagOfTasksProcessor() - { - continue_ = false; - - bagFinished_.notify_all(); // Wakes up all the pending "Join()" - - for (size_t i = 0; i < threads_.size(); i++) - { - if (threads_[i]) - { - if (threads_[i]->joinable()) - { - threads_[i]->join(); - } - - delete threads_[i]; - threads_[i] = NULL; - } - } - } - - - BagOfTasksProcessor::Handle* BagOfTasksProcessor::Submit(BagOfTasks& tasks) - { - if (tasks.GetSize() == 0) - { - return new Handle(*this, 0, true); - } - - boost::mutex::scoped_lock lock(mutex_); - - uint64_t id = countBags_; - countBags_ += 1; - - Bag bag(tasks.GetSize()); - bags_[id] = bag; - - while (!tasks.IsEmpty()) - { - queue_.Enqueue(new Task(id, tasks.Pop())); - } - - return new Handle(*this, id, false); - } -}
--- a/Core/MultiThreading/BagOfTasksProcessor.h Wed May 09 15:57:44 2018 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,150 +0,0 @@ -/** - * Orthanc - A Lightweight, RESTful DICOM Store - * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics - * Department, University Hospital of Liege, Belgium - * Copyright (C) 2017-2018 Osimis S.A., Belgium - * - * This program is free software: you can redistribute it and/or - * modify it under the terms of the GNU General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * In addition, as a special exception, the copyright holders of this - * program give permission to link the code of its release with the - * OpenSSL project's "OpenSSL" library (or with modified versions of it - * that use the same license as the "OpenSSL" library), and distribute - * the linked executables. You must obey the GNU General Public License - * in all respects for all of the code used other than "OpenSSL". If you - * 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. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see <http://www.gnu.org/licenses/>. - **/ - - -#pragma once - -#include "BagOfTasks.h" -#include "SharedMessageQueue.h" - -#include <stdint.h> -#include <map> - -namespace Orthanc -{ - class BagOfTasksProcessor : public boost::noncopyable - { - private: - enum BagStatus - { - BagStatus_Running, - BagStatus_Canceled, - BagStatus_Failed - }; - - - struct Bag - { - size_t size_; - size_t done_; - BagStatus status_; - - Bag() : - size_(0), - done_(0), - status_(BagStatus_Failed) - { - } - - explicit Bag(size_t size) : - size_(size), - done_(0), - status_(BagStatus_Running) - { - } - }; - - class Task; - - - typedef std::map<uint64_t, Bag> Bags; - typedef std::map<uint64_t, bool> ExitStatus; - - SharedMessageQueue queue_; - - boost::mutex mutex_; - uint64_t countBags_; - Bags bags_; - std::vector<boost::thread*> threads_; - ExitStatus exitStatus_; - bool continue_; - - boost::condition_variable bagFinished_; - - static void Worker(BagOfTasksProcessor* that); - - void Cancel(int64_t bag); - - bool Join(int64_t bag); - - float GetProgress(int64_t bag); - - void SignalProgress(Task& task, - Bag& bag); - - public: - class Handle : public boost::noncopyable - { - friend class BagOfTasksProcessor; - - private: - BagOfTasksProcessor& that_; - uint64_t bag_; - bool hasJoined_; - bool status_; - - Handle(BagOfTasksProcessor& that, - uint64_t bag, - bool empty) : - that_(that), - bag_(bag), - hasJoined_(empty) - { - } - - public: - ~Handle() - { - Join(); - } - - void Cancel() - { - that_.Cancel(bag_); - } - - bool Join(); - - float GetProgress() - { - return that_.GetProgress(bag_); - } - }; - - - explicit BagOfTasksProcessor(size_t countThreads); - - ~BagOfTasksProcessor(); - - Handle* Submit(BagOfTasks& tasks); - }; -}
--- a/Core/MultiThreading/Mutex.cpp Wed May 09 15:57:44 2018 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,122 +0,0 @@ -/** - * Orthanc - A Lightweight, RESTful DICOM Store - * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics - * Department, University Hospital of Liege, Belgium - * Copyright (C) 2017-2018 Osimis S.A., Belgium - * - * This program is free software: you can redistribute it and/or - * modify it under the terms of the GNU General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * In addition, as a special exception, the copyright holders of this - * program give permission to link the code of its release with the - * OpenSSL project's "OpenSSL" library (or with modified versions of it - * that use the same license as the "OpenSSL" library), and distribute - * the linked executables. You must obey the GNU General Public License - * in all respects for all of the code used other than "OpenSSL". If you - * 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. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see <http://www.gnu.org/licenses/>. - **/ - - -#include "../PrecompiledHeaders.h" -#include "Mutex.h" - -#include "../OrthancException.h" - -#if defined(_WIN32) -#include <windows.h> -#elif defined(__linux__) || defined(__FreeBSD_kernel__) || defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) -#include <pthread.h> -#else -#error Support your platform here -#endif - -namespace Orthanc -{ -#if defined (_WIN32) - - struct Mutex::PImpl - { - CRITICAL_SECTION criticalSection_; - }; - - Mutex::Mutex() - { - pimpl_ = new PImpl; - ::InitializeCriticalSection(&pimpl_->criticalSection_); - } - - Mutex::~Mutex() - { - ::DeleteCriticalSection(&pimpl_->criticalSection_); - delete pimpl_; - } - - void Mutex::Lock() - { - ::EnterCriticalSection(&pimpl_->criticalSection_); - } - - void Mutex::Unlock() - { - ::LeaveCriticalSection(&pimpl_->criticalSection_); - } - - -#elif defined(__linux__) || defined(__FreeBSD_kernel__) || defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) - - struct Mutex::PImpl - { - pthread_mutex_t mutex_; - }; - - Mutex::Mutex() - { - pimpl_ = new PImpl; - - if (pthread_mutex_init(&pimpl_->mutex_, NULL) != 0) - { - delete pimpl_; - throw OrthancException(ErrorCode_InternalError); - } - } - - Mutex::~Mutex() - { - pthread_mutex_destroy(&pimpl_->mutex_); - delete pimpl_; - } - - void Mutex::Lock() - { - if (pthread_mutex_lock(&pimpl_->mutex_) != 0) - { - throw OrthancException(ErrorCode_InternalError); - } - } - - void Mutex::Unlock() - { - if (pthread_mutex_unlock(&pimpl_->mutex_) != 0) - { - throw OrthancException(ErrorCode_InternalError); - } - } - -#else -#error Support your plateform here -#endif -}
--- a/Core/MultiThreading/Mutex.h Wed May 09 15:57:44 2018 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,57 +0,0 @@ -/** - * Orthanc - A Lightweight, RESTful DICOM Store - * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics - * Department, University Hospital of Liege, Belgium - * Copyright (C) 2017-2018 Osimis S.A., Belgium - * - * This program is free software: you can redistribute it and/or - * modify it under the terms of the GNU General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * In addition, as a special exception, the copyright holders of this - * program give permission to link the code of its release with the - * OpenSSL project's "OpenSSL" library (or with modified versions of it - * that use the same license as the "OpenSSL" library), and distribute - * the linked executables. You must obey the GNU General Public License - * in all respects for all of the code used other than "OpenSSL". If you - * 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. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see <http://www.gnu.org/licenses/>. - **/ - - -#pragma once - -#include "ILockable.h" - -namespace Orthanc -{ - class Mutex : public ILockable - { - private: - struct PImpl; - - PImpl *pimpl_; - - protected: - virtual void Lock(); - - virtual void Unlock(); - - public: - Mutex(); - - ~Mutex(); - }; -}
--- a/Core/MultiThreading/ReaderWriterLock.cpp Wed May 09 15:57:44 2018 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,126 +0,0 @@ -/** - * Orthanc - A Lightweight, RESTful DICOM Store - * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics - * Department, University Hospital of Liege, Belgium - * Copyright (C) 2017-2018 Osimis S.A., Belgium - * - * This program is free software: you can redistribute it and/or - * modify it under the terms of the GNU General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * In addition, as a special exception, the copyright holders of this - * program give permission to link the code of its release with the - * OpenSSL project's "OpenSSL" library (or with modified versions of it - * that use the same license as the "OpenSSL" library), and distribute - * the linked executables. You must obey the GNU General Public License - * in all respects for all of the code used other than "OpenSSL". If you - * 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. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see <http://www.gnu.org/licenses/>. - **/ - - -#include "../PrecompiledHeaders.h" -#include "ReaderWriterLock.h" - -#include <boost/thread/shared_mutex.hpp> - -namespace Orthanc -{ - namespace - { - // Anonymous namespace to avoid clashes between compilation - // modules. - - class ReaderLockable : public ILockable - { - private: - boost::shared_mutex& lock_; - - protected: - virtual void Lock() - { - lock_.lock_shared(); - } - - virtual void Unlock() - { - lock_.unlock_shared(); - } - - public: - explicit ReaderLockable(boost::shared_mutex& lock) : lock_(lock) - { - } - }; - - - class WriterLockable : public ILockable - { - private: - boost::shared_mutex& lock_; - - protected: - virtual void Lock() - { - lock_.lock(); - } - - virtual void Unlock() - { - lock_.unlock(); - } - - public: - explicit WriterLockable(boost::shared_mutex& lock) : lock_(lock) - { - } - }; - } - - struct ReaderWriterLock::PImpl - { - boost::shared_mutex lock_; - ReaderLockable reader_; - WriterLockable writer_; - - PImpl() : reader_(lock_), writer_(lock_) - { - } - }; - - - ReaderWriterLock::ReaderWriterLock() - { - pimpl_ = new PImpl; - } - - - ReaderWriterLock::~ReaderWriterLock() - { - delete pimpl_; - } - - - ILockable& ReaderWriterLock::ForReader() - { - return pimpl_->reader_; - } - - - ILockable& ReaderWriterLock::ForWriter() - { - return pimpl_->writer_; - } -}
--- a/Core/MultiThreading/ReaderWriterLock.h Wed May 09 15:57:44 2018 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,58 +0,0 @@ -/** - * Orthanc - A Lightweight, RESTful DICOM Store - * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics - * Department, University Hospital of Liege, Belgium - * Copyright (C) 2017-2018 Osimis S.A., Belgium - * - * This program is free software: you can redistribute it and/or - * modify it under the terms of the GNU General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * In addition, as a special exception, the copyright holders of this - * program give permission to link the code of its release with the - * OpenSSL project's "OpenSSL" library (or with modified versions of it - * that use the same license as the "OpenSSL" library), and distribute - * the linked executables. You must obey the GNU General Public License - * in all respects for all of the code used other than "OpenSSL". If you - * 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. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see <http://www.gnu.org/licenses/>. - **/ - - -#pragma once - -#include "ILockable.h" - -#include <boost/noncopyable.hpp> - -namespace Orthanc -{ - class ReaderWriterLock : public boost::noncopyable - { - private: - struct PImpl; - - PImpl *pimpl_; - - public: - ReaderWriterLock(); - - virtual ~ReaderWriterLock(); - - ILockable& ForReader(); - - ILockable& ForWriter(); - }; -}
--- a/OrthancExplorer/explorer.html Wed May 09 15:57:44 2018 +0200 +++ b/OrthancExplorer/explorer.html Wed May 09 17:56:26 2018 +0200 @@ -37,7 +37,10 @@ <div data-role="page" id="find-patients" > <div data-role="header" > <h1><span class="orthanc-name"></span>Find a patient</h1> - <a href="#plugins" data-icon="grid" class="ui-btn-left" data-direction="reverse">Plugins</a> + <div data-type="horizontal" data-role="controlgroup" class="ui-btn-left"> + <a href="#plugins" data-icon="grid" data-role="button" data-direction="reverse">Plugins</a> + <a href="#jobs" data-icon="refresh" data-role="button" data-direction="reverse">Jobs</a> + </div> <div data-type="horizontal" data-role="controlgroup" class="ui-btn-right"> <a href="#upload" data-icon="gear" data-role="button">Upload</a> <a href="#query-retrieve" data-icon="search" data-role="button">Query/Retrieve</a> @@ -418,6 +421,43 @@ </div> </div> + + <div data-role="page" id="jobs" > + <div data-role="header" > + <h1><span class="orthanc-name"></span>Jobs</h1> + <a href="#find-patients" data-icon="home" class="ui-btn-left" data-direction="reverse">Patients</a> + </div> + <div data-role="content"> + <ul id="all-jobs" data-role="listview" data-inset="true" data-filter="true"> + </ul> + </div> + </div> + + <div data-role="page" id="job" > + <div data-role="header" > + <h1><span class="orthanc-name"></span>Job</h1> + <div data-type="horizontal" data-role="controlgroup" class="ui-btn-left"> + <a href="#find-patients" data-icon="home" data-role="button" data-direction="reverse">Patients</a> + <a href="#jobs" data-icon="refresh" data-role="button" data-direction="reverse">Jobs</a> + </div> + </div> + <div data-role="content"> + <ul data-role="listview" data-inset="true" data-filter="true" id="job-info"> + </ul> + + <fieldset class="ui-grid-b"> + <div class="ui-block-a"></div> + <div class="ui-block-b"> + <button id="job-delete" data-theme="b">Delete job</button> + <button id="job-retry" data-theme="b">Retry job</button> + <button id="job-resubmit" data-theme="b">Resubmit job</button> + <button id="job-pause" data-theme="b">Pause job</button> + <button id="job-resume" data-theme="b">Resume job</button> + </div> + <div class="ui-block-c"></div> + </fieldset> + </div> + </div> <div id="peer-store" style="display:none;" class="ui-body-c"> <p align="center"><b>Sending to Orthanc peer...</b></p>
--- a/OrthancExplorer/explorer.js Wed May 09 15:57:44 2018 +0200 +++ b/OrthancExplorer/explorer.js Wed May 09 17:56:26 2018 +0200 @@ -1104,3 +1104,140 @@ } }); }); + + + +function ParseJobTime(s) +{ + var t = (s.substr(0, 4) + '-' + + s.substr(4, 2) + '-' + + s.substr(6, 5) + ':' + + s.substr(11, 2) + ':' + + s.substr(13)); + var utc = new Date(t); + + // Convert from UTC to local time + return new Date(utc.getTime() - utc.getTimezoneOffset() * 60000); +} + + +function AddJobField(target, description, field) +{ + if (!(typeof field === 'undefined')) { + target.append($('<p>') + .text(description) + .append($('<strong>').text(field))); + } +} + + +function AddJobDateField(target, description, field) +{ + if (!(typeof field === 'undefined')) { + target.append($('<p>') + .text(description) + .append($('<strong>').text(ParseJobTime(field)))); + } +} + + +$('#jobs').live('pagebeforeshow', function() { + $.ajax({ + url: '../jobs?expand', + dataType: 'json', + async: false, + cache: false, + success: function(jobs) { + var target = $('#all-jobs'); + $('li', target).remove(); + + jobs.map(function(job) { + var li = $('<li>'); + var item = $('<a>'); + li.append(item); + item.attr('href', '#job?uuid=' + job.ID); + item.append($('<h1>').text(job.Type)); + item.append($('<span>').addClass('ui-li-count').text(job.State)); + AddJobField(item, 'ID: ', job.ID); + AddJobField(item, 'Local AET: ', job.PublicContent.LocalAet); + AddJobField(item, 'Remote AET: ', job.PublicContent.RemoteAet); + AddJobDateField(item, 'Creation time: ', job.CreationTime); + AddJobDateField(item, 'Completion time: ', job.CompletionTime); + AddJobDateField(item, 'ETA: ', job.EstimatedTimeOfArrival); + target.append(li); + }); + + target.listview('refresh'); + } + }); +}); + + +$('#job').live('pagebeforeshow', function() { + if ($.mobile.pageData) { + var pageData = DeepCopy($.mobile.pageData); + + $.ajax({ + url: '../jobs/' + pageData.uuid, + dataType: 'json', + async: false, + cache: false, + success: function(job) { + var target = $('#job-info'); + $('li', target).remove(); + + target.append($('<li>') + .attr('data-role', 'list-divider') + .text('General information about the job')); + + var block = $('<li>'); + for (var i in job) { + if (i == 'CreationTime' || + i == 'CompletionTime' || + i == 'EstimatedTimeOfArrival') { + AddJobDateField(block, i + ': ', job[i]); + } else if (i != 'InternalContent' && + i != 'PublicContent' && + i != 'Timestamp') { + AddJobField(block, i + ': ', job[i]); + } + } + + target.append(block); + + target.append($('<li>') + .attr('data-role', 'list-divider') + .text('Detailed information')); + + var block = $('<li>'); + for (var i in job.PublicContent) { + AddJobField(block, i + ': ', JSON.stringify(job.PublicContent[i])); + } + + target.append(block); + + target.listview('refresh'); + + $('#job-delete').closest('.ui-btn').show(); + $('#job-retry').closest('.ui-btn').hide(); + $('#job-resubmit').closest('.ui-btn').hide(); + $('#job-pause').closest('.ui-btn').hide(); + $('#job-resume').closest('.ui-btn').hide(); + + if (job.State == 'Running' || + job.State == 'Pending' || + job.State == 'Retry') { + $('#job-pause').closest('.ui-btn').show(); + } + else if (job.State == 'Success') { + } + else if (job.State == 'Failure') { + $('#job-resubmit').closest('.ui-btn').show(); + } + else if (job.State == 'Paused') { + $('#job-resume').closest('.ui-btn').show(); + } + } + }); + } +});
--- a/OrthancServer/OrthancRestApi/OrthancRestModalities.cpp Wed May 09 15:57:44 2018 +0200 +++ b/OrthancServer/OrthancRestApi/OrthancRestModalities.cpp Wed May 09 17:56:26 2018 +0200 @@ -44,6 +44,352 @@ #include "../QueryRetrieveHandler.h" #include "../ServerToolbox.h" + + +namespace Orthanc +{ + class InstancesIteratorJob : public IJob + { + private: + bool started_; + std::vector<std::string> instances_; + size_t position_; + + public: + InstancesIteratorJob() : + started_(false), + position_(0) + { + } + + void Reserve(size_t size) + { + if (started_) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + instances_.reserve(size); + } + } + + size_t GetInstancesCount() const + { + return instances_.size(); + } + + void AddInstance(const std::string& instance) + { + if (started_) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + instances_.push_back(instance); + } + } + + virtual void Start() + { + started_ = true; + } + + virtual float GetProgress() + { + if (instances_.size() == 0) + { + return 0; + } + else + { + return (static_cast<float>(position_) / + static_cast<float>(instances_.size())); + } + } + + bool IsStarted() const + { + return started_; + } + + bool IsDone() const + { + return (position_ >= instances_.size()); + } + + void Next() + { + if (IsDone()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + position_ += 1; + } + } + + const std::string& GetCurrentInstance() const + { + if (IsDone()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + return instances_[position_]; + } + } + }; + + + class StoreScuJob : public InstancesIteratorJob + { + private: + ServerContext& context_; + std::string localAet_; + RemoteModalityParameters remote_; + bool permissive_; + std::string moveOriginatorAet_; + uint16_t moveOriginatorId_; + std::auto_ptr<DicomUserConnection> connection_; + std::set<std::string> failedInstances_; + + void Open() + { + if (connection_.get() == NULL) + { + connection_.reset(new DicomUserConnection); + connection_->SetLocalApplicationEntityTitle(localAet_); + connection_->SetRemoteModality(remote_); + connection_->Open(); + } + } + + public: + StoreScuJob(ServerContext& context) : + context_(context), + localAet_("ORTHANC"), + permissive_(false), + moveOriginatorId_(0) // By default, not a C-MOVE + { + } + + void AddResource(const std::string& publicId) + { + typedef std::list<std::string> Instances; + + Instances instances; + context_.GetIndex().GetChildInstances(instances, publicId); + + Reserve(GetInstancesCount() + instances.size()); + + for (Instances::const_iterator it = instances.begin(); + it != instances.end(); ++it) + { + AddInstance(*it); + } + } + + const std::string& GetLocalAet() const + { + return localAet_; + } + + void SetLocalAet(const std::string& aet) + { + if (IsStarted()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + localAet_ = aet; + } + } + + const RemoteModalityParameters& GetRemoteModality() const + { + return remote_; + } + + void SetRemoteModality(const RemoteModalityParameters& remote) + { + if (IsStarted()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + remote_ = remote; + } + } + + bool IsPermissive() const + { + return permissive_; + } + + void SetPermissive(bool permissive) + { + if (IsStarted()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else + { + permissive_ = permissive; + } + } + + bool HasMoveOriginator() const + { + return moveOriginatorId_ != 0; + } + + const std::string& GetMoveOriginatorAet() const + { + if (HasMoveOriginator()) + { + return moveOriginatorAet_; + } + else + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + } + + uint16_t GetMoveOriginatorId() const + { + if (HasMoveOriginator()) + { + return moveOriginatorId_; + } + else + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + } + + void SetMoveOriginator(const std::string& aet, + int id) + { + if (IsStarted()) + { + throw OrthancException(ErrorCode_BadSequenceOfCalls); + } + else if (id < 0 || + id >= 65536) + { + throw OrthancException(ErrorCode_ParameterOutOfRange); + } + else + { + moveOriginatorId_ = static_cast<uint16_t>(id); + moveOriginatorAet_ = aet; + } + } + + virtual JobStepResult* ExecuteStep() + { + if (IsDone()) + { + return new JobStepResult(JobStepCode_Success); + } + + Open(); + + bool ok = false; + + try + { + std::string dicom; + context_.ReadDicom(dicom, GetCurrentInstance()); + + if (HasMoveOriginator()) + { + connection_->Store(dicom, moveOriginatorAet_, moveOriginatorId_); + } + else + { + connection_->Store(dicom); + } + + boost::this_thread::sleep(boost::posix_time::milliseconds(300)); + + ok = true; + } + catch (OrthancException& e) + { + } + + if (!ok) + { + if (permissive_) + { + failedInstances_.insert(GetCurrentInstance()); + } + else + { + return new JobStepResult(JobStepCode_Failure); + } + } + + Next(); + + if (IsDone()) + { + return new JobStepResult(JobStepCode_Success); + } + else + { + return new JobStepResult(JobStepCode_Continue); + } + } + + virtual void ReleaseResources() // For pausing jobs + { + connection_.reset(NULL); + } + + virtual void GetJobType(std::string& target) + { + target = "C-Store"; + } + + virtual void GetPublicContent(Json::Value& value) + { + value["LocalAet"] = localAet_; + value["RemoteAet"] = remote_.GetApplicationEntityTitle(); + + if (HasMoveOriginator()) + { + value["MoveOriginatorAET"] = GetMoveOriginatorAet(); + value["MoveOriginatorID"] = GetMoveOriginatorId(); + } + + Json::Value v = Json::arrayValue; + for (std::set<std::string>::const_iterator it = failedInstances_.begin(); + it != failedInstances_.end(); ++it) + { + v.append(*it); + } + + value["FailedInstances"] = v; + } + + virtual void GetInternalContent(Json::Value& value) + { + // TODO + } + }; +} + + + + namespace Orthanc { /*************************************************************************** @@ -689,6 +1035,7 @@ bool asynchronous = Toolbox::GetJsonBooleanField(request, "Asynchronous", false); std::string moveOriginatorAET = Toolbox::GetJsonStringField(request, "MoveOriginatorAet", context.GetDefaultLocalApplicationEntityTitle()); int moveOriginatorID = Toolbox::GetJsonIntegerField(request, "MoveOriginatorID", 0 /* By default, not a C-MOVE */); + int priority = Toolbox::GetJsonIntegerField(request, "Priority", 0); if (moveOriginatorID < 0 || moveOriginatorID >= 65536) @@ -698,6 +1045,44 @@ RemoteModalityParameters p = Configuration::GetModalityUsingSymbolicName(remote); +#if 1 + std::auto_ptr<StoreScuJob> job(new StoreScuJob(context)); + job->SetLocalAet(localAet); + job->SetRemoteModality(p); + job->SetPermissive(permissive); + + if (moveOriginatorID != 0) + { + job->SetMoveOriginator(moveOriginatorAET, static_cast<uint16_t>(moveOriginatorID)); + } + + for (std::list<std::string>::const_iterator + it = instances.begin(); it != instances.end(); ++it) + { + job->AddInstance(*it); + } + + if (asynchronous) + { + // Asynchronous mode: Submit the job, but don't wait for its completion + std::string id; + context.GetJobsEngine().GetRegistry().Submit(id, job.release(), priority); + + Json::Value v; + v["ID"] = id; + call.GetOutput().AnswerJson(v); + } + else if (context.GetJobsEngine().GetRegistry().SubmitAndWait(job.release(), priority)) + { + // Synchronous mode: We have submitted and waited for completion + call.GetOutput().AnswerBuffer("{}", "application/json"); + } + else + { + call.GetOutput().SignalError(HttpStatus_500_InternalServerError); + } + +#else ServerJob job; for (std::list<std::string>::const_iterator it = instances.begin(); it != instances.end(); ++it) @@ -729,6 +1114,7 @@ { call.GetOutput().SignalError(HttpStatus_500_InternalServerError); } +#endif }
--- a/OrthancServer/OrthancRestApi/OrthancRestSystem.cpp Wed May 09 15:57:44 2018 +0200 +++ b/OrthancServer/OrthancRestApi/OrthancRestSystem.cpp Wed May 09 17:56:26 2018 +0200 @@ -146,6 +146,24 @@ } + static void GetDefaultEncoding(RestApiGetCall& call) + { + Encoding encoding = GetDefaultDicomEncoding(); + call.GetOutput().AnswerBuffer(EnumerationToString(encoding), "text/plain"); + } + + + static void SetDefaultEncoding(RestApiPutCall& call) + { + Encoding encoding = StringToEncoding(call.GetBodyData()); + + Configuration::SetDefaultEncoding(encoding); + + call.GetOutput().AnswerBuffer(EnumerationToString(encoding), "text/plain"); + } + + + // Plugins information ------------------------------------------------------ static void ListPlugins(RestApiGetCall& call) @@ -251,23 +269,65 @@ } - static void GetDefaultEncoding(RestApiGetCall& call) - { - Encoding encoding = GetDefaultDicomEncoding(); - call.GetOutput().AnswerBuffer(EnumerationToString(encoding), "text/plain"); - } - static void SetDefaultEncoding(RestApiPutCall& call) + // Jobs information ------------------------------------------------------ + + static void ListJobs(RestApiGetCall& call) { - Encoding encoding = StringToEncoding(call.GetBodyData()); + bool expand = call.HasArgument("expand"); + + Json::Value v = Json::arrayValue; + + std::set<std::string> jobs; + OrthancRestApi::GetContext(call).GetJobsEngine().GetRegistry().ListJobs(jobs); - Configuration::SetDefaultEncoding(encoding); - - call.GetOutput().AnswerBuffer(EnumerationToString(encoding), "text/plain"); + for (std::set<std::string>::const_iterator it = jobs.begin(); + it != jobs.end(); ++it) + { + if (expand) + { + JobInfo info; + if (OrthancRestApi::GetContext(call).GetJobsEngine().GetRegistry().GetJobInfo(info, *it)) + { + Json::Value tmp; + info.Serialize(tmp); + v.append(tmp); + } + } + else + { + v.append(*it); + } + } + + call.GetOutput().AnswerJson(v); } + static void GetJobInfo(RestApiGetCall& call) + { + std::string id = call.GetUriComponent("id", ""); + JobInfo info; + if (OrthancRestApi::GetContext(call).GetJobsEngine().GetRegistry().GetJobInfo(info, id)) + { + Json::Value json; + info.Serialize(json); + call.GetOutput().AnswerJson(json); + } + } + + static void PauseJob(RestApiPostCall& call) + { + std::string id = call.GetUriComponent("id", ""); + + if (OrthancRestApi::GetContext(call).GetJobsEngine().GetRegistry().Pause(id)) + { + call.GetOutput().AnswerBuffer("{}", "application/json"); + } + } + + void OrthancRestApi::RegisterSystem() { Register("/", ServeRoot); @@ -284,5 +344,9 @@ Register("/plugins", ListPlugins); Register("/plugins/{id}", GetPlugin); Register("/plugins/explorer.js", GetOrthancExplorerPlugins); + + Register("/jobs", ListJobs); + Register("/jobs/{id}", GetJobInfo); + Register("/jobs/{id}/pause", PauseJob); } }
--- a/OrthancServer/ServerContext.cpp Wed May 09 15:57:44 2018 +0200 +++ b/OrthancServer/ServerContext.cpp Wed May 09 17:56:26 2018 +0200 @@ -134,6 +134,10 @@ listeners_.push_back(ServerListener(lua_, "Lua")); + jobsEngine_.SetWorkersCount(Configuration::GetGlobalUnsignedIntegerParameter("ConcurrentJobs", 2)); + //jobsEngine_.SetMaxCompleted // TODO + jobsEngine_.Start(); + changeThread_ = boost::thread(ChangeThread, this); } @@ -168,6 +172,7 @@ scu_.Finalize(); // Do not change the order below! + jobsEngine_.Stop(); scheduler_.Stop(); index_.Stop(); }
--- a/OrthancServer/ServerContext.h Wed May 09 15:57:44 2018 +0200 +++ b/OrthancServer/ServerContext.h Wed May 09 17:56:26 2018 +0200 @@ -48,6 +48,7 @@ #include "Scheduler/ServerScheduler.h" #include "ServerIndex.h" #include "OrthancHttpHandler.h" +#include "../Core/JobsEngine/JobsEngine.h" #include <boost/filesystem.hpp> #include <boost/thread.hpp> @@ -120,6 +121,7 @@ MemoryCache dicomCache_; ReusableDicomUserConnection scu_; ServerScheduler scheduler_; + JobsEngine jobsEngine_; LuaScripting lua_; @@ -248,6 +250,11 @@ return scheduler_; } + JobsEngine& GetJobsEngine() + { + return jobsEngine_; + } + bool DeleteResource(Json::Value& target, const std::string& uuid, ResourceType expectedType);
--- a/OrthancServer/ServerIndex.cpp Wed May 09 15:57:44 2018 +0200 +++ b/OrthancServer/ServerIndex.cpp Wed May 09 17:56:26 2018 +0200 @@ -1212,7 +1212,7 @@ ResourceType type; if (!db_.LookupResource(id, type, publicId)) { - throw OrthancException(ErrorCode_InternalError); + throw OrthancException(ErrorCode_InexistentItem); } std::string patientId;
--- a/Resources/CMake/OrthancFrameworkConfiguration.cmake Wed May 09 15:57:44 2018 +0200 +++ b/Resources/CMake/OrthancFrameworkConfiguration.cmake Wed May 09 17:56:26 2018 +0200 @@ -498,9 +498,10 @@ list(APPEND ORTHANC_CORE_SOURCES_INTERNAL ${ORTHANC_ROOT}/Core/Cache/SharedArchive.cpp ${ORTHANC_ROOT}/Core/FileStorage/FilesystemStorage.cpp - ${ORTHANC_ROOT}/Core/MultiThreading/BagOfTasksProcessor.cpp - ${ORTHANC_ROOT}/Core/MultiThreading/Mutex.cpp - ${ORTHANC_ROOT}/Core/MultiThreading/ReaderWriterLock.cpp + ${ORTHANC_ROOT}/Core/JobsEngine/JobInfo.cpp + ${ORTHANC_ROOT}/Core/JobsEngine/JobStatus.cpp + ${ORTHANC_ROOT}/Core/JobsEngine/JobsEngine.cpp + ${ORTHANC_ROOT}/Core/JobsEngine/JobsRegistry.cpp ${ORTHANC_ROOT}/Core/MultiThreading/RunnableWorkersPool.cpp ${ORTHANC_ROOT}/Core/MultiThreading/Semaphore.cpp ${ORTHANC_ROOT}/Core/MultiThreading/SharedMessageQueue.cpp
--- a/Resources/Configuration.json Wed May 09 15:57:44 2018 +0200 +++ b/Resources/Configuration.json Wed May 09 17:56:26 2018 +0200 @@ -43,6 +43,11 @@ "Plugins" : [ ], + // Maximum number of processing jobs that are simultanously running + // at any given time. A value of "0" indicates to use all the + // available CPU logical cores. To emulate Orthanc <= 1.3.2, set + // this value to "1". + "ConcurrentJobs" : 2, /**
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/Graveyard/BagOfTasks.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,84 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "../ICommand.h" + +#include <list> +#include <cstddef> + +namespace Orthanc +{ + class BagOfTasks : public boost::noncopyable + { + private: + typedef std::list<ICommand*> Tasks; + + Tasks tasks_; + + public: + ~BagOfTasks() + { + for (Tasks::iterator it = tasks_.begin(); it != tasks_.end(); ++it) + { + delete *it; + } + } + + ICommand* Pop() + { + ICommand* task = tasks_.front(); + tasks_.pop_front(); + return task; + } + + void Push(ICommand* task) // Takes ownership + { + if (task != NULL) + { + tasks_.push_back(task); + } + } + + size_t GetSize() const + { + return tasks_.size(); + } + + bool IsEmpty() const + { + return tasks_.empty(); + } + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/Graveyard/BagOfTasksProcessor.cpp Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,277 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#include "../PrecompiledHeaders.h" +#include "BagOfTasksProcessor.h" + +#include "../Logging.h" +#include "../OrthancException.h" + +#include <stdio.h> + +namespace Orthanc +{ + class BagOfTasksProcessor::Task : public IDynamicObject + { + private: + uint64_t bag_; + std::auto_ptr<ICommand> command_; + + public: + Task(uint64_t bag, + ICommand* command) : + bag_(bag), + command_(command) + { + } + + bool Execute() + { + try + { + return command_->Execute(); + } + catch (OrthancException& e) + { + LOG(ERROR) << "Exception while processing a bag of tasks: " << e.What(); + return false; + } + catch (std::runtime_error& e) + { + LOG(ERROR) << "Runtime exception while processing a bag of tasks: " << e.what(); + return false; + } + catch (...) + { + LOG(ERROR) << "Native exception while processing a bag of tasks"; + return false; + } + } + + uint64_t GetBag() + { + return bag_; + } + }; + + + void BagOfTasksProcessor::SignalProgress(Task& task, + Bag& bag) + { + assert(bag.done_ < bag.size_); + + bag.done_ += 1; + + if (bag.done_ == bag.size_) + { + exitStatus_[task.GetBag()] = (bag.status_ == BagStatus_Running); + bagFinished_.notify_all(); + } + } + + void BagOfTasksProcessor::Worker(BagOfTasksProcessor* that) + { + while (that->continue_) + { + std::auto_ptr<IDynamicObject> obj(that->queue_.Dequeue(100)); + if (obj.get() != NULL) + { + Task& task = *dynamic_cast<Task*>(obj.get()); + + { + boost::mutex::scoped_lock lock(that->mutex_); + + Bags::iterator bag = that->bags_.find(task.GetBag()); + assert(bag != that->bags_.end()); + assert(bag->second.done_ < bag->second.size_); + + if (bag->second.status_ != BagStatus_Running) + { + // Do not execute this task, as its parent bag of tasks + // has failed or is tagged as canceled + that->SignalProgress(task, bag->second); + continue; + } + } + + bool success = task.Execute(); + + { + boost::mutex::scoped_lock lock(that->mutex_); + + Bags::iterator bag = that->bags_.find(task.GetBag()); + assert(bag != that->bags_.end()); + + if (!success) + { + bag->second.status_ = BagStatus_Failed; + } + + that->SignalProgress(task, bag->second); + } + } + } + } + + + void BagOfTasksProcessor::Cancel(int64_t bag) + { + boost::mutex::scoped_lock lock(mutex_); + + Bags::iterator it = bags_.find(bag); + if (it != bags_.end()) + { + it->second.status_ = BagStatus_Canceled; + } + } + + + bool BagOfTasksProcessor::Join(int64_t bag) + { + boost::mutex::scoped_lock lock(mutex_); + + while (continue_) + { + ExitStatus::iterator it = exitStatus_.find(bag); + if (it == exitStatus_.end()) // The bag is still running + { + bagFinished_.wait(lock); + } + else + { + bool status = it->second; + exitStatus_.erase(it); + return status; + } + } + + return false; // The processor is stopping + } + + + float BagOfTasksProcessor::GetProgress(int64_t bag) + { + boost::mutex::scoped_lock lock(mutex_); + + Bags::const_iterator it = bags_.find(bag); + if (it == bags_.end()) + { + // The bag of tasks has finished + return 1.0f; + } + else + { + return (static_cast<float>(it->second.done_) / + static_cast<float>(it->second.size_)); + } + } + + + bool BagOfTasksProcessor::Handle::Join() + { + if (hasJoined_) + { + return status_; + } + else + { + status_ = that_.Join(bag_); + hasJoined_ = true; + return status_; + } + } + + + BagOfTasksProcessor::BagOfTasksProcessor(size_t countThreads) : + countBags_(0), + continue_(true) + { + if (countThreads == 0) + { + throw OrthancException(ErrorCode_ParameterOutOfRange); + } + + threads_.resize(countThreads); + + for (size_t i = 0; i < threads_.size(); i++) + { + threads_[i] = new boost::thread(Worker, this); + } + } + + + BagOfTasksProcessor::~BagOfTasksProcessor() + { + continue_ = false; + + bagFinished_.notify_all(); // Wakes up all the pending "Join()" + + for (size_t i = 0; i < threads_.size(); i++) + { + if (threads_[i]) + { + if (threads_[i]->joinable()) + { + threads_[i]->join(); + } + + delete threads_[i]; + threads_[i] = NULL; + } + } + } + + + BagOfTasksProcessor::Handle* BagOfTasksProcessor::Submit(BagOfTasks& tasks) + { + if (tasks.GetSize() == 0) + { + return new Handle(*this, 0, true); + } + + boost::mutex::scoped_lock lock(mutex_); + + uint64_t id = countBags_; + countBags_ += 1; + + Bag bag(tasks.GetSize()); + bags_[id] = bag; + + while (!tasks.IsEmpty()) + { + queue_.Enqueue(new Task(id, tasks.Pop())); + } + + return new Handle(*this, id, false); + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/Graveyard/BagOfTasksProcessor.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,150 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "BagOfTasks.h" +#include "SharedMessageQueue.h" + +#include <stdint.h> +#include <map> + +namespace Orthanc +{ + class BagOfTasksProcessor : public boost::noncopyable + { + private: + enum BagStatus + { + BagStatus_Running, + BagStatus_Canceled, + BagStatus_Failed + }; + + + struct Bag + { + size_t size_; + size_t done_; + BagStatus status_; + + Bag() : + size_(0), + done_(0), + status_(BagStatus_Failed) + { + } + + explicit Bag(size_t size) : + size_(size), + done_(0), + status_(BagStatus_Running) + { + } + }; + + class Task; + + + typedef std::map<uint64_t, Bag> Bags; + typedef std::map<uint64_t, bool> ExitStatus; + + SharedMessageQueue queue_; + + boost::mutex mutex_; + uint64_t countBags_; + Bags bags_; + std::vector<boost::thread*> threads_; + ExitStatus exitStatus_; + bool continue_; + + boost::condition_variable bagFinished_; + + static void Worker(BagOfTasksProcessor* that); + + void Cancel(int64_t bag); + + bool Join(int64_t bag); + + float GetProgress(int64_t bag); + + void SignalProgress(Task& task, + Bag& bag); + + public: + class Handle : public boost::noncopyable + { + friend class BagOfTasksProcessor; + + private: + BagOfTasksProcessor& that_; + uint64_t bag_; + bool hasJoined_; + bool status_; + + Handle(BagOfTasksProcessor& that, + uint64_t bag, + bool empty) : + that_(that), + bag_(bag), + hasJoined_(empty) + { + } + + public: + ~Handle() + { + Join(); + } + + void Cancel() + { + that_.Cancel(bag_); + } + + bool Join(); + + float GetProgress() + { + return that_.GetProgress(bag_); + } + }; + + + explicit BagOfTasksProcessor(size_t countThreads); + + ~BagOfTasksProcessor(); + + Handle* Submit(BagOfTasks& tasks); + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/Graveyard/Mutex.cpp Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,122 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#include "../PrecompiledHeaders.h" +#include "Mutex.h" + +#include "../OrthancException.h" + +#if defined(_WIN32) +#include <windows.h> +#elif defined(__linux__) || defined(__FreeBSD_kernel__) || defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) +#include <pthread.h> +#else +#error Support your platform here +#endif + +namespace Orthanc +{ +#if defined (_WIN32) + + struct Mutex::PImpl + { + CRITICAL_SECTION criticalSection_; + }; + + Mutex::Mutex() + { + pimpl_ = new PImpl; + ::InitializeCriticalSection(&pimpl_->criticalSection_); + } + + Mutex::~Mutex() + { + ::DeleteCriticalSection(&pimpl_->criticalSection_); + delete pimpl_; + } + + void Mutex::Lock() + { + ::EnterCriticalSection(&pimpl_->criticalSection_); + } + + void Mutex::Unlock() + { + ::LeaveCriticalSection(&pimpl_->criticalSection_); + } + + +#elif defined(__linux__) || defined(__FreeBSD_kernel__) || defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) + + struct Mutex::PImpl + { + pthread_mutex_t mutex_; + }; + + Mutex::Mutex() + { + pimpl_ = new PImpl; + + if (pthread_mutex_init(&pimpl_->mutex_, NULL) != 0) + { + delete pimpl_; + throw OrthancException(ErrorCode_InternalError); + } + } + + Mutex::~Mutex() + { + pthread_mutex_destroy(&pimpl_->mutex_); + delete pimpl_; + } + + void Mutex::Lock() + { + if (pthread_mutex_lock(&pimpl_->mutex_) != 0) + { + throw OrthancException(ErrorCode_InternalError); + } + } + + void Mutex::Unlock() + { + if (pthread_mutex_unlock(&pimpl_->mutex_) != 0) + { + throw OrthancException(ErrorCode_InternalError); + } + } + +#else +#error Support your plateform here +#endif +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/Graveyard/Mutex.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,57 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "ILockable.h" + +namespace Orthanc +{ + class Mutex : public ILockable + { + private: + struct PImpl; + + PImpl *pimpl_; + + protected: + virtual void Lock(); + + virtual void Unlock(); + + public: + Mutex(); + + ~Mutex(); + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/Graveyard/ReaderWriterLock.cpp Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,126 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#include "../PrecompiledHeaders.h" +#include "ReaderWriterLock.h" + +#include <boost/thread/shared_mutex.hpp> + +namespace Orthanc +{ + namespace + { + // Anonymous namespace to avoid clashes between compilation + // modules. + + class ReaderLockable : public ILockable + { + private: + boost::shared_mutex& lock_; + + protected: + virtual void Lock() + { + lock_.lock_shared(); + } + + virtual void Unlock() + { + lock_.unlock_shared(); + } + + public: + explicit ReaderLockable(boost::shared_mutex& lock) : lock_(lock) + { + } + }; + + + class WriterLockable : public ILockable + { + private: + boost::shared_mutex& lock_; + + protected: + virtual void Lock() + { + lock_.lock(); + } + + virtual void Unlock() + { + lock_.unlock(); + } + + public: + explicit WriterLockable(boost::shared_mutex& lock) : lock_(lock) + { + } + }; + } + + struct ReaderWriterLock::PImpl + { + boost::shared_mutex lock_; + ReaderLockable reader_; + WriterLockable writer_; + + PImpl() : reader_(lock_), writer_(lock_) + { + } + }; + + + ReaderWriterLock::ReaderWriterLock() + { + pimpl_ = new PImpl; + } + + + ReaderWriterLock::~ReaderWriterLock() + { + delete pimpl_; + } + + + ILockable& ReaderWriterLock::ForReader() + { + return pimpl_->reader_; + } + + + ILockable& ReaderWriterLock::ForWriter() + { + return pimpl_->writer_; + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/Graveyard/ReaderWriterLock.h Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,58 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2018 Osimis S.A., Belgium + * + * This program is free software: you can redistribute it and/or + * modify it under the terms of the GNU General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * In addition, as a special exception, the copyright holders of this + * program give permission to link the code of its release with the + * OpenSSL project's "OpenSSL" library (or with modified versions of it + * that use the same license as the "OpenSSL" library), and distribute + * the linked executables. You must obey the GNU General Public License + * in all respects for all of the code used other than "OpenSSL". If you + * 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. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + **/ + + +#pragma once + +#include "ILockable.h" + +#include <boost/noncopyable.hpp> + +namespace Orthanc +{ + class ReaderWriterLock : public boost::noncopyable + { + private: + struct PImpl; + + PImpl *pimpl_; + + public: + ReaderWriterLock(); + + virtual ~ReaderWriterLock(); + + ILockable& ForReader(); + + ILockable& ForWriter(); + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/ImplementationNotes/JobsEngineStates.dot Wed May 09 17:56:26 2018 +0200 @@ -0,0 +1,23 @@ +// dot -Tpdf JobsEngineStates.dot -o JobsEngineStates.pdf + +digraph G +{ + rankdir="LR"; + init [shape=point]; + failure, success [shape=doublecircle]; + + // Internal transitions + init -> pending; + pending -> running; + running -> success; + running -> failure; + running -> retry; + retry -> pending [label="timeout"]; + + // External actions + failure -> pending [label="Resubmit()" fontcolor="red"]; + paused -> pending [label="Resume()" fontcolor="red"]; + pending -> paused [label="Pause()" fontcolor="red"]; + retry -> paused [label="Pause()" fontcolor="red"]; + running -> paused [label="Pause()" fontcolor="red"]; +}
--- a/UnitTestsSources/MultiThreadingTests.cpp Wed May 09 15:57:44 2018 +0200 +++ b/UnitTestsSources/MultiThreadingTests.cpp Wed May 09 17:56:26 2018 +0200 @@ -34,13 +34,13 @@ #include "PrecompiledHeadersUnitTests.h" #include "gtest/gtest.h" -#include "../OrthancServer/Scheduler/ServerScheduler.h" +#include "../Core/JobsEngine/JobStepRetry.h" +#include "../Core/JobsEngine/JobsEngine.h" +#include "../Core/MultiThreading/Locker.h" #include "../Core/OrthancException.h" #include "../Core/SystemToolbox.h" #include "../Core/Toolbox.h" -#include "../Core/MultiThreading/Locker.h" -#include "../Core/MultiThreading/Mutex.h" -#include "../Core/MultiThreading/ReaderWriterLock.h" +#include "../OrthancServer/Scheduler/ServerScheduler.h" using namespace Orthanc; @@ -106,27 +106,6 @@ } -TEST(MultiThreading, Mutex) -{ - Mutex mutex; - Locker locker(mutex); -} - - -TEST(MultiThreading, ReaderWriterLock) -{ - ReaderWriterLock lock; - - { - Locker locker1(lock.ForReader()); - Locker locker2(lock.ForReader()); - } - - { - Locker locker3(lock.ForWriter()); - } -} - #include "../Core/DicomNetworking/ReusableDicomUserConnection.h" @@ -259,3 +238,441 @@ t.join(); } } + + + +class DummyJob : public Orthanc::IJob +{ +private: + JobStepResult result_; + unsigned int count_; + unsigned int steps_; + +public: + DummyJob() : + result_(Orthanc::JobStepCode_Success), + count_(0), + steps_(4) + { + } + + explicit DummyJob(JobStepResult result) : + result_(result), + count_(0), + steps_(4) + { + } + + virtual void Start() + { + } + + virtual JobStepResult* ExecuteStep() + { + boost::this_thread::sleep(boost::posix_time::milliseconds(10)); + + if (count_ == steps_ - 1) + { + return new JobStepResult(result_); + } + else + { + count_++; + return new JobStepResult(JobStepCode_Continue); + } + } + + virtual void ReleaseResources() + { + } + + virtual float GetProgress() + { + 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, + const std::string& id, + Orthanc::JobState state) +{ + Orthanc::JobState s; + if (registry.GetState(s, id)) + { + return state == s; + } + else + { + return false; + } +} + + +TEST(JobsRegistry, Priority) +{ + JobsRegistry registry; + + std::string i1, i2, i3, i4; + registry.Submit(i1, new DummyJob(), 10); + registry.Submit(i2, new DummyJob(), 30); + registry.Submit(i3, new DummyJob(), 20); + registry.Submit(i4, new DummyJob(), 5); + + registry.SetMaxCompletedJobs(2); + + std::set<std::string> id; + registry.ListJobs(id); + + ASSERT_EQ(4u, id.size()); + ASSERT_TRUE(id.find(i1) != id.end()); + ASSERT_TRUE(id.find(i2) != id.end()); + ASSERT_TRUE(id.find(i3) != id.end()); + ASSERT_TRUE(id.find(i4) != id.end()); + + ASSERT_TRUE(CheckState(registry, i2, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + ASSERT_EQ(30, job.GetPriority()); + ASSERT_EQ(i2, job.GetId()); + + ASSERT_TRUE(CheckState(registry, i2, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, i2, Orthanc::JobState_Failure)); + ASSERT_TRUE(CheckState(registry, i3, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + ASSERT_EQ(20, job.GetPriority()); + ASSERT_EQ(i3, job.GetId()); + + job.MarkSuccess(); + + ASSERT_TRUE(CheckState(registry, i3, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, i3, Orthanc::JobState_Success)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + ASSERT_EQ(10, job.GetPriority()); + ASSERT_EQ(i1, job.GetId()); + } + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + ASSERT_EQ(5, job.GetPriority()); + ASSERT_EQ(i4, job.GetId()); + } + + { + JobsRegistry::RunningJob job(registry, 1); + ASSERT_FALSE(job.IsValid()); + } + + Orthanc::JobState s; + ASSERT_TRUE(registry.GetState(s, i1)); + ASSERT_FALSE(registry.GetState(s, i2)); // Removed because oldest + ASSERT_FALSE(registry.GetState(s, i3)); // Removed because second oldest + ASSERT_TRUE(registry.GetState(s, i4)); + + registry.SetMaxCompletedJobs(1); // (*) + ASSERT_FALSE(registry.GetState(s, i1)); // Just discarded by (*) + ASSERT_TRUE(registry.GetState(s, i4)); +} + + +TEST(JobsRegistry, Simultaneous) +{ + JobsRegistry registry; + + std::string i1, i2; + registry.Submit(i1, new DummyJob(), 20); + registry.Submit(i2, new DummyJob(), 10); + + ASSERT_TRUE(CheckState(registry, i1, Orthanc::JobState_Pending)); + ASSERT_TRUE(CheckState(registry, i2, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job1(registry, 0); + JobsRegistry::RunningJob job2(registry, 0); + + ASSERT_TRUE(job1.IsValid()); + ASSERT_TRUE(job2.IsValid()); + + job1.MarkFailure(); + job2.MarkSuccess(); + + ASSERT_TRUE(CheckState(registry, i1, Orthanc::JobState_Running)); + ASSERT_TRUE(CheckState(registry, i2, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, i1, Orthanc::JobState_Failure)); + ASSERT_TRUE(CheckState(registry, i2, Orthanc::JobState_Success)); +} + + +TEST(JobsRegistry, Resubmit) +{ + JobsRegistry registry; + + std::string id; + registry.Submit(id, new DummyJob(), 10); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + registry.Resubmit(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + job.MarkFailure(); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + + registry.Resubmit(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Failure)); + + registry.Resubmit(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + ASSERT_EQ(id, job.GetId()); + + job.MarkSuccess(); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Success)); + + registry.Resubmit(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Success)); +} + + +TEST(JobsRegistry, Retry) +{ + JobsRegistry registry; + + std::string id; + registry.Submit(id, new DummyJob(), 10); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + job.MarkRetry(0); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Retry)); + + registry.Resubmit(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Retry)); + + registry.ScheduleRetries(); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + job.MarkSuccess(); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Success)); +} + + +TEST(JobsRegistry, PausePending) +{ + JobsRegistry registry; + + std::string id; + registry.Submit(id, new DummyJob(), 10); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + registry.Pause(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Paused)); + + registry.Pause(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Paused)); + + registry.Resubmit(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Paused)); + + registry.Resume(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); +} + + +TEST(JobsRegistry, PauseRunning) +{ + JobsRegistry registry; + + std::string id; + registry.Submit(id, new DummyJob(), 10); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + + registry.Resubmit(id); + job.MarkPause(); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Paused)); + + registry.Resubmit(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Paused)); + + registry.Resume(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + + job.MarkSuccess(); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Success)); +} + + +TEST(JobsRegistry, PauseRetry) +{ + JobsRegistry registry; + + std::string id; + registry.Submit(id, new DummyJob(), 10); + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + + job.MarkRetry(0); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Running)); + } + + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Retry)); + + registry.Pause(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Paused)); + + registry.Resume(id); + ASSERT_TRUE(CheckState(registry, id, Orthanc::JobState_Pending)); + + { + JobsRegistry::RunningJob job(registry, 0); + ASSERT_TRUE(job.IsValid()); + + job.MarkSuccess(); + 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(100)); + + { + typedef std::set<std::string> Jobs; + + Jobs jobs; + engine.GetRegistry().ListJobs(jobs); + + Json::Value v = Json::arrayValue; + for (Jobs::const_iterator it = jobs.begin(); it != jobs.end(); ++it) + { + JobInfo info; + + if (engine.GetRegistry().GetJobInfo(info, *it)) + { + Json::Value vv; + info.Format(vv); + v.append(vv); + } + } + + std::cout << v << std::endl; + } + std::cout << "====================================================" << std::endl; + + boost::this_thread::sleep(boost::posix_time::milliseconds(100)); + + if (1) + { + printf(">> %d\n", engine.GetRegistry().SubmitAndWait(new DummyJob(JobStepResult(Orthanc::JobStepCode_Failure)), rand() % 10)); + } + + boost::this_thread::sleep(boost::posix_time::milliseconds(100)); + + + engine.Stop(); + + if (0) + { + typedef std::set<std::string> Jobs; + + Jobs jobs; + engine.GetRegistry().ListJobs(jobs); + + Json::Value v = Json::arrayValue; + for (Jobs::const_iterator it = jobs.begin(); it != jobs.end(); ++it) + { + JobInfo info; + + if (engine.GetRegistry().GetJobInfo(info, *it)) + { + Json::Value vv; + info.Format(vv); + v.append(vv); + } + } + + std::cout << v << std::endl; + } +}