Changeset 3402

Timestamp:

Mar 31, 2015, 7:02:58 AM (8 years ago)

Author:

Peter

Message:

add interface to set priority of Scheduler::Job and make the order the tasks are executed closer to FIFO. refs #800

Location:

trunk

Files:

• test/Makefile.am (modified) (2 diffs)
• test/scheduler2.cc (added)
• yat/utility/Scheduler.cc (modified) (5 diffs)
• yat/utility/Scheduler.h (modified) (9 diffs)

trunk/test/Makefile.am

@@ -5 +5 @@
 # Copyright (C) 2005 Jari Häkkinen, Peter Johansson
 # Copyright (C) 2006, 2007, 2008 Jari Häkkinen, Peter Johansson, Markus Ringnér
-# Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014 Peter Johansson
+# Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014, 2015 Peter Johansson
 #
 # This file is part of the yat library, http://dev.thep.lu.se/yat
@@ -82 +82 @@
   test/score.test \
   test/scheduler.test \
+  test/scheduler2.test \
   test/segment.test test/smart_ptr.test \
   test/smith_waterman.test \

trunk/yat/utility/Scheduler.cc

@@ -26 +26 @@
 #include <cassert>
 
+#include <iostream>
+
 namespace theplu {
 namespace yat {
@@ -31 +33 @@
 
   Scheduler::Scheduler(unsigned int threads)
-    : running_jobs_(0)
+    : running_jobs_(0), job_counter_(0)
   {
     assert(threads);
@@ -69 +71 @@
       return;
     job->status_ = Job::prepared;
+    job->id_ = job_counter_;
+    ++job_counter_;
 
     // If we have prerequisite that need to be run first, process them
@@ -128 +132 @@
   // Scheduler::Job
 
-  Scheduler::Job::Job(void)
-    : status_(pristine) {}
+  Scheduler::Job::Job(unsigned int prio)
+    : status_(pristine), priority_(prio), id_(0) {}
 
 
@@ -136 +140 @@
 
 
+  unsigned int Scheduler::Job::priority(void) const
+  {
+    return priority_;
+  }
+
+
+  bool Scheduler::LowerPriority::operator()(const JobPtr& lhs,
+                                            const JobPtr& rhs) const
+  {
+    if (rhs.get() == NULL)
+      return false;
+    if (lhs.get() == NULL)
+      return true;
+
+    if (lhs->priority() != rhs->priority())
+      return lhs->priority() < rhs->priority();
+
+    return lhs->id_ > rhs->id_;
+  }
+
+
   // Scheduler::Worker
 
-  Scheduler::Worker::Worker(Queue<boost::shared_ptr<Job> >& q,
-                            Queue<boost::shared_ptr<Job> >& c)
+  Scheduler::Worker::Worker(JobQueue& q, Queue<JobPtr>& c)
     : queue_(q), completed_(c)
   {

trunk/yat/utility/Scheduler.h

@@ -23 +23 @@
 */
 
+#include "PriorityQueue.h"
 #include "Queue.h"
 
@@ -41 +42 @@
      series of Jobs. The Jobs can have dependencies, such that Job X
      must finish before Job Y can run, and the Scheduler takes care of
-     these dependencies.Here is a small code example in which two
+     these dependencies. Here is a small code example in which two
      threads are used to process the four jobs, \c MyJob. The jobs
      have a dependency that job4 can not run until the three first
@@ -57 +58 @@
      scheduler.add_dependency(job4, job3);
      scheduler.submit(job4);
-     scheduler.launch();
+     scheduler.wait();
 
      \endcode
+
+     The Scheduler sends jobs to the workers taking into account
+     dependencies, priorities and the order in which the Scheduler has
+     seen the Jobs. Of the Jobs that are ready to run, i.e., all jobs
+     it depends on have completed, the Scheduler chose the Job with
+     highest priority. If two Jobs have the same priority, the
+     Scheduler sends them in the order of appearance, i.e., if the
+     Scheduler saw Job X before Job Y, Job X is run before Job X.
+
+     \since New in yat 0.13
    */
   class Scheduler
@@ -73 +84 @@
          \brief constructor
        */
-      Job(void);
+      explicit Job(unsigned int prio=0);
 
       /**
@@ -79 +90 @@
        */
       virtual ~Job(void);
+
+      /**
+         Jobs with greater priority are run before Jobs with less priority.
+       */
+      unsigned int priority(void) const;
 
       /**
@@ -92 +108 @@
       enum status { pristine, prepared, running, completed};
       status status_;
+      unsigned priority_;
+      unsigned id_;
     }; // end class Job
 
@@ -124 +142 @@
 
   private:
+    typedef boost::shared_ptr<Scheduler::Job> JobPtr;
+
+    struct LowerPriority
+    {
+      bool operator()(const JobPtr& lhs, const JobPtr& rhs) const;
+    };
+
+    // some typedefs for convenience
+    typedef PriorityQueue<JobPtr, LowerPriority> JobQueue;
+
     // \internal class that does the job
     //
@@ -133 +161 @@
     {
     public:
-      Worker(Queue<boost::shared_ptr<Job> >& queue,
-             Queue<boost::shared_ptr<Job> >& completed);
+      Worker(JobQueue& queue, Queue<JobPtr>& completed);
       void operator()(void);
     private:
-      Queue<boost::shared_ptr<Job> >& queue_;
-      Queue<boost::shared_ptr<Job> >& completed_;
+      JobQueue& queue_;
+      Queue<JobPtr>& completed_;
     }; // end class Worker
 
@@ -155 +182 @@
     void queue(boost::shared_ptr<Job> job);
 
-    typedef boost::shared_ptr<Scheduler::Job> JobPtr;
     int running_jobs_;
 
-    Queue<boost::shared_ptr<Job> > queue_;
-    Queue<boost::shared_ptr<Job> > completed_;
+    JobQueue queue_;
+    Queue<JobPtr> completed_;
     boost::thread_group workers_;
-
+    unsigned long int job_counter_;
   }; // end class Scheduler