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 Jbod* 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 = static_cast<Jbod*>(zone->create_host(name, speed));
25 jbod->set_num_disks(num_disks);
26 jbod->set_parity_disk_idx(num_disks -1 );
27 jbod->set_read_disk_idx(-1);
28 jbod->set_raid_level(raid_level);
29 for (unsigned int i = 0; i < num_disks; i++)
30 jbod->create_disk(name + "_disk_" + std::to_string(i), read_bandwidth, write_bandwidth);
35 JbodIoPtr Jbod::read_async(sg_size_t size)
37 auto comm = s4u::Comm::sendto_init()->set_source(const_cast<Jbod*>(this))->set_payload_size(size);
38 std::vector<s4u::IoPtr> pending_ios;
39 sg_size_t read_size = 0;
40 std::vector<s4u::Disk*> targets;
43 read_size = size / num_disks_;
44 targets = get_disks();
48 targets.push_back(get_disks().at(get_next_read_disk_idx()));
51 read_size = size / (num_disks_ - 1);
52 targets = get_disks();
56 read_size = size / (num_disks_ - 1);
57 targets = get_disks();
58 targets.erase(targets.begin() + (get_parity_disk_idx() + 1 % num_disks_));
61 read_size = size / (num_disks_ - 2);
62 targets = get_disks();
63 if ( (get_parity_disk_idx() + 2 % num_disks_) == 0 ) {
65 targets.erase(targets.begin());
66 } else if (get_parity_disk_idx() + 1 == static_cast<int>(num_disks_)) {
70 targets.erase(targets.begin() + (get_parity_disk_idx() + 1) % num_disks_,
71 targets.begin() + get_parity_disk_idx() + 3);
75 xbt_die("Unsupported RAID level. Supported level are: 0, 1, 4, 5, and 6");
77 for (const auto* disk : targets) {
78 auto io = s4u::IoPtr(disk->io_init(read_size, s4u::Io::OpType::READ));
79 io->set_name(disk->get_name())->start();
80 pending_ios.push_back(io);
83 return JbodIoPtr(new JbodIo(this, comm, nullptr, pending_ios, s4u::Io::OpType::READ));
86 sg_size_t Jbod::read(sg_size_t size)
88 read_async(size)->wait();
92 JbodIoPtr Jbod::write_async(sg_size_t size)
94 auto comm = s4u::Comm::sendto_init(s4u::Host::current(), const_cast<Jbod*>(this));
95 std::vector<s4u::IoPtr> pending_ios;
96 sg_size_t write_size = 0;
99 write_size = size / num_disks_;
105 write_size = size / (num_disks_ - 1);
108 update_parity_disk_idx();
109 write_size = size / (num_disks_ - 1);
112 update_parity_disk_idx();
113 update_parity_disk_idx();
114 write_size = size / (num_disks_ - 2);
117 xbt_die("Unsupported RAID level. Supported level are: 0, 1, 4, 5, and 6");
119 for (const auto* disk : get_disks()) {
120 auto io = s4u::IoPtr(disk->io_init(write_size, s4u::Io::OpType::WRITE));
121 io->set_name(disk->get_name());
122 pending_ios.push_back(io);
125 s4u::ExecPtr parity_block_comp = nullptr;
126 if (raid_level_ == RAID::RAID4 || raid_level_ == RAID::RAID5 || raid_level_ == RAID::RAID6) {
127 // Assume 1 flop per byte to write per parity block and two for RAID6.
128 // Do not assign the Exec yet, will be done after the completion of the CommPtr
129 if (raid_level_ == RAID::RAID6)
130 parity_block_comp = s4u::Exec::init()->set_flops_amount(200 * write_size);
132 parity_block_comp = s4u::Exec::init()->set_flops_amount(write_size);
135 comm->set_payload_size(size)->start();
136 return JbodIoPtr(new JbodIo(this, comm, parity_block_comp, pending_ios, s4u::Io::OpType::WRITE));
139 sg_size_t Jbod::write(sg_size_t size)
141 write_async(size)->wait();
147 if (type_ == s4u::Io::OpType::WRITE) {
149 XBT_DEBUG("Data received on JBOD");
150 if (parity_block_comp_) {
151 parity_block_comp_->set_host(const_cast<Jbod*>(jbod_))->wait();
152 XBT_DEBUG("Parity block computed");
154 XBT_DEBUG("Start writing");
155 for (const auto& io : pending_ios_)
159 for (const auto& io : pending_ios_) {
160 XBT_DEBUG("Wait for I/O on %s", io->get_cname());
164 if (type_ == s4u::Io::OpType::READ) {
165 XBT_DEBUG("Data read on JBOD, send it to %s", s4u::Host::current()->get_cname());
166 transfer_->set_destination(s4u::Host::current())->wait();
169 } // namespace simgrid::plugin