Mercurial > hg > orthanc
changeset 1679:15acdb19d520
RunnableWorkersPool
author | Sebastien Jodogne <s.jodogne@gmail.com> |
---|---|
date | Tue, 06 Oct 2015 13:31:40 +0200 |
parents | d7039be97eeb |
children | 4113a9a668b1 |
files | CMakeLists.txt Core/MultiThreading/IRunnableBySteps.h Core/MultiThreading/RunnableWorkersPool.cpp Core/MultiThreading/RunnableWorkersPool.h UnitTestsSources/MultiThreadingTests.cpp |
diffstat | 5 files changed, 258 insertions(+), 3 deletions(-) [+] |
line wrap: on
line diff
--- a/CMakeLists.txt Wed Sep 30 14:04:25 2015 +0200 +++ b/CMakeLists.txt Tue Oct 06 13:31:40 2015 +0200 @@ -121,6 +121,7 @@ Core/MultiThreading/ReaderWriterLock.cpp Core/MultiThreading/Semaphore.cpp Core/MultiThreading/SharedMessageQueue.cpp + Core/MultiThreading/RunnableWorkersPool.cpp Core/Images/Font.cpp Core/Images/FontRegistry.cpp Core/Images/ImageAccessor.cpp
--- a/Core/MultiThreading/IRunnableBySteps.h Wed Sep 30 14:04:25 2015 +0200 +++ b/Core/MultiThreading/IRunnableBySteps.h Tue Oct 06 13:31:40 2015 +0200 @@ -32,9 +32,11 @@ #pragma once +#include "../IDynamicObject.h" + namespace Orthanc { - class IRunnableBySteps + class IRunnableBySteps : public IDynamicObject { public: virtual ~IRunnableBySteps()
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/MultiThreading/RunnableWorkersPool.cpp Tue Oct 06 13:31:40 2015 +0200 @@ -0,0 +1,149 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, 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 "RunnableWorkersPool.h" + +#include "SharedMessageQueue.h" +#include "../OrthancException.h" +#include "../Logging.h" + +namespace Orthanc +{ + struct RunnableWorkersPool::PImpl + { + class Worker + { + private: + const bool& continue_; + SharedMessageQueue& queue_; + boost::thread thread_; + + static void WorkerThread(Worker* that) + { + while (that->continue_) + { + std::auto_ptr<IDynamicObject> obj(that->queue_.Dequeue(100)); + if (obj.get() != NULL) + { + try + { + IRunnableBySteps& runnable = *dynamic_cast<IRunnableBySteps*>(obj.get()); + + bool wishToContinue = runnable.Step(); + + if (wishToContinue) + { + // The runnable wishes to continue, reinsert it at the beginning of the queue + that->queue_.Enqueue(obj.release()); + } + } + catch (OrthancException& e) + { + LOG(ERROR) << "Exception in a pool of working threads: " << e.What(); + } + } + } + } + + public: + Worker(const bool& globalContinue, + SharedMessageQueue& queue) : + continue_(globalContinue), + queue_(queue) + { + thread_ = boost::thread(WorkerThread, this); + } + + void Join() + { + if (thread_.joinable()) + { + thread_.join(); + } + } + }; + + + bool continue_; + std::vector<Worker*> workers_; + SharedMessageQueue queue_; + }; + + + + RunnableWorkersPool::RunnableWorkersPool(size_t countWorkers) : pimpl_(new PImpl) + { + pimpl_->continue_ = true; + + if (countWorkers <= 0) + { + throw OrthancException(ErrorCode_ParameterOutOfRange); + } + + pimpl_->workers_.resize(countWorkers); + + for (size_t i = 0; i < countWorkers; i++) + { + pimpl_->workers_[i] = new PImpl::Worker(pimpl_->continue_, pimpl_->queue_); + } + } + + + RunnableWorkersPool::~RunnableWorkersPool() + { + pimpl_->continue_ = false; + + for (size_t i = 0; i < pimpl_->workers_.size(); i++) + { + PImpl::Worker* worker = pimpl_->workers_[i]; + + if (worker != NULL) + { + worker->Join(); + delete worker; + } + } + } + + + void RunnableWorkersPool::Add(IRunnableBySteps* runnable) + { + pimpl_->queue_.Enqueue(runnable); + } + + + void RunnableWorkersPool::WaitDone() + { + pimpl_->queue_.WaitEmpty(0); + } +}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Core/MultiThreading/RunnableWorkersPool.h Tue Oct 06 13:31:40 2015 +0200 @@ -0,0 +1,56 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics + * Department, University Hospital of Liege, 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 "IRunnableBySteps.h" + +#include <boost/shared_ptr.hpp> + +namespace Orthanc +{ + class RunnableWorkersPool : public boost::noncopyable + { + private: + struct PImpl; + boost::shared_ptr<PImpl> pimpl_; + + public: + RunnableWorkersPool(size_t countWorkers); + + ~RunnableWorkersPool(); + + void Add(IRunnableBySteps* runnable); // Takes the ownership + + void WaitDone(); + }; +}
--- a/UnitTestsSources/MultiThreadingTests.cpp Wed Sep 30 14:04:25 2015 +0200 +++ b/UnitTestsSources/MultiThreadingTests.cpp Tue Oct 06 13:31:40 2015 +0200 @@ -39,6 +39,7 @@ #include "../Core/MultiThreading/Locker.h" #include "../Core/MultiThreading/Mutex.h" #include "../Core/MultiThreading/ReaderWriterLock.h" +#include "../Core/MultiThreading/RunnableWorkersPool.h" using namespace Orthanc; @@ -183,7 +184,7 @@ outputs.push_back(boost::lexical_cast<std::string>(b)); } - Toolbox::USleep(100000); + Toolbox::USleep(30000); return true; } @@ -202,7 +203,7 @@ { printf(">> %s: %0.1f\n", it->c_str(), 100.0f * s->GetProgress(*it)); } - Toolbox::USleep(10000); + Toolbox::USleep(3000); } } @@ -257,3 +258,49 @@ t.join(); } } + + +namespace +{ + class MyRunnable : public IRunnableBySteps + { + private: + unsigned int& output_; + unsigned int count_; + + public: + MyRunnable(unsigned int& output) : output_(output), count_(0) + { + } + + virtual bool Step() + { + count_ ++; + output_ ++; + + boost::this_thread::sleep(boost::posix_time::milliseconds(3)); + + return (count_ < 7); + } + }; +} + + +TEST(MultiThreading, RunnableWorkersPool) +{ + unsigned int output = 0; + + { + RunnableWorkersPool pool(3); + + for (size_t i = 0; i < 11; i++) + { + pool.Add(new MyRunnable(output)); + } + + pool.WaitDone(); + } + + + ASSERT_EQ(11 * 7, output); +}