Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
s4u::Exec now has 2 child classes
[simgrid.git] / src / s4u / s4u_Exec.cpp
1 /* Copyright (c) 2006-2019. The SimGrid Team. All rights reserved.          */
2
3 /* This program is free software; you can redistribute it and/or modify it
4  * under the terms of the license (GNU LGPL) which comes with this package. */
5
6 #include "simgrid/s4u/Actor.hpp"
7 #include "simgrid/s4u/Exec.hpp"
8 #include "src/kernel/activity/ExecImpl.hpp"
9 #include "xbt/log.h"
10
11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_exec, s4u_activity, "S4U asynchronous executions");
12
13 namespace simgrid {
14 namespace s4u {
15 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Exec::on_start;
16 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Exec::on_completion;
17
18 Exec::Exec()
19 {
20   pimpl_ = kernel::activity::ExecImplPtr(new kernel::activity::ExecImpl(name_, tracing_category_));
21 }
22
23 bool Exec::test()
24 {
25   xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::FINISHED);
26
27   if (state_ == State::FINISHED)
28     return true;
29
30   if (state_ == State::INITED)
31     this->start();
32
33   if (simcall_execution_test(pimpl_)) {
34     state_ = State::FINISHED;
35     return true;
36   }
37
38   return false;
39 }
40
41 Exec* Exec::wait()
42 {
43   if (state_ == State::INITED)
44     start();
45   simcall_execution_wait(pimpl_);
46   state_ = State::FINISHED;
47   on_completion(Actor::self());
48   return this;
49 }
50
51 Exec* Exec::wait_for(double)
52 {
53   THROW_UNIMPLEMENTED;
54 }
55
56 Exec* Exec::cancel()
57 {
58   simgrid::simix::simcall([this] { boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->cancel(); });
59   state_ = State::CANCELED;
60   return this;
61 }
62
63 void intrusive_ptr_release(simgrid::s4u::Exec* e)
64 {
65   if (e->refcount_.fetch_sub(1, std::memory_order_release) == 1) {
66     std::atomic_thread_fence(std::memory_order_acquire);
67     delete e;
68   }
69 }
70
71 void intrusive_ptr_add_ref(simgrid::s4u::Exec* e)
72 {
73   e->refcount_.fetch_add(1, std::memory_order_relaxed);
74 }
75
76 /** @brief change the execution bound
77  * This means changing the maximal amount of flops per second that it may consume, regardless of what the host may
78  * deliver. Currently, this cannot be changed once the exec started.
79  */
80 ExecPtr Exec::set_bound(double bound)
81 {
82   xbt_assert(state_ == State::INITED, "Cannot change the bound of an exec after its start");
83   bound_ = bound;
84   return this;
85 }
86 ExecPtr Exec::set_timeout(double timeout)
87 {
88   xbt_assert(state_ == State::INITED, "Cannot change the bound of an exec after its start");
89   timeout_ = timeout;
90   return this;
91 }
92
93 ExecPtr Exec::set_name(std::string name)
94 {
95   xbt_assert(state_ == State::INITED, "Cannot change the name of an exec after its start");
96   name_ = std::move(name);
97   return this;
98 }
99
100 /** @brief  Change the execution priority, don't you think?
101  *
102  * An execution with twice the priority will get twice the amount of flops when the resource is shared.
103  * The default priority is 1.
104  *
105  * Currently, this cannot be changed once the exec started. */
106 ExecPtr Exec::set_priority(double priority)
107 {
108   xbt_assert(state_ == State::INITED, "Cannot change the priority of an exec after its start");
109   priority_ = priority;
110   return this;
111 }
112
113 ExecPtr Exec::set_tracing_category(std::string category)
114 {
115   xbt_assert(state_ == State::INITED, "Cannot change the tracing category of an exec after its start");
116   tracing_category_ = std::move(category);
117   return this;
118 }
119
120 ///////////// SEQUENTIAL EXECUTIONS ////////
121 ExecSeq::ExecSeq(sg_host_t host, double flops_amount) : Exec(), flops_amount_(flops_amount)
122 {
123   Activity::set_remaining(flops_amount_);
124   boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->set_host(host);
125 }
126
127 Exec* ExecSeq::start()
128 {
129   simix::simcall([this] {
130     boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->start(flops_amount_, 1. / priority_, bound_);
131   });
132   state_ = State::STARTED;
133   on_start(Actor::self());
134   return this;
135 }
136
137 /** @brief Returns whether the state of the exec is finished */
138 /** @brief Change the host on which this activity takes place.
139  *
140  * The activity cannot be terminated already (but it may be started). */
141 ExecPtr ExecSeq::set_host(Host* host)
142 {
143   xbt_assert(state_ == State::INITED || state_ == State::STARTED,
144              "Cannot change the host of an exec once it's done (state: %d)", (int)state_);
145   if (state_ == State::STARTED)
146     boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->migrate(host);
147   host_ = host;
148   boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->host_ = host;
149   return this;
150 }
151
152
153 /** @brief Retrieve the host on which this activity takes place. */
154 Host* ExecSeq::get_host()
155 {
156   return host_;
157 }
158
159 /** @brief Returns the amount of flops that remain to be done */
160 double ExecSeq::get_remaining()
161 {
162   return simgrid::simix::simcall(
163       [this]() { return boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->get_remaining(); });
164 }
165
166 /** @brief Returns the ratio of elements that are still to do
167  *
168  * The returned value is between 0 (completely done) and 1 (nothing done yet).
169  */
170 double ExecSeq::get_remaining_ratio()
171 {
172   return simgrid::simix::simcall([this]() {
173     return boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->get_seq_remaining_ratio();
174   });
175 }
176
177 ///////////// PARALLEL EXECUTIONS ////////
178 ExecPar::ExecPar(const std::vector<s4u::Host*>& hosts, const std::vector<double>& flops_amounts,
179                  const std::vector<double>& bytes_amounts)
180     : Exec(), hosts_(hosts), flops_amounts_(flops_amounts), bytes_amounts_(bytes_amounts)
181 {
182   // For parallel executions, we need a special host to run the timeout detector.
183   host_                                                                          = hosts.front();
184   boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->host_ = host_;
185 }
186
187 Exec* ExecPar::start()
188 {
189   simix::simcall([this] {
190     boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->set_timeout(timeout_);
191     boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->start(hosts_, flops_amounts_, bytes_amounts_);
192   });
193   state_ = State::STARTED;
194   on_start(Actor::self());
195   return this;
196 }
197 double ExecPar::get_remaining_ratio()
198 {
199   return simgrid::simix::simcall([this]() {
200     return boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->get_par_remaining_ratio();
201   });
202 }
203
204 double ExecPar::get_remaining()
205 {
206   XBT_WARN("Calling get_remaining() on a parallel execution is not allowed. Call get_remaining_ratio() instead.");
207   return get_remaining_ratio();
208 }
209 } // namespace s4u
210 } // namespace simgrid