1 /* Copyright (c) 2023. The SimGrid Team. All rights reserved. */
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. */
6 #include <simgrid/plugins/jbod.hpp>
7 #include <simgrid/s4u/Comm.hpp>
8 #include <simgrid/s4u/Disk.hpp>
9 #include <simgrid/s4u/Exec.hpp>
10 #include <simgrid/s4u/NetZone.hpp>
13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_jbod, s4u, "Logging specific to the JBOD implmentation");
15 namespace simgrid::plugin {
17 JbodPtr Jbod::create_jbod(s4u::NetZone* zone, const std::string& name, double speed, unsigned int num_disks,
18 RAID raid_level, double read_bandwidth, double write_bandwidth)
20 xbt_assert(not ((raid_level == RAID::RAID4 || raid_level == RAID::RAID5) && num_disks < 3),
21 "RAID%d requires at least 3 disks", (int) raid_level);
22 xbt_assert(not (raid_level == RAID::RAID6 && num_disks < 4), "RAID6 requires at least 4 disks");
24 auto* jbod = new Jbod();
25 jbod->set_controller(zone->create_host(name, speed));
26 jbod->set_num_disks(num_disks);
27 jbod->set_parity_disk_idx(num_disks -1 );
28 jbod->set_read_disk_idx(-1);
29 jbod->set_raid_level(raid_level);
30 for (unsigned int i = 0; i < num_disks; i++)
31 jbod->get_controller()->create_disk(name + "_disk_" + std::to_string(i), read_bandwidth, write_bandwidth);
36 JbodIoPtr Jbod::read_async(sg_size_t size)
38 auto comm = s4u::Comm::sendto_init()->set_source(this->controller_)->set_payload_size(size);
39 std::vector<s4u::IoPtr> pending_ios;
40 sg_size_t read_size = 0;
41 std::vector<s4u::Disk*> targets;
44 read_size = size / num_disks_;
45 targets = controller_->get_disks();
49 targets.push_back(controller_->get_disks().at(get_next_read_disk_idx()));
52 read_size = size / (num_disks_ - 1);
53 targets = controller_->get_disks();
57 read_size = size / (num_disks_ - 1);
58 targets = controller_->get_disks();
59 targets.erase(targets.begin() + (get_parity_disk_idx() + 1 % num_disks_));
62 read_size = size / (num_disks_ - 2);
63 targets = controller_->get_disks();
64 if ( (get_parity_disk_idx() + 2 % num_disks_) == 0 ) {
66 targets.erase(targets.begin());
67 } else if (get_parity_disk_idx() + 1 == static_cast<int>(num_disks_)) {
71 targets.erase(targets.begin() + (get_parity_disk_idx() + 1) % num_disks_,
72 targets.begin() + get_parity_disk_idx() + 3);
76 xbt_die("Unsupported RAID level. Supported level are: 0, 1, 4, 5, and 6");
78 for (const auto* disk : targets) {
79 auto io = s4u::IoPtr(disk->io_init(read_size, s4u::Io::OpType::READ));
80 io->set_name(disk->get_name())->start();
81 pending_ios.push_back(io);
84 return JbodIoPtr(new JbodIo(this, comm, nullptr, pending_ios, s4u::Io::OpType::READ));
87 sg_size_t Jbod::read(sg_size_t size)
89 read_async(size)->wait();
93 JbodIoPtr Jbod::write_async(sg_size_t size)
95 auto comm = s4u::Comm::sendto_init(s4u::Host::current(), this->get_controller());
96 std::vector<s4u::IoPtr> pending_ios;
97 sg_size_t write_size = 0;
100 write_size = size / num_disks_;
106 write_size = size / (num_disks_ - 1);
109 update_parity_disk_idx();
110 write_size = size / (num_disks_ - 1);
113 update_parity_disk_idx();
114 update_parity_disk_idx();
115 write_size = size / (num_disks_ - 2);
118 xbt_die("Unsupported RAID level. Supported level are: 0, 1, 4, 5, and 6");
120 for (const auto* disk : get_controller()->get_disks()) {
121 auto io = s4u::IoPtr(disk->io_init(write_size, s4u::Io::OpType::WRITE));
122 io->set_name(disk->get_name());
123 pending_ios.push_back(io);
126 s4u::ExecPtr parity_block_comp = nullptr;
127 if (raid_level_ == RAID::RAID4 || raid_level_ == RAID::RAID5 || raid_level_ == RAID::RAID6) {
128 // Assume 1 flop per byte to write per parity block and two for RAID6.
129 // Do not assign the Exec yet, will be done after the completion of the CommPtr
130 if (raid_level_ == RAID::RAID6)
131 parity_block_comp = s4u::Exec::init()->set_flops_amount(200 * write_size);
133 parity_block_comp = s4u::Exec::init()->set_flops_amount(write_size);
136 comm->set_payload_size(size)->start();
137 return JbodIoPtr(new JbodIo(this, comm, parity_block_comp, pending_ios, s4u::Io::OpType::WRITE));
140 sg_size_t Jbod::write(sg_size_t size)
142 write_async(size)->wait();
148 if (type_ == s4u::Io::OpType::WRITE) {
150 XBT_DEBUG("Data received on JBOD");
151 if (parity_block_comp_) {
152 parity_block_comp_->set_host(jbod_->get_controller())->wait();
153 XBT_DEBUG("Parity block computed");
155 XBT_DEBUG("Start writing");
156 for (const auto& io : pending_ios_)
160 for (const auto& io : pending_ios_) {
161 XBT_DEBUG("Wait for I/O on %s", io->get_cname());
165 if (type_ == s4u::Io::OpType::READ) {
166 XBT_DEBUG("Data read on JBOD, send it to %s", s4u::Host::current()->get_cname());
167 transfer_->set_destination(s4u::Host::current())->wait();
170 } // namespace simgrid::plugin