src/kernel/activity/CommImpl.cpp

   1 /* Copyright (c) 2007-2019. 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/CommImpl.hpp"
   7 #include "simgrid/kernel/resource/Action.hpp"
   8 #include "simgrid/modelchecker.h"
   9 #include "simgrid/s4u/Host.hpp"
  10 #include "src/kernel/activity/MailboxImpl.hpp"
  11 #include "src/mc/mc_replay.hpp"
  12 #include "src/surf/network_interface.hpp"
  13 #include "src/surf/surf_interface.hpp"
  14
  15 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(simix_network);
  16
  17 /******************************************************************************/
  18 /*                    SIMIX_comm_copy_data callbacks                       */
  19 /******************************************************************************/
  20 static void (*SIMIX_comm_copy_data_callback)(smx_activity_t, void*, size_t) = &SIMIX_comm_copy_pointer_callback;
  21
  22 void SIMIX_comm_set_copy_data_callback(void (*callback)(smx_activity_t, void*, size_t))
  23 {
  24   SIMIX_comm_copy_data_callback = callback;
  25 }
  26
  27 void SIMIX_comm_copy_pointer_callback(smx_activity_t synchro, void* buff, size_t buff_size)
  28 {
  29   simgrid::kernel::activity::CommImplPtr comm =
  30       boost::static_pointer_cast<simgrid::kernel::activity::CommImpl>(synchro);
  31
  32   xbt_assert((buff_size == sizeof(void*)), "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
  33   *(void**)(comm->dst_buff_) = buff;
  34 }
  35
  36 namespace simgrid {
  37 namespace kernel {
  38 namespace activity {
  39
  40 CommImpl::CommImpl(CommImpl::Type type) : type(type)
  41 {
  42   state_   = SIMIX_WAITING;
  43   src_data_ = nullptr;
  44   dst_data_ = nullptr;
  45   XBT_DEBUG("Create comm activity %p", this);
  46 }
  47
  48 CommImpl::~CommImpl()
  49 {
  50   XBT_DEBUG("Really free communication %p", this);
  51
  52   cleanupSurf();
  53
  54   if (detached && state_ != SIMIX_DONE) {
  55     /* the communication has failed and was detached:
  56      * we have to free the buffer */
  57     if (clean_fun)
  58       clean_fun(src_buff_);
  59     src_buff_ = nullptr;
  60   }
  61
  62   if (mbox)
  63     mbox->remove(this);
  64 }
  65
  66 /**  @brief Starts the simulation of a communication synchro. */
  67 void CommImpl::start()
  68 {
  69   /* If both the sender and the receiver are already there, start the communication */
  70   if (state_ == SIMIX_READY) {
  71
  72     s4u::Host* sender   = src_actor_->get_host();
  73     s4u::Host* receiver = dst_actor_->get_host();
  74
  75     surf_action_ = surf_network_model->communicate(sender, receiver, task_size_, rate_);
  76     surf_action_->set_data(this);
  77     state_ = SIMIX_RUNNING;
  78
  79     XBT_DEBUG("Starting communication %p from '%s' to '%s' (surf_action: %p)", this, sender->get_cname(),
  80               receiver->get_cname(), surf_action_);
  81
  82     /* If a link is failed, detect it immediately */
  83     if (surf_action_->get_state() == resource::Action::State::FAILED) {
  84       XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure", sender->get_cname(),
  85                 receiver->get_cname());
  86       state_ = SIMIX_LINK_FAILURE;
  87       cleanupSurf();
  88
  89     } else if (src_actor_->is_suspended() || dst_actor_->is_suspended()) {
  90       /* If any of the process is suspended, create the synchro but stop its execution,
  91          it will be restarted when the sender process resume */
  92       if (src_actor_->is_suspended())
  93         XBT_DEBUG("The communication is suspended on startup because src (%s@%s) was suspended since it initiated the "
  94                   "communication",
  95                   src_actor_->get_cname(), src_actor_->get_host()->get_cname());
  96       else
  97         XBT_DEBUG("The communication is suspended on startup because dst (%s@%s) was suspended since it initiated the "
  98                   "communication",
  99                   dst_actor_->get_cname(), dst_actor_->get_host()->get_cname());
 100
 101       surf_action_->suspend();
 102     }
 103   }
 104 }
 105
 106 /** @brief Copy the communication data from the sender's buffer to the receiver's one  */
 107 void CommImpl::copy_data()
 108 {
 109   size_t buff_size = src_buff_size_;
 110   /* If there is no data to copy then return */
 111   if (not src_buff_ || not dst_buff_ || copied)
 112     return;
 113
 114   XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)", this,
 115             src_actor_ ? src_actor_->get_host()->get_cname() : "a finished process", src_buff_,
 116             dst_actor_ ? dst_actor_->get_host()->get_cname() : "a finished process", dst_buff_, buff_size);
 117
 118   /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
 119   if (dst_buff_size_)
 120     buff_size = std::min(buff_size, *(dst_buff_size_));
 121
 122   /* Update the receiver's buffer size to the copied amount */
 123   if (dst_buff_size_)
 124     *dst_buff_size_ = buff_size;
 125
 126   if (buff_size > 0) {
 127     if (copy_data_fun)
 128       copy_data_fun(this, src_buff_, buff_size);
 129     else
 130       SIMIX_comm_copy_data_callback(this, src_buff_, buff_size);
 131   }
 132
 133   /* Set the copied flag so we copy data only once */
 134   /* (this function might be called from both communication ends) */
 135   copied = true;
 136 }
 137
 138 void CommImpl::suspend()
 139 {
 140   /* FIXME: shall we suspend also the timeout synchro? */
 141   if (surf_action_)
 142     surf_action_->suspend();
 143   /* if not created yet, the action will be suspended on creation, in CommImpl::start() */
 144 }
 145
 146 void CommImpl::resume()
 147 {
 148   /*FIXME: check what happen with the timeouts */
 149   if (surf_action_)
 150     surf_action_->resume();
 151   /* in the other case, the synchro were not really suspended yet, see CommImpl::suspend() and CommImpl::start() */
 152 }
 153
 154 void CommImpl::cancel()
 155 {
 156   /* if the synchro is a waiting state means that it is still in a mbox so remove from it and delete it */
 157   if (state_ == SIMIX_WAITING) {
 158     mbox->remove(this);
 159     state_ = SIMIX_CANCELED;
 160   } else if (not MC_is_active() /* when running the MC there are no surf actions */
 161              && not MC_record_replay_is_active() && (state_ == SIMIX_READY || state_ == SIMIX_RUNNING)) {
 162     surf_action_->cancel();
 163   }
 164 }
 165
 166 /**  @brief get the amount remaining from the communication */
 167 double CommImpl::remains()
 168 {
 169   return surf_action_->get_remains();
 170 }
 171
 172 /** @brief This is part of the cleanup process, probably an internal command */
 173 void CommImpl::cleanupSurf()
 174 {
 175   if (surf_action_) {
 176     surf_action_->unref();
 177     surf_action_ = nullptr;
 178   }
 179
 180   if (src_timeout_) {
 181     src_timeout_->unref();
 182     src_timeout_ = nullptr;
 183   }
 184
 185   if (dst_timeout_) {
 186     dst_timeout_->unref();
 187     dst_timeout_ = nullptr;
 188   }
 189 }
 190
 191 void CommImpl::post()
 192 {
 193   /* Update synchro state */
 194   if (src_timeout_ && src_timeout_->get_state() == resource::Action::State::FINISHED)
 195     state_ = SIMIX_SRC_TIMEOUT;
 196   else if (dst_timeout_ && dst_timeout_->get_state() == resource::Action::State::FINISHED)
 197     state_ = SIMIX_DST_TIMEOUT;
 198   else if (src_timeout_ && src_timeout_->get_state() == resource::Action::State::FAILED)
 199     state_ = SIMIX_SRC_HOST_FAILURE;
 200   else if (dst_timeout_ && dst_timeout_->get_state() == resource::Action::State::FAILED)
 201     state_ = SIMIX_DST_HOST_FAILURE;
 202   else if (surf_action_ && surf_action_->get_state() == resource::Action::State::FAILED) {
 203     state_ = SIMIX_LINK_FAILURE;
 204   } else
 205     state_ = SIMIX_DONE;
 206
 207   XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d", this, (int)state_,
 208             src_actor_.get(), dst_actor_.get(), detached);
 209
 210   /* destroy the surf actions associated with the Simix communication */
 211   cleanupSurf();
 212
 213   /* if there are simcalls associated with the synchro, then answer them */
 214   if (not simcalls_.empty()) {
 215     SIMIX_comm_finish(this);
 216   }
 217 }
 218
 219 } // namespace activity
 220 } // namespace kernel
 221 } // namespace simgrid