1 /* Copyright (c) 2007-2020. 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/s4u.hpp"
9 constexpr unsigned PIECE_SIZE = 65536;
10 constexpr unsigned MESSAGE_BUILD_CHAIN_SIZE = 40;
11 constexpr unsigned MESSAGE_SEND_DATA_HEADER_SIZE = 1;
13 XBT_LOG_NEW_DEFAULT_CATEGORY(s4u_chainsend, "Messages specific for chainsend");
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)
24 ~ChainMessage() = default;
29 FilePiece() = default;
30 ~FilePiece() = default;
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;
41 unsigned long long received_bytes = 0;
42 unsigned int received_pieces = 0;
43 unsigned int total_pieces = 0;
45 Peer() { me = simgrid::s4u::Mailbox::by_name(simgrid::s4u::Host::current()->get_cname()); }
50 auto msg = me->get_unique<ChainMessage>();
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);
64 simgrid::s4u::CommPtr comm = me->get_async<FilePiece>(&received);
65 pending_recvs.push_back(comm);
67 int idx = simgrid::s4u::Comm::wait_any(&pending_recvs);
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);
80 received_bytes += PIECE_SIZE;
81 XBT_DEBUG("%u pieces received, %llu bytes received", received_pieces, received_bytes);
82 if (received_pieces >= total_pieces) {
92 simgrid::s4u::Mailbox* first = nullptr;
93 std::vector<simgrid::s4u::Mailbox*> mailboxes;
94 unsigned int piece_count;
98 /* Build the chain if there's at least one peer */
99 if (not mailboxes.empty())
100 first = mailboxes.front();
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);
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);
122 simgrid::s4u::Comm::wait_all(&pending_sends);
125 Broadcaster(int hostcount, unsigned int piece_count) : piece_count(piece_count)
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));
134 ~Broadcaster() = default;
143 double start_time = simgrid::s4u::Engine::get_clock();
147 simgrid::s4u::Comm::wait_all(&p.pending_sends);
148 double end_time = simgrid::s4u::Engine::get_clock();
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));
154 static void broadcaster(int hostcount, unsigned int piece_count)
156 XBT_DEBUG("broadcaster");
158 Broadcaster bc(hostcount, piece_count);
163 int main(int argc, char* argv[])
165 simgrid::s4u::Engine e(&argc, argv);
167 e.load_platform(argv[1]);
169 simgrid::s4u::Actor::create("broadcaster", simgrid::s4u::Host::by_name("node-0.simgrid.org"), broadcaster, 8, 256);
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);
181 XBT_INFO("Total simulation time: %e", simgrid::s4u::Engine::get_clock());