X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/5089a0a98b27f5eeee62321dff4f025f1648f025..HEAD:/src/s4u/s4u_Io.cpp diff --git a/src/s4u/s4u_Io.cpp b/src/s4u/s4u_Io.cpp index cca68b0156..c20e891862 100644 --- a/src/s4u/s4u_Io.cpp +++ b/src/s4u/s4u_Io.cpp @@ -1,98 +1,182 @@ -/* Copyright (c) 2018-2019. The SimGrid Team. All rights reserved. */ +/* Copyright (c) 2018-2023. 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 "simgrid/s4u/Disk.hpp" -#include "simgrid/s4u/Io.hpp" -#include "simgrid/s4u/Storage.hpp" +#include +#include +#include +#include + #include "src/kernel/activity/IoImpl.hpp" -#include "xbt/log.h" +#include "src/kernel/actor/ActorImpl.hpp" +#include "src/kernel/actor/SimcallObserver.hpp" + +XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_io, s4u_activity, "S4U asynchronous I/Os"); + +namespace simgrid::s4u { + +Io::Io(kernel::activity::IoImplPtr pimpl) +{ + pimpl_ = pimpl; +} -namespace simgrid { -namespace s4u { +IoPtr Io::init() +{ + auto pimpl = kernel::activity::IoImplPtr(new kernel::activity::IoImpl()); + return IoPtr(static_cast(pimpl->get_iface())); +} + +IoPtr Io::streamto_init(Host* from, const Disk* from_disk, Host* to, const Disk* to_disk) +{ + auto res = Io::init()->set_source(from, from_disk)->set_destination(to, to_disk); + res->set_state(State::STARTING); + return res; +} -Io::Io(sg_disk_t disk, sg_size_t size, OpType type) : disk_(disk), size_(size), type_(type) +IoPtr Io::streamto_async(Host* from, const Disk* from_disk, Host* to, const Disk* to_disk, + uint64_t simulated_size_in_bytes) { - Activity::set_remaining(size_); - pimpl_ = kernel::activity::IoImplPtr(new kernel::activity::IoImpl()); + return Io::init()->set_size(simulated_size_in_bytes)->set_source(from, from_disk)->set_destination(to, to_disk); } -Io::Io(sg_storage_t storage, sg_size_t size, OpType type) : storage_(storage), size_(size), type_(type) +void Io::streamto(Host* from, const Disk* from_disk, Host* to, const Disk* to_disk, uint64_t simulated_size_in_bytes) { - Activity::set_remaining(size_); - pimpl_ = kernel::activity::IoImplPtr(new kernel::activity::IoImpl()); + streamto_async(from, from_disk, to, to_disk, simulated_size_in_bytes)->wait(); } -Io* Io::start() +IoPtr Io::set_source(Host* from, const Disk* from_disk) { - kernel::actor::simcall([this] { - if (storage_) { - (*boost::static_pointer_cast(pimpl_)) - .set_name(get_name()) - .set_storage(storage_->get_impl()) - .set_size(size_) - .set_type(type_) - .start(); - } else { - (*boost::static_pointer_cast(pimpl_)) - .set_name(get_name()) - .set_disk(disk_->get_impl()) - .set_size(size_) - .set_type(type_) - .start(); - } + 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_object_access(pimpl_.get(), [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()); }); - state_ = State::STARTED; + // 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 + start(); + return this; } -Io* Io::cancel() +IoPtr Io::set_destination(Host* to, const Disk* to_disk) { - simgrid::kernel::actor::simcall([this] { boost::static_pointer_cast(pimpl_)->cancel(); }); - state_ = State::CANCELED; + 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_object_access(pimpl_.get(), [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 + start(); + return this; } -Io* Io::wait() +Io* Io::do_start() { - simcall_io_wait(pimpl_); - state_ = State::FINISHED; + kernel::actor::simcall_answered( + [this] { (*boost::static_pointer_cast(pimpl_)).set_name(get_name()).start(); }); + + if (suspended_) + pimpl_->suspend(); + + state_ = State::STARTED; + fire_on_start(); + fire_on_this_start(); return this; } -Io* Io::wait_for(double) +ssize_t Io::deprecated_wait_any_for(const std::vector& ios, double timeout) // XBT_ATTRIB_DEPRECATED_v339 { - THROW_UNIMPLEMENTED; + ActivitySet set; + for (const auto& io : ios) + set.push(io); + + auto* ret = set.wait_any_for(timeout).get(); + for (size_t i = 0; i < ios.size(); i++) + if (ios[i].get() == ret) + return i; + + return -1; } -bool Io::test() +IoPtr Io::set_disk(const_sg_disk_t disk) { - xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::FINISHED); + xbt_assert(state_ == State::INITED || state_ == State::STARTING, "Cannot set disk once the Io is started"); - if (state_ == State::FINISHED) - return true; + kernel::actor::simcall_answered( + [this, disk] { boost::static_pointer_cast(pimpl_)->set_disk(disk->get_impl()); }); - if (state_ == State::INITED) - this->start(); + // Setting the disk may allow to start the activity, let's try + if (state_ == State::STARTING) + start(); - THROW_UNIMPLEMENTED; + return this; +} + +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_object_access(pimpl_.get(), [this, priority] { + boost::static_pointer_cast(pimpl_)->set_sharing_penalty(1. / priority); + }); + return this; +} + +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_object_access( + pimpl_.get(), [this, size] { boost::static_pointer_cast(pimpl_)->set_size(size); }); + set_remaining(size); + return this; +} - // return false +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_object_access( + pimpl_.get(), [this, type] { boost::static_pointer_cast(pimpl_)->set_type(type); }); + return this; +} + +IoPtr Io::update_priority(double priority) +{ + kernel::actor::simcall_answered([this, priority] { + boost::static_pointer_cast(pimpl_)->update_sharing_penalty(1. / priority); + }); + return this; } /** @brief Returns the amount of flops that remain to be done */ -double Io::get_remaining() +double Io::get_remaining() const +{ + return kernel::actor::simcall_object_access( + pimpl_.get(), [this]() { return boost::static_pointer_cast(pimpl_)->get_remaining(); }); +} + +sg_size_t Io::get_performed_ioops() const { - return kernel::actor::simcall( - [this]() { return boost::static_pointer_cast(pimpl_)->get_remaining(); }); + return boost::static_pointer_cast(pimpl_)->get_performed_ioops(); } -sg_size_t Io::get_performed_ioops() +bool Io::is_assigned() const { - return kernel::actor::simcall( - [this]() { return boost::static_pointer_cast(pimpl_)->get_performed_ioops(); }); + 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