src/s4u/s4u_comm.cpp

   1 /* Copyright (c) 2006-2018. 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/s4u/Comm.hpp"
  10 #include "simgrid/s4u/Mailbox.hpp"
  11
  12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_comm,s4u_activity,"S4U asynchronous communications");
  13
  14 namespace simgrid {
  15 namespace s4u {
  16 Comm::~Comm()
  17 {
  18   if (state_ == State::started && not detached_ && (pimpl_ == nullptr || pimpl_->state == SIMIX_RUNNING)) {
  19     XBT_INFO("Comm %p freed before its completion. Detached: %d, State: %d", this, detached_, (int)state_);
  20     if (pimpl_ != nullptr)
  21       XBT_INFO("pimpl_->state: %d", pimpl_->state);
  22     else
  23       XBT_INFO("pimpl_ is null");
  24     xbt_backtrace_display_current();
  25   }
  26 }
  27
  28 Activity* Comm::set_rate(double rate)
  29 {
  30   xbt_assert(state_ == State::inited);
  31   rate_ = rate;
  32   return this;
  33 }
  34
  35 Activity* Comm::set_src_data(void* buff)
  36 {
  37   xbt_assert(state_ == State::inited);
  38   xbt_assert(dstBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
  39   srcBuff_ = buff;
  40   return this;
  41 }
  42 Activity* Comm::set_src_data_size(size_t size)
  43 {
  44   xbt_assert(state_ == State::inited);
  45   srcBuffSize_ = size;
  46   return this;
  47 }
  48 Activity* Comm::set_src_data(void* buff, size_t size)
  49 {
  50   xbt_assert(state_ == State::inited);
  51
  52   xbt_assert(dstBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
  53   srcBuff_ = buff;
  54   srcBuffSize_ = size;
  55   return this;
  56 }
  57 Activity* Comm::set_dst_data(void** buff)
  58 {
  59   xbt_assert(state_ == State::inited);
  60   xbt_assert(srcBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
  61   dstBuff_ = buff;
  62   return this;
  63 }
  64 size_t Comm::get_dst_data_size()
  65 {
  66   xbt_assert(state_ == State::finished);
  67   return dstBuffSize_;
  68 }
  69 Activity* Comm::set_dst_data(void** buff, size_t size)
  70 {
  71   xbt_assert(state_ == State::inited);
  72
  73   xbt_assert(srcBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
  74   dstBuff_ = buff;
  75   dstBuffSize_ = size;
  76   return this;
  77 }
  78
  79 Activity* Comm::start()
  80 {
  81   xbt_assert(state_ == State::inited);
  82
  83   if (srcBuff_ != nullptr) { // Sender side
  84     pimpl_ = simcall_comm_isend(sender_, mailbox_->get_impl(), remains_, rate_, srcBuff_, srcBuffSize_, matchFunction_,
  85                                 cleanFunction_, copyDataFunction_, user_data_, detached_);
  86   } else if (dstBuff_ != nullptr) { // Receiver side
  87     xbt_assert(not detached_, "Receive cannot be detached");
  88     pimpl_ = simcall_comm_irecv(receiver_, mailbox_->get_impl(), dstBuff_, &dstBuffSize_, matchFunction_,
  89                                 copyDataFunction_, user_data_, rate_);
  90
  91   } else {
  92     xbt_die("Cannot start a communication before specifying whether we are the sender or the receiver");
  93   }
  94   state_ = State::started;
  95   return this;
  96 }
  97
  98 /** @brief Block the calling actor until the communication is finished */
  99 Activity* Comm::wait()
 100 {
 101   return this->wait(-1);
 102 }
 103
 104 /** @brief Block the calling actor until the communication is finished, or until timeout
 105  *
 106  * On timeout, an exception is thrown.
 107  *
 108  * @param timeout the amount of seconds to wait for the comm termination.
 109  *                Negative values denote infinite wait times. 0 as a timeout returns immediately. */
 110 Activity* Comm::wait(double timeout)
 111 {
 112   switch (state_) {
 113     case State::finished:
 114       return this;
 115
 116     case State::inited: // It's not started yet. Do it in one simcall
 117       if (srcBuff_ != nullptr) {
 118         simcall_comm_send(sender_, mailbox_->get_impl(), remains_, rate_, srcBuff_, srcBuffSize_, matchFunction_,
 119                           copyDataFunction_, user_data_, timeout);
 120       } else { // Receiver
 121         simcall_comm_recv(receiver_, mailbox_->get_impl(), dstBuff_, &dstBuffSize_, matchFunction_, copyDataFunction_,
 122                           user_data_, timeout, rate_);
 123       }
 124       state_ = State::finished;
 125       return this;
 126
 127     case State::started:
 128       simcall_comm_wait(pimpl_, timeout);
 129       state_ = State::finished;
 130       return this;
 131
 132     default:
 133       THROW_IMPOSSIBLE;
 134   }
 135   return this;
 136 }
 137 int Comm::test_any(std::vector<CommPtr>* comms)
 138 {
 139   smx_activity_t* array = new smx_activity_t[comms->size()];
 140   for (unsigned int i = 0; i < comms->size(); i++) {
 141     array[i] = comms->at(i)->pimpl_;
 142   }
 143   int res = simcall_comm_testany(array, comms->size());
 144   delete[] array;
 145   return res;
 146 }
 147
 148 Activity* Comm::detach()
 149 {
 150   xbt_assert(state_ == State::inited, "You cannot detach communications once they are started (not implemented).");
 151   xbt_assert(srcBuff_ != nullptr && srcBuffSize_ != 0, "You can only detach sends, not recvs");
 152   detached_ = true;
 153   return start();
 154 }
 155
 156 Activity* Comm::cancel()
 157 {
 158   simgrid::kernel::activity::CommImplPtr commPimpl =
 159       boost::static_pointer_cast<simgrid::kernel::activity::CommImpl>(pimpl_);
 160   commPimpl->cancel();
 161   return this;
 162 }
 163
 164 bool Comm::test()
 165 {
 166   xbt_assert(state_ == State::inited || state_ == State::started || state_ == State::finished);
 167
 168   if (state_ == State::finished)
 169     return true;
 170
 171   if (state_ == State::inited)
 172     this->start();
 173
 174   if(simcall_comm_test(pimpl_)){
 175     state_ = State::finished;
 176     return true;
 177   }
 178   return false;
 179 }
 180
 181 MailboxPtr Comm::get_mailbox()
 182 {
 183   return mailbox_;
 184 }
 185
 186 void intrusive_ptr_release(simgrid::s4u::Comm* c)
 187 {
 188   if (c->refcount_.fetch_sub(1, std::memory_order_release) == 1) {
 189     std::atomic_thread_fence(std::memory_order_acquire);
 190     delete c;
 191   }
 192 }
 193 void intrusive_ptr_add_ref(simgrid::s4u::Comm* c)
 194 {
 195   c->refcount_.fetch_add(1, std::memory_order_relaxed);
 196 }
 197 }
 198 } // namespaces