X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/1d381feb9b44e9d94e608c9b389c453128d44dff..d94f920eb99dc33e8c592e89bcbbb3a8e50fe919:/src/s4u/s4u_Io.cpp diff --git a/src/s4u/s4u_Io.cpp b/src/s4u/s4u_Io.cpp index 4cf33bc8d4..6d34be90e0 100644 --- a/src/s4u/s4u_Io.cpp +++ b/src/s4u/s4u_Io.cpp @@ -1,112 +1,176 @@ -/* Copyright (c) 2018-2019. 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. */ -#include "simgrid/s4u/Disk.hpp" -#include "simgrid/s4u/Io.hpp" -#include "simgrid/s4u/Storage.hpp" +#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 IOs"); +XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_io, s4u_activity, "S4U asynchronous I/Os"); -namespace simgrid { -namespace s4u { +namespace simgrid::s4u { +xbt::signal Io::on_start; -Io::Io(sg_disk_t disk, sg_size_t size, OpType type) : disk_(disk), size_(size), type_(type) +Io::Io(kernel::activity::IoImplPtr pimpl) { - Activity::set_remaining(size_); - pimpl_ = kernel::activity::IoImplPtr(new kernel::activity::IoImpl()); + pimpl_ = pimpl; } -Io::Io(sg_storage_t storage, sg_size_t size, OpType type) : storage_(storage), size_(size), type_(type) +IoPtr Io::init() { - Activity::set_remaining(size_); - pimpl_ = kernel::activity::IoImplPtr(new kernel::activity::IoImpl()); + auto pimpl = kernel::activity::IoImplPtr(new kernel::activity::IoImpl()); + return IoPtr(static_cast(pimpl->get_iface())); } -Io* Io::start() +IoPtr Io::streamto_init(Host* from, Disk* from_disk, Host* to, Disk* to_disk) { - kernel::actor::simcall([this] { - if (storage_) { - (*boost::static_pointer_cast(pimpl_)) - .set_name(name_) - .set_storage(storage_->get_impl()) - .set_size(size_) - .set_type(type_) - .start(); - } else { - (*boost::static_pointer_cast(pimpl_)) - .set_name(name_) - .set_disk(disk_->get_impl()) - .set_size(size_) - .set_type(type_) - .start(); - } - }); - state_ = State::STARTED; + 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; } -Io* Io::cancel() +IoPtr Io::set_destination(Host* to, 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_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::wait() +Io* Io::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; + on_start(*this); return this; } -Io* Io::wait_for(double) +ssize_t Io::wait_any_for(const std::vector& ios, double timeout) { - THROW_UNIMPLEMENTED; + std::vector activities; + for (const auto& io : ios) + activities.push_back(boost::dynamic_pointer_cast(io)); + return Activity::wait_any_for(activities, timeout); } -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) + vetoable_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_answered([this, priority] { + boost::static_pointer_cast(pimpl_)->set_sharing_penalty(1. / priority); + }); + return this; +} - // return false +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_answered( + [this, size] { boost::static_pointer_cast(pimpl_)->set_size(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_answered( + [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( + return kernel::actor::simcall_answered( [this]() { return boost::static_pointer_cast(pimpl_)->get_remaining(); }); } -sg_size_t Io::get_performed_ioops() +sg_size_t Io::get_performed_ioops() const { - return kernel::actor::simcall( - [this]() { return boost::static_pointer_cast(pimpl_)->get_performed_ioops(); }); + return boost::static_pointer_cast(pimpl_)->get_performed_ioops(); } -void intrusive_ptr_release(simgrid::s4u::Io* i) +bool Io::is_assigned() const { - if (i->refcount_.fetch_sub(1, std::memory_order_release) == 1) { - std::atomic_thread_fence(std::memory_order_acquire); - delete i; + 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; } } -void intrusive_ptr_add_ref(simgrid::s4u::Io* i) -{ - i->refcount_.fetch_add(1, std::memory_order_relaxed); -} -} // namespace s4u -} // namespace simgrid +} // namespace simgrid::s4u