Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
433d9bae394b66a45a1001e4dbc5bb83cf32d960
[simgrid.git] / src / kernel / activity / ExecImpl.cpp
1 /* Copyright (c) 2007-2020. 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 "src/kernel/activity/ExecImpl.hpp"
7 #include "simgrid/Exception.hpp"
8 #include "simgrid/modelchecker.h"
9 #include "src/mc/mc_replay.hpp"
10 #include "src/surf/HostImpl.hpp"
11 #include "src/surf/cpu_interface.hpp"
12 #include "src/surf/surf_interface.hpp"
13
14 #include "simgrid/s4u/Host.hpp"
15
16 #include <boost/range/algorithm.hpp>
17
18 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(simix_process);
19
20 void simcall_HANDLER_execution_wait(smx_simcall_t simcall, simgrid::kernel::activity::ExecImpl* synchro, double timeout)
21 {
22   synchro->wait_for(simcall->issuer_, timeout);
23 }
24
25 void simcall_HANDLER_execution_waitany_for(smx_simcall_t simcall, simgrid::kernel::activity::ExecImpl* execs[],
26                                            size_t count, double timeout)
27 {
28   if (timeout < 0.0) {
29     simcall->timeout_cb_ = nullptr;
30   } else {
31     simcall->timeout_cb_ = simgrid::simix::Timer::set(SIMIX_get_clock() + timeout, [simcall, execs, count]() {
32       for (size_t i = 0; i < count; i++) {
33         // Remove the first occurrence of simcall:
34         auto* exec = execs[i];
35         auto j     = boost::range::find(exec->simcalls_, simcall);
36         if (j != exec->simcalls_.end())
37           exec->simcalls_.erase(j);
38       }
39       simcall_execution_waitany_for__set__result(simcall, -1);
40       simcall->issuer_->simcall_answer();
41     });
42   }
43
44   for (size_t i = 0; i < count; i++) {
45     /* associate this simcall to the the synchro */
46     auto* exec = execs[i];
47     exec->simcalls_.push_back(simcall);
48
49     /* see if the synchro is already finished */
50     if (exec->state_ != simgrid::kernel::activity::State::WAITING &&
51         exec->state_ != simgrid::kernel::activity::State::RUNNING) {
52       exec->finish();
53       break;
54     }
55   }
56 }
57
58 namespace simgrid {
59 namespace kernel {
60 namespace activity {
61
62 ExecImpl& ExecImpl::set_host(s4u::Host* host)
63 {
64   hosts_.assign(1, host);
65   return *this;
66 }
67
68 ExecImpl& ExecImpl::set_hosts(const std::vector<s4u::Host*>& hosts)
69 {
70   hosts_ = hosts;
71   return *this;
72 }
73
74 ExecImpl& ExecImpl::set_timeout(double timeout)
75 {
76   if (timeout > 0 && not MC_is_active() && not MC_record_replay_is_active()) {
77     timeout_detector_.reset(hosts_.front()->pimpl_cpu->sleep(timeout));
78     timeout_detector_->set_activity(this);
79   }
80   return *this;
81 }
82
83 ExecImpl& ExecImpl::set_flops_amount(double flops_amount)
84 {
85   flops_amounts_.assign(1, flops_amount);
86   return *this;
87 }
88
89 ExecImpl& ExecImpl::set_flops_amounts(const std::vector<double>& flops_amounts)
90 {
91   flops_amounts_ = flops_amounts;
92   return *this;
93 }
94
95 ExecImpl& ExecImpl::set_bytes_amounts(const std::vector<double>& bytes_amounts)
96 {
97   bytes_amounts_ = bytes_amounts;
98
99   return *this;
100 }
101
102 ExecImpl* ExecImpl::start()
103 {
104   state_ = State::RUNNING;
105   if (not MC_is_active() && not MC_record_replay_is_active()) {
106     if (hosts_.size() == 1) {
107       surf_action_ = hosts_.front()->pimpl_cpu->execution_start(flops_amounts_.front());
108       surf_action_->set_sharing_penalty(sharing_penalty_);
109       surf_action_->set_category(get_tracing_category());
110
111       if (bound_ > 0)
112         surf_action_->set_bound(bound_);
113     } else {
114       surf_action_ = surf_host_model->execute_parallel(hosts_, flops_amounts_.data(), bytes_amounts_.data(), -1);
115     }
116     surf_action_->set_activity(this);
117   }
118
119   XBT_DEBUG("Create execute synchro %p: %s", this, get_cname());
120   return this;
121 }
122
123 double ExecImpl::get_seq_remaining_ratio()
124 {
125   return (surf_action_ == nullptr) ? 0 : surf_action_->get_remains() / surf_action_->get_cost();
126 }
127
128 double ExecImpl::get_par_remaining_ratio()
129 {
130   // parallel task: their remain is already between 0 and 1
131   return (surf_action_ == nullptr) ? 0 : surf_action_->get_remains();
132 }
133
134 ExecImpl& ExecImpl::set_bound(double bound)
135 {
136   bound_ = bound;
137   return *this;
138 }
139
140 ExecImpl& ExecImpl::set_sharing_penalty(double sharing_penalty)
141 {
142   sharing_penalty_ = sharing_penalty;
143   return *this;
144 }
145
146 void ExecImpl::post()
147 {
148   if (hosts_.size() == 1 && not hosts_.front()->is_on()) { /* FIXME: handle resource failure for parallel tasks too */
149     /* If the host running the synchro failed, notice it. This way, the asking
150      * process can be killed if it runs on that host itself */
151     state_ = State::FAILED;
152   } else if (surf_action_ && surf_action_->get_state() == resource::Action::State::FAILED) {
153     /* If the host running the synchro didn't fail, then the synchro was canceled */
154     state_ = State::CANCELED;
155   } else if (timeout_detector_ && timeout_detector_->get_state() == resource::Action::State::FINISHED) {
156     state_ = State::TIMEOUT;
157   } else {
158     state_ = State::DONE;
159   }
160
161   clean_action();
162   timeout_detector_.reset();
163   /* Answer all simcalls associated with the synchro */
164   finish();
165 }
166
167 void ExecImpl::finish()
168 {
169   while (not simcalls_.empty()) {
170     smx_simcall_t simcall = simcalls_.front();
171     simcalls_.pop_front();
172
173     /* If a waitany simcall is waiting for this synchro to finish, then remove it from the other synchros in the waitany
174      * list. Afterwards, get the position of the actual synchro in the waitany list and return it as the result of the
175      * simcall */
176
177     if (simcall->call_ == SIMCALL_NONE) // FIXME: maybe a better way to handle this case
178       continue;                        // if process handling comm is killed
179     if (simcall->call_ == SIMCALL_EXECUTION_WAITANY_FOR) {
180       simgrid::kernel::activity::ExecImpl** execs = simcall_execution_waitany_for__get__execs(simcall);
181       size_t count                                = simcall_execution_waitany_for__get__count(simcall);
182
183       for (size_t i = 0; i < count; i++) {
184         // Remove the first occurrence of simcall:
185         auto* exec = execs[i];
186         auto j     = boost::range::find(exec->simcalls_, simcall);
187         if (j != exec->simcalls_.end())
188           exec->simcalls_.erase(j);
189
190         if (simcall->timeout_cb_) {
191           simcall->timeout_cb_->remove();
192           simcall->timeout_cb_ = nullptr;
193         }
194       }
195
196       if (not MC_is_active() && not MC_record_replay_is_active()) {
197         ExecImpl** element = std::find(execs, execs + count, this);
198         int rank           = (element != execs + count) ? element - execs : -1;
199         simcall_execution_waitany_for__set__result(simcall, rank);
200       }
201     }
202
203     switch (state_) {
204       case State::DONE:
205         /* do nothing, synchro done */
206         XBT_DEBUG("ExecImpl::finish(): execution successful");
207         break;
208
209       case State::FAILED:
210         XBT_DEBUG("ExecImpl::finish(): host '%s' failed", simcall->issuer_->get_host()->get_cname());
211         simcall->issuer_->context_->set_wannadie();
212         if (simcall->issuer_->get_host()->is_on())
213           simcall->issuer_->exception_ =
214               std::make_exception_ptr(simgrid::HostFailureException(XBT_THROW_POINT, "Host failed"));
215         /* else, the actor will be killed with no possibility to survive */
216         break;
217
218       case State::CANCELED:
219         XBT_DEBUG("ExecImpl::finish(): execution canceled");
220         simcall->issuer_->exception_ =
221             std::make_exception_ptr(simgrid::CancelException(XBT_THROW_POINT, "Execution Canceled"));
222         break;
223
224       case State::TIMEOUT:
225         XBT_DEBUG("ExecImpl::finish(): execution timeouted");
226         simcall->issuer_->exception_ = std::make_exception_ptr(simgrid::TimeoutException(XBT_THROW_POINT, "Timeouted"));
227         break;
228
229       default:
230         xbt_die("Internal error in ExecImpl::finish(): unexpected synchro state %d", static_cast<int>(state_));
231     }
232
233     simcall->issuer_->waiting_synchro = nullptr;
234     /* Fail the process if the host is down */
235     if (simcall->issuer_->get_host()->is_on())
236       simcall->issuer_->simcall_answer();
237     else
238       simcall->issuer_->context_->set_wannadie();
239   }
240 }
241
242 ActivityImpl* ExecImpl::migrate(s4u::Host* to)
243 {
244   if (not MC_is_active() && not MC_record_replay_is_active()) {
245     resource::Action* old_action = this->surf_action_;
246     resource::Action* new_action = to->pimpl_cpu->execution_start(old_action->get_cost());
247     new_action->set_remains(old_action->get_remains());
248     new_action->set_activity(this);
249     new_action->set_sharing_penalty(old_action->get_sharing_penalty());
250
251     // FIXME: the user-defined bound seem to not be kept by LMM, that seem to overwrite it for the multi-core modeling.
252     // I hope that the user did not provide any.
253
254     old_action->set_activity(nullptr);
255     old_action->cancel();
256     old_action->unref();
257     this->surf_action_ = new_action;
258   }
259
260   on_migration(*this, to);
261   return this;
262 }
263
264 /*************
265  * Callbacks *
266  *************/
267 xbt::signal<void(ExecImpl const&, s4u::Host*)> ExecImpl::on_migration;
268
269 } // namespace activity
270 } // namespace kernel
271 } // namespace simgrid