X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/9ceefed14c83a0f6ea5f78e3acafd53181dc4fa1..d94f920eb99dc33e8c592e89bcbbb3a8e50fe919:/src/s4u/s4u_Io.cpp diff --git a/src/s4u/s4u_Io.cpp b/src/s4u/s4u_Io.cpp index 83c8a1dba7..6d34be90e0 100644 --- a/src/s4u/s4u_Io.cpp +++ b/src/s4u/s4u_Io.cpp @@ -1,4 +1,4 @@ -/* Copyright (c) 2018-2021. The SimGrid Team. All rights reserved. */ +/* Copyright (c) 2018-2022. 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. */ @@ -11,8 +11,9 @@ #include "src/kernel/actor/ActorImpl.hpp" #include "src/kernel/actor/SimcallObserver.hpp" -namespace simgrid { -namespace s4u { +XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_io, s4u_activity, "S4U asynchronous I/Os"); + +namespace simgrid::s4u { xbt::signal Io::on_start; Io::Io(kernel::activity::IoImplPtr pimpl) @@ -26,9 +27,64 @@ IoPtr Io::init() return IoPtr(static_cast(pimpl->get_iface())); } +IoPtr Io::streamto_init(Host* from, Disk* from_disk, Host* to, Disk* to_disk) +{ + auto res = Io::init()->set_source(from, from_disk)->set_destination(to, to_disk); + res->set_state(State::STARTING); + return res; +} + +IoPtr Io::streamto_async(Host* from, Disk* from_disk, Host* to, Disk* to_disk, uint64_t simulated_size_in_bytes) +{ + return Io::init()->set_size(simulated_size_in_bytes)->set_source(from, from_disk)->set_destination(to, to_disk); +} + +void Io::streamto(Host* from, Disk* from_disk, Host* to, Disk* to_disk, uint64_t simulated_size_in_bytes) +{ + streamto_async(from, from_disk, to, to_disk, simulated_size_in_bytes)->wait(); +} + +IoPtr Io::set_source(Host* from, Disk* from_disk) +{ + xbt_assert(state_ == State::INITED || state_ == State::STARTING, + "Cannot change the source of an IO stream once it's started (state: %s)", to_c_str(state_)); + kernel::actor::simcall_answered( + [this, from, from_disk] { + boost::static_pointer_cast(pimpl_)->set_host(from); + if (from_disk != nullptr) + boost::static_pointer_cast(pimpl_)->set_disk(from_disk->get_impl()); + }); + // Setting 'source' may allow to start the activity, let's try + if (state_ == State::STARTING && remains_ <= 0) + XBT_DEBUG("This IO has a size of 0 byte. It cannot start yet"); + else + vetoable_start(); + + return this; +} + +IoPtr Io::set_destination(Host* to, Disk* to_disk) +{ + xbt_assert(state_ == State::INITED || state_ == State::STARTING, + "Cannot change the source of an IO stream once it's started (state: %s)", to_c_str(state_)); + kernel::actor::simcall_answered( + [this, to, to_disk] { + boost::static_pointer_cast(pimpl_)->set_dst_host(to); + if (to_disk != nullptr) + boost::static_pointer_cast(pimpl_)->set_dst_disk(to_disk->get_impl()); + }); + // Setting 'destination' may allow to start the activity, let's try + if (state_ == State::STARTING && remains_ <= 0) + XBT_DEBUG("This IO has a size of 0 byte. It cannot start yet"); + else + vetoable_start(); + + return this; +} + Io* Io::start() { - kernel::actor::simcall( + kernel::actor::simcall_answered( [this] { (*boost::static_pointer_cast(pimpl_)).set_name(get_name()).start(); }); if (suspended_) @@ -41,27 +97,17 @@ Io* Io::start() ssize_t Io::wait_any_for(const std::vector& ios, double timeout) { - std::vector rios(ios.size()); - std::transform(begin(ios), end(ios), begin(rios), - [](const IoPtr& io) { return static_cast(io->pimpl_.get()); }); - - kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); - kernel::actor::IoWaitanySimcall observer{issuer, rios, timeout}; - ssize_t changed_pos = kernel::actor::simcall_blocking( - [&observer] { - kernel::activity::IoImpl::wait_any_for(observer.get_issuer(), observer.get_ios(), observer.get_timeout()); - }, - &observer); - if (changed_pos != -1) - ios.at(changed_pos)->complete(State::FINISHED); - return changed_pos; + std::vector activities; + for (const auto& io : ios) + activities.push_back(boost::dynamic_pointer_cast(io)); + return Activity::wait_any_for(activities, timeout); } IoPtr Io::set_disk(const_sg_disk_t disk) { xbt_assert(state_ == State::INITED || state_ == State::STARTING, "Cannot set disk once the Io is started"); - kernel::actor::simcall( + kernel::actor::simcall_answered( [this, disk] { boost::static_pointer_cast(pimpl_)->set_disk(disk->get_impl()); }); // Setting the disk may allow to start the activity, let's try @@ -75,7 +121,7 @@ IoPtr Io::set_priority(double priority) { xbt_assert(state_ == State::INITED || state_ == State::STARTING, "Cannot change the priority of an io after its start"); - kernel::actor::simcall([this, priority] { + kernel::actor::simcall_answered([this, priority] { boost::static_pointer_cast(pimpl_)->set_sharing_penalty(1. / priority); }); return this; @@ -84,23 +130,23 @@ IoPtr Io::set_priority(double priority) IoPtr Io::set_size(sg_size_t size) { xbt_assert(state_ == State::INITED || state_ == State::STARTING, "Cannot set size once the Io is started"); - kernel::actor::simcall( + kernel::actor::simcall_answered( [this, size] { boost::static_pointer_cast(pimpl_)->set_size(size); }); - Activity::set_remaining(size); + set_remaining(size); return this; } IoPtr Io::set_op_type(OpType type) { xbt_assert(state_ == State::INITED || state_ == State::STARTING, "Cannot set size once the Io is started"); - kernel::actor::simcall( + kernel::actor::simcall_answered( [this, type] { boost::static_pointer_cast(pimpl_)->set_type(type); }); return this; } IoPtr Io::update_priority(double priority) { - kernel::actor::simcall([this, priority] { + kernel::actor::simcall_answered([this, priority] { boost::static_pointer_cast(pimpl_)->update_sharing_penalty(1. / priority); }); return this; @@ -109,7 +155,7 @@ IoPtr Io::update_priority(double priority) /** @brief Returns the amount of flops that remain to be done */ double Io::get_remaining() const { - return kernel::actor::simcall( + return kernel::actor::simcall_answered( [this]() { return boost::static_pointer_cast(pimpl_)->get_remaining(); }); } @@ -120,8 +166,11 @@ sg_size_t Io::get_performed_ioops() const bool Io::is_assigned() const { - return boost::static_pointer_cast(pimpl_)->get_disk() != nullptr; + if (boost::static_pointer_cast(pimpl_)->get_host() == nullptr) { + return boost::static_pointer_cast(pimpl_)->get_disk() != nullptr; + } else { + return boost::static_pointer_cast(pimpl_)->get_dst_host() != nullptr; + } } -} // namespace s4u -} // namespace simgrid +} // namespace simgrid::s4u