src/s4u/s4u_Comm.cpp

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