Mercurial > hg > orthanc
changeset 2567:3caca43371f5 jobs
archival of BagOfTasksProcessor, Mutex, and ReaderWriterLock
author | Sebastien Jodogne <s.jodogne@gmail.com> |
---|---|
date | Mon, 07 May 2018 14:26:31 +0200 |
parents | c09ce3c038fc |
children | a46094602346 |
files | Core/MultiThreading/BagOfTasks.h Core/MultiThreading/BagOfTasksProcessor.cpp Core/MultiThreading/BagOfTasksProcessor.h Core/MultiThreading/Mutex.cpp Core/MultiThreading/Mutex.h Core/MultiThreading/ReaderWriterLock.cpp Core/MultiThreading/ReaderWriterLock.h Resources/CMake/OrthancFrameworkConfiguration.cmake Resources/Graveyard/BagOfTasks.h Resources/Graveyard/BagOfTasksProcessor.cpp Resources/Graveyard/BagOfTasksProcessor.h Resources/Graveyard/Mutex.cpp Resources/Graveyard/Mutex.h Resources/Graveyard/ReaderWriterLock.cpp Resources/Graveyard/ReaderWriterLock.h UnitTestsSources/MultiThreadingTests.cpp |
diffstat | 16 files changed, 874 insertions(+), 900 deletions(-) [+] |
line wrap: on
line diff
--- a/Core/MultiThreading/BagOfTasks.h Mon May 07 13:51:23 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 Mon May 07 13:51:23 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 Mon May 07 13:51:23 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 Mon May 07 13:51:23 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 Mon May 07 13:51:23 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 Mon May 07 13:51:23 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 Mon May 07 13:51:23 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/Resources/CMake/OrthancFrameworkConfiguration.cmake Mon May 07 13:51:23 2018 +0200 +++ b/Resources/CMake/OrthancFrameworkConfiguration.cmake Mon May 07 14:26:31 2018 +0200 @@ -498,9 +498,6 @@ 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/MultiThreading/RunnableWorkersPool.cpp ${ORTHANC_ROOT}/Core/MultiThreading/Semaphore.cpp ${ORTHANC_ROOT}/Core/MultiThreading/SharedMessageQueue.cpp
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Resources/Graveyard/BagOfTasks.h Mon May 07 14:26:31 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 Mon May 07 14:26:31 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 Mon May 07 14:26:31 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 Mon May 07 14:26:31 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 Mon May 07 14:26:31 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 Mon May 07 14:26:31 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 Mon May 07 14:26:31 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(); + }; +}
--- a/UnitTestsSources/MultiThreadingTests.cpp Mon May 07 13:51:23 2018 +0200 +++ b/UnitTestsSources/MultiThreadingTests.cpp Mon May 07 14:26:31 2018 +0200 @@ -39,8 +39,6 @@ #include "../Core/SystemToolbox.h" #include "../Core/Toolbox.h" #include "../Core/MultiThreading/Locker.h" -#include "../Core/MultiThreading/Mutex.h" -#include "../Core/MultiThreading/ReaderWriterLock.h" using namespace Orthanc; @@ -106,27 +104,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"