orthanc: UnitTestsSources/MultiThreading.cpp comparison

comparison UnitTestsSources/MultiThreading.cpp @ 768:476febedc516 lua-scripting

improvements

author	Sebastien Jodogne <s.jodogne@gmail.com>
date	Wed, 23 Apr 2014 17:23:18 +0200
parents	fc97f762834c
children	a64ca424e0e2

comparison

equal deleted inserted replaced

-:fc97f762834c
+:476febedc516
 };
 class FilterWrapper : public IDynamicObject
 {
+friend class ServerScheduler;
 private:
-IServerFilterListener *listener_;
 IServerFilter *filter_;
 std::string jobId_;
 ListOfStrings inputs_;
 std::list<FilterWrapper*> next_;
-public:
+bool Execute(IServerFilterListener& listener)
-FilterWrapper(IServerFilter *filter,
-const std::string& jobId) :
-listener_(NULL),
-filter_(filter),
-jobId_(jobId)
-{
-if (filter_ == NULL)
-{
-throw OrthancException(ErrorCode_ParameterOutOfRange);
-}
-}
-virtual ~FilterWrapper()
-{
-if (filter_ != NULL)
-{
-delete filter_;
-}
-}
-void SetListener(IServerFilterListener& listener)
-{
-listener_ = &listener;
-}
-const std::string& GetJobId() const
-{
-return jobId_;
-}
-void AddInput(const std::string& input)
-{
-inputs_.push_back(input);
-}
-bool Execute()
 {
 ListOfStrings outputs;
 if (!filter_->Apply(outputs, inputs_))
 {
-if (listener_)
+listener.SignalFailure(jobId_);
-{
-listener_->SignalFailure(jobId_);
-}
 return true;
 }
 for (std::list<FilterWrapper*>::iterator
 it = next_.begin(); it != next_.end(); it++)
 {
 (*it)->AddInput(*output);
 }
 }
-if (listener_)
+listener.SignalSuccess(jobId_);
-{
-listener_->SignalSuccess(jobId_);
-}
 return true;
 }
+public:
+FilterWrapper(IServerFilter *filter,
+const std::string& jobId) :
+filter_(filter),
+jobId_(jobId)
+{
+if (filter_ == NULL)
+{
+throw OrthancException(ErrorCode_ParameterOutOfRange);
+}
+}
+virtual ~FilterWrapper()
+{
+if (filter_ != NULL)
+{
+delete filter_;
+}
+}
+const std::string& GetJobId() const
+{
+return jobId_;
+}
+void AddInput(const std::string& input)
+{
+inputs_.push_back(input);
+}
 void ConnectNext(FilterWrapper& filter)
 {
 next_.push_back(&filter);
 }
 const std::list<FilterWrapper*>& GetNextFilters() const
 {
 return next_;
 }
-};
-enum ServerJobStatus
-{
-ServerJobStatus_Running = 1,
-ServerJobStatus_Success = 2,
-ServerJobStatus_Failure = 3
 };
 class ServerJob
 {
 size_t size = filters_.size();
 for (std::list<FilterWrapper*>::iterator
 it = filters_.begin(); it != filters_.end(); it++)
 {
-(*it)->SetListener(listener);
 target.Enqueue(*it);
 }
 filters_.clear();
 submitted_ = true;
 size_t success_;
 size_t failures_;
 std::string description_;
 };
+enum JobStatus
+{
+JobStatus_Running = 1,
+JobStatus_Success = 2,
+JobStatus_Failure = 3
+};
 typedef std::map<std::string, JobInfo> Jobs;
 boost::mutex mutex_;
 boost::condition_variable jobFinished_;
 Jobs jobs_;
 SharedMessageQueue queue_;
 bool finish_;
 boost::thread worker_;
