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_->host_;
  73     s4u::Host* receiver = dst_actor_->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
  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() || dst_actor_->is_suspended()) {
  93       if (src_actor_->is_suspended())
  94         XBT_DEBUG("The communication is suspended on startup because src (%s@%s) was suspended since it initiated the "
  95                   "communication",
  96                   src_actor_->get_cname(), src_actor_->host_->get_cname());
  97       else
  98         XBT_DEBUG("The communication is suspended on startup because dst (%s@%s) was suspended since it initiated the "
  99                   "communication",
 100                   dst_actor_->get_cname(), dst_actor_->host_->get_cname());
 101
 102       surf_action_->suspend();
 103     }
 104   }
 105 }
 106
 107 /** @brief Copy the communication data from the sender's buffer to the receiver's one  */
 108 void CommImpl::copy_data()
 109 {
 110   size_t buff_size = src_buff_size_;
 111   /* If there is no data to copy then return */
 112   if (not src_buff_ || not dst_buff_ || copied)
 113     return;
 114
 115   XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)", this,
 116             src_actor_ ? src_actor_->host_->get_cname() : "a finished process", src_buff_,
 117             dst_actor_ ? dst_actor_->host_->get_cname() : "a finished process", dst_buff_, buff_size);
 118
 119   /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
 120   if (dst_buff_size_)
 121     buff_size = std::min(buff_size, *(dst_buff_size_));
 122
 123   /* Update the receiver's buffer size to the copied amount */
 124   if (dst_buff_size_)
 125     *dst_buff_size_ = buff_size;
 126
 127   if (buff_size > 0) {
 128     if (copy_data_fun)
 129       copy_data_fun(this, src_buff_, buff_size);
 130     else
 131       SIMIX_comm_copy_data_callback(this, src_buff_, buff_size);
 132   }
 133
 134   /* Set the copied flag so we copy data only once */
 135   /* (this function might be called from both communication ends) */
 136   copied = true;
 137 }
 138
 139 void CommImpl::suspend()
 140 {
 141   /* FIXME: shall we suspend also the timeout synchro? */
 142   if (surf_action_)
 143     surf_action_->suspend();
 144   /* if not created yet, the action will be suspended on creation, in CommImpl::start() */
 145 }
 146
 147 void CommImpl::resume()
 148 {
 149   /*FIXME: check what happen with the timeouts */
 150   if (surf_action_)
 151     surf_action_->resume();
 152   /* in the other case, the synchro were not really suspended yet, see CommImpl::suspend() and CommImpl::start() */
 153 }
 154
 155 void CommImpl::cancel()
 156 {
 157   /* if the synchro is a waiting state means that it is still in a mbox so remove from it and delete it */
 158   if (state_ == SIMIX_WAITING) {
 159     mbox->remove(this);
 160     state_ = SIMIX_CANCELED;
 161   } else if (not MC_is_active() /* when running the MC there are no surf actions */
 162              && not MC_record_replay_is_active() && (state_ == SIMIX_READY || state_ == SIMIX_RUNNING)) {
 163     surf_action_->cancel();
 164   }
 165 }
 166
 167 /**  @brief get the amount remaining from the communication */
 168 double CommImpl::remains()
 169 {
 170   return surf_action_->get_remains();
 171 }
 172
 173 /** @brief This is part of the cleanup process, probably an internal command */
 174 void CommImpl::cleanupSurf()
 175 {
 176   if (surf_action_) {
 177     surf_action_->unref();
 178     surf_action_ = nullptr;
 179   }
 180
 181   if (src_timeout_) {
 182     src_timeout_->unref();
 183     src_timeout_ = nullptr;
 184   }
 185
 186   if (dst_timeout_) {
 187     dst_timeout_->unref();
 188     dst_timeout_ = nullptr;
 189   }
 190 }
 191
 192 void CommImpl::post()
 193 {
 194   /* Update synchro state */
 195   if (src_timeout_ && src_timeout_->get_state() == resource::Action::State::FINISHED)
 196     state_ = SIMIX_SRC_TIMEOUT;
 197   else if (dst_timeout_ && dst_timeout_->get_state() == resource::Action::State::FINISHED)
 198     state_ = SIMIX_DST_TIMEOUT;
 199   else if (src_timeout_ && src_timeout_->get_state() == resource::Action::State::FAILED)
 200     state_ = SIMIX_SRC_HOST_FAILURE;
 201   else if (dst_timeout_ && dst_timeout_->get_state() == resource::Action::State::FAILED)
 202     state_ = SIMIX_DST_HOST_FAILURE;
 203   else if (surf_action_ && surf_action_->get_state() == resource::Action::State::FAILED) {
 204     state_ = SIMIX_LINK_FAILURE;
 205   } else
 206     state_ = SIMIX_DONE;
 207
 208   XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d", this, (int)state_,
 209             src_actor_.get(), dst_actor_.get(), detached);
 210
 211   /* destroy the surf actions associated with the Simix communication */
 212   cleanupSurf();
 213
 214   /* if there are simcalls associated with the synchro, then answer them */
 215   if (not simcalls_.empty()) {
 216     SIMIX_comm_finish(this);
 217   }
 218 }
 219
 220 } // namespace activity
 221 } // namespace kernel
 222 } // namespace simgrid