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::string(comm->get_mailbox() == nullptr ? xbt::string("-") : comm->get_mailbox()->get_name()) +
 119            "(id:" + std::to_string(comm->get_mailbox_id()) + ") srcbuf:" + ptr_to_id<unsigned char>(comm->src_buff_) +
 120            " dstbuf:" + ptr_to_id<unsigned char>(comm->dst_buff_) + " bufsize:" + std::to_string(comm->src_buff_size_) +
 121            ")";
 122   } else {
 123     return "WaitUnknownType()";
 124   }
 125 }
 126
 127 void ActivityWaitSimcall::serialize(std::stringstream& stream) const
 128 {
 129   serialize_activity_wait(activity_, timeout_ > 0, stream);
 130 }
 131 void ActivityWaitanySimcall::serialize(std::stringstream& stream) const
 132 {
 133   stream << (short)mc::Transition::Type::WAITANY << ' ' << activities_.size() << ' ';
 134   for (auto const& act : activities_) {
 135     serialize_activity_wait(act, timeout_ > 0, stream);
 136     stream << ' ';
 137   }
 138 }
 139 std::string ActivityWaitSimcall::to_string() const
 140 {
 141   return to_string_activity_wait(activity_);
 142 }
 143 std::string ActivityWaitanySimcall::to_string() const
 144 {
 145   std::stringstream buffer("WaitAny(");
 146   for (auto const& act : activities_) {
 147     buffer << to_string_activity_wait(act);
 148   }
 149   return buffer.str();
 150 }
 151 ActivityWaitanySimcall::ActivityWaitanySimcall(ActorImpl* actor, const std::vector<activity::ActivityImpl*>& activities,
 152                                                double timeout)
 153     : ResultingSimcall(actor, -1), activities_(activities), timeout_(timeout)
 154 {
 155   // list all the activities that are ready
 156   indexes_.clear();
 157   for (unsigned i = 0; i < activities_.size(); i++)
 158     if (activities_[i]->test(get_issuer()))
 159       indexes_.push_back(i);
 160 }
 161
 162 bool ActivityWaitSimcall::is_enabled()
 163 {
 164   // FIXME: if _sg_mc_timeout == 1 and if we have either a sender or receiver timeout, the transition is enabled
 165   // because even if the communication is not ready, it can timeout and won't block.
 166
 167   return activity_->test(get_issuer());
 168 }
 169
 170 bool ActivityWaitanySimcall::is_enabled()
 171 {
 172   // list all the activities that are ready
 173   indexes_.clear();
 174   for (unsigned i = 0; i < activities_.size(); i++)
 175     if (activities_[i]->test(get_issuer()))
 176       indexes_.push_back(i);
 177
 178   //  if (_sg_mc_timeout && timeout_)  FIXME: deal with the potential timeout of the WaitAny
 179
 180   // FIXME: even if the WaitAny has no timeout, some of the activities may still have one.
 181   // we should iterate over the vector searching for them
 182   return not indexes_.empty();
 183 }
 184
 185 int ActivityWaitanySimcall::get_max_consider() const
 186 {
 187   int res = indexes_.size();
 188   //  if (_sg_mc_timeout && timeout_)
 189   //    res++;
 190
 191   return res;
 192 }
 193
 194 void ActivityWaitanySimcall::prepare(int times_considered)
 195 {
 196   if (times_considered < static_cast<int>(indexes_.size()))
 197     next_value_ = indexes_.at(times_considered);
 198   else
 199     next_value_ = -1;
 200 }
 201
 202 void CommIsendSimcall::serialize(std::stringstream& stream) const
 203 {
 204   stream << (short)mc::Transition::Type::COMM_ASYNC_SEND << ' ';
 205   stream << (uintptr_t)comm_ << ' ' << mbox_->get_id() << ' ' << (uintptr_t)src_buff_ << ' ' << src_buff_size_ << ' '
 206          << tag_;
 207   XBT_DEBUG("SendObserver comm:%p mbox:%u buff:%p size:%zu tag:%d", comm_, mbox_->get_id(), src_buff_, src_buff_size_,
 208             tag_);
 209 }
 210 std::string CommIsendSimcall::to_string() const
 211 {
 212   return std::string("CommAsyncSend(comm_id: ") + std::to_string((uintptr_t)comm_) +
 213          " mbox:" + std::to_string(mbox_->get_id()) + " srcbuf:" + ptr_to_id<unsigned char>(src_buff_) +
 214          " bufsize:" + std::to_string(src_buff_size_) + " tag: " + std::to_string(tag_) + ")";
 215 }
 216
 217 void CommIrecvSimcall::serialize(std::stringstream& stream) const
 218 {
 219   stream << (short)mc::Transition::Type::COMM_ASYNC_RECV << ' ';
 220   stream << (uintptr_t)comm_ << ' ' << mbox_->get_id() << ' ' << (uintptr_t)dst_buff_ << ' ' << tag_;
 221   XBT_DEBUG("RecvObserver comm:%p mbox:%u buff:%p tag:%d", comm_, mbox_->get_id(), dst_buff_, tag_);
 222 }
 223 std::string CommIrecvSimcall::to_string() const
 224 {
 225   return std::string("CommAsyncRecv(comm_id: ") + ptr_to_id<activity::CommImpl const>(comm_) +
 226          " mbox:" + std::to_string(mbox_->get_id()) + " dstbuf:" + ptr_to_id<unsigned char>(dst_buff_) +
 227          " tag: " + std::to_string(tag_) + ")";
 228 }
 229
 230 } // namespace simgrid::kernel::actor