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/activity/MessageQueueImpl.hpp"
  10 #include "src/kernel/actor/ActorImpl.hpp"
  11 #include "src/kernel/actor/SimcallObserver.hpp"
  12 #include "src/mc/mc_config.hpp"
  13
  14 #include <sstream>
  15
  16 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(obs_comm, mc_observer, "Logging specific to the Communication simcalls observation");
  17
  18 namespace simgrid::kernel::actor {
  19
  20 ActivityTestanySimcall::ActivityTestanySimcall(ActorImpl* actor, const std::vector<activity::ActivityImpl*>& activities,
  21                                                std::string_view fun_call)
  22     : ResultingSimcall(actor, -1), activities_(activities), fun_call_(fun_call)
  23 {
  24   indexes_.clear();
  25   // list all the activities that are ready
  26   for (unsigned i = 0; i < activities_.size(); i++)
  27     if (activities_[i]->test(get_issuer()))
  28       indexes_.push_back(i);
  29 }
  30
  31 int ActivityTestanySimcall::get_max_consider() const
  32 {
  33   return indexes_.size() + 1;
  34 }
  35
  36 void ActivityTestanySimcall::prepare(int times_considered)
  37 {
  38   if (times_considered < static_cast<int>(indexes_.size()))
  39     next_value_ = indexes_.at(times_considered);
  40   else
  41     next_value_ = -1;
  42 }
  43 static void serialize_activity_test(const activity::ActivityImpl* act, std::string const& call_location,
  44                                     std::stringstream& stream)
  45 {
  46   if (const auto* comm = dynamic_cast<activity::CommImpl const*>(act)) {
  47     stream << "  " << (short)mc::Transition::Type::COMM_TEST;
  48     stream << ' ' << comm->get_id();
  49     stream << ' ' << (comm->src_actor_ != nullptr ? comm->src_actor_->get_pid() : -1);
  50     stream << ' ' << (comm->dst_actor_ != nullptr ? comm->dst_actor_->get_pid() : -1);
  51     stream << ' ' << comm->get_mailbox_id();
  52     stream << ' ' << call_location;
  53   } else {
  54     stream << (short)mc::Transition::Type::UNKNOWN;
  55     XBT_CRITICAL("Unknown transition in a test any. Bad things may happen");
  56   }
  57 }
  58 static std::string to_string_activity_test(const activity::ActivityImpl* act)
  59 {
  60   if (const auto* comm = dynamic_cast<activity::CommImpl const*>(act)) {
  61     return "CommTest(comm_id:" + std::to_string(comm->get_id()) +
  62            " src:" + std::to_string(comm->src_actor_ != nullptr ? comm->src_actor_->get_pid() : -1) +
  63            " dst:" + std::to_string(comm->dst_actor_ != nullptr ? comm->dst_actor_->get_pid() : -1) +
  64            " mbox:" + std::to_string(comm->get_mailbox_id()) + ")";
  65   } else {
  66     return "TestUnknownType()";
  67   }
  68 }
  69 void ActivityTestanySimcall::serialize(std::stringstream& stream) const
  70 {
  71   XBT_DEBUG("Serialize %s", to_string().c_str());
  72   stream << (short)mc::Transition::Type::TESTANY << ' ' << activities_.size() << ' ';
  73   for (auto const* act : activities_) {
  74     // fun_call of each activity embedded in the TestAny is not known, so let's use the location of the TestAny itself
  75     serialize_activity_test(act, fun_call_, stream);
  76     stream << ' ';
  77   }
  78   stream << fun_call_;
  79 }
  80 std::string ActivityTestanySimcall::to_string() const
  81 {
  82   std::stringstream buffer("TestAny(");
  83   bool first = true;
  84   for (auto const* act : activities_) {
  85     if (first)
  86       first = false;
  87     else
  88       buffer << " | ";
  89     buffer << to_string_activity_test(act);
  90   }
  91   buffer << ")";
  92   return buffer.str();
  93 }
  94
  95 void ActivityTestSimcall::serialize(std::stringstream& stream) const
  96 {
  97   serialize_activity_test(activity_, fun_call_, stream);
  98 }
  99 std::string ActivityTestSimcall::to_string() const
 100 {
 101   return to_string_activity_test(activity_);
 102 }
 103 static void serialize_activity_wait(const activity::ActivityImpl* act, bool timeout, std::string const& call_location,
 104                                     std::stringstream& stream)
 105 {
 106   if (const auto* comm = dynamic_cast<activity::CommImpl const*>(act)) {
 107     stream << (short)mc::Transition::Type::COMM_WAIT << ' ';
 108     stream << timeout << ' ' << comm->get_id();
 109
 110     stream << ' ' << (comm->src_actor_ != nullptr ? comm->src_actor_->get_pid() : -1);
 111     stream << ' ' << (comm->dst_actor_ != nullptr ? comm->dst_actor_->get_pid() : -1);
 112     stream << ' ' << comm->get_mailbox_id();
 113     stream << ' ' << call_location;
 114   } else {
 115     stream << (short)mc::Transition::Type::UNKNOWN;
 116   }
 117 }
 118 static std::string to_string_activity_wait(const activity::ActivityImpl* act)
 119 {
 120   if (const auto* comm = dynamic_cast<activity::CommImpl const*>(act)) {
 121     return "CommWait(comm_id:" + std::to_string(comm->get_id()) +
 122            " src:" + std::to_string(comm->src_actor_ != nullptr ? comm->src_actor_->get_pid() : -1) +
 123            " dst:" + std::to_string(comm->dst_actor_ != nullptr ? comm->dst_actor_->get_pid() : -1) +
 124            " mbox:" + (comm->get_mailbox() == nullptr ? "-" : comm->get_mailbox()->get_name()) +
 125            "(id:" + std::to_string(comm->get_mailbox_id()) + "))";
 126   } else {
 127     return "WaitUnknownType()";
 128   }
 129 }
 130
 131 void ActivityWaitSimcall::serialize(std::stringstream& stream) const
 132 {
 133   serialize_activity_wait(activity_, timeout_ > 0, fun_call_, stream);
 134 }
 135 void ActivityWaitanySimcall::serialize(std::stringstream& stream) const
 136 {
 137   XBT_DEBUG("Serialize %s", to_string().c_str());
 138   stream << (short)mc::Transition::Type::WAITANY << ' ' << activities_.size() << ' ';
 139   for (auto const* act : activities_) {
 140     // fun_call of each activity embedded in the WaitAny is not known, so let's use the location of the WaitAny itself
 141     serialize_activity_wait(act, timeout_ > 0, fun_call_, stream);
 142     stream << ' ';
 143   }
 144   stream << fun_call_;
 145 }
 146 std::string ActivityWaitSimcall::to_string() const
 147 {
 148   return to_string_activity_wait(activity_);
 149 }
 150 std::string ActivityWaitanySimcall::to_string() const
 151 {
 152   std::stringstream buffer("WaitAny(");
 153   bool first = true;
 154   for (auto const* act : activities_) {
 155     if (first)
 156       first = false;
 157     else
 158       buffer << " | ";
 159     buffer << to_string_activity_test(act);
 160   }
 161   buffer << ")";
 162   return buffer.str();
 163 }
 164 ActivityWaitanySimcall::ActivityWaitanySimcall(ActorImpl* actor, const std::vector<activity::ActivityImpl*>& activities,
 165                                                double timeout, std::string_view fun_call)
 166     : ResultingSimcall(actor, -1), activities_(activities), timeout_(timeout), fun_call_(fun_call)
 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
 175 bool ActivityWaitSimcall::is_enabled()
 176 {
 177   // FIXME: if _sg_mc_timeout == 1 and if we have either a sender or receiver timeout, the transition is enabled
 178   // because even if the communication is not ready, it can timeout and won't block.
 179
 180   return activity_->test(get_issuer());
 181 }
 182
 183 bool ActivityWaitanySimcall::is_enabled()
 184 {
 185   // list all the activities that are ready
 186   indexes_.clear();
 187   for (unsigned i = 0; i < activities_.size(); i++)
 188     if (activities_[i]->test(get_issuer()))
 189       indexes_.push_back(i);
 190
 191   //  if (_sg_mc_timeout && timeout_)  FIXME: deal with the potential timeout of the WaitAny
 192
 193   // FIXME: even if the WaitAny has no timeout, some of the activities may still have one.
 194   // we should iterate over the vector searching for them
 195   return not indexes_.empty();
 196 }
 197
 198 int ActivityWaitanySimcall::get_max_consider() const
 199 {
 200   int res = indexes_.size();
 201   //  if (_sg_mc_timeout && timeout_)
 202   //    res++;
 203
 204   return res;
 205 }
 206
 207 void ActivityWaitanySimcall::prepare(int times_considered)
 208 {
 209   if (times_considered < static_cast<int>(indexes_.size()))
 210     next_value_ = indexes_.at(times_considered);
 211   else
 212     next_value_ = -1;
 213 }
 214
 215 void CommIsendSimcall::serialize(std::stringstream& stream) const
 216 {
 217   /* Note that the comm_ is 0 until after the execution of the simcall */
 218   stream << (short)mc::Transition::Type::COMM_ASYNC_SEND << ' ';
 219   stream << (comm_ ? comm_->get_id() : 0) << ' ' << mbox_->get_id() << ' ' << tag_;
 220   XBT_DEBUG("SendObserver comm:%p mbox:%u tag:%d", comm_, mbox_->get_id(), tag_);
 221   stream << ' ' << fun_call_;
 222 }
 223 std::string CommIsendSimcall::to_string() const
 224 {
 225   return "CommAsyncSend(comm_id: " + std::to_string((comm_ ? comm_->get_id() : 0)) + " mbox:" +
 226          std::to_string(mbox_->get_id()) + " tag: " + std::to_string(tag_) + ")";
 227 }
 228
 229 void CommIrecvSimcall::serialize(std::stringstream& stream) const
 230 {
 231   /* Note that the comm_ is 0 until after the execution of the simcall */
 232   stream << (short)mc::Transition::Type::COMM_ASYNC_RECV << ' ';
 233   stream << (comm_ ? comm_->get_id() : 0) << ' ' << mbox_->get_id() << ' ' << tag_;
 234   XBT_DEBUG("RecvObserver comm:%p mbox:%u tag:%d", comm_, mbox_->get_id(), tag_);
 235   stream << ' ' << fun_call_;
 236 }
 237
 238 std::string CommIrecvSimcall::to_string() const
 239 {
 240   return "CommAsyncRecv(comm_id: " + std::to_string((comm_ ? comm_->get_id() : 0)) + " mbox:" +
 241          std::to_string(mbox_->get_id()) + " tag: " + std::to_string(tag_) + ")";
 242 }
 243
 244 void MessIputSimcall::serialize(std::stringstream& stream) const
 245 {
 246   stream << mess_  << ' ' << queue_;
 247   XBT_DEBUG("PutObserver mess:%p queue:%p", mess_, queue_);
 248 }
 249
 250 std::string MessIputSimcall::to_string() const
 251 {
 252   return "MessAsyncPut(queue:" + queue_->get_name() + ")";
 253 }
 254
 255 void MessIgetSimcall::serialize(std::stringstream& stream) const
 256 {
 257   stream << mess_ << ' ' << queue_;
 258   XBT_DEBUG("GettObserver mess:%p queue:%p", mess_, queue_);
 259 }
 260
 261 std::string MessIgetSimcall::to_string() const
 262 {
 263   return "MessAsyncGet(queue:" + queue_->get_name() + ")";
 264 }
 265
 266
 267
 268 } // namespace simgrid::kernel::actor