FEAT: ExpressPass implementation

configuration_examples/generator/basic_ep_generator.py

@@ -0,0 +1,170 @@
+import yaml
+import argparse
+import sys
+
+def generate_topology(num_senders, num_receivers, switch_1_name="bus1", switch_2_name="bus2", link_latency="1ns", link_throughput="10Gbps", ingress_buffer_size = "32768B", egress_buffer_size = "32768B"):
+    topology = {
+        "devices": {},
+        "links": {}
+    }
+
+    # Add senders
+    for i in range(0, num_senders):
+        sender_name = f"sender{i}"
+        topology["devices"][sender_name] = {"type": "sender"}
+        base_index = 2 * i
+
+        # Add link from sender to switch
+        link_name = f"link{base_index}"
+        topology["links"][link_name] = {
+            "from": sender_name,
+            "to": switch_1_name,
+            "latency": link_latency,
+            "throughput": link_throughput,
+            "ingress_buffer_size": ingress_buffer_size,
+            "egress_buffer_size": egress_buffer_size
+        }
+
+        # Add link from switch to sender
+        link_name = f"link{base_index + 1}"
+        topology["links"][link_name] = {
+            "from": switch_1_name,
+            "to": sender_name,
+            "latency": link_latency,
+            "throughput": link_throughput,
+            "ingress_buffer_size": ingress_buffer_size,
+            "egress_buffer_size": egress_buffer_size
+        }
+
+    # Add receivers
+    for i in range(0, num_receivers):
+        receiver_name = f"receiver{i}"
+        topology["devices"][receiver_name] = {"type": "receiver"}
+        base_index = 2 * num_senders + 2 * i
+
+        # Add link from switch to receiver
+        link_name = f"link{base_index}"
+        topology["links"][link_name] = {
+            "from": switch_2_name,
+            "to": receiver_name,
+            "latency": link_latency,
+            "throughput": link_throughput,
+            "ingress_buffer_size": ingress_buffer_size,
+            "egress_buffer_size": egress_buffer_size
+        }
+
+        # Add link from receiver to switch
+        link_name = f"link{base_index + 1}"
+        topology["links"][link_name] = {
+            "from": receiver_name,
+            "to": switch_2_name,
+            "latency": link_latency,
+            "throughput": link_throughput,
+            "ingress_buffer_size": ingress_buffer_size,
+            "egress_buffer_size": egress_buffer_size
+        }
+
+    # Add the switch
+    topology["devices"][switch_1_name] = {"type": "switch"}
+    topology["devices"][switch_2_name] = {"type": "switch"}
+
+    base_index = 2 * num_senders + 2 * num_receivers
+    link_name1 = f"link{base_index}"
+    topology["links"][link_name1] = {
+        "from": switch_1_name,
+        "to": switch_2_name,
+        "latency": link_latency,
+        "throughput": link_throughput,
+        "ingress_buffer_size": ingress_buffer_size,
+        "egress_buffer_size": egress_buffer_size
+    }
+    link_name2 = f"link{base_index + 1}"
+    topology["links"][link_name2] = {
+        "from": switch_2_name,
+        "to": switch_1_name,
+        "latency": link_latency,
+        "throughput": link_throughput,
+        "ingress_buffer_size": ingress_buffer_size,
+        "egress_buffer_size": egress_buffer_size
+    }
+
+    return topology
+
+def generate_simulation(topology_path, num_senders, num_receivers, packet_size=1538, 
+                        packet_interval=100, number_of_packets=100000, algorithm="ep", simulation_time=10000000):
+    """
+    Generate a simulation YAML structure with flows between senders and receivers.
+    """
+    simulation = {
+        "topology_config_path": topology_path,
+        "flows": {},
+        "algorithm": algorithm,
+        "simulation_time": simulation_time
+    }
+
+    # Create flows - each sender connects to one receiver (1:1 mapping)
+    # If there are more senders than receivers, extra senders will connect to last receiver
+    # If there are more receivers than senders, extra receivers won't have flows
+    flow_id = 0
+    for i in range(0, num_senders):
+        flow_name = f"flow{flow_id}"
+        flow_id += 1
+        simulation["flows"][flow_name] = {
+            "sender_id": f"sender{i}",
+            "receiver_id": f"receiver{i}",
+            "packet_size": packet_size,
+            "packet_interval": packet_interval,
+            "number_of_packets": number_of_packets
+        }
+
+    return simulation
+
+def save_yaml(data, filename):
+    """Save data as YAML to a file"""
+    with open(filename, 'w') as f:
+        yaml.dump(data, f, sort_keys=False, default_flow_style=False)
+
+def parse_arguments():
+    """Parse and validate command line arguments"""
+    parser = argparse.ArgumentParser(description='Generate topology and simulation YAML files.')
+    parser.add_argument('--senders', type=int, required=True,
+                       help='Number of sender devices')
+    parser.add_argument('--receivers', type=int, required=True,
+                       help='Number of receiver devices')
+    parser.add_argument('--topology', default='stress_topology.yml',
+                       help='Output filename for topology file')
+    parser.add_argument('--simulation', default='ep_simulator.yml',
+                       help='Output filename for simulation file')
+    parser.add_argument('--topology-path', default='../topology_examples/ep_topology.yml',
+                       help='Path to topology file as referenced in simulation file')
+    parser.add_argument('--simulation-path', default='../simulation_examples/ep_simulator.yml',
+                       help='Path to topology file as referenced in simulation file')
+
+    args = parser.parse_args()
+
+    # Validate inputs
+    if args.senders < 1:
+        print("Error: Number of senders must be at least 1", file=sys.stderr)
+        sys.exit(1)
+    if args.receivers < 1:
+        print("Error: Number of receivers must be at least 1", file=sys.stderr)
+        sys.exit(1)
+
+    return args
+
+def main():
+    # Parse command line arguments
+    args = parse_arguments()
+
+    # Generate topology
+    topology = generate_topology(args.senders, args.receivers)
+    save_yaml(topology, args.topology_path)
+    print(f"Topology file saved as {args.topology} with {args.senders} senders and {args.receivers} receivers")
+
+    # Generate simulation
+    simulation = generate_simulation(args.topology_path, args.senders, args.receivers)
+    save_yaml(simulation, args.simulation_path)
+    print(f"Simulation file saved as {args.simulation}")
+
+if __name__ == "__main__":
+    main()

