teshsuite/s4u/io-stream/io-stream.cpp

   1 /* Copyright (c) 2017-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/s4u.hpp>
   7
   8 namespace sg4 = simgrid::s4u;
   9
  10 XBT_LOG_NEW_DEFAULT_CATEGORY(io_stream, "Messages specific for this simulation");
  11
  12 static void streamer(size_t size)
  13 {
  14   auto* bob        = sg4::Host::by_name("bob");
  15   auto* alice      = sg4::Host::by_name("alice");
  16   const auto* bob_disk   = bob->get_disks().front();
  17   const auto* alice_disk = alice->get_disks().front();
  18   double clock = sg4::Engine::get_clock();
  19
  20   XBT_INFO("[Bob -> Alice] Store and Forward (1 block)");
  21   bob_disk->read(size);
  22   XBT_INFO("    Read  : %.6f seconds", sg4::Engine::get_clock() - clock);
  23   clock = sg4::Engine::get_clock();
  24   sg4::Comm::sendto(bob, alice, size);
  25   XBT_INFO("    Send  : %.6f seconds", sg4::Engine::get_clock() - clock);
  26   clock = sg4::Engine::get_clock();
  27   alice_disk->write(size);
  28   XBT_INFO("    Write : %.6f seconds", sg4::Engine::get_clock() - clock);
  29   XBT_INFO("    Total : %.6f seconds", sg4::Engine::get_clock());
  30
  31   XBT_INFO("[Bob -> Alice] Store and Forward (100 blocks)");
  32   size_t block_size = size / 100;
  33   sg4::IoPtr read       = bob_disk->read_async(block_size);
  34   sg4::CommPtr transfer = sg4::Comm::sendto_async(bob, alice, block_size);
  35   sg4::IoPtr write      = alice_disk->write_async(block_size);
  36
  37   clock = sg4::Engine::get_clock();
  38
  39   for (int i = 0; i < 99; i++){
  40     read->wait();
  41     read = bob_disk->read_async(block_size);
  42     transfer->wait();
  43     transfer = sg4::Comm::sendto_async(bob, alice, block_size);
  44     write->wait();
  45     write = alice_disk->write_async(block_size);
  46   }
  47
  48   read->wait();
  49   transfer->wait();
  50   write->wait();
  51   XBT_INFO("    Total : %.6f seconds", sg4::Engine::get_clock() - clock);
  52
  53   XBT_INFO("[Bob -> Alice] Streaming (Read bottleneck)");
  54   clock = sg4::Engine::get_clock();
  55   sg4::Io::streamto(bob, bob_disk, alice, alice_disk, size);
  56   XBT_INFO("    Total : %.6f seconds", sg4::Engine::get_clock() - clock);
  57
  58   XBT_INFO("[Alice -> Bob] Streaming (Write bottleneck)");
  59   clock = sg4::Engine::get_clock();
  60   sg4::Io::streamto(alice, alice_disk, bob, bob_disk, size);
  61   XBT_INFO("    Total : %.6f seconds", sg4::Engine::get_clock() - clock);
  62
  63   XBT_INFO("Start two 10-second background traffic between Bob and Alice");
  64   sg4::CommPtr bt1 = sg4::Comm::sendto_async(bob, alice, 2e7);
  65   XBT_INFO("[Bob -> Alice] Streaming (Transfer bottleneck)");
  66   clock = sg4::Engine::get_clock();
  67   sg4::Io::streamto(bob, bob_disk, alice, alice_disk, size);
  68   XBT_INFO("    Total : %.6f seconds", sg4::Engine::get_clock() - clock);
  69   bt1->wait();
  70
  71   XBT_INFO("[Bob -> Alice] Streaming \"from disk to memory\" (no write)");
  72   clock = sg4::Engine::get_clock();
  73   sg4::Io::streamto(bob, bob_disk, alice, nullptr, size);
  74   XBT_INFO("    Total : %.6f seconds", sg4::Engine::get_clock() - clock);
  75
  76   XBT_INFO("[Bob -> Alice] Streaming \"from memory to disk\" (no read)");
  77   clock = sg4::Engine::get_clock();
  78   sg4::Io::streamto(bob, nullptr, alice, alice_disk, size);
  79   XBT_INFO("    Total : %.6f seconds", sg4::Engine::get_clock() - clock);
  80
  81   XBT_INFO("[Bob -> Bob] Disk to disk (no transfer)");
  82   clock = sg4::Engine::get_clock();
  83   sg4::Io::streamto(bob, bob_disk, bob, bob_disk, size);
  84   XBT_INFO("    Total : %.6f seconds", sg4::Engine::get_clock() - clock);
  85 }
  86
  87 static void background_send() {
  88   sg4::this_actor::sleep_for(23.000150);
  89   sg4::Mailbox::by_name("mbox")->put(new double(1), 2e7);
  90 }
  91
  92 static void background_recv() {
  93   double* res;
  94   sg4::CommPtr comm = sg4::Mailbox::by_name("mbox")->get_async<double>(&res);
  95   sg4::this_actor::sleep_for(33.1);
  96   comm->wait();
  97   delete res;
  98 }
  99
 100 int main(int argc, char** argv)
 101 {
 102   sg4::Engine e(&argc, argv);
 103
 104   /* simple platform containing 2 hosts and 2 disks */
 105   auto* zone = sg4::create_full_zone("");
 106   auto* bob  = zone->create_host("bob", 1e6);
 107   auto* alice  = zone->create_host("alice", 1e6);
 108
 109   auto* link = zone->create_link("link", "2MBps")->set_latency("50us");
 110   zone->add_route(bob, alice, {link});
 111
 112   bob->create_disk("bob_disk", "1MBps", "500kBps");
 113   alice->create_disk("alice_disk", "4MBps", "4MBps");
 114
 115   zone->seal();
 116
 117   sg4::Actor::create("streamer", bob, streamer, 4e6);
 118   sg4::Actor::create("background send", bob, background_send);
 119   sg4::Actor::create("background recv", alice, background_recv);
 120
 121   e.run();
 122
 123   return 0;
 124 }