# HG changeset patch
# User Sebastien Jodogne <s.jodogne@gmail.com>
# Date 1528474128 -7200
# Node ID d26dd081df97f9d6136ad58fb5f0bd0542021c0c
# Parent  5fa2f2ce74f0d8b31b6396cbb1a5439c92d3e563
saving jobs engine on exit

diff -r 5fa2f2ce74f0 -r d26dd081df97 Core/JobsEngine/JobsEngine.cpp
--- a/Core/JobsEngine/JobsEngine.cpp	Fri Jun 08 15:48:35 2018 +0200
+++ b/Core/JobsEngine/JobsEngine.cpp	Fri Jun 08 18:08:48 2018 +0200
@@ -37,6 +37,8 @@
 #include "../Logging.h"
 #include "../OrthancException.h"
 
+#include <json/reader.h>
+
 namespace Orthanc
 {
   bool JobsEngine::IsRunning()
@@ -156,6 +158,7 @@
 
   JobsEngine::JobsEngine() :
     state_(State_Setup),
+    registry_(new JobsRegistry),
     threadSleep_(200),
     workers_(1)
   {
@@ -172,7 +175,49 @@
     }
   }
 
-    
+ 
+  JobsRegistry& JobsEngine::GetRegistry()
+  {
+    if (registry_.get() == NULL)
+    {
+      throw OrthancException(ErrorCode_InternalError);
+    }
+
+    return *registry_;
+  }
+  
+   
+  void JobsEngine::LoadRegistryFromJson(IJobUnserializer& unserializer,
+                                        const Json::Value& serialized)
+  {
+    boost::mutex::scoped_lock lock(stateMutex_);
+      
+    if (state_ != State_Setup)
+    {
+      // Can only be invoked before calling "Start()"
+      throw OrthancException(ErrorCode_BadSequenceOfCalls);
+    }
+
+    registry_.reset(new JobsRegistry(unserializer, serialized));
+  }
+
+
+  void JobsEngine::LoadRegistryFromString(IJobUnserializer& unserializer,
+                                          const std::string& serialized)
+  {
+    Json::Value value;
+    Json::Reader reader;
+    if (reader.parse(serialized, value))
+    {
+      LoadRegistryFromJson(unserializer, value);
+    }
+    else
+    {
+      throw OrthancException(ErrorCode_BadFileFormat);
+    }
+  }
+
+
   void JobsEngine::SetWorkersCount(size_t count)
   {
     boost::mutex::scoped_lock lock(stateMutex_);
diff -r 5fa2f2ce74f0 -r d26dd081df97 Core/JobsEngine/JobsEngine.h
--- a/Core/JobsEngine/JobsEngine.h	Fri Jun 08 15:48:35 2018 +0200
+++ b/Core/JobsEngine/JobsEngine.h	Fri Jun 08 18:08:48 2018 +0200
@@ -52,7 +52,7 @@
 
     boost::mutex                 stateMutex_;
     State                        state_;
-    JobsRegistry                 registry_;
+    std::auto_ptr<JobsRegistry>  registry_;
     boost::thread                retryHandler_;
     unsigned int                 threadSleep_;
     std::vector<boost::thread*>  workers_;
@@ -72,14 +72,17 @@
 
     ~JobsEngine();
 
+    JobsRegistry& GetRegistry();
+
+    void LoadRegistryFromJson(IJobUnserializer& unserializer,
+                              const Json::Value& serialized);
+
+    void LoadRegistryFromString(IJobUnserializer& unserializer,
+                                const std::string& serialized);
+
     void SetWorkersCount(size_t count);
 
     void SetThreadSleep(unsigned int sleep);
-    
-    JobsRegistry& GetRegistry()
-    {
-      return registry_;
-    }
 
     void Start();
 
diff -r 5fa2f2ce74f0 -r d26dd081df97 Core/JobsEngine/JobsRegistry.cpp
--- a/Core/JobsEngine/JobsRegistry.cpp	Fri Jun 08 15:48:35 2018 +0200
+++ b/Core/JobsEngine/JobsRegistry.cpp	Fri Jun 08 18:08:48 2018 +0200
@@ -109,6 +109,7 @@
       }
 
       job->GetJobType(jobType_);
+      job->Start();
 
       lastStatus_ = JobStatus(ErrorCode_Success, *job_);
     }
