Changeset 3401

Timestamp:

Mar 31, 2015, 2:56:34 AM (8 years ago)

Author:

Peter

Message:

refs #800

Avoid using parents and children terminologoy to describe Job
relationships as it was confusing. Move add_dependency function from
class Scheduler::Job to class Scheduler. Likewise move observers to
class Job rather than holding an extra map<Job, observers> in
Scheduler.

Location:

trunk

Files:

• test/scheduler.cc (modified) (1 diff)
• yat/utility/Scheduler.cc (modified) (4 diffs)
• yat/utility/Scheduler.h (modified) (6 diffs)

trunk/test/scheduler.cc

@@ -62 +62 @@
   scheduler.submit(sleeper[1]);
   sleeper.push_back(boost::make_shared<Sleeper>(2));
-  sleeper.back()->add_dependency(sleeper[0]);
+  scheduler.add_dependency(sleeper[2], sleeper[0]);
   sleeper.push_back(boost::make_shared<Sleeper>(3));
 
   // not strictly needed since sleeper2 depends on sleeper0,
   // but... this is a test
-  sleeper.back()->add_dependency(sleeper[0]);
-  sleeper.back()->add_dependency(sleeper[1]);
-  sleeper.back()->add_dependency(sleeper[2]);
+  scheduler.add_dependency(sleeper[3], sleeper[0]);
+  scheduler.add_dependency(sleeper[3], sleeper[1]);
+  scheduler.add_dependency(sleeper[3], sleeper[2]);
 
   scheduler.submit(sleeper[3]);

trunk/yat/utility/Scheduler.cc

@@ -39 +39 @@
 
 
+  void Scheduler::add_dependency(boost::shared_ptr<Job> job,
+                                 boost::shared_ptr<Job> prerequisite)
+  {
+    assert(job->status_ == Job::pristine);
+    job->prerequisite_.insert(prerequisite);
+    prerequisite->observers_.push_back(job);
+  }
+
+
   void Scheduler::post_process(JobPtr job)
   {
@@ -44 +53 @@
     assert(running_jobs_>=0);
     job->status_ = Job::completed;
-    // check if there are jobs waiting for job
-    std::map<JobPtr, std::vector<JobPtr> >::iterator it = children_.find(job);
-    if (it == children_.end())
-      return;
     // for convenience
-    const std::vector<JobPtr>& vec = it->second;
+    const std::vector<JobPtr>& vec = job->observers_;
+    // notify obervers, jobs that have 'job' as prerequisite
     for (size_t i=0; i<vec.size(); ++i) {
-      vec[i]->parents_.erase(job);
-      if (vec[i]->parents_.empty())
+      vec[i]->prerequisite_.erase(job);
+      if (vec[i]->prerequisite_.empty())
         queue(vec[i]);
     }
-
-    // entry no longer needed
-    children_.erase(it);
   }
 
@@ -67 +70 @@
     job->status_ = Job::prepared;
 
-    // If we have parents that need to be run first, well process them
+    // If we have prerequisite that need to be run first, process them
 
     typedef std::set<JobPtr>::iterator iterator;
-    for (iterator j = job->parents_.begin(); j!=job->parents_.end(); ) {
+    for (iterator j=job->prerequisite_.begin(); j!=job->prerequisite_.end();) {
       if ((*j)->status_ == Job::completed)
-        job->parents_.erase(j++);
+        job->prerequisite_.erase(j++);
       else {
         process(*j);
-        // record that job is a child of j, so we can notify job when
-        // j has completed
-        children_[*j].push_back(job);
         ++j;
       }
     }
 
-    // If job has no parents (or all parents are completed) send job to queue
-    if (job->parents_.empty()) {
+    // If all prerequisite are finished, send job to queue
+    if (job->prerequisite_.empty()) {
       queue(job);
     }
@@ -136 +136 @@
 
 
-  void Scheduler::Job::add_dependency(boost::shared_ptr<Job> job)
-  {
-    assert(status_==pristine);
-    parents_.insert(job);
-  }
-
-
   // Scheduler::Worker
 

trunk/yat/utility/Scheduler.h

@@ -50 +50 @@
      Scheduler scheduler(2);
      boost::shared_ptr<MyJob> job1(new MyJob("Hello"));
-     boost::shared_ptr<MyJob> job1(new MyJob(" "));
+     boost::shared_ptr<MyJob> job2(new MyJob(" "));
      boost::shared_ptr<MyJob> job3(new MyJob("World"));
      boost::shared_ptr<MyJob> job4(new MyJob("\n"));
-     job4.add_dependency(job1);
-     job4.add_dependency(job2);
-     job4.add_dependency(job3);
+     scheduler.add_dependency(job4, job1);
+     scheduler.add_dependency(job4, job2);
+     scheduler.add_dependency(job4, job3);
      scheduler.submit(job4);
      scheduler.launch();
@@ -81 +81 @@
 
       /**
-         \brief add a dependency
-
-         This job will not be processed until \a job has finished.
-       */
-      void add_dependency(boost::shared_ptr<Job> job);
-
-      /**
          This function defines the work done in thread.
        */
@@ -94 +87 @@
       friend class Scheduler;
       // set of jobs that have to finish before this can run
-      std::set<boost::shared_ptr<Job> > parents_;
+      std::set<boost::shared_ptr<Job> > prerequisite_;
+      // jobs that have *this as prerequisite
+      std::vector<boost::shared_ptr<Job> > observers_;
       enum status { pristine, prepared, running, completed};
       status status_;
@@ -105 +100 @@
      */
     Scheduler(unsigned int threads);
+
+    /**
+       \brief add a dependency rule
+
+       Add a dependency that Job \a prerequisite has to complete
+       before Job \a job is run.
+     */
+    void add_dependency(boost::shared_ptr<Job> job,
+                        boost::shared_ptr<Job> prerequisite);
 
     /**
@@ -138 +142 @@
 
     // function called when job has finished and returned from
-    // worker. If there are any jobs that depends on \a job those jobs
+    // worker. If there are any jobs that depend on \a job, those jobs
     // are notified and if it makes them ready to be processed they
     // are sent to queue.
@@ -154 +158 @@
     int running_jobs_;
 
-    // value lists all jobs that depend on key, i.e., key has to
-    // finish before jobs in value can start
-    std::map<JobPtr, std::vector<JobPtr> > children_;
-
     Queue<boost::shared_ptr<Job> > queue_;
     Queue<boost::shared_ptr<Job> > completed_;