Mercurial > hg > orthanc
changeset 6638:a92513c29da2
LoaderThreads now also used in CStore and PeerStore jobs
line wrap: on
line diff
--- a/NEWS Thu Mar 26 11:56:40 2026 +0100 +++ b/NEWS Fri Mar 27 16:32:46 2026 +0100 @@ -4,7 +4,22 @@ General ------- -* New experimental configuration "PatientLevelEnabled" (TODO: work in progree) +* New configuration "LoaderThreads" that is a general configuration for all actions + that require loading files, possibly in parallel: + - downloading zip archive/media, + - performing a C-Store to a remote modality + - performing a peer transfer to a remote Orthanc. + Note: the old "ZipLoaderThreads" configuration is still available for backward + compatibility reasons. +* When performing a C-Store or a peer transfer, the instances are now + grouped per series and ordered by their InstanceNumber instead of using + a random order in previous Orthanc versions. +TODO before release: - reconfigure the instancesLoader when unserializing a C-Store or peer job + - serialize the ParentResources in C-Store and peer job + - use ThreadedInstancesLoader in OrthancGetRequestHandler + + +* New experimental configuration "PatientLevelEnabled". REST API --------
--- a/OrthancServer/CMakeLists.txt Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/CMakeLists.txt Fri Mar 27 16:32:46 2026 +0100 @@ -160,7 +160,9 @@ ${CMAKE_SOURCE_DIR}/Sources/ServerJobs/OrthancPeerStoreJob.cpp ${CMAKE_SOURCE_DIR}/Sources/ServerJobs/ResourceModificationJob.cpp ${CMAKE_SOURCE_DIR}/Sources/ServerJobs/SplitStudyJob.cpp + ${CMAKE_SOURCE_DIR}/Sources/ServerJobs/StoreJob.cpp ${CMAKE_SOURCE_DIR}/Sources/ServerJobs/StorageCommitmentScpJob.cpp + ${CMAKE_SOURCE_DIR}/Sources/ServerJobs/ThreadedInstancesLoader.cpp ${CMAKE_SOURCE_DIR}/Sources/ServerJobs/ThreadedSetOfInstancesJob.cpp ${CMAKE_SOURCE_DIR}/Sources/ServerToolbox.cpp ${CMAKE_SOURCE_DIR}/Sources/SimpleInstanceOrdering.cpp
--- a/OrthancServer/Resources/Configuration.json Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Resources/Configuration.json Fri Mar 27 16:32:46 2026 +0100 @@ -1009,12 +1009,19 @@ // as soon as one DICOM file gets compressed (new in Orthanc 1.9.4) "SynchronousZipStream" : true, - // Default number of loader threads when generating Zip archive/media. - // A value of 0 means reading and writing are performed in sequence - // (default behaviour). A value > 1 is meaningful only if the storage + // Default number of loader threads when reading files from disk when + // executing some tasks: + // - generating archive/media, + // - executing a C-Store + // - transmitting resources through Orthanc peering. + // A value of 0 and 1 are equivalent: a single thread is used. + // A value > 1 is meaningful only if the storage // is a distributed network storage (e.g object storage plugin). - // (new experimental feature in Orthanc 1.10.0) - "ZipLoaderThreads": 0, + // (new in Orthanc 1.12.11) + // Note, from 1.10.0 to 1.12.10, a "ZipLoaderThreads" configuration + // was available only for the zip archive/media. It is still valid + // for backward compatibility. + "LoaderThreads": 1, // Extra Main Dicom tags that are stored in DB together with all default // Main Dicom tags that are already stored.
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/OrthancServer/Resources/ImplementationNotes/JobsEngineClasses.txt Fri Mar 27 16:32:46 2026 +0100 @@ -0,0 +1,27 @@ +class IJob : + virtual void Start() = 0; + virtual JobStepResult Step(const std::string& jobId) = 0; + virtual bool NeedsProgressUpdateBetweenSteps() const // only for jobs whose progress is updated by outside events (like C-Move and C-Get) + virtual float GetProgress() const = 0; + ... + +class SetOfCommandsJob : IJob + class ICommand + virtual bool Execute(const std::string& jobId) = 0; + virtual void Serialize(Json::Value& target) const = 0; + class ICommandUnserializer + virtual ICommand* Unserialize(const Json::Value& source) const = 0; + + virtual void Start() ORTHANC_OVERRIDE; + virtual float GetProgress() ORTHANC_OVERRIDE { returns position_/commands_.size()} + void SetPermissive(bool permissive); // if a command fails, the complete job does not fails + virtual JobStepResult Step(const std::string& jobId) = commands_[position_]->Execute() + +class SetOfInstancesJob : SetOfCommandsJob + virtual bool HandleInstance(const std::string& instance) = 0; + virtual bool HandleTrailingStep() = 0; + + +class StoreJob : SetOfInstancesJob (contains a ThreadedInstancesLoader) + +Class ModalityStoreJob : StoreJob \ No newline at end of file
--- a/OrthancServer/Sources/OrthancConfiguration.cpp Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/OrthancConfiguration.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -47,6 +47,8 @@ static const char* const WARNINGS = "Warnings"; static const char* const JOBS_ENGINE_THREADS_COUNT = "JobsEngineThreadsCount"; static const char* const DICOM_LOSSY_TRANSCODING_QUALITY = "DicomLossyTranscodingQuality"; +static const char* const CONFIG_LOADER_THREADS = "LoaderThreads"; +static const char* const CONFIG_ZIP_LOADER_THREADS = "ZipLoaderThreads"; // for backward compatibility only namespace Orthanc { @@ -1347,4 +1349,18 @@ throw OrthancException(ErrorCode_BadFileFormat); } } + + unsigned int OrthancConfiguration::GetLoaderThreads() const + { + // from 1.10.0 to 1.12.10, only CONFIG_ZIP_LOADER_THREADS was available -> read from it if CONFIG_LOADER_THREADS is not specified. + unsigned int loaderThreads = GetUnsignedIntegerParameter(CONFIG_LOADER_THREADS, GetUnsignedIntegerParameter(CONFIG_ZIP_LOADER_THREADS, 1)); + + if (loaderThreads <= 1) + { + return 1; // 0 is not a valid internal value anymore + } + + return loaderThreads; + } + }
--- a/OrthancServer/Sources/OrthancConfiguration.h Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/OrthancConfiguration.h Fri Mar 27 16:32:46 2026 +0100 @@ -248,6 +248,7 @@ void RemovePeer(const std::string& symbolicName); + unsigned int GetLoaderThreads() const; void Format(std::string& result) const;
--- a/OrthancServer/Sources/OrthancMoveRequestHandler.cpp Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/OrthancMoveRequestHandler.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -33,6 +33,7 @@ #include "OrthancConfiguration.h" #include "ServerContext.h" #include "ServerJobs/DicomModalityStoreJob.h" +#include "ServerJobs/ThreadedInstancesLoader.h" namespace Orthanc @@ -46,7 +47,9 @@ private: ServerContext& context_; const std::string& localAet_; - std::vector<std::string> instances_; + std::vector<std::string> instancesIds_; + // std::vector<FileInfo> filesInfo_; + std::unique_ptr<ThreadedInstancesLoader> instancesLoader_; size_t position_; RemoteModalityParameters remote_; std::string originatorAet_; @@ -57,6 +60,7 @@ SynchronousMove(ServerContext& context, const std::string& targetAet, const std::vector<std::string>& publicIds, + ResourceType resourceType, const std::string& originatorAet, uint16_t originatorId) : context_(context), @@ -65,43 +69,49 @@ originatorAet_(originatorAet), originatorId_(originatorId) { + unsigned int loaderThreads; + { OrthancConfiguration::ReaderLock lock; remote_ = lock.GetConfiguration().GetModalityUsingAet(targetAet); + loaderThreads = lock.GetConfiguration().GetLoaderThreads(); } + instancesLoader_.reset(new ThreadedInstancesLoader(context_, loaderThreads, false, DicomTransferSyntax_BigEndianExplicit /* dummy value*/, 0, "CSTO")); + + std::vector<FileInfo> filesInfo; for (size_t i = 0; i < publicIds.size(); i++) { - CLOG(INFO, DICOM) << "Sending resource " << publicIds[i] << " to modality \"" + const std::string resourceId = publicIds[i]; + + CLOG(INFO, DICOM) << "Sending resource " << resourceId << " to modality \"" << targetAet << "\" in synchronous mode"; - std::list<std::string> tmp; - context_.GetIndex().GetChildInstances(tmp, publicIds[i]); + context.GetOrderedChildInstances(instancesIds_, filesInfo, resourceId, resourceType); + } - instances_.reserve(tmp.size()); - for (std::list<std::string>::iterator it = tmp.begin(); it != tmp.end(); ++it) - { - instances_.push_back(*it); - } + for (size_t i = 0; i < instancesIds_.size(); ++i) + { + instancesLoader_->PrepareDicom(instancesIds_[i], filesInfo[i]); } } virtual unsigned int GetSubOperationCount() const ORTHANC_OVERRIDE { - return instances_.size(); + return instancesIds_.size(); } virtual Status DoNext() ORTHANC_OVERRIDE { - if (position_ >= instances_.size()) + if (position_ >= instancesIds_.size()) { return Status_Failure; } - const std::string& id = instances_[position_++]; + const std::string& id = instancesIds_[position_++]; std::string dicom; - context_.ReadDicom(dicom, id); + instancesLoader_->GetDicom(dicom, id); if (connection_.get() == NULL) { @@ -130,6 +140,7 @@ AsynchronousMove(ServerContext& context, const std::string& targetAet, const std::vector<std::string>& publicIds, + ResourceType resourceType, const std::string& originatorAet, uint16_t originatorId) : context_(context), @@ -151,23 +162,23 @@ job_->SetMoveOriginator(originatorAet, originatorId); } + std::vector<std::string> instancesIds; + std::vector<FileInfo> filesInfo; + for (size_t i = 0; i < publicIds.size(); i++) { - CLOG(INFO, DICOM) << "Sending resource " << publicIds[i] << " to modality \"" + const std::string resourceId = publicIds[i]; + + CLOG(INFO, DICOM) << "Sending resource " << resourceId << " to modality \"" << targetAet << "\" in asynchronous mode"; - std::list<std::string> tmp; - context_.GetIndex().GetChildInstances(tmp, publicIds[i]); - - countInstances_ = tmp.size(); + job_->AddParentResource(resourceId, resourceType); - job_->Reserve(job_->GetCommandsCount() + tmp.size()); + context.GetOrderedChildInstances(instancesIds, filesInfo, resourceId, resourceType); + } - for (std::list<std::string>::iterator it = tmp.begin(); it != tmp.end(); ++it) - { - job_->AddInstance(*it); - } - } + job_->AddInstances(instancesIds, filesInfo); + countInstances_ = instancesIds.size(); } virtual unsigned int GetSubOperationCount() const ORTHANC_OVERRIDE @@ -289,6 +300,7 @@ static IMoveRequestIterator* CreateIterator(ServerContext& context, const std::string& targetAet, const std::vector<std::string>& publicIds, + ResourceType resourceType, const std::string& originatorAet, uint16_t originatorId) { @@ -307,11 +319,11 @@ if (synchronous) { - return new SynchronousMove(context, targetAet, publicIds, originatorAet, originatorId); + return new SynchronousMove(context, targetAet, publicIds, resourceType, originatorAet, originatorId); } else { - return new AsynchronousMove(context, targetAet, publicIds, originatorAet, originatorId); + return new AsynchronousMove(context, targetAet, publicIds, resourceType, originatorAet, originatorId); } } @@ -356,12 +368,21 @@ std::vector<std::string> publicIds; - if (LookupIdentifiers(publicIds, ResourceType_Instance, input) || - LookupIdentifiers(publicIds, ResourceType_Series, input) || - LookupIdentifiers(publicIds, ResourceType_Study, input) || - LookupIdentifiers(publicIds, ResourceType_Patient, input)) + if (LookupIdentifiers(publicIds, ResourceType_Instance, input)) + { + return CreateIterator(context_, targetAet, publicIds, ResourceType_Instance, connection.GetRemoteAet(), originatorId); + } + else if (LookupIdentifiers(publicIds, ResourceType_Series, input)) { - return CreateIterator(context_, targetAet, publicIds, connection.GetRemoteAet(), originatorId); + return CreateIterator(context_, targetAet, publicIds, ResourceType_Series, connection.GetRemoteAet(), originatorId); + } + else if (LookupIdentifiers(publicIds, ResourceType_Study, input)) + { + return CreateIterator(context_, targetAet, publicIds, ResourceType_Study, connection.GetRemoteAet(), originatorId); + } + else if (LookupIdentifiers(publicIds, ResourceType_Patient, input)) + { + return CreateIterator(context_, targetAet, publicIds, ResourceType_Patient, connection.GetRemoteAet(), originatorId); } else { @@ -382,7 +403,7 @@ if (LookupIdentifiers(publicIds, level, input)) { - return CreateIterator(context_, targetAet, publicIds, connection.GetRemoteAet(), originatorId); + return CreateIterator(context_, targetAet, publicIds, level, connection.GetRemoteAet(), originatorId); } else {
--- a/OrthancServer/Sources/OrthancRestApi/OrthancRestArchive.cpp Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/OrthancRestApi/OrthancRestArchive.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -52,7 +52,6 @@ static const char* const GET_FILENAME = "filename"; static const char* const GET_RESOURCES = "resources"; - static const char* const CONFIG_LOADER_THREADS = "ZipLoaderThreads"; static const char* const CONFIG_ALLOW_UTF8 = "ZipUseUtf8"; @@ -188,7 +187,7 @@ { OrthancConfiguration::ReaderLock lock; - loaderThreads = lock.GetConfiguration().GetUnsignedIntegerParameter(CONFIG_LOADER_THREADS, 0); // New in Orthanc 1.10.0 + loaderThreads = lock.GetConfiguration().GetLoaderThreads(); } // New in Orthanc 1.12.11 @@ -783,7 +782,7 @@ { OrthancConfiguration::ReaderLock lock; - unsigned int loaderThreads = lock.GetConfiguration().GetUnsignedIntegerParameter(CONFIG_LOADER_THREADS, 0); // New in Orthanc 1.10.0 + unsigned int loaderThreads = lock.GetConfiguration().GetLoaderThreads(); job->SetLoaderThreads(loaderThreads); }
--- a/OrthancServer/Sources/OrthancRestApi/OrthancRestModalities.cpp Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/OrthancRestApi/OrthancRestModalities.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -1360,7 +1360,7 @@ ***************************************************************************/ static void GetInstancesToExport(Json::Value& otherArguments, - SetOfInstancesJob& job, + StoreJob& job, const std::string& remote, RestApiPostCall& call) { @@ -1439,23 +1439,27 @@ "Resources to be exported must be specified as a JSON array of strings"); } - std::string stripped = Toolbox::StripSpaces((*resources) [i].asString()); - if (!Toolbox::IsSHA1(stripped)) + std::string resourceId = Toolbox::StripSpaces((*resources) [i].asString()); + if (!Toolbox::IsSHA1(resourceId)) { throw OrthancException(ErrorCode_BadFileFormat, - "This string is not a valid Orthanc identifier: " + stripped); + "This string is not a valid Orthanc identifier: " + resourceId); } ResourceType level; - context.GetIndex().LookupResourceType(level, stripped); - - job.AddParentResource(stripped, level); // New in Orthanc 1.5.7 - - context.AddChildInstances(job, stripped, level); + context.GetIndex().LookupResourceType(level, resourceId); + + job.AddParentResource(resourceId, level); // New in Orthanc 1.5.7 + + std::vector<std::string> instancesIds; + std::vector<FileInfo> filesInfo; + + context.GetOrderedChildInstances(instancesIds, filesInfo, resourceId, level); + job.AddInstances(instancesIds, filesInfo); if (logExportedResources) { - context.GetIndex().LogExportedResource(stripped, remote); + context.GetIndex().LogExportedResource(resourceId, remote); } } }
--- a/OrthancServer/Sources/ServerContext.cpp Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/ServerContext.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -49,6 +49,7 @@ #include "ServerToolbox.h" #include "StorageCommitmentReports.h" #include "OutgoingDicomInstance.h" +#include "SimpleInstanceOrdering.h" #include <dcmtk/dcmdata/dcfilefo.h> #include <dcmtk/dcmnet/dimse.h> @@ -1809,20 +1810,53 @@ } - void ServerContext::AddChildInstances(SetOfInstancesJob& job, - const std::string& publicId, - ResourceType level) + void ServerContext::GetOrderedChildInstances(std::vector<std::string>& instancesIds, + std::vector<FileInfo>& filesInfo, + const std::string& publicId, + ResourceType level) { - std::list<std::string> instances; - GetIndex().GetChildInstances(instances, publicId, level); + // don't mix series and, inside a series, order the instances per instance number + switch (level) + { + case ResourceType_Patient: + case ResourceType_Study: + { + std::list<std::string> childrenIds; + GetIndex().GetChildren(childrenIds, level, publicId); + + for (std::list<std::string>::const_iterator it = childrenIds.begin(); it != childrenIds.end(); ++it) + { + GetOrderedChildInstances(instancesIds, filesInfo, *it, GetChildResourceType(level)); + } + }; break; + case ResourceType_Series: + { + SimpleInstanceOrdering orderedInstances(GetIndex(), publicId); - job.Reserve(job.GetInstancesCount() + instances.size()); + instancesIds.reserve(instancesIds.size() + orderedInstances.GetInstancesCount()); + filesInfo.reserve(filesInfo.size() + orderedInstances.GetInstancesCount()); + + for (size_t i = 0; i < orderedInstances.GetInstancesCount(); ++i) + { + instancesIds.push_back(orderedInstances.GetInstanceId(i)); + filesInfo.push_back(orderedInstances.GetInstanceFileInfo(i)); + } + }; break; + case ResourceType_Instance: + { + FileInfo fileInfo; + int64_t revisionNotUsed; - for (std::list<std::string>::const_iterator - it = instances.begin(); it != instances.end(); ++it) - { - job.AddInstance(*it); + if (GetIndex().LookupAttachment(fileInfo, revisionNotUsed, level, publicId, FileContentType_Dicom)) + { + instancesIds.push_back(publicId); + filesInfo.push_back(fileInfo); + } + }; break; + default: + throw OrthancException(ErrorCode_InternalError); } + }
--- a/OrthancServer/Sources/ServerContext.h Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/ServerContext.h Fri Mar 27 16:32:46 2026 +0100 @@ -513,9 +513,10 @@ bool HasPlugins() const; - void AddChildInstances(SetOfInstancesJob& job, - const std::string& publicId, - ResourceType level); + void GetOrderedChildInstances(std::vector<std::string>& instancesIds, + std::vector<FileInfo>& filesInfo, + const std::string& publicId, + ResourceType level); void SignalUpdatedModalities();
--- a/OrthancServer/Sources/ServerJobs/ArchiveJob.cpp Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/ServerJobs/ArchiveJob.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -36,6 +36,7 @@ #include "../OrthancConfiguration.h" #include "../ServerContext.h" #include "../SimpleInstanceOrdering.h" +#include "ThreadedInstancesLoader.h" #include <stdio.h> #include <boost/range/algorithm/count.hpp> @@ -57,8 +58,6 @@ static const char* const KEY_TRANSCODE = "Transcode"; static const char* const KEY_ALLOW_UTF8 = "Utf8"; -static boost::mutex loaderThreadsCounterMutex; -static uint32_t loaderThreadsCounter = 0; namespace Orthanc { @@ -86,263 +85,6 @@ } - class ArchiveJob::InstanceLoader : public boost::noncopyable - { - protected: - ServerContext& context_; - bool transcode_; - DicomTransferSyntax transferSyntax_; - unsigned int lossyQuality_; - public: - explicit InstanceLoader(ServerContext& context, bool transcode, DicomTransferSyntax transferSyntax, unsigned int lossyQuality) - : context_(context), - transcode_(transcode), - transferSyntax_(transferSyntax), - lossyQuality_(lossyQuality) - { - } - - virtual ~InstanceLoader() - { - } - - virtual void PrepareDicom(const std::string& instanceId, const FileInfo& fileInfo) - { - } - - bool TranscodeDicom(std::string& transcodedBuffer, const std::string& sourceBuffer, const std::string& instanceId) - { - if (transcode_) - { - std::set<DicomTransferSyntax> syntaxes; - syntaxes.insert(transferSyntax_); - - IDicomTranscoder::DicomImage source, transcoded; - source.SetExternalBuffer(sourceBuffer); - - if (context_.Transcode(transcoded, source, syntaxes, TranscodingSopInstanceUidMode_AllowNew, lossyQuality_)) - { - transcodedBuffer.assign(reinterpret_cast<const char*>(transcoded.GetBufferData()), transcoded.GetBufferSize()); - return true; - } - else - { - LOG(INFO) << "Cannot transcode instance " << instanceId - << " to transfer syntax: " << GetTransferSyntaxUid(transferSyntax_); - } - } - - return false; - } - - virtual void GetDicom(std::string& dicom, const std::string& instanceId, const FileInfo& fileInfo) = 0; - - virtual void Clear(bool isAbort) - { - } - }; - - class ArchiveJob::SynchronousInstanceLoader : public ArchiveJob::InstanceLoader - { - public: - explicit SynchronousInstanceLoader(ServerContext& context, bool transcode, DicomTransferSyntax transferSyntax, unsigned int lossyQuality) - : InstanceLoader(context, transcode, transferSyntax, lossyQuality) - { - } - - virtual void GetDicom(std::string& dicom, const std::string& instanceId, const FileInfo& fileInfo) ORTHANC_OVERRIDE - { - context_.ReadAttachment(dicom, fileInfo, true); - - if (transcode_) - { - std::string transcoded; - if (TranscodeDicom(transcoded, dicom, instanceId)) - { - dicom.swap(transcoded); - } - } - - } - }; - - class InstanceToPreload : public Orthanc::IDynamicObject - { - private: - std::string id_; - FileInfo fileInfo_; - - public: - explicit InstanceToPreload(const std::string& id, const FileInfo& fileInfo) : - id_(id), - fileInfo_(fileInfo) - { - } - - virtual ~InstanceToPreload() ORTHANC_OVERRIDE - { - } - - const std::string& GetId() const {return id_;}; - const FileInfo& GetFileInfo() const {return fileInfo_;}; - }; - - class ArchiveJob::ThreadedInstanceLoader : public ArchiveJob::InstanceLoader - { - boost::condition_variable condInstanceAvailable_; - std::map<std::string, boost::shared_ptr<std::string> > availableInstances_; - boost::mutex availableInstancesMutex_; - BlockingSharedMessageQueue instancesToPreload_; - std::vector<boost::thread*> threads_; - bool loadersShouldStop_; - - - public: - ThreadedInstanceLoader(ServerContext& context, size_t threadCount, bool transcode, DicomTransferSyntax transferSyntax, unsigned int lossyQuality) - : InstanceLoader(context, transcode, transferSyntax, lossyQuality), - instancesToPreload_ (3*threadCount), - loadersShouldStop_(false) - { - for (size_t i = 0; i < threadCount; i++) - { - threads_.push_back(new boost::thread(PreloaderWorkerThread, this)); - } - } - - virtual ~ThreadedInstanceLoader() ORTHANC_OVERRIDE - { - ThreadedInstanceLoader::Clear(false); - } - - virtual void Clear(bool isAbort) ORTHANC_OVERRIDE - { - if (threads_.size() > 0) - { - loadersShouldStop_ = true; // not need to protect this by a mutex. This is the only "writer" and all loaders are "readers" - - if (isAbort) - { - LOG(INFO) << "Cancelling the loader threads"; - instancesToPreload_.Clear(); - } - else - { - LOG(INFO) << "Waiting for loader threads to complete"; - } - - // unlock the loaders if they are waiting on this message queue (this happens when the job completes sucessfully) - for (size_t i = 0; i < threads_.size(); i++) - { - instancesToPreload_.Enqueue(NULL); - } - - for (size_t i = 0; i < threads_.size(); i++) - { - if (threads_[i]->joinable()) - { - threads_[i]->join(); - } - delete threads_[i]; - } - - threads_.clear(); - availableInstances_.clear(); - - LOG(INFO) << "Waiting for loader threads to complete - done"; - } - } - - static void PreloaderWorkerThread(ThreadedInstanceLoader* that) - { - { - boost::mutex::scoped_lock lock(loaderThreadsCounterMutex); - Logging::SetCurrentThreadName(std::string("ARCH-LOAD-") + boost::lexical_cast<std::string>(loaderThreadsCounter++)); - loaderThreadsCounter %= 1000000; - } - - LOG(INFO) << "Loader thread has started"; - - while (true) - { - std::unique_ptr<InstanceToPreload> instanceToPreload(dynamic_cast<InstanceToPreload*>(that->instancesToPreload_.Dequeue(0))); - if (instanceToPreload.get() == NULL || that->loadersShouldStop_) // that's the signal to exit the thread - { - LOG(INFO) << "Loader thread has completed"; - return; - } - - try - { - boost::shared_ptr<std::string> dicomContent(new std::string()); - that->context_.ReadAttachment(*dicomContent, instanceToPreload->GetFileInfo(), true); - - if (that->transcode_) - { - boost::shared_ptr<std::string> transcodedDicom(new std::string()); - if (that->TranscodeDicom(*transcodedDicom, *dicomContent, instanceToPreload->GetId())) - { - dicomContent = transcodedDicom; - } - } - - { - boost::mutex::scoped_lock lock(that->availableInstancesMutex_); - that->availableInstances_[instanceToPreload->GetId()] = dicomContent; - that->condInstanceAvailable_.notify_one(); - } - } - catch (OrthancException& e) - { - LOG(ERROR) << "Failed to load instance " << instanceToPreload->GetId() << " error: " << e.GetDetails(); - boost::mutex::scoped_lock lock(that->availableInstancesMutex_); - // store a NULL result to notify that we could not read the instance - that->availableInstances_[instanceToPreload->GetId()] = boost::shared_ptr<std::string>(); - that->condInstanceAvailable_.notify_one(); - } - catch (...) - { - LOG(ERROR) << "Failed to load instance " << instanceToPreload->GetId() << " unknown error"; - boost::mutex::scoped_lock lock(that->availableInstancesMutex_); - // store a NULL result to notify that we could not read the instance - that->availableInstances_[instanceToPreload->GetId()] = boost::shared_ptr<std::string>(); - that->condInstanceAvailable_.notify_one(); - } - } - } - - virtual void PrepareDicom(const std::string& instanceId, const FileInfo& fileInfo) ORTHANC_OVERRIDE - { - instancesToPreload_.Enqueue(new InstanceToPreload(instanceId, fileInfo)); - } - - virtual void GetDicom(std::string& dicom, const std::string& instanceId, const FileInfo& fileInfo) ORTHANC_OVERRIDE - { - boost::mutex::scoped_lock lock(availableInstancesMutex_); - - while (true) - { - // wait for this instance to be available but this might not be the one we are waiting for ! - while (availableInstances_.find(instanceId) == availableInstances_.end()) - { - condInstanceAvailable_.wait(lock); - } - - boost::shared_ptr<std::string> dicomContent; - - // this is the instance we were waiting for - dicomContent = availableInstances_[instanceId]; - availableInstances_.erase(instanceId); - - if (dicomContent.get() == NULL) // there has been an error while reading the file - { - throw OrthancException(ErrorCode_InexistentItem); - } - dicom.swap(*dicomContent); - - return; - } - } - }; // This enum defines specific resource types to be used when exporting the archive. // It defines if we should use the PatientInfo from the Patient or from the Study. @@ -774,7 +516,7 @@ } void Apply(HierarchicalZipWriter& writer, - InstanceLoader& instanceLoader, + ThreadedInstancesLoader& instancesLoader, DicomDirWriter* dicomDir, const std::string& dicomDirFolder, bool transcode, @@ -797,7 +539,7 @@ try { LOG(INFO) << "Adding instance " << instanceId_ << " in zip"; - instanceLoader.GetDicom(content, instanceId_, fileInfo_); + instancesLoader.GetDicom(content, instanceId_); } catch (OrthancException& e) { @@ -828,11 +570,11 @@ std::deque<Command*> commands_; uint64_t uncompressedSize_; unsigned int instancesCount_; - InstanceLoader& instanceLoader_; + ThreadedInstancesLoader& instancesLoader_; void ApplyInternal(HierarchicalZipWriter& writer, - InstanceLoader& instanceLoader, + ThreadedInstancesLoader& instancesLoader, size_t index, DicomDirWriter* dicomDir, const std::string& dicomDirFolder, @@ -844,14 +586,14 @@ throw OrthancException(ErrorCode_ParameterOutOfRange); } - commands_[index]->Apply(writer, instanceLoader, dicomDir, dicomDirFolder, transcode, transferSyntax); + commands_[index]->Apply(writer, instancesLoader, dicomDir, dicomDirFolder, transcode, transferSyntax); } public: - explicit ZipCommands(InstanceLoader& instanceLoader) : + explicit ZipCommands(ThreadedInstancesLoader& instancesLoader) : uncompressedSize_(0), instancesCount_(0), - instanceLoader_(instanceLoader) + instancesLoader_(instancesLoader) { } @@ -882,24 +624,24 @@ // "media" flavor (with DICOMDIR) void Apply(HierarchicalZipWriter& writer, - InstanceLoader& instanceLoader, + ThreadedInstancesLoader& instancesLoader, size_t index, DicomDirWriter& dicomDir, const std::string& dicomDirFolder, bool transcode, DicomTransferSyntax transferSyntax) const { - ApplyInternal(writer, instanceLoader, index, &dicomDir, dicomDirFolder, transcode, transferSyntax); + ApplyInternal(writer, instancesLoader, index, &dicomDir, dicomDirFolder, transcode, transferSyntax); } // "archive" flavor (without DICOMDIR) void Apply(HierarchicalZipWriter& writer, - InstanceLoader& instanceLoader, + ThreadedInstancesLoader& instancesLoader, size_t index, bool transcode, DicomTransferSyntax transferSyntax) const { - ApplyInternal(writer, instanceLoader, index, NULL, "", transcode, transferSyntax); + ApplyInternal(writer, instancesLoader, index, NULL, "", transcode, transferSyntax); } void AddOpenDirectory(const std::string& filename) @@ -916,7 +658,7 @@ const std::string& instanceId, const FileInfo& fileInfo) { - instanceLoader_.PrepareDicom(instanceId, fileInfo); + instancesLoader_.PrepareDicom(instanceId, fileInfo); commands_.push_back(new Command(Type_WriteInstance, filename, instanceId, fileInfo)); instancesCount_ ++; uncompressedSize_ += fileInfo.GetUncompressedSize(); @@ -1090,7 +832,7 @@ { private: ServerContext& context_; - InstanceLoader& instanceLoader_; + ThreadedInstancesLoader& instancesLoader_; ZipCommands commands_; std::unique_ptr<HierarchicalZipWriter> zip_; std::unique_ptr<DicomDirWriter> dicomDir_; @@ -1100,14 +842,14 @@ public: ZipWriterIterator(ServerContext& context, - InstanceLoader& instanceLoader, + ThreadedInstancesLoader& instancesLoader, ArchiveIndex& archive, bool isMedia, bool enableExtendedSopClass, bool allowUtf8) : context_(context), - instanceLoader_(instanceLoader), - commands_(instanceLoader), + instancesLoader_(instancesLoader), + commands_(instancesLoader), isMedia_(isMedia), isStream_(false), allowUtf8_(allowUtf8) @@ -1234,13 +976,13 @@ if (isMedia_) { assert(dicomDir_.get() != NULL); - commands_.Apply(*zip_, instanceLoader_, index, *dicomDir_, + commands_.Apply(*zip_, instancesLoader_, index, *dicomDir_, MEDIA_IMAGES_FOLDER, transcode, transferSyntax); } else { assert(dicomDir_.get() == NULL); - commands_.Apply(*zip_, instanceLoader_, index, transcode, transferSyntax); + commands_.Apply(*zip_, instancesLoader_, index, transcode, transferSyntax); } } } @@ -1273,7 +1015,7 @@ transcode_(false), transferSyntax_(DicomTransferSyntax_LittleEndianImplicit), lossyQuality_(100), - loaderThreads_(0), + loaderThreads_(1), allowUtf8_(false) { } @@ -1395,7 +1137,7 @@ } else { - loaderThreads_ = loaderThreads; + loaderThreads_ = std::max(1u, loaderThreads); } } @@ -1422,15 +1164,7 @@ void ArchiveJob::Start() { - if (loaderThreads_ == 0) - { - // default behaviour before loaderThreads was introducted in 1.10.0 - instanceLoader_.reset(new SynchronousInstanceLoader(context_, transcode_, transferSyntax_, lossyQuality_)); - } - else - { - instanceLoader_.reset(new ThreadedInstanceLoader(context_, loaderThreads_, transcode_, transferSyntax_, lossyQuality_)); - } + instancesLoader_.reset(new ThreadedInstancesLoader(context_, loaderThreads_, transcode_, transferSyntax_, lossyQuality_, "ARCH")); if (writer_.get() != NULL) { @@ -1453,7 +1187,7 @@ assert(asynchronousTarget_.get() != NULL); asynchronousTarget_->Touch(); // Make sure we can write to the temporary file - writer_.reset(new ZipWriterIterator(context_, *instanceLoader_, *archive_, isMedia_, enableExtendedSopClass_, allowUtf8_)); + writer_.reset(new ZipWriterIterator(context_, *instancesLoader_, *archive_, isMedia_, enableExtendedSopClass_, allowUtf8_)); writer_->SetOutputFile(asynchronousTarget_->GetPath()); } } @@ -1461,7 +1195,7 @@ { assert(synchronousTarget_.get() != NULL); - writer_.reset(new ZipWriterIterator(context_, *instanceLoader_, *archive_, isMedia_, enableExtendedSopClass_, allowUtf8_)); + writer_.reset(new ZipWriterIterator(context_, *instancesLoader_, *archive_, isMedia_, enableExtendedSopClass_, allowUtf8_)); writer_->AcquireOutputStream(synchronousTarget_.release()); } @@ -1506,9 +1240,9 @@ writer_.reset(); } - if (instanceLoader_.get() != NULL) + if (instancesLoader_.get() != NULL) { - instanceLoader_->Clear(false); + instancesLoader_->Clear(false); } if (asynchronousTarget_.get() != NULL) @@ -1578,9 +1312,9 @@ asynchronousTarget_.reset(); // clear the loader threads - if (instanceLoader_.get() != NULL) + if (instancesLoader_.get() != NULL) { - instanceLoader_->Clear(true); + instancesLoader_->Clear(true); } } }
--- a/OrthancServer/Sources/ServerJobs/ArchiveJob.h Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/ServerJobs/ArchiveJob.h Fri Mar 27 16:32:46 2026 +0100 @@ -34,6 +34,7 @@ namespace Orthanc { class ServerContext; + class ThreadedInstancesLoader; class ArchiveJob : public IJob { @@ -45,14 +46,11 @@ class ResourceIdentifiers; class ZipCommands; class ZipWriterIterator; - class InstanceLoader; - class SynchronousInstanceLoader; - class ThreadedInstanceLoader; std::unique_ptr<ZipWriter::IOutputStream> synchronousTarget_; // Only valid before "Start()" std::unique_ptr<TemporaryFile> asynchronousTarget_; ServerContext& context_; - std::unique_ptr<InstanceLoader> instanceLoader_; + std::unique_ptr<ThreadedInstancesLoader> instancesLoader_; boost::shared_ptr<ArchiveIndex> archive_; bool isMedia_; bool enableExtendedSopClass_;
--- a/OrthancServer/Sources/ServerJobs/DicomModalityStoreJob.cpp Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/ServerJobs/DicomModalityStoreJob.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -55,7 +55,7 @@ try { - context_.ReadDicom(dicom, instance); + instancesLoader_->GetDicom(dicom, instance); } catch (OrthancException& e) { @@ -112,7 +112,7 @@ DicomModalityStoreJob::DicomModalityStoreJob(ServerContext& context) : - context_(context), + StoreJob(context), moveOriginatorId_(0), // By default, not a C-MOVE storageCommitment_(false) // By default, no storage commitment { @@ -207,9 +207,10 @@ void DicomModalityStoreJob::Stop(JobStopReason reason) // For pausing jobs { connection_.reset(NULL); + + StoreJob::Stop(reason); } - void DicomModalityStoreJob::ResetStorageCommitment() { if (storageCommitment_) @@ -269,8 +270,7 @@ DicomModalityStoreJob::DicomModalityStoreJob(ServerContext& context, const Json::Value& serialized) : - SetOfInstancesJob(serialized), - context_(context) + StoreJob(context, serialized) { moveOriginatorAet_ = SerializationToolbox::ReadString(serialized, MOVE_ORIGINATOR_AET); moveOriginatorId_ = static_cast<uint16_t>
--- a/OrthancServer/Sources/ServerJobs/DicomModalityStoreJob.h Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/ServerJobs/DicomModalityStoreJob.h Fri Mar 27 16:32:46 2026 +0100 @@ -24,7 +24,7 @@ #pragma once #include "../../../OrthancFramework/Sources/Compatibility.h" -#include "../../../OrthancFramework/Sources/JobsEngine/SetOfInstancesJob.h" +#include "StoreJob.h" #include "../../../OrthancFramework/Sources/DicomNetworking/DicomStoreUserConnection.h" #include <list> @@ -33,10 +33,9 @@ { class ServerContext; - class DicomModalityStoreJob : public SetOfInstancesJob + class DicomModalityStoreJob : public StoreJob { private: - ServerContext& context_; DicomAssociationParameters parameters_; std::string moveOriginatorAet_; uint16_t moveOriginatorId_; @@ -93,6 +92,11 @@ target = "DicomModalityStore"; } + virtual const char* GetLoaderPrefix() const + { + return "CSTO"; + } + virtual void GetPublicContent(Json::Value& value) const ORTHANC_OVERRIDE; virtual bool Serialize(Json::Value& target) const ORTHANC_OVERRIDE;
--- a/OrthancServer/Sources/ServerJobs/OrthancPeerStoreJob.cpp Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/ServerJobs/OrthancPeerStoreJob.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -61,7 +61,7 @@ if (transcode_) { std::string dicom; - context_.ReadDicom(dicom, instance); + instancesLoader_->GetDicom(dicom, instance); std::set<DicomTransferSyntax> syntaxes; syntaxes.insert(transferSyntax_); @@ -217,6 +217,8 @@ void OrthancPeerStoreJob::Stop(JobStopReason reason) // For pausing jobs { client_.reset(NULL); + + StoreJob::Stop(reason); } @@ -249,8 +251,7 @@ OrthancPeerStoreJob::OrthancPeerStoreJob(ServerContext& context, const Json::Value& serialized) : - SetOfInstancesJob(serialized), - context_(context) + StoreJob(context, serialized) { assert(serialized.type() == Json::objectValue); peer_ = WebServiceParameters(serialized[PEER]);
--- a/OrthancServer/Sources/ServerJobs/OrthancPeerStoreJob.h Thu Mar 26 11:56:40 2026 +0100 +++ b/OrthancServer/Sources/ServerJobs/OrthancPeerStoreJob.h Fri Mar 27 16:32:46 2026 +0100 @@ -24,7 +24,7 @@ #pragma once #include "../../../OrthancFramework/Sources/Compatibility.h" -#include "../../../OrthancFramework/Sources/JobsEngine/SetOfInstancesJob.h" +#include "StoreJob.h" #include "../../../OrthancFramework/Sources/HttpClient.h" #include <stdint.h> @@ -34,10 +34,9 @@ { class ServerContext; - class OrthancPeerStoreJob : public SetOfInstancesJob + class OrthancPeerStoreJob : public StoreJob { private: - ServerContext& context_; WebServiceParameters peer_; std::unique_ptr<HttpClient> client_; bool transcode_; @@ -52,7 +51,7 @@ public: explicit OrthancPeerStoreJob(ServerContext& context) : - context_(context), + StoreJob(context), transcode_(false), transferSyntax_(DicomTransferSyntax_LittleEndianExplicit), // Dummy value compress_(false), @@ -97,6 +96,11 @@ target = "OrthancPeerStore"; } + virtual const char* GetLoaderPrefix() const + { + return "PSTO"; + } + virtual void GetPublicContent(Json::Value& value) const ORTHANC_OVERRIDE; virtual bool Serialize(Json::Value& target) const ORTHANC_OVERRIDE;
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/OrthancServer/Sources/ServerJobs/StoreJob.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -0,0 +1,102 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2023 Osimis S.A., Belgium + * Copyright (C) 2024-2026 Orthanc Team SRL, Belgium + * Copyright (C) 2021-2026 Sebastien Jodogne, ICTEAM UCLouvain, 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. + * + * 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 "StoreJob.h" + +#include <algorithm> + +#include "../../../OrthancFramework/Sources/Compatibility.h" +#include "../../../OrthancFramework/Sources/Logging.h" +#include "../../../OrthancFramework/Sources/SerializationToolbox.h" +#include "../ServerContext.h" +#include "../OrthancConfiguration.h" + + +namespace Orthanc +{ + StoreJob::StoreJob(ServerContext& context) : + context_(context) + { + } + + StoreJob::StoreJob(ServerContext& context, + const Json::Value& serialized) : + SetOfInstancesJob(serialized), + context_(context) + { + } + + void StoreJob::AddInstances(const std::vector<std::string>& instancesIds, + const std::vector<FileInfo>& filesInfo) + { + assert(instancesIds.size() == filesInfo.size()); + + instancesIds_.reserve(instancesIds_.size() + instancesIds.size()); + filesInfo_.reserve(filesInfo.size() + filesInfo.size()); + + std::copy(instancesIds.begin(), instancesIds.end(), std::back_inserter(instancesIds_)); + std::copy(filesInfo.begin(), filesInfo.end(), std::back_inserter(filesInfo_)); + + // create the commands (they must be in the same order as the instancesIds_ such that the instancesLoader loads them in the right consume order) + for (size_t i = 0; i < instancesIds.size(); ++i) + { + AddInstance(instancesIds[i]); + } + } + + void StoreJob::Start() + { + size_t loaderThreads = 1; + { + OrthancConfiguration::ReaderLock lock; + loaderThreads = lock.GetConfiguration().GetLoaderThreads(); + } + + instancesLoader_.reset(new ThreadedInstancesLoader(context_, loaderThreads, false, DicomTransferSyntax_LittleEndianImplicit /* dummy value not used*/, 0, GetLoaderPrefix())); + + for (size_t i = 0; i < instancesIds_.size(); ++i) + { + instancesLoader_->PrepareDicom(instancesIds_[i], filesInfo_[i]); + } + + SetOfInstancesJob::Start(); + } + + void StoreJob::Stop(JobStopReason reason) // For pausing jobs + { + if (reason == JobStopReason_Canceled || + reason == JobStopReason_Failure || + reason == JobStopReason_Retry) + { + // clear the loader threads + if (instancesLoader_.get() != NULL) + { + instancesLoader_->Clear(true); + } + } + } + + + +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/OrthancServer/Sources/ServerJobs/StoreJob.h Fri Mar 27 16:32:46 2026 +0100 @@ -0,0 +1,62 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2023 Osimis S.A., Belgium + * Copyright (C) 2024-2026 Orthanc Team SRL, Belgium + * Copyright (C) 2021-2026 Sebastien Jodogne, ICTEAM UCLouvain, 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. + * + * 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 "../../../OrthancFramework/Sources/Compatibility.h" +#include "../../../OrthancFramework/Sources/JobsEngine/SetOfInstancesJob.h" +#include "../../../OrthancFramework/Sources/DicomNetworking/DicomStoreUserConnection.h" +#include "ThreadedInstancesLoader.h" + +#include <vector> + +namespace Orthanc +{ + class ServerContext; + class ThreadedInstancesLoader; + + class StoreJob : public SetOfInstancesJob + { + protected: + ServerContext& context_; + std::unique_ptr<ThreadedInstancesLoader> instancesLoader_; + // TODO: re-create when unserializing (in AddParentResources ?) + std::vector<std::string> instancesIds_; + std::vector<FileInfo> filesInfo_; + + public: + explicit StoreJob(ServerContext& context); + + StoreJob(ServerContext& context, + const Json::Value& serialized); + + virtual void Start() ORTHANC_OVERRIDE; + + virtual void Stop(JobStopReason reason) ORTHANC_OVERRIDE; + + void AddInstances(const std::vector<std::string>& instancesIds, + const std::vector<FileInfo>& filesInfo); + + virtual const char* GetLoaderPrefix() const = 0; + }; +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/OrthancServer/Sources/ServerJobs/ThreadedInstancesLoader.cpp Fri Mar 27 16:32:46 2026 +0100 @@ -0,0 +1,249 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2023 Osimis S.A., Belgium + * Copyright (C) 2024-2026 Orthanc Team SRL, Belgium + * Copyright (C) 2021-2026 Sebastien Jodogne, ICTEAM UCLouvain, 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. + * + * 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 "ThreadedInstancesLoader.h" + +#include "../ServerContext.h" +#include "../../../OrthancFramework/Sources/Logging.h" +#include "../../../OrthancFramework/Sources/DicomParsing/IDicomTranscoder.h" +#include "../../../OrthancFramework/Sources/DicomParsing/FromDcmtkBridge.h" + +static boost::mutex loaderThreadsCounterMutex; +static uint32_t loaderThreadsCounter = 0; + + +namespace Orthanc +{ + class InstanceToPreload : public Orthanc::IDynamicObject + { + private: + std::string id_; + FileInfo fileInfo_; + + public: + explicit InstanceToPreload(const std::string& id, const FileInfo& fileInfo) : + id_(id), + fileInfo_(fileInfo) + { + } + + virtual ~InstanceToPreload() ORTHANC_OVERRIDE + { + } + + const std::string& GetId() const {return id_;}; + const FileInfo& GetFileInfo() const {return fileInfo_;}; + }; + + + ThreadedInstancesLoader::ThreadedInstancesLoader(ServerContext& context, size_t threadCount, bool transcode, DicomTransferSyntax transferSyntax, unsigned int lossyQuality, const std::string& nameForLogs4Char) + : availableInstancesSemaphore_(3*threadCount), + instancesToPreload_(0), // no limit on the message queue, the flow control is performed by the availableInstancesSemaphore_ + loadersShouldStop_(false), + nameForLogs4Char_(nameForLogs4Char), + context_(context), + transcode_(transcode), + transferSyntax_(transferSyntax), + lossyQuality_(lossyQuality) + + { + assert(nameForLogs4Char_.size() <= 4); + + if (threadCount < 1) + { + throw OrthancException(ErrorCode_InternalError); + } + + for (size_t i = 0; i < threadCount; i++) + { + threads_.push_back(new boost::thread(PreloaderWorkerThread, this)); + } + } + + ThreadedInstancesLoader::~ThreadedInstancesLoader() + { + ThreadedInstancesLoader::Clear(false); + } + + void ThreadedInstancesLoader::Clear(bool isAbort) + { + if (threads_.size() > 0) + { + loadersShouldStop_ = true; // not need to protect this by a mutex. This is the only "writer" and all loaders are "readers" + + if (isAbort) + { + LOG(INFO) << "Cancelling the loader threads"; + instancesToPreload_.Clear(); + } + else + { + LOG(INFO) << "Waiting for loader threads to complete"; + } + + // unlock the loaders if they are waiting on this message queue (this happens when the job completes sucessfully) + for (size_t i = 0; i < threads_.size(); i++) + { + instancesToPreload_.Enqueue(NULL); + } + + // unlock the loaders if they are waiting for room in the availableInstances (this happens when the job is interrupted) + availableInstancesSemaphore_.Release(threads_.size()); + + for (size_t i = 0; i < threads_.size(); i++) + { + if (threads_[i]->joinable()) + { + threads_[i]->join(); + } + delete threads_[i]; + } + + threads_.clear(); + availableInstances_.clear(); + + LOG(INFO) << "Waiting for loader threads to complete - done"; + } + } + + void ThreadedInstancesLoader::PreloaderWorkerThread(ThreadedInstancesLoader* that) + { + { + boost::mutex::scoped_lock lock(loaderThreadsCounterMutex); + Logging::SetCurrentThreadName(that->nameForLogs4Char_ + std::string("-LOAD-") + boost::lexical_cast<std::string>(loaderThreadsCounter++)); + loaderThreadsCounter %= 1000000; + } + + LOG(INFO) << "Loader thread has started"; + + while (true) + { + that->availableInstancesSemaphore_.Acquire(1); // reserve the slot early (since the instances are ordered, it is important that a single worker does not push dozens of small instances while we are waiting for a slot for a big instance) + + std::unique_ptr<InstanceToPreload> instanceToPreload(dynamic_cast<InstanceToPreload*>(that->instancesToPreload_.Dequeue(0))); + if (instanceToPreload.get() == NULL || that->loadersShouldStop_) // that's the signal to exit the thread + { + LOG(INFO) << "Loader thread has completed"; + return; + } + + const std::string& instanceId = instanceToPreload->GetId(); + + try + { + boost::shared_ptr<std::string> dicomContent(new std::string()); + that->context_.ReadAttachment(*dicomContent, instanceToPreload->GetFileInfo(), true); + + if (that->transcode_) + { + boost::shared_ptr<std::string> transcodedDicom(new std::string()); + if (that->TranscodeDicom(*transcodedDicom, *dicomContent, instanceId)) + { + dicomContent = transcodedDicom; + } + } + + { + boost::mutex::scoped_lock lock(that->availableInstancesMutex_); + that->availableInstances_[instanceId] = dicomContent; + that->condInstanceAvailable_.notify_one(); + } + } + catch (OrthancException& e) + { + LOG(ERROR) << "Failed to load instance " << instanceId << " error: " << e.GetDetails(); + boost::mutex::scoped_lock lock(that->availableInstancesMutex_); + // store a NULL result to notify that we could not read the instance + that->availableInstances_[instanceId] = boost::shared_ptr<std::string>(); + that->condInstanceAvailable_.notify_one(); + } + catch (...) + { + LOG(ERROR) << "Failed to load instance " << instanceId << " unknown error"; + boost::mutex::scoped_lock lock(that->availableInstancesMutex_); + // store a NULL result to notify that we could not read the instance + that->availableInstances_[instanceId] = boost::shared_ptr<std::string>(); + that->condInstanceAvailable_.notify_one(); + } + } + } + + void ThreadedInstancesLoader::PrepareDicom(const std::string& instanceId, const FileInfo& fileInfo) + { + instancesToPreload_.Enqueue(new InstanceToPreload(instanceId, fileInfo)); + } + + void ThreadedInstancesLoader::GetDicom(std::string& dicom, const std::string& instanceId) + { + boost::mutex::scoped_lock lock(availableInstancesMutex_); + + while (true) + { + // wait for this instance to be available but this might not be the one we are waiting for ! + while (availableInstances_.find(instanceId) == availableInstances_.end()) + { + condInstanceAvailable_.wait(lock); + } + + boost::shared_ptr<std::string> dicomContent; + + // this is the instance we were waiting for + dicomContent = availableInstances_[instanceId]; + availableInstances_.erase(instanceId); + availableInstancesSemaphore_.Release(1); + + if (dicomContent.get() == NULL) // there has been an error while reading the file + { + throw OrthancException(ErrorCode_InexistentItem); + } + dicom.swap(*dicomContent); + + return; + } + }; + + bool ThreadedInstancesLoader::TranscodeDicom(std::string& transcodedBuffer, const std::string& sourceBuffer, const std::string& instanceId) + { + if (transcode_) + { + std::set<DicomTransferSyntax> syntaxes; + syntaxes.insert(transferSyntax_); + + IDicomTranscoder::DicomImage source, transcoded; + source.SetExternalBuffer(sourceBuffer); + + if (context_.Transcode(transcoded, source, syntaxes, TranscodingSopInstanceUidMode_AllowNew, lossyQuality_)) + { + transcodedBuffer.assign(reinterpret_cast<const char*>(transcoded.GetBufferData()), transcoded.GetBufferSize()); + return true; + } + else + { + LOG(INFO) << "Cannot transcode instance " << instanceId + << " to transfer syntax: " << GetTransferSyntaxUid(transferSyntax_); + } + } + + return false; + } + +} \ No newline at end of file
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/OrthancServer/Sources/ServerJobs/ThreadedInstancesLoader.h Fri Mar 27 16:32:46 2026 +0100 @@ -0,0 +1,72 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2016 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, Belgium + * Copyright (C) 2017-2023 Osimis S.A., Belgium + * Copyright (C) 2024-2026 Orthanc Team SRL, Belgium + * Copyright (C) 2021-2026 Sebastien Jodogne, ICTEAM UCLouvain, 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. + * + * 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 <string> +#include <boost/noncopyable.hpp> +#include "../../../OrthancFramework/Sources/Compatibility.h" +#include "../../../OrthancFramework/Sources/MultiThreading/BlockingSharedMessageQueue.h" +#include "../../../OrthancFramework/Sources/Enumerations.h" +#include "../../../OrthancFramework/Sources/MultiThreading/Semaphore.h" + + +namespace Orthanc +{ + class ServerContext; + class FileInfo; + + class ThreadedInstancesLoader : public boost::noncopyable + { + boost::condition_variable condInstanceAvailable_; + std::map<std::string, boost::shared_ptr<std::string> > availableInstances_; + boost::mutex availableInstancesMutex_; + Semaphore availableInstancesSemaphore_; + BlockingSharedMessageQueue instancesToPreload_; + std::vector<boost::thread*> threads_; + bool loadersShouldStop_; + std::string nameForLogs4Char_; + + protected: + ServerContext& context_; + bool transcode_; + DicomTransferSyntax transferSyntax_; + unsigned int lossyQuality_; + + public: + ThreadedInstancesLoader(ServerContext& context, size_t threadCount, bool transcode, DicomTransferSyntax transferSyntax, unsigned int lossyQuality, const std::string& nameForLogs4Char); + + ~ThreadedInstancesLoader(); + + void Clear(bool isAbort); + + static void PreloaderWorkerThread(ThreadedInstancesLoader* that); + + void PrepareDicom(const std::string& instanceId, const FileInfo& fileInfo); + + void GetDicom(std::string& dicom, const std::string& instanceId); + + protected: + bool TranscodeDicom(std::string& transcodedBuffer, const std::string& sourceBuffer, const std::string& instanceId); + }; +} + \ No newline at end of file