X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/a16ca05f4f79be7979ee3425e9bcae5633201d06..35c78ad8f81fb892c32e2a16d3cbb0ed84480e51:/src/kernel/activity/MailboxImpl.cpp diff --git a/src/kernel/activity/MailboxImpl.cpp b/src/kernel/activity/MailboxImpl.cpp index 936b249718..8ad9e02ce9 100644 --- a/src/kernel/activity/MailboxImpl.cpp +++ b/src/kernel/activity/MailboxImpl.cpp @@ -1,10 +1,9 @@ -/* Copyright (c) 2007-2017. The SimGrid Team. All rights reserved. */ +/* Copyright (c) 2007-2019. The SimGrid Team. All rights reserved. */ /* This program is free software; you can redistribute it and/or modify it * under the terms of the license (GNU LGPL) which comes with this package. */ #include "src/kernel/activity/MailboxImpl.hpp" - #include "src/kernel/activity/CommImpl.hpp" XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_mailbox, simix, "Mailbox implementation"); @@ -13,7 +12,7 @@ static std::map* mailboxes = new std::mapfind(name); if (mbox != mailboxes->end()) @@ -34,26 +33,29 @@ MailboxImpl* MailboxImpl::byNameOrNull(const char* name) else return nullptr; } + /** @brief Returns the mailbox of that name, newly created on need */ -MailboxImpl* MailboxImpl::byNameOrCreate(const char* name) +MailboxImpl* MailboxImpl::by_name_or_create(const std::string& name) { - xbt_assert(name, "Mailboxes must have a name"); /* two processes may have pushed the same mbox_create simcall at the same time */ auto m = mailboxes->find(name); if (m == mailboxes->end()) { smx_mailbox_t mbox = new MailboxImpl(name); - XBT_DEBUG("Creating a mailbox at %p with name %s", mbox, name); + XBT_DEBUG("Creating a mailbox at %p with name %s", mbox, name.c_str()); (*mailboxes)[mbox->name_] = mbox; return mbox; } else return m->second; } /** @brief set the receiver of the mailbox to allow eager sends - * \param actor The receiving dude + * @param actor The receiving dude */ -void MailboxImpl::setReceiver(s4u::ActorPtr actor) +void MailboxImpl::set_receiver(s4u::ActorPtr actor) { - this->permanent_receiver = actor.get()->getImpl(); + if (actor != nullptr) + this->permanent_receiver_ = actor->get_impl(); + else + this->permanent_receiver_ = nullptr; } /** @brief Pushes a communication activity into a mailbox * @param comm What to add @@ -61,7 +63,7 @@ void MailboxImpl::setReceiver(s4u::ActorPtr actor) void MailboxImpl::push(activity::CommImplPtr comm) { comm->mbox = this; - this->comm_queue.push_back(std::move(comm)); + this->comm_queue_.push_back(std::move(comm)); } /** @brief Removes a communication activity from a mailbox @@ -73,14 +75,87 @@ void MailboxImpl::remove(smx_activity_t activity) boost::static_pointer_cast(activity); xbt_assert(comm->mbox == this, "Comm %p is in mailbox %s, not mailbox %s", comm.get(), - (comm->mbox ? comm->mbox->name_ : "(null)"), this->name_); + (comm->mbox ? comm->mbox->get_cname() : "(null)"), this->get_cname()); comm->mbox = nullptr; - for (auto it = this->comm_queue.begin(); it != this->comm_queue.end(); it++) + for (auto it = this->comm_queue_.begin(); it != this->comm_queue_.end(); it++) if (*it == comm) { - this->comm_queue.erase(it); + this->comm_queue_.erase(it); return; } - xbt_die("Comm %p not found in mailbox %s", comm.get(), this->name_); + xbt_die("Comm %p not found in mailbox %s", comm.get(), this->get_cname()); +} + +smx_activity_t MailboxImpl::iprobe(int type, int (*match_fun)(void*, void*, CommImpl*), void* data) +{ + XBT_DEBUG("iprobe from %p %p", this, &comm_queue_); + + CommImplPtr this_comm; + CommImpl::Type smx_type; + if (type == 1) { + this_comm = CommImplPtr(new CommImpl(CommImpl::Type::SEND)); + smx_type = CommImpl::Type::RECEIVE; + } else { + this_comm = CommImplPtr(new CommImpl(CommImpl::Type::RECEIVE)); + smx_type = CommImpl::Type::SEND; + } + smx_activity_t other_synchro = nullptr; + if (permanent_receiver_ != nullptr && not done_comm_queue_.empty()) { + XBT_DEBUG("first check in the permanent recv mailbox, to see if we already got something"); + other_synchro = find_matching_comm(smx_type, match_fun, data, this_comm, /*done*/ true, /*remove_matching*/ false); + } + if (not other_synchro) { + XBT_DEBUG("check if we have more luck in the normal mailbox"); + other_synchro = find_matching_comm(smx_type, match_fun, data, this_comm, /*done*/ false, /*remove_matching*/ false); + } + + return other_synchro; +} + +/** + * @brief Checks if there is a communication activity queued in comm_queue_ matching our needs + * @param type The type of communication we are looking for (comm_send, comm_recv) + * @param match_fun the function to apply + * @param this_user_data additional parameter to the match_fun + * @param my_synchro what to compare against + * @param remove_matching whether or not to clean the found object from the queue + * @return The communication activity if found, nullptr otherwise + */ +CommImplPtr MailboxImpl::find_matching_comm(CommImpl::Type type, int (*match_fun)(void*, void*, CommImpl*), + void* this_user_data, CommImplPtr my_synchro, bool done, + bool remove_matching) +{ + void* other_user_data = nullptr; + boost::circular_buffer_space_optimized* deque; + if (done) + deque = &done_comm_queue_; + else + deque = &comm_queue_; + + for (auto it = deque->begin(); it != deque->end(); it++) { + CommImplPtr comm = boost::dynamic_pointer_cast(std::move(*it)); + + if (comm->type == CommImpl::Type::SEND) { + other_user_data = comm->src_data_; + } else if (comm->type == CommImpl::Type::RECEIVE) { + other_user_data = comm->dst_data_; + } + if (comm->type == type && (match_fun == nullptr || match_fun(this_user_data, other_user_data, comm.get())) && + (not comm->match_fun || comm->match_fun(other_user_data, this_user_data, my_synchro.get()))) { + XBT_DEBUG("Found a matching communication synchro %p", comm.get()); + if (remove_matching) + deque->erase(it); +#if SIMGRID_HAVE_MC + comm->mbox_cpy = comm->mbox; +#endif + comm->mbox = nullptr; + return comm; + } + XBT_DEBUG("Sorry, communication synchro %p does not match our needs:" + " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)", + comm.get(), (int)comm->type, (int)type); + } + XBT_DEBUG("No matching communication synchro found"); + return nullptr; } } }