orthanc: Resources/Graveyard/OldScheduler/ServerScheduler.cpp comparison

comparison Resources/Graveyard/OldScheduler/ServerScheduler.cpp @ 2610:3ff4c50647ea jobs

moving the old scheduler to the graveyard

author	Sebastien Jodogne <s.jodogne@gmail.com>
date	Sat, 19 May 2018 16:40:26 +0200
parents	OrthancServer/Scheduler/ServerScheduler.cpp@878b59270859
children	4e43e67f8ecf

comparison

equal deleted inserted replaced

-:f7a84b551ee4
+:3ff4c50647ea
+/**
+* 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 "../PrecompiledHeadersServer.h"
+#include "ServerScheduler.h"
+#include "../../Core/OrthancException.h"
+#include "../../Core/Logging.h"
+namespace Orthanc
+{
+namespace
+{
+// Anonymous namespace to avoid clashes between compilation modules
+class Sink : public IServerCommand
+{
+private:
+ListOfStrings& target_;
+public:
+explicit Sink(ListOfStrings& target) : target_(target)
+{
+}
+virtual bool Apply(ListOfStrings& outputs,
+const ListOfStrings& inputs)
+{
+for (ListOfStrings::const_iterator
+it = inputs.begin(); it != inputs.end(); ++it)
+{
+target_.push_back(*it);
+}
+return true;
+}
+};
+}
+ServerScheduler::JobInfo& ServerScheduler::GetJobInfo(const std::string& jobId)
+{
+Jobs::iterator info = jobs_.find(jobId);
+if (info == jobs_.end())
+{
+throw OrthancException(ErrorCode_InternalError);
+}
+return info->second;
+}
+void ServerScheduler::SignalSuccess(const std::string& jobId)
+{
+boost::mutex::scoped_lock lock(mutex_);
+JobInfo& info = GetJobInfo(jobId);
+info.success_++;
+assert(info.failures_ == 0);
+if (info.success_ >= info.size_)
+{
+if (info.watched_)
+{
+watchedJobStatus_[jobId] = JobStatus_Success;
+watchedJobFinished_.notify_all();
+}
+LOG(INFO) << "Job successfully finished (" << info.description_ << ")";
+jobs_.erase(jobId);
+availableJob_.Release();
+}
+}
+void ServerScheduler::SignalFailure(const std::string& jobId)
+{
+boost::mutex::scoped_lock lock(mutex_);
+JobInfo& info = GetJobInfo(jobId);
+info.failures_++;
+if (info.success_ + info.failures_ >= info.size_)
+{
+if (info.watched_)
+{
+watchedJobStatus_[jobId] = JobStatus_Failure;
+watchedJobFinished_.notify_all();
+}
+LOG(ERROR) << "Job has failed (" << info.description_ << ")";
+jobs_.erase(jobId);
+availableJob_.Release();
+}
+}
+void ServerScheduler::Worker(ServerScheduler* that)
+{
+static const int32_t TIMEOUT = 100;
+LOG(WARNING) << "The server scheduler has started";
+while (!that->finish_)
+{
+std::auto_ptr<IDynamicObject> object(that->queue_.Dequeue(TIMEOUT));
+if (object.get() != NULL)
+{
+ServerCommandInstance& filter = dynamic_cast<ServerCommandInstance&>(*object);
+// Skip the execution of this filter if its parent job has
+// previously failed.
+bool jobHasFailed;
+{
+boost::mutex::scoped_lock lock(that->mutex_);
+JobInfo& info = that->GetJobInfo(filter.GetJobId());
+jobHasFailed = (info.failures_ > 0 || info.cancel_);
+}
+if (jobHasFailed)
+{
+that->SignalFailure(filter.GetJobId());
+}
+else
+{
+filter.Execute(*that);
+}
+}
+}
+}
+void ServerScheduler::SubmitInternal(ServerJob& job,
+bool watched)
+{
+availableJob_.Acquire();
+boost::mutex::scoped_lock lock(mutex_);
+JobInfo info;
+info.size_ = job.Submit(queue_, *this);
+info.cancel_ = false;
+info.success_ = 0;
+info.failures_ = 0;
+info.description_ = job.GetDescription();
+info.watched_ = watched;
+assert(info.size_ > 0);
+if (watched)
+{
+watchedJobStatus_[job.GetId()] = JobStatus_Running;
+}
+jobs_[job.GetId()] = info;
+LOG(INFO) << "New job submitted (" << job.description_ << ")";
+}
+ServerScheduler::ServerScheduler(unsigned int maxJobs) : availableJob_(maxJobs)
+{
+if (maxJobs == 0)
+{
+throw OrthancException(ErrorCode_ParameterOutOfRange);
+}
+finish_ = false;
+worker_ = boost::thread(Worker, this);
+}
+ServerScheduler::~ServerScheduler()
+{
+if (!finish_)
+{
+LOG(ERROR) << "INTERNAL ERROR: ServerScheduler::Finalize() should be invoked manually to avoid mess in the destruction order!";
+Stop();
+}
+}
+void ServerScheduler::Stop()
+{
+if (!finish_)
+{
+finish_ = true;
+if (worker_.joinable())
+{
+worker_.join();
+}
+}
+}
+void ServerScheduler::Submit(ServerJob& job)
+{
+if (job.filters_.empty())
+{
+return;
+}
+SubmitInternal(job, false);
+}
+bool ServerScheduler::SubmitAndWait(ListOfStrings& outputs,
+ServerJob& job)
+{
+std::string jobId = job.GetId();
+outputs.clear();
+if (job.filters_.empty())
+{
+return true;
+}
+// Add a sink filter to collect all the results of the filters
+// that have no next filter.
+ServerCommandInstance& sink = job.AddCommand(new Sink(outputs));
+for (std::list<ServerCommandInstance*>::iterator
+it = job.filters_.begin(); it != job.filters_.end(); ++it)
+{
+if ((*it) != &sink &&
+(*it)->IsConnectedToSink())
+{
+(*it)->ConnectOutput(sink);
+}
+}
+// Submit the job
+SubmitInternal(job, true);
+// Wait for the job to complete (either success or failure)
+JobStatus status;
+{
+boost::mutex::scoped_lock lock(mutex_);
+assert(watchedJobStatus_.find(jobId) != watchedJobStatus_.end());
+while (watchedJobStatus_[jobId] == JobStatus_Running)
+{
+watchedJobFinished_.wait(lock);
+}
+status = watchedJobStatus_[jobId];
+watchedJobStatus_.erase(jobId);
+}
+return (status == JobStatus_Success);
+}
+bool ServerScheduler::SubmitAndWait(ServerJob& job)
+{
+ListOfStrings ignoredSink;
+return SubmitAndWait(ignoredSink, job);
+}
+bool ServerScheduler::IsRunning(const std::string& jobId)
+{
+boost::mutex::scoped_lock lock(mutex_);
+return jobs_.find(jobId) != jobs_.end();
+}
+void ServerScheduler::Cancel(const std::string& jobId)
+{
+boost::mutex::scoped_lock lock(mutex_);
+Jobs::iterator job = jobs_.find(jobId);
+if (job != jobs_.end())
+{
+job->second.cancel_ = true;
+LOG(WARNING) << "Canceling a job (" << job->second.description_ << ")";
+}
+}
+float ServerScheduler::GetProgress(const std::string& jobId)
+{
+boost::mutex::scoped_lock lock(mutex_);
+Jobs::iterator job = jobs_.find(jobId);
+if (job == jobs_.end() ||
+job->second.size_ == 0  /* should never happen */)
+{
+// This job is not running
+return 1;
+}
+if (job->second.failures_ != 0)
+{
+return 1;
+}
+if (job->second.size_ == 1)
+{
+return static_cast<float>(job->second.success_);
+}
+return (static_cast<float>(job->second.success_) /
+static_cast<float>(job->second.size_ - 1));
+}
+void ServerScheduler::GetListOfJobs(ListOfStrings& jobs)
+{
+boost::mutex::scoped_lock lock(mutex_);
+jobs.clear();
+for (Jobs::const_iterator
+it = jobs_.begin(); it != jobs_.end(); ++it)
+{
+jobs.push_back(it->first);
+}
+}
+}

Mercurial > hg > orthanc

comparison Resources/Graveyard/OldScheduler/ServerScheduler.cpp @ 2610:3ff4c50647ea jobs