include/simgrid/s4u/Task.hpp

   1 #ifndef SIMGRID_S4U_TASK_H_
   2 #define SIMGRID_S4U_TASK_H_
   3
   4 #include <simgrid/s4u/Activity.hpp>
   5 #include <simgrid/s4u/Io.hpp>
   6 #include <xbt/Extendable.hpp>
   7
   8 #include <atomic>
   9 #include <deque>
  10 #include <map>
  11 #include <memory>
  12 #include <set>
  13
  14 namespace simgrid::s4u {
  15
  16 class Task;
  17 using TaskPtr = boost::intrusive_ptr<Task>;
  18 XBT_PUBLIC void intrusive_ptr_release(Task* o);
  19 XBT_PUBLIC void intrusive_ptr_add_ref(Task* o);
  20 class ExecTask;
  21 using ExecTaskPtr = boost::intrusive_ptr<ExecTask>;
  22 class CommTask;
  23 using CommTaskPtr = boost::intrusive_ptr<CommTask>;
  24 class IoTask;
  25 using IoTaskPtr = boost::intrusive_ptr<IoTask>;
  26
  27 class XBT_PUBLIC Token : public xbt::Extendable<Token> {};
  28
  29 class Task {
  30   std::set<Task*> successors_                 = {};
  31   std::map<Task*, unsigned int> predecessors_ = {};
  32
  33   bool ready_to_run() const;
  34   void receive(Task* source);
  35
  36 protected:
  37   std::string name_;
  38   double amount_;
  39   std::shared_ptr<Token> token_ = nullptr;
  40   std::deque<std::map<TaskPtr, std::shared_ptr<Token>>> tokens_received_;
  41   int queued_execs_ = 0;
  42   int count_        = 0;
  43   bool working_     = false;
  44   s4u::ActivityPtr previous_activity_;
  45   s4u::ActivityPtr current_activity_;
  46   xbt::signal<void(Task*)> on_this_start_;
  47   xbt::signal<void(Task*)> on_this_end_;
  48   explicit Task(const std::string& name);
  49   virtual ~Task()     = default;
  50   virtual void fire() = 0;
  51   void complete();
  52
  53   static xbt::signal<void(Task*)> on_start;
  54   static xbt::signal<void(Task*)> on_end;
  55   std::atomic_int_fast32_t refcount_{0};
  56
  57 public:
  58   const std::string& get_name() const { return name_; }
  59   const char* get_cname() const { return name_.c_str(); }
  60   void enqueue_execs(int n);
  61   void set_amount(double amount);
  62   double get_amount() const { return amount_; }
  63   void set_token(std::shared_ptr<Token> token);
  64   std::shared_ptr<Token> get_next_token_from(TaskPtr t);
  65   void add_successor(TaskPtr t);
  66   void remove_successor(TaskPtr t);
  67   void remove_all_successors();
  68   const std::set<Task*>& get_successors() const { return successors_; }
  69   void on_this_start_cb(const std::function<void(Task*)>& func);
  70   void on_this_end_cb(const std::function<void(Task*)>& func);
  71   int get_count() const;
  72
  73   /** Add a callback fired before a task activity start.
  74    * Triggered after the on_this_start function**/
  75   static void on_start_cb(const std::function<void(Task*)>& cb) { on_start.connect(cb); }
  76   /** Add a callback fired after a task activity end.
  77    * Triggered after the on_this_end function, but before
  78    * sending tokens to successors.**/
  79   static void on_end_cb(const std::function<void(Task*)>& cb) { on_end.connect(cb); }
  80
  81 #ifndef DOXYGEN
  82   friend void intrusive_ptr_release(Task* o)
  83   {
  84     if (o->refcount_.fetch_sub(1, std::memory_order_release) == 1) {
  85       std::atomic_thread_fence(std::memory_order_acquire);
  86       delete o;
  87     }
  88   }
  89   friend void intrusive_ptr_add_ref(Task* o) { o->refcount_.fetch_add(1, std::memory_order_relaxed); }
  90 #endif
  91 };
  92
  93 class ExecTask : public Task {
  94   s4u::Host* host_;
  95
  96   explicit ExecTask(const std::string& name);
  97   void fire() override;
  98
  99 public:
 100   static ExecTaskPtr init(const std::string& name);
 101   static ExecTaskPtr init(const std::string& name, double flops, s4u::Host* host);
 102   ExecTaskPtr set_host(s4u::Host* host);
 103   s4u::Host* get_host() const { return host_; }
 104   ExecTaskPtr set_flops(double flops);
 105   double get_flops() const { return get_amount(); }
 106 };
 107
 108 class CommTask : public Task {
 109   s4u::Host* source_;
 110   s4u::Host* destination_;
 111
 112   explicit CommTask(const std::string& name);
 113   void fire() override;
 114
 115 public:
 116   static CommTaskPtr init(const std::string& name);
 117   static CommTaskPtr init(const std::string& name, double bytes, s4u::Host* source, s4u::Host* destination);
 118   CommTaskPtr set_source(s4u::Host* source);
 119   s4u::Host* get_source() const { return source_; }
 120   CommTaskPtr set_destination(s4u::Host* destination);
 121   s4u::Host* get_destination() const { return destination_; }
 122   CommTaskPtr set_bytes(double bytes);
 123   double get_bytes() const { return get_amount(); }
 124 };
 125
 126 class IoTask : public Task {
 127   s4u::Disk* disk_;
 128   s4u::Io::OpType type_;
 129   explicit IoTask(const std::string& name);
 130   void fire() override;
 131
 132 public:
 133   static IoTaskPtr init(const std::string& name);
 134   static IoTaskPtr init(const std::string& name, double bytes, s4u::Disk* disk, s4u::Io::OpType type);
 135   IoTaskPtr set_disk(s4u::Disk* disk);
 136   s4u::Disk* get_disk() const { return disk_; }
 137   IoTaskPtr set_bytes(double bytes);
 138   double get_bytes() const { return get_amount(); }
 139   IoTaskPtr set_op_type(s4u::Io::OpType type);
 140   s4u::Io::OpType get_op_type() const { return type_; }
 141 };
 142 } // namespace simgrid::s4u
 143 #endif