xbt::signal<void(Comm const&, bool is_sender)> Comm::on_start;
xbt::signal<void(Comm const&)> Comm::on_completion;
+void Comm::complete(Activity::State state)
+{
+ Activity::complete(state);
+ on_completion(*this);
+}
+
Comm::~Comm()
{
if (state_ == State::STARTED && not detached_ &&
std::transform(begin(*comms), end(*comms), begin(rcomms),
[](const CommPtr& comm) { return static_cast<kernel::activity::CommImpl*>(comm->pimpl_.get()); });
int changed_pos = simcall_comm_waitany(rcomms.data(), rcomms.size(), timeout);
- if (changed_pos != -1) {
- on_completion(*(comms->at(changed_pos)));
- comms->at(changed_pos)->release_dependencies();
- }
+ if (changed_pos != -1)
+ comms->at(changed_pos)->complete(State::FINISHED);
return changed_pos;
}
return res;
}
+void Comm::sendto(Host* from, Host* to, double simulated_size_in_bytes)
+{
+ sendto_async(from, to, simulated_size_in_bytes)->wait();
+}
+
Comm* Comm::start()
{
xbt_assert(get_state() == State::INITED || get_state() == State::STARTING,
return this;
}
-/** @brief Block the calling actor until the communication is finished */
-Comm* Comm::wait()
-{
- return this->wait_for(-1);
-}
-
/** @brief Block the calling actor until the communication is finished, or until timeout
*
* On timeout, an exception is thrown and the communication is invalidated.
simcall_comm_recv(receiver_, mailbox_->get_impl(), dst_buff_, &dst_buff_size_, match_fun_, copy_data_function_,
get_user_data(), timeout, rate_);
}
- state_ = State::FINISHED;
- this->release_dependencies();
break;
case State::STARTED:
simcall_comm_wait(get_impl(), timeout);
- state_ = State::FINISHED;
- this->release_dependencies();
break;
case State::CANCELED:
default:
THROW_IMPOSSIBLE;
}
- on_completion(*this);
+ complete(State::FINISHED);
return this;
}
[](const CommPtr& comm) { return static_cast<kernel::activity::CommImpl*>(comm->pimpl_.get()); });
int changed_pos = simcall_comm_testany(rcomms.data(), rcomms.size());
if (changed_pos != -1)
- comms->at(changed_pos)->release_dependencies();
+ comms->at(changed_pos)->complete(State::FINISHED);
return changed_pos;
}
return this;
}
-Comm* Comm::cancel()
-{
- kernel::actor::simcall([this] {
- if (pimpl_)
- boost::static_pointer_cast<kernel::activity::CommImpl>(pimpl_)->cancel();
- });
- state_ = State::CANCELED;
- return this;
-}
-
-bool Comm::test()
+bool Comm::test() // TODO: merge with Activity::test, once modernized
{
xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::STARTING ||
- state_ == State::FINISHED);
+ state_ == State::CANCELED || state_ == State::FINISHED);
- if (state_ == State::FINISHED)
+ if (state_ == State::CANCELED || state_ == State::FINISHED)
return true;
if (state_ == State::INITED || state_ == State::STARTING)
this->vetoable_start();
if (simcall_comm_test(get_impl())) {
- state_ = State::FINISHED;
- this->release_dependencies();
+ complete(State::FINISHED);
return true;
}
return false;