Changeset 3823 for trunk/yat/utility/Scheduler.cc

Timestamp:

Jul 16, 2019, 4:11:48 AM (4 years ago)

Author:

Peter

Message:

Merge release 0.16 into trunk

Location:

trunk

Files:

• . (modified) (1 prop)
• yat/utility/Scheduler.cc (modified) (14 diffs)

trunk/yat/utility/Scheduler.cc

@@ -2 +2 @@
 
 /*
-  Copyright (C) 2014, 2015, 2017 Peter Johansson
+  Copyright (C) 2014, 2015, 2017, 2019 Peter Johansson
 
   This file is part of the yat library, http://dev.thep.lu.se/yat
@@ -33 +33 @@
 
   Scheduler::Scheduler(unsigned int threads)
-    : running_jobs_(0),
-      job_handler_(JobHandler(threads, queue_, jobs_, running_jobs_, error_))
+    : data_(threads), job_handler_(JobHandler(data_))
   {
     assert(threads);
@@ -56 +55 @@
     // first signal to JobHandler that Scheduler is waiting
     boost::shared_ptr<Job> end;
-    jobs_.push(end);
+    data_.jobs().push(end);
 
     job_handler_.interrupt();
@@ -66 +65 @@
   {
     throw_if_error();
-    return running_jobs_ + queue_.size();
+    return data_.running_jobs().get() + data_.queue().size();
   }
 
@@ -73 +72 @@
   {
     throw_if_error();
-    jobs_.push(job);
+    data_.jobs().push(job);
   }
 
@@ -80 +79 @@
   {
     boost::exception_ptr error;
-    if (error_.try_pop(error))
+    if (data_.error().try_pop(error))
       boost::rethrow_exception(error);
   }
@@ -96 +95 @@
     // first signal to JobHandler that Scheduler is waiting
     boost::shared_ptr<Job> end;
-    jobs_.push(end);
+    data_.jobs().push(end);
 
     // wait for job handler to finish
@@ -197 +196 @@
 
 
+  // Scheduler::JobHandlerData
+  Scheduler::JobHandlerData::JobHandlerData(unsigned int threads)
+    : job_count_(0), running_jobs_(0), threads_(threads)
+  {}
+
+
+  Queue<boost::exception_ptr>&
+  Scheduler::JobHandlerData::error(void) const
+  {
+    return error_;
+  }
+
+
+  const Queue<Scheduler::JobPtr>& Scheduler::JobHandlerData::jobs(void) const
+  {
+    return jobs_;
+  }
+
+
+  Queue<Scheduler::JobPtr>& Scheduler::JobHandlerData::jobs(void)
+  {
+    return jobs_;
+  }
+
+
+  const Scheduler::JobQueue& Scheduler::JobHandlerData::queue(void) const
+  {
+    return queue_;
+  }
+
+
+  Scheduler::JobQueue& Scheduler::JobHandlerData::queue(void)
+  {
+    return queue_;
+  }
+
+
+  const Scheduler::JobHandlerData::Count&
+  Scheduler::JobHandlerData::job_count(void) const
+  {
+    return job_count_;
+  }
+
+
+  Scheduler::JobHandlerData::Count&
+  Scheduler::JobHandlerData::job_count(void)
+  {
+    return job_count_;
+  }
+
+
+  const Scheduler::JobHandlerData::Count&
+  Scheduler::JobHandlerData::running_jobs(void) const
+  {
+    return running_jobs_;
+  }
+
+
+  Scheduler::JobHandlerData::Count&
+  Scheduler::JobHandlerData::running_jobs(void)
+  {
+    return running_jobs_;
+  }
+
+
+  const Scheduler::JobHandlerData::Count&
+  Scheduler::JobHandlerData::threads(void) const
+  {
+    return threads_;
+  }
+
+
+  Scheduler::JobHandlerData::Count&
+  Scheduler::JobHandlerData::threads(void)
+  {
+    return threads_;
+  }
+
+
+  // Scheduler::JobHandlerData::Count
+  Scheduler::JobHandlerData::Count::Count(int x)
+    : x_(x)
+  {
+  }
+
+
+  void Scheduler::JobHandlerData::Count::decrement(void)
+  {
+    boost::unique_lock<boost::mutex> lock(mutex_);
+    --x_;
+  }
+
+
+  int Scheduler::JobHandlerData::Count::get(void) const
+  {
+    boost::unique_lock<boost::mutex> lock(mutex_);
+    return x_;
+  }
+
+
+  void Scheduler::JobHandlerData::Count::increment(void)
+  {
+    boost::unique_lock<boost::mutex> lock(mutex_);
+    ++x_;
+  }
+
+
+  void Scheduler::JobHandlerData::Count::set(int x)
+  {
+    boost::unique_lock<boost::mutex> lock(mutex_);
+    x_ = x;
+  }
+
+
   // Scheduler::JobHandler
 
-  Scheduler::JobHandler::JobHandler(unsigned int threads,
-                                    JobQueue& queue,
-                                    Queue<JobPtr>& jobs,
-                                    running_jobs_type& running_jobs,
-                                    Queue<boost::exception_ptr>& error)
-    : threads_(threads),
-      queue_(queue), jobs_(jobs), running_jobs_(running_jobs),
-      error_(error), job_counter_(0)
-  {
-  }
+  Scheduler::JobHandler::JobHandler(Scheduler::JobHandlerData& data)
+    : data_(&data)
+  {}
+
 
 
   void Scheduler::JobHandler::post_process(JobPtr job)
   {
-    --running_jobs_;
-    assert(running_jobs_>=0);
+    assert(job);
+    assert(data_);
+    data_->running_jobs().decrement();
+    assert(data_->running_jobs().get() >= 0);
     job->status_ = Job::completed;
 
     if (job->error_) {
-      error_.push(job->error_);
+      data_->error().push(job->error_);
       return;
     }
@@ -244 +353 @@
     assert(job->status_ == Job::pristine);
     job->status_ = Job::prepared;
-    job->id_ = job_counter_;
-    ++job_counter_;
+    job->id_ = data_->job_count().get();
+    data_->job_count().increment();
 
     // If job have prerequisite that need to be run first, process them
@@ -276 +385 @@
   {
     job->status_ = Job::running;
-    ++running_jobs_;
-    assert(running_jobs_>0);
-    queue_.push(job);
+    data_->running_jobs().increment();
+    assert(data_->running_jobs().get() > 0);
+    data_->queue().push(job);
   }
 
@@ -300 +409 @@
   void Scheduler::JobHandler::operator()(void)
   {
+    assert(data_);
     boost::thread_group workers;
-    for (size_t i=0; i<threads_; ++i)
-      workers.create_thread(Worker(queue_, jobs_));
+    for (int i=0; i < data_->threads().get(); ++i)
+      workers.create_thread(Worker(data_->queue(), data_->jobs()));
     // Process jobs (in jobs_) coming both from Scheduler and
     // completed jobs from Workers until Scheduler is waiting
@@ -312 +422 @@
     JobPtr job;
 
-    while (!scheduler_is_waiting || running_jobs_ || jobs_.size()) {
-      jobs_.pop(job);
+    while (!scheduler_is_waiting || data_->running_jobs().get() ||
+           !data_->jobs().empty()) {
+      data_->jobs().pop(job);
       if (job)
         process(job);
@@ -323 +434 @@
       // Since we are in a background thread, we cannot throw from
       // here, instead interrupt workers and return early.
-      if (!error_.empty()) {
+      if (!data_->error().empty()) {
         // In case queue is empty, workers might be be stuck
         // waiting for job queue to pop, then send them a poison pill so
         // they stop.
         boost::shared_ptr<Job> end;
-        queue_.push(end);
+        data_->queue().push(end);
         // For other cases (queue is not empty) send workers a
         // interrupt signal and wait for them to wrap up.
         workers.interrupt_all();
-        running_jobs_ = 0;
-        queue_.clear();
+        data_->running_jobs().set(0);
+        data_->queue().clear();
         return;
       }
@@ -343 +454 @@
     // kill workers
     boost::shared_ptr<Job> end;
-    queue_.push(end);
+    data_->queue().push(end);
     workers.join_all();
   }