X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/435eac9875f286908c7ead8f6b50fd9bb23690b8..d94f920eb99dc33e8c592e89bcbbb3a8e50fe919:/src/s4u/s4u_Io.cpp diff --git a/src/s4u/s4u_Io.cpp b/src/s4u/s4u_Io.cpp index 4dfbfa2c03..6d34be90e0 100644 --- a/src/s4u/s4u_Io.cpp +++ b/src/s4u/s4u_Io.cpp @@ -1,89 +1,176 @@ -/* Copyright (c) 2018. 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/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 I/Os"); + +namespace simgrid::s4u { +xbt::signal Io::on_start; + +Io::Io(kernel::activity::IoImplPtr pimpl) +{ + pimpl_ = pimpl; +} -XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_io, s4u_activity, "S4U asynchronous IOs"); +IoPtr Io::init() +{ + auto pimpl = kernel::activity::IoImplPtr(new kernel::activity::IoImpl()); + return IoPtr(static_cast(pimpl->get_iface())); +} -namespace simgrid { -namespace s4u { +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; +} -Activity* Io::start() +IoPtr Io::streamto_async(Host* from, Disk* from_disk, Host* to, Disk* to_disk, uint64_t simulated_size_in_bytes) { - pimpl_ = simcall_io_start("", size_, storage_); + 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_answered( + [this] { (*boost::static_pointer_cast(pimpl_)).set_name(get_name()).start(); }); + + if (suspended_) + pimpl_->suspend(); + state_ = State::STARTED; + on_start(*this); + return this; +} + +ssize_t Io::wait_any_for(const std::vector& ios, double timeout) +{ + 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_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 + if (state_ == State::STARTING) + vetoable_start(); + + 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; } -Activity* Io::cancel() +IoPtr Io::set_size(sg_size_t size) { - simgrid::simix::simcall([this] { dynamic_cast(pimpl_.get())->cancel(); }); - state_ = State::CANCELED; + 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; } -Activity* Io::wait() +IoPtr Io::set_op_type(OpType type) { - // simcall_execution_wait(pimpl_); - state_ = State::FINISHED; + 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; } -Activity* Io::wait(double timeout) +IoPtr Io::update_priority(double priority) { - THROW_UNIMPLEMENTED; + kernel::actor::simcall_answered([this, priority] { + boost::static_pointer_cast(pimpl_)->update_sharing_penalty(1. / priority); + }); return this; } -// -///** @brief Returns whether the state of the exec is finished */ -// bool Exec::test() -//{ -// xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::FINISHED); -// -// if (state_ == State::FINISHED) -// return true; -// -// if (state_ == State::INITED) -// this->start(); -// -// if (simcall_execution_test(pimpl_)) { -// state_ = State::FINISHED; -// return true; -// } -// -// return false; -//} /** @brief Returns the amount of flops that remain to be done */ -double Io::get_remaining() +double Io::get_remaining() const { - return simgrid::simix::simcall( - [this]() { return boost::static_pointer_cast(pimpl_)->get_remaining(); }); + return kernel::actor::simcall_answered( + [this]() { return boost::static_pointer_cast(pimpl_)->get_remaining(); }); } -// double Io::get_remaining_ratio() -//{ -// return simgrid::simix::simcall([this]() { -// return boost::static_pointer_cast(pimpl_)->get_remaining_ratio(); -// }); -//} - -void intrusive_ptr_release(simgrid::s4u::Io* e) +sg_size_t Io::get_performed_ioops() const { - if (e->refcount_.fetch_sub(1, std::memory_order_release) == 1) { - std::atomic_thread_fence(std::memory_order_acquire); - delete e; - } + return boost::static_pointer_cast(pimpl_)->get_performed_ioops(); } -void intrusive_ptr_add_ref(simgrid::s4u::Io* e) +bool Io::is_assigned() const { - e->refcount_.fetch_add(1, std::memory_order_relaxed); + 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