copy_data_callback_ = callback;
}
+CommImpl& CommImpl::set_type(CommImplType type)
+{
+ type_ = type;
+ return *this;
+}
+
+CommImpl& CommImpl::set_source(s4u::Host* from)
+{
+ from_ = from;
+ return *this;
+}
+
+CommImpl& CommImpl::set_destination(s4u::Host* to)
+{
+ to_ = to;
+ return *this;
+}
+
CommImpl& CommImpl::set_size(double size)
{
size_ = size;
return *this;
}
-CommImpl::CommImpl(s4u::Host* from, s4u::Host* to, double bytes) : size_(bytes), detached_(true), from_(from), to_(to)
-{
- set_state(State::READY);
-}
-
CommImpl::~CommImpl()
{
XBT_DEBUG("Really free communication %p in state %s (detached = %d)", this, get_state_str(), detached_);
/* If both the sender and the receiver are already there, start the communication */
if (get_state() == State::READY) {
from_ = from_ != nullptr ? from_ : src_actor_->get_host();
+ xbt_assert(from_->is_on());
to_ = to_ != nullptr ? to_ : dst_actor_->get_host();
+ xbt_assert(to_->is_on());
+
/* Getting the network_model from the origin host
* Valid while we have a single network model, otherwise we would need to change this function to first get the
* routes and later create the respective surf actions */
XBT_DEBUG("send from mailbox %p", mbox);
/* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
- CommImplPtr this_comm(new CommImpl(CommImpl::Type::SEND));
+ CommImplPtr this_comm(new CommImpl());
+ this_comm->set_type(CommImplType::SEND);
/* Look for communication synchro matching our needs. We also provide a description of
* ourself so that the other side also gets a chance of choosing if it wants to match with us.
*
* If it is not found then push our communication into the rendez-vous point */
CommImplPtr other_comm =
- mbox->find_matching_comm(CommImpl::Type::RECEIVE, observer->get_match_fun(), observer->get_payload(), this_comm,
+ mbox->find_matching_comm(CommImplType::RECEIVE, observer->get_match_fun(), observer->get_payload(), this_comm,
/*done*/ false, /*remove_matching*/ true);
if (not other_comm) {
ActivityImplPtr CommImpl::irecv(actor::CommIrecvSimcall* observer)
{
- CommImplPtr this_synchro(new CommImpl(CommImpl::Type::RECEIVE));
+ CommImplPtr this_synchro(new CommImpl());
+ this_synchro->set_type(CommImplType::RECEIVE);
+
auto* mbox = observer->get_mailbox();
XBT_DEBUG("recv from mbox %p. this_synchro=%p", mbox, this_synchro.get());
if (mbox->is_permanent() && mbox->has_some_done_comm()) {
XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication");
// find a match in the list of already received comms
- other_comm = mbox->find_matching_comm(CommImpl::Type::SEND, observer->get_match_fun(), observer->get_payload(),
+ other_comm = mbox->find_matching_comm(CommImplType::SEND, observer->get_match_fun(), observer->get_payload(),
this_synchro, /*done*/ true, /*remove_matching*/ true);
// if not found, assume the receiver came first, register it to the mailbox in the classical way
if (not other_comm) {
* ourself so that the other side also gets a chance of choosing if it wants to match with us.
*
* If it is not found then push our communication into the rendez-vous point */
- other_comm = mbox->find_matching_comm(CommImpl::Type::SEND, observer->get_match_fun(), observer->get_payload(),
+ other_comm = mbox->find_matching_comm(CommImplType::SEND, observer->get_match_fun(), observer->get_payload(),
this_synchro, /*done*/ false, /*remove_matching*/ true);
if (other_comm == nullptr) {
set_state(State::SRC_TIMEOUT);
else if (dst_timeout_ && dst_timeout_->get_state() == resource::Action::State::FINISHED)
set_state(State::DST_TIMEOUT);
- else if (src_timeout_ && src_timeout_->get_state() == resource::Action::State::FAILED)
+ else if ((from_ && not from_->is_on()) || (src_timeout_ && src_timeout_->get_state() == resource::Action::State::FAILED))
set_state(State::SRC_HOST_FAILURE);
- else if (dst_timeout_ && dst_timeout_->get_state() == resource::Action::State::FAILED)
+ else if ((to_ && not to_->is_on()) || (dst_timeout_ && dst_timeout_->get_state() == resource::Action::State::FAILED))
set_state(State::DST_HOST_FAILURE);
else if (surf_action_ && surf_action_->get_state() == resource::Action::State::FAILED) {
set_state(State::LINK_FAILURE);
- } else
+ } else if (get_state() == State::RUNNING) {
+ xbt_assert(from_ && from_->is_on());
+ xbt_assert(to_ && to_->is_on());
set_state(State::DONE);
+ }
XBT_DEBUG("CommImpl::post(): comm %p, state %s, src_proc %p, dst_proc %p, detached: %d", this, get_state_str(),
src_actor_.get(), dst_actor_.get(), detached_);
unsigned MailboxImpl::next_id_ = 0;
+ MailboxImpl::~MailboxImpl()
+ {
+ clear();
+ set_receiver(nullptr);
+ }
+
/** @brief set the receiver of the mailbox to allow eager sends
* @param actor The receiving dude
*/
void MailboxImpl::set_receiver(s4u::ActorPtr actor)
{
+ if (this->permanent_receiver_) {
+ std::vector<MailboxImpl*>& mboxes = this->permanent_receiver_->mailboxes;
+ mboxes.erase(std::remove(mboxes.begin(), mboxes.end(), this), mboxes.end());
+ }
+
if (actor != nullptr)
this->permanent_receiver_ = actor->get_impl();
else
xbt_die("Comm %p not found in mailbox %s", comm.get(), this->get_cname());
}
+ /** @brief Removes all communication activities from a mailbox
+ */
+ void MailboxImpl::clear()
+ {
+ for (auto comm : done_comm_queue_) {
+ comm->cancel();
+ comm->set_state(State::DST_HOST_FAILURE);
+ }
+ done_comm_queue_.clear();
+
+ // CommImpl::cancel() will remove the comm from the mailbox..
+ while (not comm_queue_.empty()) {
+ auto comm = comm_queue_.back();
+ if (comm->get_state() == State::WAITING && not comm->detached()) {
+ comm->cancel();
+ comm->set_state(State::DST_HOST_FAILURE);
+ } else
+ comm_queue_.pop_back();
+ }
+ }
+
CommImplPtr MailboxImpl::iprobe(int type, bool (*match_fun)(void*, void*, CommImpl*), void* data)
{
XBT_DEBUG("iprobe from %p %p", this, &comm_queue_);
- CommImplPtr this_comm;
- CommImpl::Type smx_type;
+ CommImplPtr this_comm(new CommImpl);
+ CommImplType other_type;
if (type == 1) {
- this_comm = CommImplPtr(new CommImpl(CommImpl::Type::SEND));
- smx_type = CommImpl::Type::RECEIVE;
+ this_comm->set_type(CommImplType::SEND);
+ other_type = CommImplType::RECEIVE;
} else {
- this_comm = CommImplPtr(new CommImpl(CommImpl::Type::RECEIVE));
- smx_type = CommImpl::Type::SEND;
+ this_comm->set_type(CommImplType::RECEIVE);
+ other_type = CommImplType::SEND;
}
CommImplPtr other_comm = 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_comm = find_matching_comm(smx_type, match_fun, data, this_comm, /*done*/ true, /*remove_matching*/ false);
+ other_comm = find_matching_comm(other_type, match_fun, data, this_comm, /*done*/ true, /*remove_matching*/ false);
}
if (not other_comm) {
XBT_DEBUG("check if we have more luck in the normal mailbox");
- other_comm = find_matching_comm(smx_type, match_fun, data, this_comm, /*done*/ false, /*remove_matching*/ false);
+ other_comm = find_matching_comm(other_type, match_fun, data, this_comm, /*done*/ false, /*remove_matching*/ false);
}
return other_comm;
* @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, bool (*match_fun)(void*, void*, CommImpl*),
+CommImplPtr MailboxImpl::find_matching_comm(CommImplType type, bool (*match_fun)(void*, void*, CommImpl*),
void* this_user_data, const CommImplPtr& my_synchro, bool done,
bool remove_matching)
{
auto iter = std::find_if(
comm_queue.begin(), comm_queue.end(), [&type, &match_fun, &this_user_data, &my_synchro](const CommImplPtr& comm) {
- void* other_user_data = (comm->type_ == CommImpl::Type::SEND ? comm->src_data_ : comm->dst_data_);
- return (comm->type_ == type && (not match_fun || match_fun(this_user_data, other_user_data, comm.get())) &&
+ void* other_user_data = (comm->get_type() == CommImplType::SEND ? comm->src_data_ : comm->dst_data_);
+ return (comm->get_type() == type && (not match_fun || 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())));
});
if (iter == comm_queue.end()) {
/** @brief Public interface */
unsigned get_id() const { return id_; }
+ ~MailboxImpl();
+
const s4u::Mailbox* get_iface() const { return &piface_; }
s4u::Mailbox* get_iface() { return &piface_; }
void push(CommImplPtr comm);
void push_done(CommImplPtr done_comm) { done_comm_queue_.push_back(done_comm); }
void remove(const CommImplPtr& comm);
+ void clear();
CommImplPtr iprobe(int type, bool (*match_fun)(void*, void*, CommImpl*), void* data);
- CommImplPtr find_matching_comm(CommImpl::Type type, bool (*match_fun)(void*, void*, CommImpl*), void* this_user_data,
+ CommImplPtr find_matching_comm(CommImplType type, bool (*match_fun)(void*, void*, CommImpl*), void* this_user_data,
const CommImplPtr& my_synchro, bool done, bool remove_matching);
bool is_permanent() const { return permanent_receiver_ != nullptr; }
actor::ActorImplPtr get_permanent_receiver() const { return permanent_receiver_; }
