Mercurial > hg > orthanc
diff UnitTestsSources/MultiThreadingTests.cpp @ 994:b3d4f8a30324 lua-scripting
integration mainline->lua-scripting
author | Sebastien Jodogne <s.jodogne@gmail.com> |
---|---|
date | Wed, 02 Jul 2014 14:42:49 +0200 |
parents | UnitTestsSources/MultiThreading.cpp@394a19d44f9d UnitTestsSources/MultiThreading.cpp@dfc076546821 |
children | 8c67382f44a7 |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/UnitTestsSources/MultiThreadingTests.cpp Wed Jul 02 14:42:49 2014 +0200 @@ -0,0 +1,383 @@ +/** + * Orthanc - A Lightweight, RESTful DICOM Store + * Copyright (C) 2012-2014 Medical Physics Department, CHU 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 "PrecompiledHeadersUnitTests.h" +#include "gtest/gtest.h" +#include <glog/logging.h> + +#include "../OrthancServer/Scheduler/ServerScheduler.h" + +#include "../Core/OrthancException.h" +#include "../Core/Toolbox.h" +#include "../Core/MultiThreading/ArrayFilledByThreads.h" +#include "../Core/MultiThreading/Locker.h" +#include "../Core/MultiThreading/Mutex.h" +#include "../Core/MultiThreading/ReaderWriterLock.h" +#include "../Core/MultiThreading/ThreadedCommandProcessor.h" + +using namespace Orthanc; + +namespace +{ + class DynamicInteger : public ICommand + { + private: + int value_; + std::set<int>& target_; + + public: + DynamicInteger(int value, std::set<int>& target) : + value_(value), target_(target) + { + } + + int GetValue() const + { + return value_; + } + + virtual bool Execute() + { + static boost::mutex mutex; + boost::mutex::scoped_lock lock(mutex); + target_.insert(value_); + return true; + } + }; + + class MyFiller : public ArrayFilledByThreads::IFiller + { + private: + int size_; + unsigned int created_; + std::set<int> set_; + + public: + MyFiller(int size) : size_(size), created_(0) + { + } + + virtual size_t GetFillerSize() + { + return size_; + } + + virtual IDynamicObject* GetFillerItem(size_t index) + { + static boost::mutex mutex; + boost::mutex::scoped_lock lock(mutex); + created_++; + return new DynamicInteger(index * 2, set_); + } + + unsigned int GetCreatedCount() const + { + return created_; + } + + std::set<int> GetSet() + { + return set_; + } + }; +} + + + + +TEST(MultiThreading, SharedMessageQueueBasic) +{ + std::set<int> s; + + SharedMessageQueue q; + ASSERT_TRUE(q.WaitEmpty(0)); + q.Enqueue(new DynamicInteger(10, s)); + ASSERT_FALSE(q.WaitEmpty(1)); + q.Enqueue(new DynamicInteger(20, s)); + q.Enqueue(new DynamicInteger(30, s)); + q.Enqueue(new DynamicInteger(40, s)); + + std::auto_ptr<DynamicInteger> i; + i.reset(dynamic_cast<DynamicInteger*>(q.Dequeue(1))); ASSERT_EQ(10, i->GetValue()); + i.reset(dynamic_cast<DynamicInteger*>(q.Dequeue(1))); ASSERT_EQ(20, i->GetValue()); + i.reset(dynamic_cast<DynamicInteger*>(q.Dequeue(1))); ASSERT_EQ(30, i->GetValue()); + ASSERT_FALSE(q.WaitEmpty(1)); + i.reset(dynamic_cast<DynamicInteger*>(q.Dequeue(1))); ASSERT_EQ(40, i->GetValue()); + ASSERT_TRUE(q.WaitEmpty(0)); + ASSERT_EQ(NULL, q.Dequeue(1)); +} + + +TEST(MultiThreading, SharedMessageQueueClean) +{ + std::set<int> s; + + try + { + SharedMessageQueue q; + q.Enqueue(new DynamicInteger(10, s)); + q.Enqueue(new DynamicInteger(20, s)); + throw OrthancException("Nope"); + } + catch (OrthancException&) + { + } +} + + +TEST(MultiThreading, ArrayFilledByThreadEmpty) +{ + MyFiller f(0); + ArrayFilledByThreads a(f); + a.SetThreadCount(1); + ASSERT_EQ(0, a.GetSize()); +} + + +TEST(MultiThreading, ArrayFilledByThread1) +{ + MyFiller f(100); + ArrayFilledByThreads a(f); + a.SetThreadCount(1); + ASSERT_EQ(100, a.GetSize()); + for (size_t i = 0; i < a.GetSize(); i++) + { + ASSERT_EQ(2 * i, dynamic_cast<DynamicInteger&>(a.GetItem(i)).GetValue()); + } +} + + +TEST(MultiThreading, ArrayFilledByThread4) +{ + MyFiller f(100); + ArrayFilledByThreads a(f); + a.SetThreadCount(4); + ASSERT_EQ(100, a.GetSize()); + for (size_t i = 0; i < a.GetSize(); i++) + { + ASSERT_EQ(2 * i, dynamic_cast<DynamicInteger&>(a.GetItem(i)).GetValue()); + } + + ASSERT_EQ(100u, f.GetCreatedCount()); + + a.Invalidate(); + + ASSERT_EQ(100, a.GetSize()); + ASSERT_EQ(200u, f.GetCreatedCount()); + ASSERT_EQ(4u, a.GetThreadCount()); + ASSERT_TRUE(f.GetSet().empty()); + + for (size_t i = 0; i < a.GetSize(); i++) + { + ASSERT_EQ(2 * i, dynamic_cast<DynamicInteger&>(a.GetItem(i)).GetValue()); + } +} + + +TEST(MultiThreading, CommandProcessor) +{ + ThreadedCommandProcessor p(4); + + std::set<int> s; + + for (size_t i = 0; i < 100; i++) + { + p.Post(new DynamicInteger(i * 2, s)); + } + + p.Join(); + + for (size_t i = 0; i < 200; i++) + { + if (i % 2) + ASSERT_TRUE(s.find(i) == s.end()); + else + ASSERT_TRUE(s.find(i) != s.end()); + } +} + + +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 "../OrthancServer/DicomProtocol/ReusableDicomUserConnection.h" + +TEST(ReusableDicomUserConnection, DISABLED_Basic) +{ + ReusableDicomUserConnection c; + c.SetMillisecondsBeforeClose(200); + printf("START\n"); fflush(stdout); + + { + ReusableDicomUserConnection::Locker lock(c, "STORESCP", "localhost", 2000, ModalityManufacturer_Generic); + lock.GetConnection().StoreFile("/home/jodogne/DICOM/Cardiac/MR.X.1.2.276.0.7230010.3.1.4.2831157719.2256.1336386844.676281"); + } + + printf("**\n"); fflush(stdout); + Toolbox::USleep(1000000); + printf("**\n"); fflush(stdout); + + { + ReusableDicomUserConnection::Locker lock(c, "STORESCP", "localhost", 2000, ModalityManufacturer_Generic); + lock.GetConnection().StoreFile("/home/jodogne/DICOM/Cardiac/MR.X.1.2.276.0.7230010.3.1.4.2831157719.2256.1336386844.676277"); + } + + Toolbox::ServerBarrier(); + printf("DONE\n"); fflush(stdout); +} + + + +class Tutu : public IServerFilter +{ +private: + int factor_; + +public: + Tutu(int f) : factor_(f) + { + } + + virtual bool Apply(ListOfStrings& outputs, + const ListOfStrings& inputs) + { + for (ListOfStrings::const_iterator + it = inputs.begin(); it != inputs.end(); it++) + { + int a = boost::lexical_cast<int>(*it); + int b = factor_ * a; + + printf("%d * %d = %d\n", a, factor_, b); + + //if (a == 84) { printf("BREAK\n"); return false; } + + outputs.push_back(boost::lexical_cast<std::string>(b)); + } + + Toolbox::USleep(1000000); + + return true; + } + + virtual bool SendOutputsToSink() const + { + return true; + } +}; + + +static void Tata(ServerScheduler* s, ServerJob* j, bool* done) +{ + typedef IServerFilter::ListOfStrings ListOfStrings; + +#if 1 + while (!(*done)) + { + ListOfStrings l; + s->GetListOfJobs(l); + for (ListOfStrings::iterator i = l.begin(); i != l.end(); i++) + printf(">> %s: %0.1f\n", i->c_str(), 100.0f * s->GetProgress(*i)); + Toolbox::USleep(100000); + } +#else + ListOfStrings l; + s->GetListOfJobs(l); + for (ListOfStrings::iterator i = l.begin(); i != l.end(); i++) + printf(">> %s\n", i->c_str()); + Toolbox::USleep(1500000); + s->Cancel(*j); + Toolbox::USleep(1000000); + s->GetListOfJobs(l); + for (ListOfStrings::iterator i = l.begin(); i != l.end(); i++) + printf(">> %s\n", i->c_str()); +#endif +} + + +TEST(Toto, Toto) +{ + ServerScheduler scheduler; + + ServerJob job; + ServerFilterInstance& f2 = job.AddFilter(new Tutu(2)); + ServerFilterInstance& f3 = job.AddFilter(new Tutu(3)); + ServerFilterInstance& f4 = job.AddFilter(new Tutu(4)); + ServerFilterInstance& f5 = job.AddFilter(new Tutu(5)); + f2.AddInput(boost::lexical_cast<std::string>(42)); + //f3.AddInput(boost::lexical_cast<std::string>(42)); + //f4.AddInput(boost::lexical_cast<std::string>(42)); + f2.ConnectNext(f3); + f3.ConnectNext(f4); + f4.ConnectNext(f5); + + job.SetDescription("tutu"); + + bool done = false; + boost::thread t(Tata, &scheduler, &job, &done); + + + //scheduler.Submit(job); + + IServerFilter::ListOfStrings l; + scheduler.SubmitAndWait(l, job); + + for (IServerFilter::ListOfStrings::iterator i = l.begin(); i != l.end(); i++) + { + printf("** %s\n", i->c_str()); + } + + //Toolbox::ServerBarrier(); + //Toolbox::USleep(3000000); + + done = true; + t.join(); +}