@@ -592,6 +593,11 @@
   void JobsRegistry::SubmitInternal(std::string& id,
                                     JobHandler* handlerRaw)
   {
+    if (handlerRaw == NULL)
+    {
+      throw OrthancException(ErrorCode_NullPointer);
+    }
+    
     std::auto_ptr<JobHandler>  handler(handlerRaw);
 
     boost::mutex::scoped_lock lock(mutex_);
@@ -600,8 +606,32 @@
     id = handler->GetId();
     int priority = handler->GetPriority();
 
-    pendingJobs_.push(handler.get());
-    pendingJobAvailable_.notify_one();
+    switch (handler->GetState())
+    {
+      case JobState_Pending:
+      case JobState_Retry:
+      case JobState_Running:
+        handler->SetState(JobState_Pending);
+        pendingJobs_.push(handler.get());
+        pendingJobAvailable_.notify_one();
+        break;
+ 
+      case JobState_Success:
+        SetCompletedJob(*handler, true);
+        break;
+        
+      case JobState_Failure:
+        SetCompletedJob(*handler, false);
+        break;
+
+      case JobState_Paused:
+        break;
+        
+      default:
+        LOG(ERROR) << "A job should not be loaded from state: "
+                   << EnumerationToString(handler->GetState());
+        throw OrthancException(ErrorCode_InternalError);
+    }
 
     jobsIndex_.insert(std::make_pair(id, handler.release()));
 
diff -r 5fa2f2ce74f0 -r d26dd081df97 OrthancServer/ServerContext.cpp
--- a/OrthancServer/ServerContext.cpp	Fri Jun 08 15:48:35 2018 +0200
+++ b/OrthancServer/ServerContext.cpp	Fri Jun 08 18:08:48 2018 +0200
@@ -34,20 +34,21 @@
 #include "PrecompiledHeadersServer.h"
 #include "ServerContext.h"
 
+#include "../Core/DicomParsing/FromDcmtkBridge.h"
 #include "../Core/FileStorage/StorageAccessor.h"
 #include "../Core/HttpServer/FilesystemHttpSender.h"
 #include "../Core/HttpServer/HttpStreamTranscoder.h"
 #include "../Core/Logging.h"
-#include "../Core/DicomParsing/FromDcmtkBridge.h"
+#include "../Plugins/Engine/OrthancPlugins.h"
+#include "OrthancInitialization.h"
+#include "OrthancRestApi/OrthancRestApi.h"
+#include "Search/LookupResource.h"
+#include "ServerJobs/OrthancJobUnserializer.h"
 #include "ServerToolbox.h"
-#include "OrthancInitialization.h"
 
 #include <EmbeddedResources.h>
 #include <dcmtk/dcmdata/dcfilefo.h>
 
-#include "OrthancRestApi/OrthancRestApi.h"
-#include "../Plugins/Engine/OrthancPlugins.h"
-#include "Search/LookupResource.h"
 
 
 #define ENABLE_DICOM_CACHE  1
@@ -65,11 +66,12 @@
 
 namespace Orthanc
 {
-  void ServerContext::ChangeThread(ServerContext* that)
+  void ServerContext::ChangeThread(ServerContext* that,
+                                   unsigned int sleepDelay)
   {
     while (!that->done_)
     {
-      std::auto_ptr<IDynamicObject> obj(that->pendingChanges_.Dequeue(100));
+      std::auto_ptr<IDynamicObject> obj(that->pendingChanges_.Dequeue(sleepDelay));
         
       if (obj.get() != NULL)
       {
@@ -106,6 +108,58 @@
   }
 
 
+  void ServerContext::SetupJobsEngine(bool unitTesting)
+  {
+    jobsEngine_.SetWorkersCount(Configuration::GetGlobalUnsignedIntegerParameter("ConcurrentJobs", 2));
+    jobsEngine_.SetThreadSleep(unitTesting ? 20 : 200);
+
+    std::string serialized;
+    if (index_.LookupGlobalProperty(serialized, GlobalProperty_JobsRegistry))
+    {
+      LOG(WARNING) << "Reloading the jobs from the last execution of Orthanc";
+      OrthancJobUnserializer unserializer(*this);
+
+      try
+      {
+        jobsEngine_.LoadRegistryFromString(unserializer, serialized);
+      }
+      catch (OrthancException& e)
+      {
+        LOG(ERROR) << "Cannot unserialize the jobs engine: " << e.What();
+        throw;
+      }
+    }
+    else
+    {
+      LOG(INFO) << "The last execution of Orthanc has archived no job";
+      //jobsEngine_.GetRegistry().SetMaxCompleted   // TODO
+    }
+
+    jobsEngine_.Start();
+  }
+
+
+  void ServerContext::SaveJobsEngine()
+  {
+    LOG(INFO) << "Serializing the content of the jobs engine";
+    
+    try
+    {
+      Json::Value value;
+      jobsEngine_.GetRegistry().Serialize(value);
+
+      Json::FastWriter writer;
+      std::string serialized = writer.write(value);
+
+      index_.SetGlobalProperty(GlobalProperty_JobsRegistry, serialized);
+    }
+    catch (OrthancException& e)
+    {
+      LOG(ERROR) << "Cannot serialize the jobs engine: " << e.What();
+    }
+  }
+
+
   ServerContext::ServerContext(IDatabaseWrapper& database,
                                IStorageArea& area,
                                bool unitTesting) :
@@ -125,12 +179,9 @@
   {
     listeners_.push_back(ServerListener(lua_, "Lua"));
 
-    jobsEngine_.SetWorkersCount(Configuration::GetGlobalUnsignedIntegerParameter("ConcurrentJobs", 2));
-    //jobsEngine_.SetMaxCompleted   // TODO
-    jobsEngine_.SetThreadSleep(unitTesting ? 20 : 200);
-    jobsEngine_.Start();
+    SetupJobsEngine(unitTesting);
 
-    changeThread_ = boost::thread(ChangeThread, this);
+    changeThread_ = boost::thread(ChangeThread, this, (unitTesting ? 20 : 100));
   }
 
 
@@ -161,6 +212,8 @@
         changeThread_.join();
       }
 
+      SaveJobsEngine();
+
       // Do not change the order below!
       jobsEngine_.Stop();
       index_.Stop();
diff -r 5fa2f2ce74f0 -r d26dd081df97 OrthancServer/ServerContext.h
--- a/OrthancServer/ServerContext.h	Fri Jun 08 15:48:35 2018 +0200
+++ b/OrthancServer/ServerContext.h	Fri Jun 08 18:08:48 2018 +0200
@@ -104,11 +104,16 @@
     typedef std::list<ServerListener>  ServerListeners;
 
 
-    static void ChangeThread(ServerContext* that);
+    static void ChangeThread(ServerContext* that,
+                             unsigned int sleepDelay);
 
     void ReadDicomAsJsonInternal(std::string& result,
                                  const std::string& instancePublicId);
 
+    void SetupJobsEngine(bool unitTesting);
+
+    void SaveJobsEngine();
+
     ServerIndex index_;
     IStorageArea& area_;
 
diff -r 5fa2f2ce74f0 -r d26dd081df97 OrthancServer/ServerEnumerations.h
--- a/OrthancServer/ServerEnumerations.h	Fri Jun 08 15:48:35 2018 +0200
+++ b/OrthancServer/ServerEnumerations.h	Fri Jun 08 18:08:48 2018 +0200
@@ -76,7 +76,8 @@
     GlobalProperty_DatabaseSchemaVersion = 1,   // Unused in the Orthanc core as of Orthanc 0.9.5
     GlobalProperty_FlushSleep = 2,
     GlobalProperty_AnonymizationSequence = 3,
-    GlobalProperty_DatabasePatchLevel = 4       // Reserved for internal use of the database plugins
+    GlobalProperty_DatabasePatchLevel = 4,      // Reserved for internal use of the database plugins
+    GlobalProperty_JobsRegistry = 5
   };
 
   enum MetadataType
diff -r 5fa2f2ce74f0 -r d26dd081df97 OrthancServer/ServerIndex.cpp
--- a/OrthancServer/ServerIndex.cpp	Fri Jun 08 15:48:35 2018 +0200
+++ b/OrthancServer/ServerIndex.cpp	Fri Jun 08 18:08:48 2018 +0200
@@ -2096,13 +2096,20 @@
   }
 
 
+  bool ServerIndex::LookupGlobalProperty(std::string& value,
+                                         GlobalProperty property)
+  {
+    boost::mutex::scoped_lock lock(mutex_);
+    return db_.LookupGlobalProperty(value, property);
+  }
+  
+
   std::string ServerIndex::GetGlobalProperty(GlobalProperty property,
                                              const std::string& defaultValue)
   {
-    boost::mutex::scoped_lock lock(mutex_);
-
     std::string value;
-    if (db_.LookupGlobalProperty(value, property))
+
+    if (LookupGlobalProperty(value, property))
     {
       return value;
     }
diff -r 5fa2f2ce74f0 -r d26dd081df97 OrthancServer/ServerIndex.h
--- a/OrthancServer/ServerIndex.h	Fri Jun 08 15:48:35 2018 +0200
+++ b/OrthancServer/ServerIndex.h	Fri Jun 08 18:08:48 2018 +0200
@@ -258,6 +258,9 @@
     void SetGlobalProperty(GlobalProperty property,
                            const std::string& value);
 
+    bool LookupGlobalProperty(std::string& value,
+                              GlobalProperty property);
+
     std::string GetGlobalProperty(GlobalProperty property,
                                   const std::string& defaultValue);