SimGrid/C++中基于邮箱的环形节点流缓冲区开发求助：实现20条消息传输

Hey there! As someone who’s tinkered with SimGrid for a while, I can help you put together this ring topology message-passing setup with a stream buffer. Let’s start by breaking down the core pieces, then jump into a complete, runnable example.

First, let’s make sure we’re on the same page about the key components:

• Ring Topology: Each node connects to exactly two others, forming a closed loop. We’ll set up hosts where each sends messages to the next node (and receives from the previous one).
• Mailboxes: SimGrid’s mailboxes are named channels that processes use to send/receive messages—think of them as dedicated mail slots for each node.
• Stream Buffer: For this task, we’ll implement a temporary storage buffer at each node to hold incoming messages before forwarding or processing them. Since you want 20 total messages, we’ll ensure each one traverses the ring as needed.

Here’s a working code snippet that creates a 4-node ring, sends 20 messages around the loop, and uses a basic stream buffer at each node:

#include <simgrid/s4u.hpp>
#include <vector>
#include <string>

XBT_LOG_NEW_DEFAULT_CATEGORY(ring_stream_example, "Logs for our ring stream buffer demo");

// Process logic for each node in the ring
void ring_node_process(int node_id, int total_nodes, int total_messages) {
  simgrid::s4u::Host* my_host = simgrid::s4u::Host::current();
  std::string my_mailbox = "node-" + std::to_string(node_id);
  
  // Calculate next node in the ring (wrap around with modulo)
  int next_node_id = (node_id + 1) % total_nodes;
  std::string next_mailbox = "node-" + std::to_string(next_node_id);

  // Stream buffer: temporary storage for incoming messages
  std::vector<std::string> stream_buffer;

  // Node 0 is the initiator: sends all 20 messages to the next node
  if (node_id == 0) {
    for (int msg_idx = 0; msg_idx < total_messages; ++msg_idx) {
      std::string msg_content = "Message #" + std::to_string(msg_idx) + " | Origin: Node " + std::to_string(node_id);
      // Send message with 1KB size (adjust as needed)
      simgrid::s4u::Mailbox::by_name(next_mailbox)->put(new std::string(msg_content), 1024);
      XBT_INFO("Node %d sent: %s", node_id, msg_content.c_str());
    }
  }

  // All nodes listen for messages and handle them
  int messages_handled = 0;
  while (messages_handled < total_messages) {
    // Receive message from our mailbox
    auto received_msg = simgrid::s4u::Mailbox::by_name(my_mailbox)->get<std::string>();
    XBT_INFO("Node %d received: %s", node_id, received_msg->c_str());

    // Add to stream buffer before processing/forwarding
    stream_buffer.push_back(*received_msg);
    delete received_msg; // Clean up the dynamically allocated message

    // If we're not the origin node, forward the message to the next node
    if (node_id != 0) {
      std::string forwarded_msg = stream_buffer.back() + " | Forwarded by Node " + std::to_string(node_id);
      simgrid::s4u::Mailbox::by_name(next_mailbox)->put(new std::string(forwarded_msg), 1024);
      XBT_INFO("Node %d forwarded: %s", node_id, forwarded_msg.c_str());
    }

    messages_handled++;
    stream_buffer.clear(); // Clear buffer after processing this message (tweak for longer buffering)
  }

  XBT_INFO("Node %d finished handling all messages!", node_id);
}

int main(int argc, char** argv) {
  simgrid::s4u::Engine sim_engine(&argc, argv);

  // Create 4 hosts (nodes) for the ring
  const int total_nodes = 4;
  std::vector<simgrid::s4u::Host*> ring_hosts;
  for (int i = 0; i < total_nodes; ++i) {
    ring_hosts.push_back(sim_engine.host_create("node-" + std::to_string(i)));
  }

  // Create links to form the ring (each node connects to the next, last to first)
  for (int i = 0; i < total_nodes; ++i) {
    int next_node = (i + 1) % total_nodes;
    // 1Gbps link with 10ms latency (adjust these values for your simulation)
    simgrid::s4u::Link* inter_node_link = sim_engine.link_create(
      "link-" + std::to_string(i) + "-" + std::to_string(next_node),
      1e9, 10e-6
    );
    sim_engine.hosts_route(ring_hosts[i], ring_hosts[next_node], {inter_node_link});
  }

  // Launch the process on each node
  const int total_messages = 20;
  for (int i = 0; i < total_nodes; ++i) {
    simgrid::s4u::Actor::create(
      "process-node-" + std::to_string(i),
      ring_hosts[i],
      ring_node_process,
      i, total_nodes, total_messages
    );
  }

  // Run the simulation
  sim_engine.run();

  return 0;
}

Let’s break down what this code does:

1. Topology Setup: We create 4 hosts and connect each to the next with a network link (the last host loops back to the first to form a ring).
2. Mailbox Naming: Each node uses a mailbox named node-X (where X is its ID), making it trivial to target the next node in the loop.
3. Stream Buffer: The stream_buffer vector acts as temporary storage for incoming messages. In this example, we clear it after processing each message, but you could modify this to buffer multiple messages (e.g., hold 5 messages before forwarding) for more advanced stream control.
4. Message Flow: Node 0 sends all 20 messages to Node 1. Each subsequent node receives the message, adds it to the buffer, forwards it to the next node, and clears the buffer. When Node 0 receives the forwarded messages, it just processes them without forwarding (since the ring is complete).
5. Network Parameters: We set links to 1Gbps with 10ms latency to simulate real-world network conditions, but you can adjust these values to match your needs.

1. Save the code as ring_stream_buffer.cpp
2. Compile it using SimGrid’s compiler wrapper: sg++ ring_stream_buffer.cpp -o ring_stream_buffer
3. Run the executable: ./ring_stream_buffer

You’ll see detailed logs showing each node sending, receiving, and forwarding messages until all 20 have completed the ring.

内容的提问来源于stack exchange，提问作者Iago Tonello