examples/s4u/app-chainsend/s4u-app-chainsend.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/s4u.hpp"
   7 #include <vector>
   8
   9 constexpr unsigned PIECE_SIZE                    = 65536;
  10 constexpr unsigned MESSAGE_BUILD_CHAIN_SIZE      = 40;
  11 constexpr unsigned MESSAGE_SEND_DATA_HEADER_SIZE = 1;
  12
  13 XBT_LOG_NEW_DEFAULT_CATEGORY(s4u_chainsend, "Messages specific for chainsend");
  14
  15 class ChainMessage {
  16 public:
  17   simgrid::s4u::Mailbox* prev_   = nullptr;
  18   simgrid::s4u::Mailbox* next_   = nullptr;
  19   unsigned int num_pieces        = 0;
  20   explicit ChainMessage(simgrid::s4u::Mailbox* prev, simgrid::s4u::Mailbox* next, const unsigned int num_pieces)
  21       : prev_(prev), next_(next), num_pieces(num_pieces)
  22   {
  23   }
  24   ~ChainMessage() = default;
  25 };
  26
  27 class FilePiece {
  28 public:
  29   FilePiece()  = default;
  30   ~FilePiece() = default;
  31 };
  32
  33 class Peer {
  34 public:
  35   simgrid::s4u::Mailbox* prev = nullptr;
  36   simgrid::s4u::Mailbox* next = nullptr;
  37   simgrid::s4u::Mailbox* me   = nullptr;
  38   std::vector<simgrid::s4u::CommPtr> pending_recvs;
  39   std::vector<simgrid::s4u::CommPtr> pending_sends;
  40
  41   unsigned long long received_bytes = 0;
  42   unsigned int received_pieces      = 0;
  43   unsigned int total_pieces         = 0;
  44
  45   Peer() { me = simgrid::s4u::Mailbox::by_name(simgrid::s4u::Host::current()->get_cname()); }
  46   ~Peer()     = default;
  47
  48   void joinChain()
  49   {
  50     auto msg     = me->get_unique<ChainMessage>();
  51     prev         = msg->prev_;
  52     next         = msg->next_;
  53     total_pieces = msg->num_pieces;
  54     XBT_DEBUG("Peer %s got a 'BUILD_CHAIN' message (prev: %s / next: %s)", me->get_cname(),
  55               prev ? prev->get_cname() : nullptr, next ? next->get_cname() : nullptr);
  56   }
  57
  58   void forwardFile()
  59   {
  60     FilePiece* received;
  61     bool done = false;
  62
  63     while (not done) {
  64       simgrid::s4u::CommPtr comm = me->get_async<FilePiece>(&received);
  65       pending_recvs.push_back(comm);
  66
  67       int idx = simgrid::s4u::Comm::wait_any(&pending_recvs);
  68       if (idx != -1) {
  69         comm = pending_recvs.at(idx);
  70         XBT_DEBUG("Peer %s got a 'SEND_DATA' message", me->get_cname());
  71         pending_recvs.erase(pending_recvs.begin() + idx);
  72         if (next != nullptr) {
  73           XBT_DEBUG("Sending (asynchronously) from %s to %s", me->get_cname(), next->get_cname());
  74           simgrid::s4u::CommPtr send = next->put_async(received, MESSAGE_SEND_DATA_HEADER_SIZE + PIECE_SIZE);
  75           pending_sends.push_back(send);
  76         } else
  77           delete received;
  78
  79         received_pieces++;
  80         received_bytes += PIECE_SIZE;
  81         XBT_DEBUG("%u pieces received, %llu bytes received", received_pieces, received_bytes);
  82         if (received_pieces >= total_pieces) {
  83           done = true;
  84         }
  85       }
  86     }
  87   }
  88 };
  89
  90 class Broadcaster {
  91 public:
  92   simgrid::s4u::Mailbox* first = nullptr;
  93   std::vector<simgrid::s4u::Mailbox*> mailboxes;
  94   unsigned int piece_count;
  95
  96   void buildChain()
  97   {
  98     /* Build the chain if there's at least one peer */
  99     if (not mailboxes.empty())
 100       first = mailboxes.front();
 101
 102     for (unsigned i = 0; i < mailboxes.size(); i++) {
 103       simgrid::s4u::Mailbox* prev = i > 0 ? mailboxes[i - 1] : nullptr;
 104       simgrid::s4u::Mailbox* next = i < mailboxes.size() - 1 ? mailboxes[i + 1] : nullptr;
 105       XBT_DEBUG("Building chain--broadcaster:\"%s\" dest:\"%s\" prev:\"%s\" next:\"%s\"",
 106                 simgrid::s4u::Host::current()->get_cname(), mailboxes[i]->get_cname(),
 107                 prev ? prev->get_cname() : nullptr, next ? next->get_cname() : nullptr);
 108       /* Send message to current peer */
 109       mailboxes[i]->put(new ChainMessage(prev, next, piece_count), MESSAGE_BUILD_CHAIN_SIZE);
 110     }
 111   }
 112
 113   void sendFile()
 114   {
 115     std::vector<simgrid::s4u::CommPtr> pending_sends;
 116     for (unsigned int current_piece = 0; current_piece < piece_count; current_piece++) {
 117       XBT_DEBUG("Sending (send) piece %u from %s into mailbox %s", current_piece,
 118                 simgrid::s4u::Host::current()->get_cname(), first->get_cname());
 119       simgrid::s4u::CommPtr comm = first->put_async(new FilePiece(), MESSAGE_SEND_DATA_HEADER_SIZE + PIECE_SIZE);
 120       pending_sends.push_back(comm);
 121     }
 122     simgrid::s4u::Comm::wait_all(&pending_sends);
 123   }
 124
 125   Broadcaster(int hostcount, unsigned int piece_count) : piece_count(piece_count)
 126   {
 127     for (int i = 1; i <= hostcount; i++) {
 128       std::string name = std::string("node-") + std::to_string(i) + ".simgrid.org";
 129       XBT_DEBUG("%s", name.c_str());
 130       mailboxes.push_back(simgrid::s4u::Mailbox::by_name(name));
 131     }
 132   }
 133
 134   ~Broadcaster() = default;
 135 };
 136
 137 static void peer()
 138 {
 139   XBT_DEBUG("peer");
 140
 141   Peer p;
 142
 143   double start_time = simgrid::s4u::Engine::get_clock();
 144   p.joinChain();
 145   p.forwardFile();
 146
 147   simgrid::s4u::Comm::wait_all(&p.pending_sends);
 148   double end_time = simgrid::s4u::Engine::get_clock();
 149
 150   XBT_INFO("### %f %llu bytes (Avg %f MB/s); copy finished (simulated).", end_time - start_time, p.received_bytes,
 151            p.received_bytes / 1024.0 / 1024.0 / (end_time - start_time));
 152 }
 153
 154 static void broadcaster(int hostcount, unsigned int piece_count)
 155 {
 156   XBT_DEBUG("broadcaster");
 157
 158   Broadcaster bc(hostcount, piece_count);
 159   bc.buildChain();
 160   bc.sendFile();
 161 }
 162
 163 int main(int argc, char* argv[])
 164 {
 165   simgrid::s4u::Engine e(&argc, argv);
 166
 167   e.load_platform(argv[1]);
 168
 169   simgrid::s4u::Actor::create("broadcaster", simgrid::s4u::Host::by_name("node-0.simgrid.org"), broadcaster, 8, 256);
 170
 171   simgrid::s4u::Actor::create("peer", simgrid::s4u::Host::by_name("node-1.simgrid.org"), peer);
 172   simgrid::s4u::Actor::create("peer", simgrid::s4u::Host::by_name("node-2.simgrid.org"), peer);
 173   simgrid::s4u::Actor::create("peer", simgrid::s4u::Host::by_name("node-3.simgrid.org"), peer);
 174   simgrid::s4u::Actor::create("peer", simgrid::s4u::Host::by_name("node-4.simgrid.org"), peer);
 175   simgrid::s4u::Actor::create("peer", simgrid::s4u::Host::by_name("node-5.simgrid.org"), peer);
 176   simgrid::s4u::Actor::create("peer", simgrid::s4u::Host::by_name("node-6.simgrid.org"), peer);
 177   simgrid::s4u::Actor::create("peer", simgrid::s4u::Host::by_name("node-7.simgrid.org"), peer);
 178   simgrid::s4u::Actor::create("peer", simgrid::s4u::Host::by_name("node-8.simgrid.org"), peer);
 179
 180   e.run();
 181   XBT_INFO("Total simulation time: %e", simgrid::s4u::Engine::get_clock());
 182
 183   return 0;
 184 }