Changeset 3808

Timestamp:

Jul 5, 2019, 1:46:34 AM (4 years ago)

Author:

Peter

Message:

Move the data member of the JobHandler? into its own data class,
JobHandlerData?, which is thread safe. This class is now owned by the
Scheduler and the JobHandler? has access via a pointer. This fixes #924
as the class is thread safe and since the sharing between Scheduler
and the JobHandler? class is now more complete also allow future
additions of the interface such as mentions in tickets:

refs #899, #910, #914 and #916.

Location:

branches/0.16-stable

Files:

• test/Makefile.am (modified) (1 diff)
• test/scheduler5.cc (added)
• yat/utility/Scheduler.cc (modified) (13 diffs)
• yat/utility/Scheduler.h (modified) (6 diffs)

branches/0.16-stable/test/Makefile.am

@@ -102 +102 @@
   test/scheduler3.test \
   test/scheduler4.test \
+  test/scheduler5.test \
   test/segment.test test/smart_ptr.test \
   test/smith_waterman.test \

branches/0.16-stable/yat/utility/Scheduler.cc

@@ -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 (size_t 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();
   }

branches/0.16-stable/yat/utility/Scheduler.h

@@ -23 +23 @@
 */
 
-#include "config_public.h"
 #include "PriorityQueue.h"
 #include "Queue.h"
@@ -31 +30 @@
 #include <boost/shared_ptr.hpp>
 
-#ifdef YAT_HAVE_ATOMIC
-#include <atomic>
-#endif
 #include <set>
 #include <deque>
@@ -87 +83 @@
   class Scheduler
   {
-#ifdef YAT_HAVE_ATOMIC
-    typedef std::atomic<int> running_jobs_type;
-#else
-    typedef int running_jobs_type;
-#endif
   public:
     /**
@@ -218 +209 @@
     }; // end class Worker
 
+
+    class JobHandlerData
+    {
+    public:
+      /// thread-safe class around int
+      class Count
+      {
+      public:
+        /// Constructor
+        explicit Count(int x=0);
+        /// increase value with 1
+        void decrement(void);
+        /// return value
+        int get(void) const;
+        /// decrease value with 1
+        void increment(void);
+        /// modify value
+        void set(int x);
+      private:
+        mutable boost::mutex mutex_;
+        int x_;
+      };
+
+      JobHandlerData(unsigned int threads);
+      Queue<boost::exception_ptr>& error(void) const;
+
+      const Queue<JobPtr>& jobs(void) const;
+      Queue<JobPtr>& jobs(void);
+      const JobQueue& queue(void) const;
+      JobQueue& queue(void);
+
+      const Count& job_count(void) const;
+      Count& job_count(void);
+
+      const Count& running_jobs(void) const;
+      Count& running_jobs(void);
+
+      const Count& threads(void) const;
+      Count& threads(void);
+    private:
+      mutable Queue<boost::exception_ptr> error_;
+      Queue<JobPtr> jobs_;
+      JobQueue queue_;
+
+      Count job_count_;
+      Count running_jobs_;
+      Count threads_;
+    };
+
+
     // \internal Class that handles job
     class JobHandler
     {
     public:
-      JobHandler(unsigned int threads, JobQueue& queue, Queue<JobPtr>& jobs,
-                 running_jobs_type& running_jobs,
-                 Queue<boost::exception_ptr>& error);
+      JobHandler(JobHandlerData& data);
 
       void operator()(void);
@@ -237 +276 @@
 
       void send2queue(JobPtr& job);
-      unsigned int threads_;
-      JobQueue& queue_;
-      Queue<JobPtr>& jobs_;
-      running_jobs_type& running_jobs_;
-      Queue<boost::exception_ptr>& error_;
-      int job_counter_;
+
+      JobHandlerData* data_;
     };
 
@@ -263 +298 @@
     void throw_if_error(void) const;
 
-    JobQueue queue_;
-    Queue<JobPtr> jobs_;
-    running_jobs_type running_jobs_;
-    mutable Queue<boost::exception_ptr> error_;
+    JobHandlerData data_;
     boost::thread job_handler_;
   }; // end class Scheduler