src/plugins/jbod.cpp

   1 /* Copyright (c) 2023. 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/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>
  11
  12
  13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_jbod, s4u, "Logging specific to the JBOD implmentation");
  14
  15 namespace simgrid::plugin {
  16
  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)
  19 {
  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");
  23
  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);
  31
  32   return jbod;
  33 }
  34
  35 JbodIoPtr Jbod::read_async(sg_size_t size)
  36 {
  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;
  41   switch(raid_level_) {
  42     case RAID::RAID0:
  43       read_size = size / num_disks_;
  44       targets = get_disks();
  45       break;
  46     case RAID::RAID1:
  47       read_size = size;
  48       targets.push_back(get_disks().at(get_next_read_disk_idx()));
  49       break;
  50     case RAID::RAID4:
  51       read_size = size / (num_disks_ - 1);
  52       targets = get_disks();
  53       targets.pop_back();
  54       break;
  55     case RAID::RAID5:
  56       read_size = size / (num_disks_ - 1);
  57       targets = get_disks();
  58       targets.erase(targets.begin() + (get_parity_disk_idx() + 1 % num_disks_));
  59       break;
  60     case RAID::RAID6:
  61       read_size = size / (num_disks_ - 2);
  62       targets = get_disks();
  63       if ( (get_parity_disk_idx() + 2 % num_disks_) == 0 ) {
  64         targets.pop_back();
  65         targets.erase(targets.begin());
  66       } else if (get_parity_disk_idx() + 1 == static_cast<int>(num_disks_)) {
  67         targets.pop_back();
  68         targets.pop_back();
  69       } else {
  70         targets.erase(targets.begin() + (get_parity_disk_idx() + 1) % num_disks_,
  71                       targets.begin() + get_parity_disk_idx() + 3);
  72       }
  73       break;
  74     default:
  75       xbt_die("Unsupported RAID level. Supported level are: 0, 1, 4, 5, and 6");
  76   }
  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);
  81   }
  82
  83   return JbodIoPtr(new JbodIo(this, comm, nullptr, pending_ios, s4u::Io::OpType::READ));
  84 }
  85
  86 sg_size_t Jbod::read(sg_size_t size)
  87 {
  88   read_async(size)->wait();
  89   return size;
  90 }
  91
  92 JbodIoPtr Jbod::write_async(sg_size_t size)
  93 {
  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;
  97   switch(raid_level_) {
  98     case RAID::RAID0:
  99       write_size = size / num_disks_;
 100       break;
 101     case RAID::RAID1:
 102       write_size = size;
 103       break;
 104     case RAID::RAID4:
 105       write_size = size / (num_disks_ - 1);
 106       break;
 107     case RAID::RAID5:
 108       update_parity_disk_idx();
 109       write_size = size / (num_disks_ - 1);
 110       break;
 111     case RAID::RAID6:
 112       update_parity_disk_idx();
 113       update_parity_disk_idx();
 114       write_size = size / (num_disks_ - 2);
 115       break;
 116     default:
 117       xbt_die("Unsupported RAID level. Supported level are: 0, 1, 4, 5, and 6");
 118   }
 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);
 123   }
 124
 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);
 131     else
 132       parity_block_comp = s4u::Exec::init()->set_flops_amount(write_size);
 133   }
 134
 135   comm->set_payload_size(size)->start();
 136   return JbodIoPtr(new JbodIo(this, comm, parity_block_comp, pending_ios, s4u::Io::OpType::WRITE));
 137 }
 138
 139 sg_size_t Jbod::write(sg_size_t size)
 140 {
 141   write_async(size)->wait();
 142   return size;
 143 }
 144
 145 void JbodIo::wait()
 146 {
 147   if (type_ == s4u::Io::OpType::WRITE) {
 148     transfer_->wait();
 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");
 153     }
 154     XBT_DEBUG("Start writing");
 155     for (const auto& io : pending_ios_)
 156       io->start();
 157   }
 158
 159   for (const auto& io : pending_ios_) {
 160     XBT_DEBUG("Wait for I/O on %s", io->get_cname());
 161     io->wait();
 162   }
 163
 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();
 167   }
 168 }
 169 } // namespace simgrid::plugin