Changeset 753ea6d in nscp for modules/Scheduler/simple_scheduler.hpp

Timestamp:

01/07/10 08:10:43 (3 years ago)

Author:

Michael Medin <michael@…>

Branches:

master, 0.4.0, 0.4.1, 0.4.2

Children:

cc2efd6

Parents:

031acbf

Message:

improved the scheduled module and re-added the settings command line toy (so now you can migrate settings again)
Schedule module is 80% finished but notification channel subsystem is still missing.

File:

• modules/Scheduler/simple_scheduler.hpp (modified) (6 diffs)

modules/Scheduler/simple_scheduler.hpp

@@ -8 +8 @@
 #include <boost/thread.hpp>
 #include <boost/date_time/local_time/local_time.hpp>
+#include <boost/optional.hpp>
 
 namespace scheduler {
 
+  class task_not_found {
+    int id_;
+  public:
+    task_not_found(int id) : id_(id) {}
+    int get_id() {return id_; }
+  };
   class target {
   public:
@@ -18 +25 @@
     std::wstring tag;
     boost::posix_time::time_duration duration;
+    std::wstring  channel;
+    //std::wstring duration;
 
-    target() {}
-    target(const target &other) : id(other.id), command(other.command), duration(other.duration) {}
+
+    void set_duration(boost::posix_time::time_duration duration_) {
+      duration = duration_;
+      // TODO!
+    }
+
+    target() : duration(boost::posix_time::minutes(0))
+    {}
+    
+    target(const target &other) : id(other.id), command(other.command), duration(other.duration) {}
     target& operator=(target const& other) {
       command = other.command;
@@ -27 +44 @@
       return *this;
     }
-    static target empty() {
-      return target();
-    }
+    ~target() {}
   };
   struct schedule_instance {
@@ -39 +54 @@
   };
 
+  template<typename T>
+  class safe_schedule_queue {
+  public:
+    typedef boost::optional<T> value_type;
+  private:
+    typedef std::priority_queue<T> schedule_queue_type;
+    schedule_queue_type queue_;
+    boost::shared_mutex mutex_;
+  public:
+    bool empty(unsigned int timeout = 5) {
+      boost::shared_lock<boost::shared_mutex> lock(mutex_, boost::get_system_time() + boost::posix_time::seconds(5));
+      if (!lock.owns_lock()) 
+        return false;
+      return queue_.empty();
+    }
+
+    boost::optional<T> top(unsigned int timeout = 5) {
+      boost::shared_lock<boost::shared_mutex> lock(mutex_, boost::get_system_time() + boost::posix_time::seconds(5));
+      if (!lock || queue_.empty())
+        return boost::optional<T>();
+      return boost::optional<T>(queue_.top());
+    }
+
+    boost::optional<T> pop(unsigned int timeout = 5) {
+      boost::unique_lock<boost::shared_mutex> lock(mutex_, boost::get_system_time() + boost::posix_time::seconds(timeout));
+      if (!lock || queue_.empty())
+        return boost::optional<T>();
+      boost::optional<T>  ret = queue_.top();
+      queue_.pop();
+      return ret;
+    }
+
+    bool push(T instance, unsigned int timeout = 5) {
+      boost::unique_lock<boost::shared_mutex> lock(mutex_, boost::get_system_time() + boost::posix_time::seconds(timeout));
+      if (!lock) {
+        return false;
+      }
+      queue_.push(instance);
+      return true;
+    }
+  };
+
   class simple_scheduler {
   private:
     typedef std::map<int,target> target_list_type;
-    typedef std::priority_queue<schedule_instance> schedule_queue_type;
+    typedef safe_schedule_queue<schedule_instance> schedule_queue_type;
     target_list_type targets_;
     unsigned int target_id_;
     schedule_queue_type queue_;
+    unsigned int thread_count_;
 
 
     // thread variables
     volatile bool stop_requested_;
-    boost::shared_ptr<boost::thread> thread_;
+    volatile bool running_;
+    boost::thread_group threads_;
+    //boost::shared_ptr<boost::thread> thread_;
     boost::mutex mutex_;
 
   public:
 
-    simple_scheduler() : target_id_(0) {}
+    simple_scheduler() : target_id_(0), stop_requested_(false), running_(false), thread_count_(10) {}
     ~simple_scheduler() {}
 
@@ -61 +121 @@
     int add_task(target item);
     void remove_task(int id);
-    target get_task(int id);
+    boost::optional<target> get_task(int id);
     
     void start();
     void stop();
+
+    void set_threads(int threads) {
+      thread_count_ = threads;
+      start_thread();
+    }
 
 
@@ -72 +137 @@
     void reschedule(target item);
     void reschedule(target item, boost::posix_time::ptime now);
+    void reschedule_wnext(target item, boost::posix_time::ptime next);
     void execute(target item);
     void start_thread();
 
-    boost::posix_time::ptime now() {
+    inline boost::posix_time::ptime now() {
       return boost::get_system_time();
     }