Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Merge branch 'master' of https://framagit.org/simgrid/simgrid
[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 xbt::signal<void(Actor const&)> Exec::on_start;
16 xbt::signal<void(Actor const&)> Exec::on_completion;
17
18 Exec::Exec()
19 {
20   pimpl_ = kernel::activity::ExecImplPtr(new kernel::activity::ExecImpl());
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   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(const std::string& name)
94 {
95   xbt_assert(state_ == State::INITED, "Cannot change the name of an exec after its start");
96   name_ = 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(const std::string& category)
114 {
115   xbt_assert(state_ == State::INITED, "Cannot change the tracing category of an exec after its start");
116   tracing_category_ = 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_))
131         .set_name(name_)
132         .set_tracing_category(tracing_category_)
133         .start(flops_amount_, 1. / priority_, bound_);
134   });
135   state_ = State::STARTED;
136   on_start(*Actor::self());
137   return this;
138 }
139
140 /** @brief Returns whether the state of the exec is finished */
141 /** @brief Change the host on which this activity takes place.
142  *
143  * The activity cannot be terminated already (but it may be started). */
144 ExecPtr ExecSeq::set_host(Host* host)
145 {
146   xbt_assert(state_ == State::INITED || state_ == State::STARTED,
147              "Cannot change the host of an exec once it's done (state: %d)", (int)state_);
148   if (state_ == State::STARTED)
149     boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->migrate(host);
150   host_ = host;
151   boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->host_ = host;
152   return this;
153 }
154
155
156 /** @brief Retrieve the host on which this activity takes place. */
157 Host* ExecSeq::get_host()
158 {
159   return host_;
160 }
161
162 /** @brief Returns the amount of flops that remain to be done */
163 double ExecSeq::get_remaining()
164 {
165   return simgrid::simix::simcall(
166       [this]() { return boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->get_remaining(); });
167 }
168
169 /** @brief Returns the ratio of elements that are still to do
170  *
171  * The returned value is between 0 (completely done) and 1 (nothing done yet).
172  */
173 double ExecSeq::get_remaining_ratio()
174 {
175   return simgrid::simix::simcall([this]() {
176     return boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->get_seq_remaining_ratio();
177   });
178 }
179
180 ///////////// PARALLEL EXECUTIONS ////////
181 ExecPar::ExecPar(const std::vector<s4u::Host*>& hosts, const std::vector<double>& flops_amounts,
182                  const std::vector<double>& bytes_amounts)
183     : Exec(), hosts_(hosts), flops_amounts_(flops_amounts), bytes_amounts_(bytes_amounts)
184 {
185   // For parallel executions, we need a special host to run the timeout detector.
186   host_                                                                          = hosts.front();
187   boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_)->host_ = host_;
188 }
189
190 Exec* ExecPar::start()
191 {
192   simix::simcall([this] {
193     boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->set_timeout(timeout_);
194     boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->start(hosts_, flops_amounts_, bytes_amounts_);
195   });
196   state_ = State::STARTED;
197   on_start(*Actor::self());
198   return this;
199 }
200 double ExecPar::get_remaining_ratio()
201 {
202   return simix::simcall(
203       [this]() { return boost::static_pointer_cast<kernel::activity::ExecImpl>(pimpl_)->get_par_remaining_ratio(); });
204 }
205
206 double ExecPar::get_remaining()
207 {
208   XBT_WARN("Calling get_remaining() on a parallel execution is not allowed. Call get_remaining_ratio() instead.");
209   return get_remaining_ratio();
210 }
211 } // namespace s4u
212 } // namespace simgrid