src/s4u/s4u_Comm.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 "src/msg/msg_private.hpp"
   7 #include "xbt/log.h"
   8
   9 #include "simgrid/Exception.hpp"
  10 #include "simgrid/s4u/Comm.hpp"
  11 #include "simgrid/s4u/Engine.hpp"
  12 #include "simgrid/s4u/Mailbox.hpp"
  13
  14 #include <simgrid/comm.h>
  15
  16 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_comm, s4u_activity, "S4U asynchronous communications");
  17
  18 namespace simgrid {
  19 namespace s4u {
  20 xbt::signal<void(Comm const&, bool is_sender)> Comm::on_start;
  21 xbt::signal<void(Comm const&)> Comm::on_completion;
  22
  23 void Comm::complete(Activity::State state)
  24 {
  25   Activity::complete(state);
  26   on_completion(*this);
  27 }
  28
  29 Comm::~Comm()
  30 {
  31   if (state_ == State::STARTED && not detached_ &&
  32       (pimpl_ == nullptr || pimpl_->state_ == kernel::activity::State::RUNNING)) {
  33     XBT_INFO("Comm %p freed before its completion. Did you forget to detach it? (state: %s)", this, get_state_str());
  34     if (pimpl_ != nullptr)
  35       XBT_INFO("pimpl_->state: %s", pimpl_->get_state_str());
  36     else
  37       XBT_INFO("pimpl_ is null");
  38     xbt_backtrace_display_current();
  39   }
  40 }
  41
  42 ssize_t Comm::wait_any_for(const std::vector<CommPtr>& comms, double timeout)
  43 {
  44   std::vector<kernel::activity::CommImpl*> rcomms(comms.size());
  45   std::transform(begin(comms), end(comms), begin(rcomms),
  46                  [](const CommPtr& comm) { return static_cast<kernel::activity::CommImpl*>(comm->pimpl_.get()); });
  47   ssize_t changed_pos = -1;
  48   try {
  49     changed_pos = simcall_comm_waitany(rcomms.data(), rcomms.size(), timeout);
  50   } catch (const NetworkFailureException& e) {
  51     for (auto c : comms) {
  52       if (c->pimpl_->state_ == kernel::activity::State::FAILED) {
  53         c->complete(State::FAILED);
  54       }
  55     }
  56     e.rethrow_nested(XBT_THROW_POINT, boost::core::demangle(typeid(e).name()) + " raised in kernel mode.");
  57   }
  58   if (changed_pos != -1)
  59     comms.at(changed_pos)->complete(State::FINISHED);
  60   return changed_pos;
  61 }
  62
  63 void Comm::wait_all(const std::vector<CommPtr>& comms)
  64 {
  65   // TODO: this should be a simcall or something
  66   for (auto& comm : comms)
  67     comm->wait();
  68 }
  69
  70 size_t Comm::wait_all_for(const std::vector<CommPtr>& comms, double timeout)
  71 {
  72   if (timeout < 0.0) {
  73     wait_all(comms);
  74     return comms.size();
  75   }
  76
  77   double deadline = Engine::get_clock() + timeout;
  78   std::vector<CommPtr> waited_comm(1, nullptr);
  79   for (size_t i = 0; i < comms.size(); i++) {
  80     double wait_timeout = std::max(0.0, deadline - Engine::get_clock());
  81     waited_comm[0]      = comms[i];
  82     // Using wait_any_for() here (and not wait_for) because we don't want comms to be invalidated on timeout
  83     if (wait_any_for(waited_comm, wait_timeout) == -1) {
  84       XBT_DEBUG("Timeout (%g): i = %zu", wait_timeout, i);
  85       return i;
  86     }
  87   }
  88   return comms.size();
  89 }
  90
  91 CommPtr Comm::set_rate(double rate)
  92 {
  93   xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
  94              __FUNCTION__);
  95   rate_ = rate;
  96   return this;
  97 }
  98
  99 CommPtr Comm::set_src_data(void* buff)
 100 {
 101   xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
 102              __FUNCTION__);
 103   xbt_assert(dst_buff_ == nullptr, "Cannot set the src and dst buffers at the same time");
 104   src_buff_ = buff;
 105   return this;
 106 }
 107
 108 CommPtr Comm::set_src_data_size(size_t size)
 109 {
 110   xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
 111              __FUNCTION__);
 112   src_buff_size_ = size;
 113   return this;
 114 }
 115
 116 CommPtr Comm::set_src_data(void* buff, size_t size)
 117 {
 118   xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
 119              __FUNCTION__);
 120
 121   xbt_assert(dst_buff_ == nullptr, "Cannot set the src and dst buffers at the same time");
 122   src_buff_      = buff;
 123   src_buff_size_ = size;
 124   return this;
 125 }
 126
 127 CommPtr Comm::set_dst_data(void** buff)
 128 {
 129   xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
 130              __FUNCTION__);
 131   xbt_assert(src_buff_ == nullptr, "Cannot set the src and dst buffers at the same time");
 132   dst_buff_ = buff;
 133   return this;
 134 }
 135 void* Comm::get_dst_data()
 136 {
 137   return dst_buff_;
 138 }
 139
 140 size_t Comm::get_dst_data_size() const
 141 {
 142   return dst_buff_size_;
 143 }
 144 CommPtr Comm::set_dst_data(void** buff, size_t size)
 145 {
 146   xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)",
 147              __FUNCTION__);
 148
 149   xbt_assert(src_buff_ == nullptr, "Cannot set the src and dst buffers at the same time");
 150   dst_buff_      = buff;
 151   dst_buff_size_ = size;
 152   return this;
 153 }
 154 CommPtr Comm::set_payload_size(uint64_t bytes)
 155 {
 156   Activity::set_remaining(bytes);
 157   return this;
 158 }
 159
 160 CommPtr Comm::sendto_init(Host* from, Host* to)
 161 {
 162   CommPtr res(new Comm());
 163   res->from_ = from;
 164   res->to_   = to;
 165
 166   return res;
 167 }
 168
 169 CommPtr Comm::sendto_async(Host* from, Host* to, uint64_t simulated_size_in_bytes)
 170 {
 171   auto res = Comm::sendto_init(from, to)->set_payload_size(simulated_size_in_bytes);
 172   res->vetoable_start();
 173   return res;
 174 }
 175
 176 void Comm::sendto(Host* from, Host* to, uint64_t simulated_size_in_bytes)
 177 {
 178   sendto_async(from, to, simulated_size_in_bytes)->wait();
 179 }
 180
 181 Comm* Comm::start()
 182 {
 183   xbt_assert(get_state() == State::INITED || get_state() == State::STARTING,
 184              "You cannot use %s() once your communication started (not implemented)", __FUNCTION__);
 185   if (from_ != nullptr || to_ != nullptr) {
 186     xbt_assert(from_ != nullptr && to_ != nullptr, "When either from_ or to_ is specified, both must be.");
 187     xbt_assert(src_buff_ == nullptr && dst_buff_ == nullptr,
 188                "Direct host-to-host communications cannot carry any data.");
 189     pimpl_ = kernel::actor::simcall([this] {
 190       kernel::activity::CommImplPtr res(new kernel::activity::CommImpl(this->from_, this->to_, this->get_remaining()));
 191       res->start();
 192       return res;
 193     });
 194
 195   } else if (src_buff_ != nullptr) { // Sender side
 196     on_start(*this, true /* is_sender*/);
 197     pimpl_ = simcall_comm_isend(sender_, mailbox_->get_impl(), remains_, rate_, src_buff_, src_buff_size_, match_fun_,
 198                                 clean_fun_, copy_data_function_, get_user_data(), detached_);
 199   } else if (dst_buff_ != nullptr) { // Receiver side
 200     xbt_assert(not detached_, "Receive cannot be detached");
 201     on_start(*this, false /*is_sender*/);
 202     pimpl_ = simcall_comm_irecv(receiver_, mailbox_->get_impl(), dst_buff_, &dst_buff_size_, match_fun_,
 203                                 copy_data_function_, get_user_data(), rate_);
 204
 205   } else {
 206     xbt_die("Cannot start a communication before specifying whether we are the sender or the receiver");
 207   }
 208
 209   if (suspended_)
 210     pimpl_->suspend();
 211
 212   state_ = State::STARTED;
 213   return this;
 214 }
 215
 216 /** @brief Block the calling actor until the communication is finished, or until timeout
 217  *
 218  * On timeout, an exception is thrown and the communication is invalidated.
 219  *
 220  * @param timeout the amount of seconds to wait for the comm termination.
 221  *                Negative values denote infinite wait times. 0 as a timeout returns immediately. */
 222 Comm* Comm::wait_for(double timeout)
 223 {
 224   switch (state_) {
 225     case State::FINISHED:
 226       break;
 227     case State::FAILED:
 228       throw NetworkFailureException(XBT_THROW_POINT, "Cannot wait for a failed communication");
 229
 230     case State::INITED:
 231     case State::STARTING: // It's not started yet. Do it in one simcall if it's a regular communication
 232       if (from_ != nullptr || to_ != nullptr) {
 233         return vetoable_start()->wait_for(timeout); // In the case of host2host comm, do it in two simcalls
 234       } else if (src_buff_ != nullptr) {
 235         on_start(*this, true /*is_sender*/);
 236         simcall_comm_send(sender_, mailbox_->get_impl(), remains_, rate_, src_buff_, src_buff_size_, match_fun_,
 237                           copy_data_function_, get_user_data(), timeout);
 238
 239       } else { // Receiver
 240         on_start(*this, false /*is_sender*/);
 241         simcall_comm_recv(receiver_, mailbox_->get_impl(), dst_buff_, &dst_buff_size_, match_fun_, copy_data_function_,
 242                           get_user_data(), timeout, rate_);
 243       }
 244       break;
 245
 246     case State::STARTED:
 247       try {
 248         simcall_comm_wait(get_impl(), timeout);
 249       } catch (const NetworkFailureException& e) {
 250         complete(State::FAILED);
 251         e.rethrow_nested(XBT_THROW_POINT, boost::core::demangle(typeid(e).name()) + " raised in kernel mode.");
 252       }
 253       break;
 254
 255     case State::CANCELED:
 256       throw CancelException(XBT_THROW_POINT, "Communication canceled");
 257
 258     default:
 259       THROW_IMPOSSIBLE;
 260   }
 261   complete(State::FINISHED);
 262   return this;
 263 }
 264
 265 ssize_t Comm::test_any(const std::vector<CommPtr>& comms)
 266 {
 267   std::vector<kernel::activity::CommImpl*> rcomms(comms.size());
 268   std::transform(begin(comms), end(comms), begin(rcomms),
 269                  [](const CommPtr& comm) { return static_cast<kernel::activity::CommImpl*>(comm->pimpl_.get()); });
 270   ssize_t changed_pos = simcall_comm_testany(rcomms.data(), rcomms.size());
 271   if (changed_pos != -1)
 272     comms.at(changed_pos)->complete(State::FINISHED);
 273   return changed_pos;
 274 }
 275
 276 Comm* Comm::detach()
 277 {
 278   xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication is %s (not implemented)",
 279              __FUNCTION__, get_state_str());
 280   xbt_assert(dst_buff_ == nullptr && dst_buff_size_ == 0, "You can only detach sends, not recvs");
 281   detached_ = true;
 282   vetoable_start();
 283   return this;
 284 }
 285
 286 bool Comm::test() // TODO: merge with Activity::test, once modernized
 287 {
 288   xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::STARTING ||
 289              state_ == State::CANCELED || state_ == State::FINISHED);
 290
 291   if (state_ == State::CANCELED || state_ == State::FINISHED)
 292     return true;
 293
 294   if (state_ == State::INITED || state_ == State::STARTING)
 295     this->vetoable_start();
 296
 297   if (simcall_comm_test(get_impl())) {
 298     complete(State::FINISHED);
 299     return true;
 300   }
 301   return false;
 302 }
 303
 304 Mailbox* Comm::get_mailbox() const
 305 {
 306   return mailbox_;
 307 }
 308
 309 Actor* Comm::get_sender() const
 310 {
 311   kernel::actor::ActorImplPtr sender = nullptr;
 312   if (pimpl_)
 313     sender = boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->src_actor_;
 314   return sender ? sender->get_ciface() : nullptr;
 315 }
 316
 317 CommPtr Comm::set_copy_data_callback(void (*callback)(kernel::activity::CommImpl*, void*, size_t))
 318 {
 319   copy_data_function_ = callback;
 320   return this;
 321 }
 322
 323 } // namespace s4u
 324 } // namespace simgrid
 325 /* **************************** Public C interface *************************** */
 326 void sg_comm_detach(sg_comm_t comm, void (*clean_function)(void*))
 327 {
 328   comm->detach(clean_function);
 329   comm->unref();
 330 }
 331 void sg_comm_unref(sg_comm_t comm)
 332 {
 333   comm->unref();
 334 }
 335 int sg_comm_test(sg_comm_t comm)
 336 {
 337   bool finished = comm->test();
 338   if (finished)
 339     comm->unref();
 340   return finished;
 341 }
 342
 343 sg_error_t sg_comm_wait(sg_comm_t comm)
 344 {
 345   return sg_comm_wait_for(comm, -1);
 346 }
 347
 348 sg_error_t sg_comm_wait_for(sg_comm_t comm, double timeout)
 349 {
 350   sg_error_t status = SG_OK;
 351
 352   simgrid::s4u::CommPtr s4u_comm(comm, false);
 353   try {
 354     s4u_comm->wait_for(timeout);
 355   } catch (const simgrid::TimeoutException&) {
 356     status = SG_ERROR_TIMEOUT;
 357   } catch (const simgrid::CancelException&) {
 358     status = SG_ERROR_CANCELED;
 359   } catch (const simgrid::NetworkFailureException&) {
 360     status = SG_ERROR_NETWORK;
 361   }
 362   return status;
 363 }
 364
 365 void sg_comm_wait_all(sg_comm_t* comms, size_t count)
 366 {
 367   sg_comm_wait_all_for(comms, count, -1);
 368 }
 369
 370 size_t sg_comm_wait_all_for(sg_comm_t* comms, size_t count, double timeout)
 371 {
 372   std::vector<simgrid::s4u::CommPtr> s4u_comms;
 373   for (size_t i = 0; i < count; i++)
 374     s4u_comms.emplace_back(comms[i], false);
 375
 376   size_t pos = simgrid::s4u::Comm::wait_all_for(s4u_comms, timeout);
 377   for (size_t i = pos; i < count; i++)
 378     s4u_comms[i]->add_ref();
 379   return pos;
 380 }
 381
 382 ssize_t sg_comm_wait_any(sg_comm_t* comms, size_t count)
 383 {
 384   return sg_comm_wait_any_for(comms, count, -1);
 385 }
 386
 387 ssize_t sg_comm_wait_any_for(sg_comm_t* comms, size_t count, double timeout)
 388 {
 389   std::vector<simgrid::s4u::CommPtr> s4u_comms;
 390   for (size_t i = 0; i < count; i++)
 391     s4u_comms.emplace_back(comms[i], false);
 392
 393   ssize_t pos = simgrid::s4u::Comm::wait_any_for(s4u_comms, timeout);
 394   for (size_t i = 0; i < count; i++) {
 395     if (pos != -1 && static_cast<size_t>(pos) != i)
 396       s4u_comms[i]->add_ref();
 397   }
 398   return pos;
 399 }