-std::map<std::string, ServerJobStatus> watchedJobStatus_;
+std::map<std::string, JobStatus> watchedJobStatus_;
 JobInfo& GetJobInfo(const std::string& jobId)
 {
 Jobs::iterator info = jobs_.find(jobId);
 if (info.success_ >= info.size_)
 {
 if (info.watched_)
 {
-watchedJobStatus_[jobId] = ServerJobStatus_Success;
+watchedJobStatus_[jobId] = JobStatus_Success;
 jobFinished_.notify_all();
 }
 LOG(INFO) << "Job successfully finished (" << info.description_ << ")";
 jobs_.erase(jobId);
 if (info.success_ + info.failures_ >= info.size_)
 {
 if (info.watched_)
 {
-watchedJobStatus_[jobId] = ServerJobStatus_Failure;
+watchedJobStatus_[jobId] = JobStatus_Failure;
 jobFinished_.notify_all();
 }
 LOG(ERROR) << "Job has failed (" << info.description_ << ")";
 jobs_.erase(jobId);
 {
 that->SignalFailure(filter.GetJobId());
 }
 else
 {
-filter.Execute();
+filter.Execute(*that);
 }
 }
 }
 }
 assert(info.size_ > 0);
 if (watched)
 {
-watchedJobStatus_[job.GetId()] = ServerJobStatus_Running;
+watchedJobStatus_[job.GetId()] = JobStatus_Running;
 }
 jobs_[job.GetId()] = info;
 LOG(INFO) << "New job submitted (" << job.description_ << ")";
 // Submit the job
 SubmitInternal(job, true);
 // Wait for the job to complete (either success or failure)
-ServerJobStatus status;
+JobStatus status;
 {
 boost::mutex::scoped_lock lock(mutex_);
+assert(watchedJobStatus_.find(jobId) != watchedJobStatus_.end());
-while (watchedJobStatus_[jobId] == ServerJobStatus_Running)
+while (watchedJobStatus_[jobId] == JobStatus_Running)
 {
 jobFinished_.wait(lock);
 }
 status = watchedJobStatus_[jobId];
 watchedJobStatus_.erase(jobId);
 }
-return (status == ServerJobStatus_Success);
+return (status == JobStatus_Success);
 }
 bool IsRunning(const std::string& jobId)
 {
 {
 // This job is not running
 return 1;
 }
-float n = static_cast<float>(job->second.failures_ + job->second.success_);
+if (job->second.failures_ != 0)
-float d = static_cast<float>(job->second.size_);
+{
-return n / d;
+return 1;
+}
+if (job->second.size_ == 1)
+{
+return job->second.success_;
+}
+return (static_cast<float>(job->second.success_) /
+static_cast<float>(job->second.size_ - 1));
 }
 bool IsRunning(const ServerJob& job)
 {
 return IsRunning(job.GetId());
 }
 float GetProgress(const ServerJob& job)
 {
 return GetProgress(job);
+}
+void GetListOfJobs(ListOfStrings& jobs)
+{
+boost::mutex::scoped_lock lock(mutex_);
+jobs.clear();
+for (Jobs::const_iterator
+it = jobs_.begin(); it != jobs_.end(); it++)
+{
+jobs.push_back(it->first);
+}
 }
 };
 }
 Toolbox::USleep(1000000);
 return true;
 }
 };
+static void Tata(ServerScheduler* s, ServerJob* j)
+{
+#if 1
+for (;;)
+{
+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;
 FilterWrapper& f2 = job.AddFilter(new Tutu(2));
 FilterWrapper& f3 = job.AddFilter(new Tutu(3));
 FilterWrapper& f4 = job.AddFilter(new Tutu(4));
+FilterWrapper& 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");
+boost::thread t(Tata, &scheduler, &job);
 //scheduler.Submit(job);
 ListOfStrings l;
 scheduler.SubmitAndWait(l, job);
 {
 printf("** %s\n", i->c_str());
 }
 //Toolbox::ServerBarrier();
-Toolbox::USleep(30000);
+//Toolbox::USleep(3000000);
-}
+//t.join();
+}

Mercurial > hg > orthanc

comparison UnitTestsSources/MultiThreading.cpp @ 768:476febedc516 lua-scripting