src/kernel/actor/CommObserver.cpp

   1 /* Copyright (c) 2019-2023. 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 "simgrid/s4u/Host.hpp"
   7 #include "src/kernel/activity/CommImpl.hpp"
   8 #include "src/kernel/activity/MailboxImpl.hpp"
   9 #include "src/kernel/actor/ActorImpl.hpp"
  10 #include "src/kernel/actor/SimcallObserver.hpp"
  11 #include "src/mc/mc_config.hpp"
  12
  13 #include <sstream>
  14
  15 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(obs_comm, mc_observer, "Logging specific to the Communication simcalls observation");
  16
  17 namespace simgrid::kernel::actor {
  18
  19 ActivityTestanySimcall::ActivityTestanySimcall(ActorImpl* actor, const std::vector<activity::ActivityImpl*>& activities)
  20     : ResultingSimcall(actor, -1), activities_(activities)
  21 {
  22   indexes_.clear();
  23   // list all the activities that are ready
  24   for (unsigned i = 0; i < activities_.size(); i++)
  25     if (activities_[i]->test(get_issuer()))
  26       indexes_.push_back(i);
  27 }
  28
  29 int ActivityTestanySimcall::get_max_consider() const
  30 {
  31   return indexes_.size() + 1;
  32 }
  33
  34 void ActivityTestanySimcall::prepare(int times_considered)
  35 {
  36   if (times_considered < static_cast<int>(indexes_.size()))
  37     next_value_ = indexes_.at(times_considered);
  38   else
  39     next_value_ = -1;
  40 }
  41 static void serialize_activity_test(const activity::ActivityImpl* act, std::stringstream& stream)
  42 {
  43   if (const auto* comm = dynamic_cast<activity::CommImpl const*>(act)) {
  44     stream << "  " << (short)mc::Transition::Type::COMM_TEST;
  45     stream << ' ' << (uintptr_t)comm;
  46     stream << ' ' << (comm->src_actor_ != nullptr ? comm->src_actor_->get_pid() : -1);
  47     stream << ' ' << (comm->dst_actor_ != nullptr ? comm->dst_actor_->get_pid() : -1);
  48     stream << ' ' << comm->get_mailbox_id();
  49     stream << ' ' << (uintptr_t)comm->src_buff_ << ' ' << (uintptr_t)comm->dst_buff_ << ' ' << comm->src_buff_size_;
  50   } else {
  51     stream << (short)mc::Transition::Type::UNKNOWN;
  52   }
  53 }
  54 static std::string to_string_activity_test(const activity::ActivityImpl* act)
  55 {
  56   if (const auto* comm = dynamic_cast<activity::CommImpl const*>(act)) {
  57     const std::string src_buff_id = ptr_to_id<unsigned char>(comm->src_buff_);
  58     const std::string dst_buff_id = ptr_to_id<unsigned char>(comm->dst_buff_);
  59     return "CommTest(comm_id:" + ptr_to_id<activity::CommImpl const>(comm) +
  60            " src:" + std::to_string(comm->src_actor_ != nullptr ? comm->src_actor_->get_pid() : -1) +
  61            " dst:" + std::to_string(comm->dst_actor_ != nullptr ? comm->dst_actor_->get_pid() : -1) +
  62            " mbox:" + std::to_string(comm->get_mailbox_id()) + " srcbuf:" + src_buff_id + " dstbuf:" + dst_buff_id +
  63            " bufsize:" + std::to_string(comm->src_buff_size_);
  64   } else {
  65     return "TestUnknownType()";
  66   }
  67 }
  68 void ActivityTestanySimcall::serialize(std::stringstream& stream) const
  69 {
  70   stream << (short)mc::Transition::Type::TESTANY << ' ' << activities_.size() << ' ';
  71   for (auto const* act : activities_) {
  72     serialize_activity_test(act, stream);
  73     stream << ' ';
  74   }
  75 }
  76 std::string ActivityTestanySimcall::to_string() const
  77 {
  78   std::stringstream buffer("TestAny(");
  79   for (auto const* act : activities_) {
  80     buffer << to_string_activity_test(act);
  81   }
  82   return buffer.str();
  83 }
  84
  85 void ActivityTestSimcall::serialize(std::stringstream& stream) const
  86 {
  87   serialize_activity_test(activity_, stream);
  88 }
  89 std::string ActivityTestSimcall::to_string() const
  90 {
  91   return to_string_activity_test(activity_);
  92 }
  93 static void serialize_activity_wait(const activity::ActivityImpl* act, bool timeout, std::stringstream& stream)
  94 {
  95   if (const auto* comm = dynamic_cast<activity::CommImpl const*>(act)) {
  96     stream << (short)mc::Transition::Type::COMM_WAIT << ' ';
  97     stream << timeout << ' ' << (uintptr_t)comm;
  98
  99     stream << ' ' << (comm->src_actor_ != nullptr ? comm->src_actor_->get_pid() : -1);
 100     stream << ' ' << (comm->dst_actor_ != nullptr ? comm->dst_actor_->get_pid() : -1);
 101     stream << ' ' << comm->get_mailbox_id();
 102     stream << ' ' << (uintptr_t)comm->src_buff_ << ' ' << (uintptr_t)comm->dst_buff_ << ' ' << comm->src_buff_size_;
 103   } else {
 104     stream << (short)mc::Transition::Type::UNKNOWN;
 105   }
 106 }
 107 static std::string to_string_activity_wait(const activity::ActivityImpl* act)
 108 {
 109   if (const auto* comm = dynamic_cast<activity::CommImpl const*>(act)) {
 110     const std::string src_buff_id = ptr_to_id<unsigned char>(comm->src_buff_);
 111     const std::string dst_buff_id = ptr_to_id<unsigned char>(comm->dst_buff_);
 112     return "CommWait(comm_id:" + ptr_to_id<activity::CommImpl const>(comm) +
 113            " src:" + std::to_string(comm->src_actor_ != nullptr ? comm->src_actor_->get_pid() : -1) +
 114            " dst:" + std::to_string(comm->dst_actor_ != nullptr ? comm->dst_actor_->get_pid() : -1) +
 115            " mbox:" + (comm->get_mailbox() == nullptr ? "-" : comm->get_mailbox()->get_name()) +
 116            "(id:" + std::to_string(comm->get_mailbox_id()) + ") srcbuf:" + src_buff_id + " dstbuf:" + dst_buff_id +
 117            " bufsize:" + std::to_string(comm->src_buff_size_) + ")";
 118   } else {
 119     return "WaitUnknownType()";
 120   }
 121 }
 122
 123 void ActivityWaitSimcall::serialize(std::stringstream& stream) const
 124 {
 125   serialize_activity_wait(activity_, timeout_ > 0, stream);
 126 }
 127 void ActivityWaitanySimcall::serialize(std::stringstream& stream) const
 128 {
 129   stream << (short)mc::Transition::Type::WAITANY << ' ' << activities_.size() << ' ';
 130   for (auto const* act : activities_) {
 131     serialize_activity_wait(act, timeout_ > 0, stream);
 132     stream << ' ';
 133   }
 134 }
 135 std::string ActivityWaitSimcall::to_string() const
 136 {
 137   return to_string_activity_wait(activity_);
 138 }
 139 std::string ActivityWaitanySimcall::to_string() const
 140 {
 141   std::stringstream buffer("WaitAny(");
 142   for (auto const* act : activities_) {
 143     buffer << to_string_activity_wait(act);
 144   }
 145   return buffer.str();
 146 }
 147 ActivityWaitanySimcall::ActivityWaitanySimcall(ActorImpl* actor, const std::vector<activity::ActivityImpl*>& activities,
 148                                                double timeout)
 149     : ResultingSimcall(actor, -1), activities_(activities), timeout_(timeout)
 150 {
 151   // list all the activities that are ready
 152   indexes_.clear();
 153   for (unsigned i = 0; i < activities_.size(); i++)
 154     if (activities_[i]->test(get_issuer()))
 155       indexes_.push_back(i);
 156 }
 157
 158 bool ActivityWaitSimcall::is_enabled()
 159 {
 160   // FIXME: if _sg_mc_timeout == 1 and if we have either a sender or receiver timeout, the transition is enabled
 161   // because even if the communication is not ready, it can timeout and won't block.
 162
 163   return activity_->test(get_issuer());
 164 }
 165
 166 bool ActivityWaitanySimcall::is_enabled()
 167 {
 168   // list all the activities that are ready
 169   indexes_.clear();
 170   for (unsigned i = 0; i < activities_.size(); i++)
 171     if (activities_[i]->test(get_issuer()))
 172       indexes_.push_back(i);
 173
 174   //  if (_sg_mc_timeout && timeout_)  FIXME: deal with the potential timeout of the WaitAny
 175
 176   // FIXME: even if the WaitAny has no timeout, some of the activities may still have one.
 177   // we should iterate over the vector searching for them
 178   return not indexes_.empty();
 179 }
 180
 181 int ActivityWaitanySimcall::get_max_consider() const
 182 {
 183   int res = indexes_.size();
 184   //  if (_sg_mc_timeout && timeout_)
 185   //    res++;
 186
 187   return res;
 188 }
 189
 190 void ActivityWaitanySimcall::prepare(int times_considered)
 191 {
 192   if (times_considered < static_cast<int>(indexes_.size()))
 193     next_value_ = indexes_.at(times_considered);
 194   else
 195     next_value_ = -1;
 196 }
 197
 198 void CommIsendSimcall::serialize(std::stringstream& stream) const
 199 {
 200   /* Note that the comm_ is 0 until after the execution of the simcall */
 201   stream << (short)mc::Transition::Type::COMM_ASYNC_SEND << ' ';
 202   stream << (uintptr_t)comm_ << ' ' << mbox_->get_id() << ' ' << (uintptr_t)src_buff_ << ' ' << src_buff_size_ << ' '
 203          << tag_;
 204   XBT_DEBUG("SendObserver comm:%p mbox:%u buff:%p size:%zu tag:%d", comm_, mbox_->get_id(), src_buff_, src_buff_size_,
 205             tag_);
 206 }
 207 std::string CommIsendSimcall::to_string() const
 208 {
 209   return "CommAsyncSend(comm_id: " + std::to_string((uintptr_t)comm_) + " mbox:" + std::to_string(mbox_->get_id()) +
 210          " srcbuf:" + ptr_to_id<unsigned char>(src_buff_) + " bufsize:" + std::to_string(src_buff_size_) +
 211          " tag: " + std::to_string(tag_) + ")";
 212 }
 213
 214 void CommIrecvSimcall::serialize(std::stringstream& stream) const
 215 {
 216   /* Note that the comm_ is 0 until after the execution of the simcall */
 217   stream << (short)mc::Transition::Type::COMM_ASYNC_RECV << ' ';
 218   stream << (uintptr_t)comm_ << ' ' << mbox_->get_id() << ' ' << (uintptr_t)dst_buff_ << ' ' << tag_;
 219   XBT_DEBUG("RecvObserver comm:%p mbox:%u buff:%p tag:%d", comm_, mbox_->get_id(), dst_buff_, tag_);
 220 }
 221 std::string CommIrecvSimcall::to_string() const
 222 {
 223   return "CommAsyncRecv(comm_id: " + ptr_to_id<activity::CommImpl const>(comm_) +
 224          " mbox:" + std::to_string(mbox_->get_id()) + " dstbuf:" + ptr_to_id<unsigned char>(dst_buff_) +
 225          " tag: " + std::to_string(tag_) + ")";
 226 }
 227
 228 } // namespace simgrid::kernel::actor