# HG changeset patch # User Sebastien Jodogne # Date 1543594797 -3600 # Node ID dc18d5804746946ebc8563116448547c8817aebc # Parent e6204cd21443f0df14b4ebdac5e09675cd282819 support of JobsHistorySize set to zero diff -r e6204cd21443 -r dc18d5804746 Core/JobsEngine/JobsEngine.cpp --- a/Core/JobsEngine/JobsEngine.cpp Thu Nov 29 20:36:55 2018 +0100 +++ b/Core/JobsEngine/JobsEngine.cpp Fri Nov 30 17:19:57 2018 +0100 @@ -156,9 +156,9 @@ } - JobsEngine::JobsEngine() : + JobsEngine::JobsEngine(size_t maxCompletedJobs) : state_(State_Setup), - registry_(new JobsRegistry), + registry_(new JobsRegistry(maxCompletedJobs)), threadSleep_(200), workers_(1) { @@ -198,7 +198,9 @@ throw OrthancException(ErrorCode_BadSequenceOfCalls); } - registry_.reset(new JobsRegistry(unserializer, serialized)); + assert(registry_.get() != NULL); + const size_t maxCompletedJobs = registry_->GetMaxCompletedJobs(); + registry_.reset(new JobsRegistry(unserializer, serialized, maxCompletedJobs)); } diff -r e6204cd21443 -r dc18d5804746 Core/JobsEngine/JobsEngine.h --- a/Core/JobsEngine/JobsEngine.h Thu Nov 29 20:36:55 2018 +0100 +++ b/Core/JobsEngine/JobsEngine.h Fri Nov 30 17:19:57 2018 +0100 @@ -68,7 +68,7 @@ size_t workerIndex); public: - JobsEngine(); + JobsEngine(size_t maxCompletedJobs); ~JobsEngine(); diff -r e6204cd21443 -r dc18d5804746 Core/JobsEngine/JobsRegistry.cpp --- a/Core/JobsEngine/JobsRegistry.cpp Thu Nov 29 20:36:55 2018 +0100 +++ b/Core/JobsEngine/JobsRegistry.cpp Fri Nov 30 17:19:57 2018 +0100 @@ -47,7 +47,6 @@ static const char* JOB = "Job"; static const char* JOBS = "Jobs"; static const char* JOBS_REGISTRY = "JobsRegistry"; - static const char* MAX_COMPLETED_JOBS = "MaxCompletedJobs"; static const char* CREATION_TIME = "CreationTime"; static const char* LAST_CHANGE_TIME = "LastChangeTime"; static const char* RUNTIME = "Runtime"; @@ -435,20 +434,19 @@ void JobsRegistry::ForgetOldCompletedJobs() { - if (maxCompletedJobs_ != 0) + while (completedJobs_.size() > maxCompletedJobs_) { - while (completedJobs_.size() > maxCompletedJobs_) - { - assert(completedJobs_.front() != NULL); + assert(completedJobs_.front() != NULL); + + std::string id = completedJobs_.front()->GetId(); + assert(jobsIndex_.find(id) != jobsIndex_.end()); - std::string id = completedJobs_.front()->GetId(); - assert(jobsIndex_.find(id) != jobsIndex_.end()); + jobsIndex_.erase(id); + delete(completedJobs_.front()); + completedJobs_.pop_front(); + } - jobsIndex_.erase(id); - delete(completedJobs_.front()); - completedJobs_.pop_front(); - } - } + CheckInvariants(); } @@ -458,26 +456,48 @@ job.SetState(success ? JobState_Success : JobState_Failure); completedJobs_.push_back(&job); - ForgetOldCompletedJobs(); - someJobComplete_.notify_all(); } void JobsRegistry::MarkRunningAsCompleted(JobHandler& job, - bool success) + CompletedReason reason) { - LOG(INFO) << "Job has completed with " << (success ? "success" : "failure") - << ": " << job.GetId(); + const char* tmp; + + switch (reason) + { + case CompletedReason_Success: + tmp = "success"; + break; + + case CompletedReason_Failure: + tmp = "success"; + break; + + case CompletedReason_Canceled: + tmp = "cancel"; + break; + + default: + throw OrthancException(ErrorCode_InternalError); + } + + LOG(INFO) << "Job has completed with " << tmp << ": " << job.GetId(); CheckInvariants(); assert(job.GetState() == JobState_Running); - SetCompletedJob(job, success); + SetCompletedJob(job, reason == CompletedReason_Success); + + if (reason == CompletedReason_Canceled) + { + job.SetLastErrorCode(ErrorCode_CanceledJob); + } if (observer_ != NULL) { - if (success) + if (reason == CompletedReason_Success) { observer_->SignalJobSuccess(job.GetId()); } @@ -487,7 +507,9 @@ } } - CheckInvariants(); + // WARNING: The following call might make "job" invalid if the job + // history size is empty + ForgetOldCompletedJobs(); } @@ -558,8 +580,14 @@ maxCompletedJobs_ = n; ForgetOldCompletedJobs(); + } + + size_t JobsRegistry::GetMaxCompletedJobs() + { + boost::mutex::scoped_lock lock(mutex_); CheckInvariants(); + return maxCompletedJobs_; } @@ -604,17 +632,14 @@ void JobsRegistry::SubmitInternal(std::string& id, - JobHandler* handlerRaw, - bool keepLastChangeTime) + JobHandler* handler) { - if (handlerRaw == NULL) + if (handler == NULL) { throw OrthancException(ErrorCode_NullPointer); } - std::auto_ptr handler(handlerRaw); - - boost::posix_time::ptime lastChangeTime = handler->GetLastStateChangeTime(); + std::auto_ptr protection(handler); { boost::mutex::scoped_lock lock(mutex_); @@ -623,13 +648,15 @@ id = handler->GetId(); int priority = handler->GetPriority(); + jobsIndex_.insert(std::make_pair(id, protection.release())); + switch (handler->GetState()) { case JobState_Pending: case JobState_Retry: case JobState_Running: handler->SetState(JobState_Pending); - pendingJobs_.push(handler.get()); + pendingJobs_.push(handler); pendingJobAvailable_.notify_one(); break; @@ -650,13 +677,6 @@ throw OrthancException(ErrorCode_InternalError); } - if (keepLastChangeTime) - { - handler->SetLastStateChangeTime(lastChangeTime); - } - - jobsIndex_.insert(std::make_pair(id, handler.release())); - LOG(INFO) << "New job submitted with priority " << priority << ": " << id; if (observer_ != NULL) @@ -664,7 +684,9 @@ observer_->SignalJobSubmitted(id); } - CheckInvariants(); + // WARNING: The following call might make "handler" invalid if + // the job history size is empty + ForgetOldCompletedJobs(); } } @@ -673,7 +695,7 @@ IJob* job, // Takes ownership int priority) { - SubmitInternal(id, new JobHandler(job, priority), false); + SubmitInternal(id, new JobHandler(job, priority)); } @@ -681,7 +703,7 @@ int priority) { std::string id; - SubmitInternal(id, new JobHandler(job, priority), false); + SubmitInternal(id, new JobHandler(job, priority)); } @@ -904,7 +926,10 @@ throw OrthancException(ErrorCode_InternalError); } - CheckInvariants(); + // WARNING: The following call might make "handler" invalid if + // the job history size is empty + ForgetOldCompletedJobs(); + return true; } } @@ -1091,17 +1116,12 @@ switch (targetState_) { case JobState_Failure: - registry_.MarkRunningAsCompleted(*handler_, false); - - if (canceled_) - { - handler_->SetLastErrorCode(ErrorCode_CanceledJob); - } - + registry_.MarkRunningAsCompleted + (*handler_, canceled_ ? CompletedReason_Canceled : CompletedReason_Failure); break; case JobState_Success: - registry_.MarkRunningAsCompleted(*handler_, true); + registry_.MarkRunningAsCompleted(*handler_, CompletedReason_Success); break; case JobState_Paused: @@ -1293,7 +1313,6 @@ target = Json::objectValue; target[TYPE] = JOBS_REGISTRY; - target[MAX_COMPLETED_JOBS] = static_cast(maxCompletedJobs_); target[JOBS] = Json::objectValue; for (JobsIndex::const_iterator it = jobsIndex_.begin(); @@ -1309,7 +1328,9 @@ JobsRegistry::JobsRegistry(IJobUnserializer& unserializer, - const Json::Value& s) : + const Json::Value& s, + size_t maxCompletedJobs) : + maxCompletedJobs_(maxCompletedJobs), observer_(NULL) { if (SerializationToolbox::ReadString(s, TYPE) != JOBS_REGISTRY || @@ -1319,17 +1340,28 @@ throw OrthancException(ErrorCode_BadFileFormat); } - maxCompletedJobs_ = SerializationToolbox::ReadUnsignedInteger(s, MAX_COMPLETED_JOBS); - Json::Value::Members members = s[JOBS].getMemberNames(); for (Json::Value::Members::const_iterator it = members.begin(); it != members.end(); ++it) { std::auto_ptr job(new JobHandler(unserializer, s[JOBS][*it], *it)); - + + const boost::posix_time::ptime lastChangeTime = job->GetLastStateChangeTime(); + std::string id; - SubmitInternal(id, job.release(), true); + SubmitInternal(id, job.release()); + + // Check whether the job has not been removed (which could be + // the case if the "maxCompletedJobs_" value gets smaller) + JobsIndex::iterator found = jobsIndex_.find(id); + if (found != jobsIndex_.end()) + { + // The job still lies in the history: Update the time of its + // last change to the time that was serialized + assert(found->second != NULL); + found->second->SetLastStateChangeTime(lastChangeTime); + } } } } diff -r e6204cd21443 -r dc18d5804746 Core/JobsEngine/JobsRegistry.h --- a/Core/JobsEngine/JobsRegistry.h Thu Nov 29 20:36:55 2018 +0100 +++ b/Core/JobsEngine/JobsRegistry.h Fri Nov 30 17:19:57 2018 +0100 @@ -71,6 +71,13 @@ }; private: + enum CompletedReason + { + CompletedReason_Success, + CompletedReason_Failure, + CompletedReason_Canceled + }; + class JobHandler; struct PriorityComparator @@ -115,7 +122,7 @@ bool success); void MarkRunningAsCompleted(JobHandler& job, - bool success); + CompletedReason reason); void MarkRunningAsRetry(JobHandler& job, unsigned int timeout); @@ -130,23 +137,25 @@ void RemoveRetryJob(JobHandler* handler); void SubmitInternal(std::string& id, - JobHandler* handler, - bool keepLastChangeTime); + JobHandler* handler); public: - JobsRegistry() : - maxCompletedJobs_(10), + JobsRegistry(size_t maxCompletedJobs) : + maxCompletedJobs_(maxCompletedJobs), observer_(NULL) { } JobsRegistry(IJobUnserializer& unserializer, - const Json::Value& s); + const Json::Value& s, + size_t maxCompletedJobs); ~JobsRegistry(); void SetMaxCompletedJobs(size_t i); + size_t GetMaxCompletedJobs(); + void ListJobs(std::set& target); bool GetJobInfo(JobInfo& target, diff -r e6204cd21443 -r dc18d5804746 Core/JobsEngine/Operations/SequenceOfOperationsJob.h --- a/Core/JobsEngine/Operations/SequenceOfOperationsJob.h Thu Nov 29 20:36:55 2018 +0100 +++ b/Core/JobsEngine/Operations/SequenceOfOperationsJob.h Fri Nov 30 17:19:57 2018 +0100 @@ -71,6 +71,8 @@ std::list observers_; TimeoutDicomConnectionManager connectionManager_; + void NotifyDone() const; + public: SequenceOfOperationsJob(); @@ -96,8 +98,8 @@ public: Lock(SequenceOfOperationsJob& that) : - that_(that), - lock_(that.mutex_) + that_(that), + lock_(that.mutex_) { } diff -r e6204cd21443 -r dc18d5804746 NEWS --- a/NEWS Thu Nov 29 20:36:55 2018 +0100 +++ b/NEWS Fri Nov 30 17:19:57 2018 +0100 @@ -39,6 +39,7 @@ * Orthanc starts even if jobs from a previous execution cannot be unserialized * New CMake option "ENABLE_DCMTK_LOG" to disable logging internal to DCMTK * Fix issue 114 (Boost 1.68 doesn't support SHA-1 anymore) +* Support of "JobsHistorySize" set to zero * Upgraded dependencies for static and Windows builds: - boost 1.68.0 - lua 5.3.5 diff -r e6204cd21443 -r dc18d5804746 OrthancServer/ServerContext.cpp --- a/OrthancServer/ServerContext.cpp Thu Nov 29 20:36:55 2018 +0100 +++ b/OrthancServer/ServerContext.cpp Fri Nov 30 17:19:57 2018 +0100 @@ -214,7 +214,8 @@ ServerContext::ServerContext(IDatabaseWrapper& database, IStorageArea& area, - bool unitTesting) : + bool unitTesting, + size_t maxCompletedJobs) : index_(*this, database, (unitTesting ? 20 : 500)), area_(area), compressionEnabled_(false), @@ -224,6 +225,7 @@ mainLua_(*this), filterLua_(*this), luaListener_(*this), + jobsEngine_(maxCompletedJobs), #if ORTHANC_ENABLE_PLUGINS == 1 plugins_(NULL), #endif diff -r e6204cd21443 -r dc18d5804746 OrthancServer/ServerContext.h --- a/OrthancServer/ServerContext.h Thu Nov 29 20:36:55 2018 +0100 +++ b/OrthancServer/ServerContext.h Fri Nov 30 17:19:57 2018 +0100 @@ -160,12 +160,17 @@ DicomCacheProvider provider_; boost::mutex dicomCacheMutex_; MemoryCache dicomCache_; - JobsEngine jobsEngine_; LuaScripting mainLua_; LuaScripting filterLua_; LuaServerListener luaListener_; + // The "JobsEngine" must be *after* "LuaScripting", as + // "LuaScripting" embeds "LuaJobManager" that registers as an + // observer to "SequenceOfOperationsJob", whose lifetime + // corresponds to that of "JobsEngine" + JobsEngine jobsEngine_; + #if ORTHANC_ENABLE_PLUGINS == 1 OrthancPlugins* plugins_; #endif @@ -206,7 +211,8 @@ ServerContext(IDatabaseWrapper& database, IStorageArea& area, - bool unitTesting); + bool unitTesting, + size_t maxCompletedJobs); ~ServerContext(); diff -r e6204cd21443 -r dc18d5804746 OrthancServer/ServerJobs/LuaJobManager.cpp --- a/OrthancServer/ServerJobs/LuaJobManager.cpp Thu Nov 29 20:36:55 2018 +0100 +++ b/OrthancServer/ServerJobs/LuaJobManager.cpp Fri Nov 30 17:19:57 2018 +0100 @@ -143,6 +143,7 @@ // Need to create a new job, as the previous one is either // finished, or is getting too long that_.currentJob_ = new SequenceOfOperationsJob; + that_.currentJob_->Register(that_); that_.currentJob_->SetDescription("Lua"); { diff -r e6204cd21443 -r dc18d5804746 OrthancServer/main.cpp --- a/OrthancServer/main.cpp Thu Nov 29 20:36:55 2018 +0100 +++ b/OrthancServer/main.cpp Fri Nov 30 17:19:57 2018 +0100 @@ -1085,8 +1085,8 @@ OrthancPlugins *plugins, bool loadJobsFromDatabase) { - ServerContext context(database, storageArea, false /* not running unit tests */); - + size_t maxCompletedJobs; + { OrthancConfiguration::ReaderLock lock; @@ -1101,6 +1101,15 @@ HttpClient::SetDefaultProxy(lock.GetConfiguration().GetStringParameter("HttpProxy", "")); DicomUserConnection::SetDefaultTimeout(lock.GetConfiguration().GetUnsignedIntegerParameter("DicomScuTimeout", 10)); + + maxCompletedJobs = lock.GetConfiguration().GetUnsignedIntegerParameter("JobsHistorySize", 10); + } + + ServerContext context(database, storageArea, false /* not running unit tests */, maxCompletedJobs); + + { + OrthancConfiguration::ReaderLock lock; + context.SetCompressionEnabled(lock.GetConfiguration().GetBooleanParameter("StorageCompression", false)); context.SetStoreMD5ForAttachments(lock.GetConfiguration().GetBooleanParameter("StoreMD5ForAttachments", true)); @@ -1125,9 +1134,6 @@ { context.GetIndex().SetMaximumStorageSize(0); } - - context.GetJobsEngine().GetRegistry().SetMaxCompletedJobs - (lock.GetConfiguration().GetUnsignedIntegerParameter("JobsHistorySize", 10)); } { diff -r e6204cd21443 -r dc18d5804746 Resources/Configuration.json --- a/Resources/Configuration.json Thu Nov 29 20:36:55 2018 +0100 +++ b/Resources/Configuration.json Fri Nov 30 17:19:57 2018 +0100 @@ -430,7 +430,9 @@ // Maximum number of completed jobs that are kept in memory. A // processing job is considered as complete once it is tagged as - // "Success" or "Failure". + // "Success" or "Failure". Since Orthanc 1.4.3, a value of "0" + // indicates to keep no job in memory (i.e. jobs are removed from + // the history as soon as they are completed). "JobsHistorySize" : 10, // Specifies how Orthanc reacts when it receives a DICOM instance diff -r e6204cd21443 -r dc18d5804746 UnitTestsSources/MultiThreadingTests.cpp --- a/UnitTestsSources/MultiThreadingTests.cpp Thu Nov 29 20:36:55 2018 +0100 +++ b/UnitTestsSources/MultiThreadingTests.cpp Fri Nov 30 17:19:57 2018 +0100 @@ -329,7 +329,7 @@ TEST(JobsRegistry, Priority) { - JobsRegistry registry; + JobsRegistry registry(10); std::string i1, i2, i3, i4; registry.Submit(i1, new DummyJob(), 10); @@ -408,7 +408,7 @@ TEST(JobsRegistry, Simultaneous) { - JobsRegistry registry; + JobsRegistry registry(10); std::string i1, i2; registry.Submit(i1, new DummyJob(), 20); @@ -438,7 +438,7 @@ TEST(JobsRegistry, Resubmit) { - JobsRegistry registry; + JobsRegistry registry(10); std::string id; registry.Submit(id, new DummyJob(), 10); @@ -482,7 +482,7 @@ TEST(JobsRegistry, Retry) { - JobsRegistry registry; + JobsRegistry registry(10); std::string id; registry.Submit(id, new DummyJob(), 10); @@ -519,7 +519,7 @@ TEST(JobsRegistry, PausePending) { - JobsRegistry registry; + JobsRegistry registry(10); std::string id; registry.Submit(id, new DummyJob(), 10); @@ -542,7 +542,7 @@ TEST(JobsRegistry, PauseRunning) { - JobsRegistry registry; + JobsRegistry registry(10); std::string id; registry.Submit(id, new DummyJob(), 10); @@ -580,7 +580,7 @@ TEST(JobsRegistry, PauseRetry) { - JobsRegistry registry; + JobsRegistry registry(10); std::string id; registry.Submit(id, new DummyJob(), 10); @@ -617,7 +617,7 @@ TEST(JobsRegistry, Cancel) { - JobsRegistry registry; + JobsRegistry registry(10); std::string id; registry.Submit(id, new DummyJob(), 10); @@ -711,7 +711,7 @@ TEST(JobsEngine, SubmitAndWait) { - JobsEngine engine; + JobsEngine engine(10); engine.SetThreadSleep(10); engine.SetWorkersCount(3); engine.Start(); @@ -731,7 +731,7 @@ TEST(JobsEngine, DISABLED_SequenceOfOperationsJob) { - JobsEngine engine; + JobsEngine engine(10); engine.SetThreadSleep(10); engine.SetWorkersCount(3); engine.Start(); @@ -771,7 +771,7 @@ TEST(JobsEngine, DISABLED_Lua) { - JobsEngine engine; + JobsEngine engine(10); engine.SetThreadSleep(10); engine.SetWorkersCount(2); engine.Start(); @@ -1282,7 +1282,7 @@ OrthancJobsSerialization() { db_.Open(); - context_.reset(new ServerContext(db_, storage_, true /* running unit tests */)); + context_.reset(new ServerContext(db_, storage_, true /* running unit tests */, 10)); context_->SetupJobsEngine(true, false); } @@ -1704,7 +1704,7 @@ std::string i1, i2; { - JobsRegistry registry; + JobsRegistry registry(10); registry.Submit(i1, new DummyJob(), 10); registry.Submit(i2, new SequenceOfOperationsJob(), 30); registry.Serialize(s); @@ -1712,7 +1712,7 @@ { DummyUnserializer unserializer; - JobsRegistry registry(unserializer, s); + JobsRegistry registry(unserializer, s, 10); Json::Value t; registry.Serialize(t); diff -r e6204cd21443 -r dc18d5804746 UnitTestsSources/ServerIndexTests.cpp --- a/UnitTestsSources/ServerIndexTests.cpp Thu Nov 29 20:36:55 2018 +0100 +++ b/UnitTestsSources/ServerIndexTests.cpp Fri Nov 30 17:19:57 2018 +0100 @@ -677,7 +677,7 @@ FilesystemStorage storage(path); DatabaseWrapper db; // The SQLite DB is in memory db.Open(); - ServerContext context(db, storage, true /* running unit tests */); + ServerContext context(db, storage, true /* running unit tests */, 10); context.SetupJobsEngine(true, false); ServerIndex& index = context.GetIndex(); @@ -777,7 +777,7 @@ FilesystemStorage storage(path); DatabaseWrapper db; // The SQLite DB is in memory db.Open(); - ServerContext context(db, storage, true /* running unit tests */); + ServerContext context(db, storage, true /* running unit tests */, 10); context.SetupJobsEngine(true, false); ServerIndex& index = context.GetIndex(); @@ -865,7 +865,7 @@ MemoryStorageArea storage; DatabaseWrapper db; // The SQLite DB is in memory db.Open(); - ServerContext context(db, storage, true /* running unit tests */); + ServerContext context(db, storage, true /* running unit tests */, 10); context.SetupJobsEngine(true, false); context.SetCompressionEnabled(true);