src/kernel/actor/CommObserver.cpp

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