# HG changeset patch # User Sebastien Jodogne # Date 1558517322 -7200 # Node ID a4bfb420b869864b45336a0f8d14b30ae2f3f1a7 # Parent bcd3ea868bcd7af0561ff8772418cfbe44a95c64 SleepOracleCommand diff -r bcd3ea868bcd -r a4bfb420b869 Samples/Sdl/Loader.cpp --- a/Samples/Sdl/Loader.cpp Wed May 22 09:55:01 2019 +0200 +++ b/Samples/Sdl/Loader.cpp Wed May 22 11:28:42 2019 +0200 @@ -69,6 +69,7 @@ public: enum Type { + Type_Sleep, Type_OrthancRestApi, Type_GetOrthancImage, Type_GetOrthancWebViewerJpeg @@ -153,6 +154,18 @@ throw Orthanc::OrthancException(Orthanc::ErrorCode_BadSequenceOfCalls); } } + + Orthanc::IDynamicObject* ReleasePayload() + { + if (HasPayload()) + { + return payload_.release(); + } + else + { + throw Orthanc::OrthancException(Orthanc::ErrorCode_BadSequenceOfCalls); + } + } }; @@ -190,6 +203,32 @@ return exception_; } }; + + + class SleepOracleCommand : public OracleCommandWithPayload + { + private: + unsigned int milliseconds_; + + public: + ORTHANC_STONE_DEFINE_ORIGIN_MESSAGE(__FILE__, __LINE__, TimeoutMessage, SleepOracleCommand); + + SleepOracleCommand(unsigned int milliseconds) : + milliseconds_(milliseconds) + { + } + + virtual Type GetType() const + { + return Type_Sleep; + } + + unsigned int GetDelay() const + { + return milliseconds_; + } + }; + typedef std::map HttpHeaders; @@ -1937,7 +1976,7 @@ - class NativeOracle : public IOracle + class ThreadedOracle : public IOracle { private: class Item : public Orthanc::IDynamicObject @@ -1963,7 +2002,7 @@ return receiver_; } - const IOracleCommand& GetCommand() const + IOracleCommand& GetCommand() { assert(command_.get() != NULL); return *command_; @@ -1979,13 +2018,108 @@ }; + class SleepingCommands : public boost::noncopyable + { + private: + class Item + { + private: + const OrthancStone::IObserver& receiver_; + std::auto_ptr command_; + boost::posix_time::ptime expiration_; + + public: + Item(const OrthancStone::IObserver& receiver, + SleepOracleCommand* command) : + receiver_(receiver), + command_(command) + { + if (command == NULL) + { + throw Orthanc::OrthancException(Orthanc::ErrorCode_NullPointer); + } + + expiration_ = (boost::posix_time::second_clock::local_time() + + boost::posix_time::milliseconds(command_->GetDelay())); + } + + const boost::posix_time::ptime& GetExpirationTime() const + { + return expiration_; + } + + void Awake(IMessageEmitter& emitter) + { + assert(command_.get() != NULL); + + SleepOracleCommand::TimeoutMessage message(*command_); + emitter.EmitMessage(receiver_, message); + } + }; + + typedef std::list Content; + + boost::mutex mutex_; + Content content_; + + public: + ~SleepingCommands() + { + for (Content::iterator it = content_.begin(); it != content_.end(); ++it) + { + if (*it != NULL) + { + delete *it; + } + } + } + + void Add(const OrthancStone::IObserver& receiver, + SleepOracleCommand* command) // Takes ownership + { + boost::mutex::scoped_lock lock(mutex_); + + content_.push_back(new Item(receiver, command)); + } + + void AwakeExpired(IMessageEmitter& emitter) + { + boost::mutex::scoped_lock lock(mutex_); + + const boost::posix_time::ptime now = boost::posix_time::second_clock::local_time(); + + Content stillSleeping; + + for (Content::iterator it = content_.begin(); it != content_.end(); ++it) + { + if (*it != NULL && + (*it)->GetExpirationTime() <= now) + { + (*it)->Awake(emitter); + delete *it; + *it = NULL; + } + else + { + stillSleeping.push_back(*it); + } + } + + // Compact the still-sleeping commands + content_ = stillSleeping; + } + }; + + IMessageEmitter& emitter_; Orthanc::WebServiceParameters orthanc_; Orthanc::SharedMessageQueue queue_; State state_; boost::mutex mutex_; std::vector workers_; - + SleepingCommands sleepingCommands_; + boost::thread sleepingWorker_; + unsigned int sleepingTimeResolution_; void CopyHttpHeaders(Orthanc::HttpClient& client, const HttpHeaders& headers) @@ -2036,6 +2170,20 @@ void Execute(const OrthancStone::IObserver& receiver, + SleepOracleCommand& command) + { + std::auto_ptr copy(new SleepOracleCommand(command.GetDelay())); + + if (command.HasPayload()) + { + copy->SetPayload(command.ReleasePayload()); + } + + sleepingCommands_.Add(receiver, copy.release()); + } + + + void Execute(const OrthancStone::IObserver& receiver, const OrthancRestApiCommand& command) { Orthanc::HttpClient client(orthanc_, command.GetUri()); @@ -2103,12 +2251,17 @@ if (object.get() != NULL) { - const Item& item = dynamic_cast(*object); + Item& item = dynamic_cast(*object); try { switch (item.GetCommand().GetType()) { + case IOracleCommand::Type_Sleep: + Execute(item.GetReceiver(), + dynamic_cast(item.GetCommand())); + break; + case IOracleCommand::Type_OrthancRestApi: Execute(item.GetReceiver(), dynamic_cast(item.GetCommand())); @@ -2143,7 +2296,7 @@ } - static void Worker(NativeOracle* that) + static void Worker(ThreadedOracle* that) { assert(that != NULL); @@ -2162,6 +2315,27 @@ } + static void SleepingWorker(ThreadedOracle* that) + { + assert(that != NULL); + + for (;;) + { + { + boost::mutex::scoped_lock lock(that->mutex_); + if (that->state_ != State_Running) + { + return; + } + } + + that->sleepingCommands_.AwakeExpired(that->emitter_); + + boost::this_thread::sleep(boost::posix_time::milliseconds(that->sleepingTimeResolution_)); + } + } + + void StopInternal() { { @@ -2178,6 +2352,11 @@ } } + if (sleepingWorker_.joinable()) + { + sleepingWorker_.join(); + } + for (size_t i = 0; i < workers_.size(); i++) { if (workers_[i] != NULL) @@ -2194,14 +2373,15 @@ public: - NativeOracle(IMessageEmitter& emitter) : + ThreadedOracle(IMessageEmitter& emitter) : emitter_(emitter), state_(State_Setup), - workers_(4) + workers_(4), + sleepingTimeResolution_(50) // By default, time resolution of 50ms { } - virtual ~NativeOracle() + virtual ~ThreadedOracle() { StopInternal(); } @@ -2238,6 +2418,24 @@ } } + void SetSleepingTimeResolution(unsigned int milliseconds) + { + boost::mutex::scoped_lock lock(mutex_); + + if (milliseconds <= 0) + { + throw Orthanc::OrthancException(Orthanc::ErrorCode_ParameterOutOfRange); + } + else if (state_ != State_Setup) + { + throw Orthanc::OrthancException(Orthanc::ErrorCode_BadSequenceOfCalls); + } + else + { + sleepingTimeResolution_ = milliseconds; + } + } + void Start() { boost::mutex::scoped_lock lock(mutex_); @@ -2254,6 +2452,8 @@ { workers_[i] = new boost::thread(Worker, this); } + + sleepingWorker_ = boost::thread(SleepingWorker, this); } } @@ -2345,6 +2545,11 @@ class Toto : public OrthancStone::IObserver { private: + void Handle(const Refactoring::SleepOracleCommand::TimeoutMessage& message) + { + printf("TIMEOUT! %d\n", dynamic_cast& >(message.GetOrigin().GetPayload()).GetValue()); + } + void Handle(const Refactoring::OrthancRestApiCommand::SuccessMessage& message) { Json::Value v; @@ -2385,6 +2590,10 @@ { oracle.RegisterObserverCallback (new OrthancStone::Callable + (*this, &Toto::Handle)); + + oracle.RegisterObserverCallback + (new OrthancStone::Callable (*this, &Toto::Handle)); oracle.RegisterObserverCallback @@ -2415,7 +2624,7 @@ loader2.reset(new Refactoring::VolumeSeriesOrthancLoader(oracle, lock.GetOracleObservable())); } - if (1) + if (0) { Json::Value v = Json::objectValue; v["Level"] = "Series"; @@ -2429,7 +2638,7 @@ oracle.Schedule(*toto, command.release()); } - if (1) + if (0) { std::auto_ptr command(new Refactoring::GetOrthancImageCommand); command->SetHttpHeader("Accept", std::string(Orthanc::EnumerationToString(Orthanc::MimeType_Jpeg))); @@ -2437,7 +2646,7 @@ oracle.Schedule(*toto, command.release()); } - if (1) + if (0) { std::auto_ptr command(new Refactoring::GetOrthancImageCommand); command->SetHttpHeader("Accept", std::string(Orthanc::EnumerationToString(Orthanc::MimeType_Png))); @@ -2445,7 +2654,7 @@ oracle.Schedule(*toto, command.release()); } - if (1) + if (0) { std::auto_ptr command(new Refactoring::GetOrthancImageCommand); command->SetHttpHeader("Accept", std::string(Orthanc::EnumerationToString(Orthanc::MimeType_Png))); @@ -2453,7 +2662,7 @@ oracle.Schedule(*toto, command.release()); } - if (1) + if (0) { std::auto_ptr command(new Refactoring::GetOrthancImageCommand); command->SetHttpHeader("Accept-Encoding", "gzip"); @@ -2462,7 +2671,7 @@ oracle.Schedule(*toto, command.release()); } - if (1) + if (0) { std::auto_ptr command(new Refactoring::GetOrthancImageCommand); command->SetHttpHeader("Accept", std::string(Orthanc::EnumerationToString(Orthanc::MimeType_Pam))); @@ -2470,7 +2679,7 @@ oracle.Schedule(*toto, command.release()); } - if (1) + if (0) { std::auto_ptr command(new Refactoring::GetOrthancWebViewerJpegCommand); command->SetHttpHeader("Accept-Encoding", "gzip"); @@ -2480,12 +2689,23 @@ } + if (0) + { + for (unsigned int i = 0; i < 10; i++) + { + std::auto_ptr command(new Refactoring::SleepOracleCommand(i * 1000)); + command->SetPayload(new Orthanc::SingleValueObject(42 * i)); + oracle.Schedule(*toto, command.release()); + } + } + // 2017-11-17-Anonymized //loader1->LoadSeries("cb3ea4d1-d08f3856-ad7b6314-74d88d77-60b05618"); // CT //loader2->LoadInstance("41029085-71718346-811efac4-420e2c15-d39f99b6"); // RT-DOSE // Delphine - loader1->LoadSeries("5990e39c-51e5f201-fe87a54c-31a55943-e59ef80e"); // CT + //loader1->LoadSeries("5990e39c-51e5f201-fe87a54c-31a55943-e59ef80e"); // CT + loader1->LoadSeries("67f1b334-02c16752-45026e40-a5b60b6b-030ecab5"); // Lung 1/10mm LOG(WARNING) << "...Waiting for Ctrl-C..."; Orthanc::SystemToolbox::ServerBarrier(); @@ -2508,7 +2728,7 @@ { Refactoring::NativeApplicationContext context; - Refactoring::NativeOracle oracle(context); + Refactoring::ThreadedOracle oracle(context); { Orthanc::WebServiceParameters p;