Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Enable access to Exec from ExecImpl and fix get_finish_time()
[simgrid.git] / src / s4u / s4u_Exec.cpp
1 /* Copyright (c) 2006-2021. 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/Exception.hpp"
7 #include "simgrid/exec.h"
8 #include "simgrid/s4u/Actor.hpp"
9 #include "simgrid/s4u/Exec.hpp"
10 #include "src/kernel/activity/ExecImpl.hpp"
11 #include "xbt/log.h"
12
13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_exec, s4u_activity, "S4U asynchronous executions");
14
15 namespace simgrid {
16 namespace s4u {
17 xbt::signal<void(Exec const&)> Exec::on_start;
18 xbt::signal<void(Exec const&)> Exec::on_completion;
19
20 Exec::Exec(kernel::activity::ExecImplPtr pimpl)
21 {
22   pimpl_ = pimpl;
23 }
24
25 ExecPtr Exec::init()
26 {
27   auto pimpl = kernel::activity::ExecImplPtr(new kernel::activity::ExecImpl());
28   return ExecPtr(pimpl->get_iface());
29 }
30
31 Exec* Exec::wait()
32 {
33   return this->wait_for(-1);
34 }
35
36 Exec* Exec::wait_for(double timeout)
37 {
38   if (state_ == State::INITED)
39     vetoable_start();
40
41   kernel::actor::ActorImpl* issuer = Actor::self()->get_impl();
42   kernel::actor::simcall_blocking<void>([this, issuer, timeout] { this->get_impl()->wait_for(issuer, timeout); });
43   state_ = State::FINISHED;
44   on_completion(*this);
45   this->release_dependencies();
46   return this;
47 }
48
49 int Exec::wait_any_for(std::vector<ExecPtr>* execs, double timeout)
50 {
51   std::vector<kernel::activity::ExecImpl*> rexecs(execs->size());
52   std::transform(begin(*execs), end(*execs), begin(rexecs),
53                  [](const ExecPtr& exec) { return static_cast<kernel::activity::ExecImpl*>(exec->pimpl_.get()); });
54
55   int changed_pos = simcall_execution_waitany_for(rexecs.data(), rexecs.size(), timeout);
56   if (changed_pos != -1)
57     execs->at(changed_pos)->release_dependencies();
58   return changed_pos;
59 }
60
61 Exec* Exec::cancel()
62 {
63   kernel::actor::simcall([this] { boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->cancel(); });
64   state_ = State::CANCELED;
65   return this;
66 }
67
68 /** @brief change the execution bound
69  * This means changing the maximal amount of flops per second that it may consume, regardless of what the host may
70  * deliver. Currently, this cannot be changed once the exec started.
71  */
72 ExecPtr Exec::set_bound(double bound)
73 {
74   xbt_assert(state_ == State::INITED, "Cannot change the bound of an exec after its start");
75   bound_ = bound;
76   return this;
77 }
78 ExecPtr Exec::set_timeout(double timeout) // XBT_ATTRIB_DEPRECATED_v329
79 {
80   xbt_assert(state_ == State::INITED, "Cannot change the bound of an exec after its start");
81   timeout_ = timeout;
82   return this;
83 }
84
85 ExecPtr Exec::set_flops_amount(double flops_amount)
86 {
87   xbt_assert(state_ == State::INITED, "Cannot change the flop_amount of an exec after its start");
88   flops_amounts_.assign(1, flops_amount);
89   Activity::set_remaining(flops_amounts_.front());
90   return this;
91 }
92
93 ExecPtr Exec::set_flops_amounts(const std::vector<double>& flops_amounts)
94 {
95   xbt_assert(state_ == State::INITED, "Cannot change the flops_amounts of an exec after its start");
96   flops_amounts_ = flops_amounts;
97   parallel_      = true;
98   return this;
99 }
100
101 ExecPtr Exec::set_bytes_amounts(const std::vector<double>& bytes_amounts)
102 {
103   xbt_assert(state_ == State::INITED, "Cannot change the bytes_amounts of an exec after its start");
104   bytes_amounts_ = bytes_amounts;
105   parallel_      = true;
106   return this;
107 }
108
109 /** @brief Retrieve the host on which this activity takes place.
110  *  If it runs on more than one host, only the first host is returned.
111  */
112 Host* Exec::get_host() const
113 {
114   return static_cast<kernel::activity::ExecImpl*>(pimpl_.get())->get_host();
115 }
116 unsigned int Exec::get_host_number() const
117 {
118   return static_cast<kernel::activity::ExecImpl*>(pimpl_.get())->get_host_number();
119 }
120 double Exec::get_cost() const
121 {
122   return (pimpl_->surf_action_ == nullptr) ? -1 : pimpl_->surf_action_->get_cost();
123 }
124
125 /** @brief  Change the execution priority, don't you think?
126  *
127  * An execution with twice the priority will get twice the amount of flops when the resource is shared.
128  * The default priority is 1.
129  *
130  * Currently, this cannot be changed once the exec started. */
131 ExecPtr Exec::set_priority(double priority)
132 {
133   xbt_assert(state_ == State::INITED, "Cannot change the priority of an exec after its start");
134   priority_ = priority;
135   return this;
136 }
137
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 Exec::set_host(Host* host)
142 {
143   xbt_assert(state_ == State::INITED || state_ == State::STARTING || state_ == State::STARTED,
144              "Cannot change the host of an exec once it's done (state: %d)", (int)state_);
145   hosts_.assign(1, host);
146
147   if (state_ == State::STARTED)
148     boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->migrate(host);
149
150   boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->set_host(host);
151
152   if (state_ == State::STARTING)
153   // Setting the host may allow to start the activity, let's try
154     vetoable_start();
155
156   return this;
157 }
158
159 ExecPtr Exec::set_hosts(const std::vector<Host*>& hosts)
160 {
161   xbt_assert(state_ == State::INITED || state_ == State::STARTING,
162       "Cannot change the hosts of an exec once it's done (state: %d)", (int)state_);
163
164   hosts_    = hosts;
165   parallel_ = true;
166
167   // Setting the host may allow to start the activity, let's try
168   if (state_ == State::STARTING)
169      vetoable_start();
170
171   return this;
172 }
173
174 ///////////// SEQUENTIAL EXECUTIONS ////////
175 Exec* Exec::start()
176 {
177   if (is_parallel())
178     kernel::actor::simcall([this] {
179       (*boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_))
180           .set_hosts(hosts_)
181           .set_timeout(timeout_)
182           .set_flops_amounts(flops_amounts_)
183           .set_bytes_amounts(bytes_amounts_)
184           .start();
185     });
186   else
187     kernel::actor::simcall([this] {
188       (*boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_))
189           .set_name(get_name())
190           .set_tracing_category(get_tracing_category())
191           .set_sharing_penalty(1. / priority_)
192           .set_bound(bound_)
193           .set_flops_amount(flops_amounts_.front())
194           .start();
195     });
196
197   if (suspended_)
198     pimpl_->suspend();
199
200   state_ = State::STARTED;
201   start_time_ = pimpl_->surf_action_->get_start_time();
202   on_start(*this);
203   return this;
204 }
205
206 double Exec::get_remaining() const
207 {
208   if (is_parallel()) {
209     XBT_WARN("Calling get_remaining() on a parallel execution is not allowed. Call get_remaining_ratio() instead.");
210     return get_remaining_ratio();
211   } else
212     return kernel::actor::simcall(
213         [this]() { return boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->get_remaining(); });
214 }
215
216 /** @brief Returns the ratio of elements that are still to do
217  *
218  * The returned value is between 0 (completely done) and 1 (nothing done yet).
219  */
220 double Exec::get_remaining_ratio() const
221 {
222   if (is_parallel())
223     return kernel::actor::simcall(
224         [this]() { return boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->get_par_remaining_ratio(); });
225   else
226     return kernel::actor::simcall(
227         [this]() { return boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->get_seq_remaining_ratio(); });
228 }
229
230 } // namespace s4u
231 } // namespace simgrid
232 /* **************************** Public C interface *************************** */
233 void sg_exec_set_bound(sg_exec_t exec, double bound)
234 {
235   exec->set_bound(bound);
236 }
237
238 const char* sg_exec_get_name(const_sg_exec_t exec)
239 {
240   return exec->get_cname();
241 }
242
243 void sg_exec_set_name(sg_exec_t exec, const char* name)
244 {
245   exec->set_name(name);
246 }
247
248 void sg_exec_set_host(sg_exec_t exec, sg_host_t new_host)
249 {
250   exec->set_host(new_host);
251 }
252
253 double sg_exec_get_remaining(const_sg_exec_t exec)
254 {
255   return exec->get_remaining();
256 }
257
258 double sg_exec_get_remaining_ratio(const_sg_exec_t exec)
259 {
260   return exec->get_remaining_ratio();
261 }
262
263 void sg_exec_start(sg_exec_t exec)
264 {
265   exec->start();
266 }
267
268 void sg_exec_cancel(sg_exec_t exec)
269 {
270   exec->cancel();
271   exec->unref();
272 }
273
274 int sg_exec_test(sg_exec_t exec)
275 {
276   bool finished = exec->test();
277   if (finished)
278     exec->unref();
279   return finished;
280 }
281
282 sg_error_t sg_exec_wait(sg_exec_t exec)
283 {
284   sg_error_t status = SG_OK;
285
286   simgrid::s4u::ExecPtr s4u_exec(exec, false);
287   try {
288     s4u_exec->wait_for(-1);
289   } catch (const simgrid::TimeoutException&) {
290     status = SG_ERROR_TIMEOUT;
291   } catch (const simgrid::CancelException&) {
292     status = SG_ERROR_CANCELED;
293   } catch (const simgrid::HostFailureException&) {
294     status = SG_ERROR_HOST;
295   }
296   return status;
297 }
298
299 sg_error_t sg_exec_wait_for(sg_exec_t exec, double timeout)
300 {
301   sg_error_t status = SG_OK;
302
303   simgrid::s4u::ExecPtr s4u_exec(exec, false);
304   try {
305     s4u_exec->wait_for(timeout);
306   } catch (const simgrid::TimeoutException&) {
307     status = SG_ERROR_TIMEOUT;
308   } catch (const simgrid::CancelException&) {
309     status = SG_ERROR_CANCELED;
310   } catch (const simgrid::HostFailureException&) {
311     status = SG_ERROR_HOST;
312   }
313   return status;
314 }
315
316 int sg_exec_wait_any(sg_exec_t* execs, size_t count)
317 {
318   return sg_exec_wait_any_for(execs, count, -1);
319 }
320
321 int sg_exec_wait_any_for(sg_exec_t* execs, size_t count, double timeout)
322 {
323   std::vector<simgrid::s4u::ExecPtr> s4u_execs;
324   for (unsigned int i = 0; i < count; i++)
325     s4u_execs.emplace_back(execs[i], false);
326
327   int pos = simgrid::s4u::Exec::wait_any_for(&s4u_execs, timeout);
328   for (unsigned i = 0; i < count; i++) {
329     if (pos != -1 && static_cast<unsigned>(pos) != i)
330       s4u_execs[i]->add_ref();
331   }
332   return pos;
333 }