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