Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
add capacity to set priorities on I/Os + example
[simgrid.git] / src / kernel / activity / IoImpl.cpp
1 /* Copyright (c) 2007-2021. The SimGrid Team. All rights reserved.          */
2
3 /* This program is free software; you can redistribute it and/or modify it
4  * under the terms of the license (GNU LGPL) which comes with this package. */
5
6 #include <simgrid/Exception.hpp>
7 #include <simgrid/kernel/routing/NetPoint.hpp>
8 #include <simgrid/s4u/Engine.hpp>
9 #include <simgrid/s4u/Host.hpp>
10
11 #include "mc/mc.h"
12 #include "src/kernel/activity/IoImpl.hpp"
13 #include "src/kernel/actor/ActorImpl.hpp"
14 #include "src/kernel/actor/SimcallObserver.hpp"
15 #include "src/kernel/context/Context.hpp"
16 #include "src/kernel/resource/DiskImpl.hpp"
17 #include "src/mc/mc_replay.hpp"
18 #include "src/surf/cpu_interface.hpp"
19
20 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_io, simix, "Logging specific to SIMIX (io)");
21
22 namespace simgrid {
23 namespace kernel {
24 namespace activity {
25
26 IoImpl::IoImpl()
27 {
28   piface_ = new s4u::Io(this);
29 }
30
31 IoImpl& IoImpl::set_sharing_penalty(double sharing_penalty)
32 {
33   sharing_penalty_ = sharing_penalty;
34   return *this;
35 }
36
37 IoImpl& IoImpl::set_timeout(double timeout)
38 {
39   const s4u::Host* host = get_disk()->get_host();
40   timeout_detector_     = host->get_cpu()->sleep(timeout);
41   timeout_detector_->set_activity(this);
42   return *this;
43 }
44
45 IoImpl& IoImpl::set_type(s4u::Io::OpType type)
46 {
47   type_ = type;
48   return *this;
49 }
50
51 IoImpl& IoImpl::set_size(sg_size_t size)
52 {
53   size_ = size;
54   return *this;
55 }
56
57 IoImpl& IoImpl::set_disk(resource::DiskImpl* disk)
58 {
59   disk_ = disk;
60   return *this;
61 }
62
63 IoImpl* IoImpl::start()
64 {
65   state_ = State::RUNNING;
66   surf_action_ =
67       disk_->get_host()->get_netpoint()->get_englobing_zone()->get_disk_model()->io_start(disk_, size_, type_);
68   surf_action_->set_sharing_penalty(sharing_penalty_);
69   surf_action_->set_activity(this);
70
71   XBT_DEBUG("Create IO synchro %p %s", this, get_cname());
72
73   return this;
74 }
75
76 void IoImpl::post()
77 {
78   performed_ioops_ = surf_action_->get_cost();
79   if (surf_action_->get_state() == resource::Action::State::FAILED) {
80     if (disk_ && not disk_->is_on())
81       state_ = State::FAILED;
82     else
83       state_ = State::CANCELED;
84   } else if (timeout_detector_ && timeout_detector_->get_state() == resource::Action::State::FINISHED) {
85     if (surf_action_->get_remains() > 0.0) {
86       surf_action_->set_state(resource::Action::State::FAILED);
87       state_ = State::TIMEOUT;
88     } else {
89       state_ = State::DONE;
90     }
91   } else {
92     state_ = State::DONE;
93   }
94
95   clean_action();
96   if (timeout_detector_) {
97     timeout_detector_->unref();
98     timeout_detector_ = nullptr;
99   }
100
101   /* Answer all simcalls associated with the synchro */
102   finish();
103 }
104
105 void IoImpl::finish()
106 {
107   XBT_DEBUG("IoImpl::finish() in state %s", to_c_str(state_));
108   while (not simcalls_.empty()) {
109     smx_simcall_t simcall = simcalls_.front();
110     simcalls_.pop_front();
111
112     /* If a waitany simcall is waiting for this synchro to finish, then remove it from the other synchros in the waitany
113      * list. Afterwards, get the position of the actual synchro in the waitany list and return it as the result of the
114      * simcall */
115
116     if (simcall->call_ == simix::Simcall::NONE) // FIXME: maybe a better way to handle this case
117       continue;                                 // if process handling comm is killed
118     if (auto* observer = dynamic_cast<kernel::actor::IoWaitanySimcall*>(simcall->observer_)) { // simcall is a wait_any?
119       const auto& ios = observer->get_ios();
120
121       for (auto* io : ios) {
122         io->unregister_simcall(simcall);
123
124         if (simcall->timeout_cb_) {
125           simcall->timeout_cb_->remove();
126           simcall->timeout_cb_ = nullptr;
127         }
128       }
129
130       if (not MC_is_active() && not MC_record_replay_is_active()) {
131         auto element = std::find(ios.begin(), ios.end(), this);
132         int rank     = element != ios.end() ? static_cast<int>(std::distance(ios.begin(), element)) : -1;
133         observer->set_result(rank);
134       }
135     }
136
137     switch (state_) {
138       case State::FAILED:
139         simcall->issuer_->context_->set_wannadie();
140         piface_->complete(s4u::Activity::State::FAILED);
141         simcall->issuer_->exception_ =
142             std::make_exception_ptr(StorageFailureException(XBT_THROW_POINT, "Storage failed"));
143         break;
144       case State::CANCELED:
145         simcall->issuer_->exception_ = std::make_exception_ptr(CancelException(XBT_THROW_POINT, "I/O Canceled"));
146         break;
147       case State::TIMEOUT:
148         simcall->issuer_->exception_ = std::make_exception_ptr(TimeoutException(XBT_THROW_POINT, "Timeouted"));
149         break;
150       default:
151         xbt_assert(state_ == State::DONE, "Internal error in IoImpl::finish(): unexpected synchro state %s",
152                    to_c_str(state_));
153     }
154
155     simcall->issuer_->waiting_synchro_ = nullptr;
156     simcall->issuer_->simcall_answer();
157   }
158 }
159
160 void IoImpl::wait_any_for(actor::ActorImpl* issuer, const std::vector<IoImpl*>& ios, double timeout)
161 {
162   if (timeout < 0.0) {
163     issuer->simcall_.timeout_cb_ = nullptr;
164   } else {
165     issuer->simcall_.timeout_cb_ = timer::Timer::set(s4u::Engine::get_clock() + timeout, [issuer, &ios]() {
166       issuer->simcall_.timeout_cb_ = nullptr;
167       for (auto* io : ios)
168         io->unregister_simcall(&issuer->simcall_);
169       // default result (-1) is set in actor::IoWaitanySimcall
170       issuer->simcall_answer();
171     });
172   }
173
174   for (auto* io : ios) {
175     /* associate this simcall to the the synchro */
176     io->simcalls_.push_back(&issuer->simcall_);
177
178     /* see if the synchro is already finished */
179     if (io->state_ != State::WAITING && io->state_ != State::RUNNING) {
180       io->finish();
181       break;
182     }
183   }
184 }
185
186 } // namespace activity
187 } // namespace kernel
188 } // namespace simgrid