source/device/device.hpp

@@ -30,6 +30,7 @@ class IRoutingDevice : public Identifiable {
 public:
     virtual ~IRoutingDevice() = default;
 
+    virtual Id get_id() const = 0;
     virtual bool add_inlink(std::shared_ptr<ILink> link) = 0;
     virtual bool add_outlink(std::shared_ptr<ILink> link) = 0;
     virtual bool update_routing_table(Id dest_id, std::shared_ptr<ILink> link, size_t paths_count = 1) = 0;
@@ -44,13 +45,15 @@ class IRoutingDevice : public Identifiable {
 class IReceiver : public IRoutingDevice, public IProcessingDevice {
 public:
     virtual ~IReceiver() = default;
+    virtual Time send_system_packet(Packet packet) = 0;
 };
 
 class ISender : public IRoutingDevice, public IProcessingDevice {
 public:
     virtual ~ISender() = default;
     virtual void enqueue_packet(Packet packet) = 0;
     virtual Time send_data() = 0;
+    virtual Time send_system_packet(Packet packet) = 0;
 };
 
 class ISwitch : public IRoutingDevice,

source/device/receiver.cpp

@@ -75,10 +75,15 @@ Time Receiver::process() {
     LOG_INFO("Processing packet from link on receiver. Packet: " +
              data_packet.to_string());
 
-    if (data_packet.type == DATA && data_packet.dest_id == get_id()) {
+    if (data_packet.dest_id == get_id()) {
         // TODO: think about processing time
         // Not sure if we want to send ack before processing or after it
+        // TODO: move to TCP flow
         total_processing_time += send_ack(data_packet);
+        data_packet.flow->update(data_packet, get_type());
+        if (data_packet.type == PacketType::DATA) {
+            ++m_cnt;
+        }
     } else {
         LOG_WARN(
             "Packet arrived to Receiver that is not its destination; using "
@@ -101,6 +106,24 @@ Time Receiver::process() {
     return total_processing_time;
 }
 
+Time Receiver::send_system_packet(Packet packet) {
+    Time total_processing_time = 1;
+
+    auto next_link = get_link_to_destination(packet);
+    if (next_link == nullptr) {
+        LOG_WARN("Link to send packet does not exist. Packet: " + packet.to_string());
+        return total_processing_time;
+    }
+
+    // TODO: add some sender ID for easier packet path tracing
+    LOG_INFO("Sent new system packet from receiver. Data packet: " +
+        packet.to_string() + ". Receiver id: " + get_id() + " Time: " + std::to_string(Scheduler::get_instance().get_current_time()));
+
+    next_link->schedule_arrival(packet);
+    // total_processing_time += sending_data_time;
+    return total_processing_time;
+}
+
 Time Receiver::send_ack(Packet data_packet) {
     Time processing_time = 1;
     Time current_time = Scheduler::get_instance().get_current_time();

source/device/receiver.hpp

@@ -1,5 +1,6 @@
 #pragma once
 #include <memory>
+#include <iostream>
 
 #include "packet.hpp"
 #include "event.hpp"
@@ -16,7 +17,9 @@ class Receiver : public IReceiver,
                  public std::enable_shared_from_this<Receiver> {
 public:
     Receiver(Id a_id);
-    ~Receiver() = default;
+    ~Receiver() {
+        std::cout << "Arrived to " + get_id() + ": " << m_cnt << std::endl;
+    };
 
     bool add_inlink(std::shared_ptr<ILink> link) final;
     bool add_outlink(std::shared_ptr<ILink> link) final;
@@ -35,13 +38,15 @@ class Receiver : public IReceiver,
     // Packets are taken from ingress buffers on a round-robin basis.
     // The iterator over ingress buffers is stored in m_next_link.
     Time process() final;
+    Time send_system_packet(Packet packet) final;
 
     Id get_id() const final;
 
 private:
     Time send_ack(Packet data_packet);
     std::unique_ptr<IRoutingDevice> m_router;
     SchedulingModule<IReceiver, Process> m_process_scheduler;
+    int m_cnt = 0;
 };
 
 }  // namespace sim

source/device/sender.cpp

@@ -84,7 +84,7 @@ Time Sender::process() {
     LOG_INFO("Processing packet from link on sender. Packet: " +
              packet.to_string());
 
-    if (packet.type == PacketType::ACK && packet.dest_id == get_id()) {
+    if (packet.dest_id == get_id()) {
         packet.flow->update(packet, get_type());
     } else {
         LOG_WARN(
@@ -142,6 +142,24 @@ Time Sender::send_data() {
     return total_processing_time;
 }
 
+Time Sender::send_system_packet(Packet packet) {
+    Time total_processing_time = 1;
+
+    auto next_link = get_link_to_destination(packet);
+    if (next_link == nullptr) {
+        LOG_WARN("Link to send packet does not exist");
+        return total_processing_time;
+    }
+
+    // TODO: add some sender ID for easier packet path tracing
+    LOG_INFO("Sent new system packet from sender. Data packet: " +
+        packet.to_string() + ". Sender id: " + get_id() + ". Time: " + std::to_string(Scheduler::get_instance().get_current_time()));
+
+    next_link->schedule_arrival(packet);
+    // total_processing_time += sending_data_time;
+    return total_processing_time;
+}
+
 std::set<std::shared_ptr<ILink>> Sender::get_outlinks() {
     return m_router->get_outlinks();
 }

source/device/sender.hpp

@@ -38,6 +38,7 @@ class Sender : public ISender,
     // The iterator over ingress buffers is stored in m_next_link.
     Time process() final;
     Time send_data() final;
+    Time send_system_packet(Packet packet) final;
 
     void enqueue_packet(Packet packet) final;
 

source/express_pass/ep_event.cpp

@@ -0,0 +1,42 @@
+#include "event.hpp"
+#include "express_pass/ep_event.hpp"
+
+#include "scheduler.hpp"
+
+namespace sim {
+
+SendCredit::SendCredit(Time a_time, std::weak_ptr<EPFlow> a_flow, Size a_packet_size) : Event(a_time), m_flow(a_flow), m_packet_size(a_packet_size) {}
+
+void SendCredit::operator()() {
+    if (m_flow.expired()) {
+        return;
+    }
+
+    Time sending_delay = m_flow.lock()->send_credit();
+    if (sending_delay == 0) {
+        return;
+    }
+
+    std::unique_ptr<Event> new_event = std::make_unique<SendCredit>(
+        m_time + sending_delay, m_flow, m_packet_size);
+    Scheduler::get_instance().add(std::move(new_event));
+}
+
+RunFeedbackControlLoop::RunFeedbackControlLoop(Time a_time, std::weak_ptr<EPFlow> a_flow) : Event(a_time), m_flow(a_flow) {}
+
+void RunFeedbackControlLoop::operator()() {
+    if (m_flow.expired()) {
+        return;
+    }
+
+    Time rtt = m_flow.lock()->feedback_control_loop();
+    if (rtt == 0) {
+        return;
+    }
+
+    std::unique_ptr<Event> new_event = std::make_unique<RunFeedbackControlLoop>(
+        m_time + rtt, m_flow);
+    Scheduler::get_instance().add(std::move(new_event));
+}
+
+}  // namespace sim

source/express_pass/ep_event.hpp

@@ -0,0 +1,27 @@
+#include "event.hpp"
+#include "express_pass/ep_flow.hpp"
+
+namespace sim {
+
+class SendCredit : public Event {
+public:
+    SendCredit(Time a_time, std::weak_ptr<EPFlow> a_flow, Size a_packet_size);
+    virtual ~SendCredit() = default;
+    void operator()() final;
+
+private:
+    std::weak_ptr<EPFlow> m_flow;
+    Size m_packet_size;
+};
+
+class RunFeedbackControlLoop : public Event {
+public:
+    RunFeedbackControlLoop(Time a_time, std::weak_ptr<EPFlow> a_flow);
+    virtual ~RunFeedbackControlLoop() = default;
+    void operator()() final;
+
+private:
+    std::weak_ptr<EPFlow> m_flow;
+};
+
+} // namespace sim