src/kernel/activity/MailboxImpl.cpp

   1 /* Copyright (c) 2007-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 "src/kernel/activity/MailboxImpl.hpp"
   7 #include "src/kernel/activity/CommImpl.hpp"
   8
   9 #include <unordered_map>
  10
  11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(ker_mailbox, kernel, "Mailbox implementation");
  12
  13 /******************************************************************************/
  14 /*                           Rendez-Vous Points                               */
  15 /******************************************************************************/
  16
  17 namespace simgrid {
  18 namespace kernel {
  19 namespace activity {
  20
  21 unsigned MailboxImpl::next_id_ = 0;
  22
  23 /** @brief set the receiver of the mailbox to allow eager sends
  24  *  @param actor The receiving dude
  25  */
  26 void MailboxImpl::set_receiver(s4u::ActorPtr actor)
  27 {
  28   if (actor != nullptr)
  29     this->permanent_receiver_ = actor->get_impl();
  30   else
  31     this->permanent_receiver_ = nullptr;
  32 }
  33 /** @brief Pushes a communication activity into a mailbox
  34  *  @param comm What to add
  35  */
  36 void MailboxImpl::push(CommImplPtr comm)
  37 {
  38   comm->set_mailbox(this);
  39   this->comm_queue_.push_back(std::move(comm));
  40 }
  41
  42 /** @brief Removes a communication activity from a mailbox
  43  *  @param comm What to remove
  44  */
  45 void MailboxImpl::remove(const CommImplPtr& comm)
  46 {
  47   xbt_assert(comm->get_mailbox() == this, "Comm %p is in mailbox %s, not mailbox %s", comm.get(),
  48              (comm->get_mailbox() ? comm->get_mailbox()->get_cname() : "(null)"), this->get_cname());
  49
  50   comm->set_mailbox(nullptr);
  51   for (auto it = this->comm_queue_.begin(); it != this->comm_queue_.end(); it++)
  52     if (*it == comm) {
  53       this->comm_queue_.erase(it);
  54       return;
  55     }
  56   xbt_die("Comm %p not found in mailbox %s", comm.get(), this->get_cname());
  57 }
  58
  59 CommImplPtr MailboxImpl::iprobe(int type, bool (*match_fun)(void*, void*, CommImpl*), void* data)
  60 {
  61   XBT_DEBUG("iprobe from %p %p", this, &comm_queue_);
  62
  63   CommImplPtr this_comm(new CommImpl);
  64   CommImplType other_type;
  65   if (type == 1) {
  66     this_comm->set_type(CommImplType::SEND);
  67     other_type = CommImplType::RECEIVE;
  68   } else {
  69     this_comm->set_type(CommImplType::RECEIVE);
  70     other_type = CommImplType::SEND;
  71   }
  72   CommImplPtr other_comm = nullptr;
  73   if (permanent_receiver_ != nullptr && not done_comm_queue_.empty()) {
  74     XBT_DEBUG("first check in the permanent recv mailbox, to see if we already got something");
  75     other_comm = find_matching_comm(other_type, match_fun, data, this_comm, /*done*/ true, /*remove_matching*/ false);
  76   }
  77   if (not other_comm) {
  78     XBT_DEBUG("check if we have more luck in the normal mailbox");
  79     other_comm = find_matching_comm(other_type, match_fun, data, this_comm, /*done*/ false, /*remove_matching*/ false);
  80   }
  81
  82   return other_comm;
  83 }
  84
  85 /**
  86  *  @brief Checks if there is a communication activity queued in comm_queue_ matching our needs
  87  *  @param type The type of communication we are looking for (comm_send, comm_recv)
  88  *  @param match_fun the function to apply
  89  *  @param this_user_data additional parameter to the match_fun
  90  *  @param my_synchro what to compare against
  91  *  @param remove_matching whether or not to clean the found object from the queue
  92  *  @return The communication activity if found, nullptr otherwise
  93  */
  94 CommImplPtr MailboxImpl::find_matching_comm(CommImplType type, bool (*match_fun)(void*, void*, CommImpl*),
  95                                             void* this_user_data, const CommImplPtr& my_synchro, bool done,
  96                                             bool remove_matching)
  97 {
  98   auto& comm_queue      = done ? done_comm_queue_ : comm_queue_;
  99
 100   auto iter = std::find_if(
 101       comm_queue.begin(), comm_queue.end(), [&type, &match_fun, &this_user_data, &my_synchro](const CommImplPtr& comm) {
 102         void* other_user_data = (comm->get_type() == CommImplType::SEND ? comm->src_data_ : comm->dst_data_);
 103         return (comm->get_type() == type && (not match_fun || match_fun(this_user_data, other_user_data, comm.get())) &&
 104                 (not comm->match_fun || comm->match_fun(other_user_data, this_user_data, my_synchro.get())));
 105       });
 106   if (iter == comm_queue.end()) {
 107     XBT_DEBUG("No matching communication synchro found");
 108     return nullptr;
 109   }
 110
 111   const CommImplPtr& comm = *iter;
 112   XBT_DEBUG("Found a matching communication synchro %p", comm.get());
 113 #if SIMGRID_HAVE_MC
 114   comm->mbox_cpy = comm->get_mailbox();
 115 #endif
 116   comm->set_mailbox(nullptr);
 117   CommImplPtr comm_cpy = comm;
 118   if (remove_matching)
 119     comm_queue.erase(iter);
 120   return comm_cpy;
 121 }
 122 } // namespace activity
 123 } // namespace kernel
 124 } // namespace simgrid