diff --git a/configuration_examples/generator/basic_ep_generator.py b/configuration_examples/generator/basic_ep_generator.py new file mode 100644 index 0000000000..e7c83ccb15 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/source/device/device.hpp b/source/device/device.hpp index da39fa9cf2..0d70916c83 100644 --- a/source/device/device.hpp +++ b/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 link) = 0; virtual bool add_outlink(std::shared_ptr link) = 0; virtual bool update_routing_table(Id dest_id, std::shared_ptr link, size_t paths_count = 1) = 0; @@ -44,6 +45,7 @@ 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 { @@ -51,6 +53,7 @@ class ISender : public IRoutingDevice, public IProcessingDevice { 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, diff --git a/source/device/receiver.cpp b/source/device/receiver.cpp index 909d9e488f..e8483f39fd 100644 --- a/source/device/receiver.cpp +++ b/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(); diff --git a/source/device/receiver.hpp b/source/device/receiver.hpp index 176916cd1c..25a1f3c488 100644 --- a/source/device/receiver.hpp +++ b/source/device/receiver.hpp @@ -1,5 +1,6 @@ #pragma once #include +#include #include "packet.hpp" #include "event.hpp" @@ -16,7 +17,9 @@ class Receiver : public IReceiver, public std::enable_shared_from_this { public: Receiver(Id a_id); - ~Receiver() = default; + ~Receiver() { + std::cout << "Arrived to " + get_id() + ": " << m_cnt << std::endl; + }; bool add_inlink(std::shared_ptr link) final; bool add_outlink(std::shared_ptr link) final; @@ -35,6 +38,7 @@ 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; @@ -42,6 +46,7 @@ class Receiver : public IReceiver, Time send_ack(Packet data_packet); std::unique_ptr m_router; SchedulingModule m_process_scheduler; + int m_cnt = 0; }; } // namespace sim diff --git a/source/device/sender.cpp b/source/device/sender.cpp index 0dca532c79..89f578bbf7 100644 --- a/source/device/sender.cpp +++ b/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> Sender::get_outlinks() { return m_router->get_outlinks(); } diff --git a/source/device/sender.hpp b/source/device/sender.hpp index 661303a78b..10572d199e 100644 --- a/source/device/sender.hpp +++ b/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; diff --git a/source/express_pass/ep_event.cpp b/source/express_pass/ep_event.cpp new file mode 100644 index 0000000000..c04d8c086e --- /dev/null +++ b/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 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 new_event = std::make_unique( + 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 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 new_event = std::make_unique( + m_time + rtt, m_flow); + Scheduler::get_instance().add(std::move(new_event)); +} + +} // namespace sim \ No newline at end of file diff --git a/source/express_pass/ep_event.hpp b/source/express_pass/ep_event.hpp new file mode 100644 index 0000000000..5621e01188 --- /dev/null +++ b/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 a_flow, Size a_packet_size); + virtual ~SendCredit() = default; + void operator()() final; + +private: + std::weak_ptr m_flow; + Size m_packet_size; +}; + +class RunFeedbackControlLoop : public Event { +public: + RunFeedbackControlLoop(Time a_time, std::weak_ptr a_flow); + virtual ~RunFeedbackControlLoop() = default; + void operator()() final; + +private: + std::weak_ptr m_flow; +}; + +} // namespace sim \ No newline at end of file diff --git a/source/express_pass/ep_flow.cpp b/source/express_pass/ep_flow.cpp new file mode 100644 index 0000000000..375caffce4 --- /dev/null +++ b/source/express_pass/ep_flow.cpp @@ -0,0 +1,230 @@ +#include "express_pass/ep_flow.hpp" + +#include +#include +#include + +#include "scheduler.hpp" +#include "logger/logger.hpp" +#include "express_pass/ep_event.hpp" + +namespace sim { + +EPFlow::EPFlow(Id a_id, std::shared_ptr a_src, std::shared_ptr a_dest, + Size a_packet_size, Time a_delay_between_packets, + std::uint32_t a_packets_to_send) + : m_id(a_id), + m_src(a_src), + m_dest(a_dest), + m_sender_status(SenderStatus::CREDIT_STOP_S), + m_receiver_status(ReceiverStatus::CREDIT_STOP_R), + m_packet_size(a_packet_size), + m_delay_between_packets(a_delay_between_packets), + m_packets_to_send(a_packets_to_send) { + + auto link = m_dest.lock()->get_link_to_destination(generate_packet(PacketType::CREDIT, true)); + m_max_rate = (double)(m_packet_size * 8) / 10.0; // TODO: / link.get_speed(); + } + +void EPFlow::schedule_packet_generation(Time time) { + auto generate_event_ptr = + std::make_unique(time, shared_from_this(), m_packet_size); + Scheduler::get_instance().add(std::move(generate_event_ptr)); +} + +Packet EPFlow::generate_packet(PacketType type, bool from_receiver, std::uint32_t packet_num, Time RTT) { + sim::Packet packet; + packet.type = type; + if (type == PacketType::DATA) { + packet.size_byte = m_packet_size; + } else { + packet.size_byte = m_system_packet_size + (rand() % 20); + } + packet.flow = this; + packet.source_id = (from_receiver ? m_dest.lock()->get_id() : m_src.lock()->get_id()); + packet.dest_id = (from_receiver ? m_src.lock()->get_id() : m_dest.lock()->get_id()); + packet.packet_num = packet_num; + packet.RTT = RTT; + return packet; +} + +std::uint32_t EPFlow::get_next_packet_num() { + return m_current_packet_num++; +} + +void EPFlow::start() { + schedule_packet_generation(Scheduler::get_instance().get_current_time()); + + sim::Packet packet = generate_packet(PacketType::CREDIT_REQUEST, false); + m_src.lock()->send_system_packet(packet); + m_sender_status = SenderStatus::CREQ_SENT; +} + +Time EPFlow::send_credit() { + if (m_stopped) { + return 0; + } + + feedback_control_loop(); + + auto packet = generate_packet(PacketType::CREDIT, true, get_next_packet_num(), Scheduler::get_instance().get_current_time()); + m_dest.lock()->send_system_packet(packet); + m_sent_credits++; + + std::uint32_t rate = m_max_rate / m_current_rate_coeff; + int rnd_part = (std::rand() % (int)(rate * 0.08)) - (int)(rate * 0.04); + m_accum += rate + rnd_part; + return rate + rnd_part; + +} + + +void EPFlow::process_getting_data_packet(Packet data_packet) { + double alpha = 0.2; + m_RTT = (1 - alpha) * m_RTT + alpha * (Scheduler::get_instance().get_current_time() - data_packet.RTT); + + int distance = data_packet.packet_num - m_last_obtained_packet_num; + + if (distance < 0) { + // TODO: log error credit packet reordering or credit sequence number overflow happend. + throw "ERR"; + } + m_sent_credits += (distance + 1); + m_lost_credits += distance; + + m_last_obtained_packet_num = data_packet.packet_num + 1; + + // LOG_ERROR("Got DATA " + std::to_string(get_id()) + " Packet: " + data_packet.to_string() + " RTT: " + std::to_string(m_RTT) + " REAL RTT: " + std::to_string(Scheduler::get_instance().get_current_time() - data_packet.RTT)); +} + +// feedback control launch once per RTT (scheduled by event) +Time EPFlow::feedback_control_loop() { + if (m_sent_credits == 0) { + return m_RTT; + } + + if (Scheduler::get_instance().get_current_time() - m_last_credit_rate_update < m_RTT) { + return m_RTT; + } + + // std::cout << "Lost: " << m_lost_credits << std::endl; + // std::cout << "Sent: " << m_sent_credits << std::endl; + + auto old_coeff = m_current_rate_coeff; + double credit_loss = (double)m_lost_credits / (double)m_sent_credits; + + if (credit_loss <= m_target_loss && m_incr_cnt <= 10) { + if (m_was_increasing) { + m_w = (m_w + m_w_max) / 2; + m_incr_cnt++; + } + m_was_increasing = true; + m_current_rate_coeff = (1 - m_w) * m_current_rate_coeff + m_w * m_max_rate_coeff * (1 + m_target_loss); + } else { + m_max_meg_incr_in_a_row = std::max(m_max_meg_incr_in_a_row, m_incr_cnt); + m_was_increasing = false; + if (credit_loss >= 1.0) { + m_current_rate_coeff = m_max_rate / (double)m_RTT; + } else { + if (m_incr_cnt > 10) { + m_current_rate_coeff /= (8.0 * (1 - credit_loss)); + } else { + m_current_rate_coeff = m_current_rate_coeff * (1 - credit_loss) * (1 + m_target_loss); + } + + + // TEST PART + // m_current_rate_coeff = m_max_rate*(m_sent_credits - m_lost_credits) + // / (double)(Scheduler::get_instance().get_current_time() - m_last_credit_rate_update) + // * (1.0 + m_target_loss); + // m_current_rate_coeff = m_current_rate_coeff * (1 - credit_loss) * (1 + m_target_loss); + } + if (m_current_rate_coeff > old_coeff) { + m_current_rate_coeff = old_coeff; + } + + m_incr_cnt = 0; + m_w = std::max(m_w / 2, m_w_min); + } + + // std::cout << "Rate: " << m_current_rate_coeff << std::endl; + + m_lost_credits = 0; + m_sent_credits = 0; + m_last_credit_rate_update = Scheduler::get_instance().get_current_time(); + + return m_RTT; +} + +void EPFlow::update(Packet packet, DeviceType type) { + if (packet.type == PacketType::CREDIT_REQUEST && type == DeviceType::RECEIVER && m_receiver_status == ReceiverStatus::CREDIT_STOP_R) { + m_receiver_status = ReceiverStatus::CREDIT_SENDING; + + auto send_credit_event_ptr = + std::make_unique(Scheduler::get_instance().get_current_time(), shared_from_this(), m_system_packet_size); + Scheduler::get_instance().add(std::move(send_credit_event_ptr)); + + // auto feedback_control_event_ptr = + // std::make_unique(Scheduler::get_instance().get_current_time(), shared_from_this()); + // Scheduler::get_instance().add(std::move(feedback_control_event_ptr)); + } + else if (packet.type == PacketType::CREDIT_STOP_P && type == DeviceType::RECEIVER && m_receiver_status == ReceiverStatus::CREDIT_SENDING) { + m_receiver_status = ReceiverStatus::CREDIT_STOP_R; + m_stopped = true; + } + else if (packet.type == PacketType::DATA && type == DeviceType::RECEIVER) { + // calculate data statistics + // check corresponding credit id and update state + process_getting_data_packet(packet); + } else if (packet.type == PacketType::CREDIT && type == DeviceType::SENDER) { + // LOG_ERROR("Got CREDIT " + std::to_string(get_id()) + " TIME: " + std::to_string(Scheduler::get_instance().get_current_time())); + if (m_sender_status != SenderStatus::CREQ_SENT && m_sender_status != SenderStatus::CREDIT_RECEIVING) { + LOG_ERROR("Unexpected sender status when got credit"); + return; + } + if (m_sender_status == SenderStatus::CREQ_SENT) { + m_sender_status = SenderStatus::CREDIT_RECEIVING; + } + + if (m_sending_buffer.empty()) { + m_sender_status = SenderStatus::CSTOP_SENT; + + auto packet = generate_packet(PacketType::CREDIT_STOP_P, false); + m_src.lock()->send_system_packet(packet); + } else { + m_sending_buffer.front().RTT = packet.RTT; + m_sending_buffer.front().packet_num = packet.packet_num; + put_data_to_device(); + } + } + else { + LOG_ERROR("Unexpected update state"); + } +} + + +Time EPFlow::create_new_data_packet() { + if (m_packets_to_send == 0) { + return 0; + } + --m_packets_to_send; + Packet data = generate_packet(PacketType::DATA); + LOG_INFO("Created packet: " + data.to_string()); + m_sending_buffer.push(data); + return m_delay_between_packets; +} + +Time EPFlow::put_data_to_device() { + m_src.lock()->enqueue_packet(m_sending_buffer.front()); + LOG_INFO("Put data to device: " + m_sending_buffer.front().to_string()); + m_sending_buffer.pop(); + return m_delay_between_packets; +} + +std::shared_ptr EPFlow::get_sender() const { return m_src.lock(); } + +std::shared_ptr EPFlow::get_receiver() const { return m_dest.lock(); } + +Id EPFlow::get_id() const { return m_id; } + +} // namespace sim \ No newline at end of file diff --git a/source/express_pass/ep_flow.hpp b/source/express_pass/ep_flow.hpp new file mode 100644 index 0000000000..a55b3d0984 --- /dev/null +++ b/source/express_pass/ep_flow.hpp @@ -0,0 +1,98 @@ +#pragma once +#include +#include + +#include "flow/flow.hpp" +#include "packet.hpp" +#include "utils/identifier_factory.hpp" + +namespace sim { + +class IReceiver; +class ISender; + +enum SenderStatus { + CREDIT_STOP_S, + CREQ_SENT, + CREDIT_RECEIVING, + CSTOP_SENT +}; + +enum ReceiverStatus { + CREDIT_STOP_R, + CREDIT_SENDING +}; + +class EPFlow : public IFlow, public std::enable_shared_from_this { +public: + EPFlow(Id a_id, std::shared_ptr a_src, std::shared_ptr a_dest, + Size a_packet_size, Time a_delay_between_packets, + std::uint32_t a_packets_to_send); + virtual ~EPFlow() { + std::cout << "Max incr: " << std::max(m_max_meg_incr_in_a_row, m_incr_cnt) << std::endl; + }; + + // Start at time + void start() final; + + Time create_new_data_packet() final; + Time put_data_to_device(); + + // Update the internal state according to some congestion control algorithm + // Call try_to_generate upon the update + void update(Packet packet, DeviceType type) final; + + std::shared_ptr get_sender() const final; + std::shared_ptr get_receiver() const final; + + Time send_credit(); + Time feedback_control_loop(); + + Id get_id() const final; + +private: + void schedule_packet_generation(Time time); + Packet generate_packet(PacketType type, bool from_receiver = false, std::uint32_t packet_num = 0, Time RTT = 0); + void sender_send_stop(); + std::uint32_t get_next_packet_num(); + void process_getting_data_packet(Packet data_packet); + + Id m_id; + std::weak_ptr m_src; + std::weak_ptr m_dest; + SenderStatus m_sender_status; + ReceiverStatus m_receiver_status; + + bool m_stopped = false; + std::uint32_t m_current_packet_num = 0; + + Size m_packet_size; + Time m_delay_between_packets; + std::uint32_t m_packets_to_send; + Size m_system_packet_size = 84; + + + double m_max_rate_coeff = 1.0; + double m_current_rate_coeff = 1.0 / 4.0; + double m_max_rate; + double m_w = 1.0 / 8.0; + double m_w_max = 0.5; + double m_w_min = 0.01; + double m_target_loss = 0.05; + bool m_was_increasing = false; + int m_incr_cnt = 0; + int m_max_meg_incr_in_a_row = 0; + Time m_last_credit_rate_update = 0; + + Time m_RTT = 9000; + std::uint32_t m_last_obtained_packet_num = 0; + std::uint32_t m_last_received_credit = 0; + std::uint32_t m_sent_credits = 0; + std::uint32_t m_lost_credits = 0; + std::uint32_t m_last_feedback_run = 0; + + std::queue m_sending_buffer; + Time m_accum; +}; + +} // namespace sim \ No newline at end of file diff --git a/source/express_pass/ep_link.cpp b/source/express_pass/ep_link.cpp new file mode 100644 index 0000000000..29bbb02d4c --- /dev/null +++ b/source/express_pass/ep_link.cpp @@ -0,0 +1,168 @@ +#include "express_pass/ep_link.hpp" + +#include "device/switch.hpp" +#include "scheduler.hpp" +#include "logger/logger.hpp" + +namespace sim { + +EPLink::EPLink(Id a_id, std::weak_ptr a_from, + std::weak_ptr a_to, std::uint32_t a_speed_gbps, + Time a_delay, Size a_max_src_egress_buffer_size_byte, + Size a_max_ingress_buffer_size) + : m_id(a_id), + m_from(a_from), + m_to(a_to), + m_speed_gbps(a_speed_gbps), + m_src_egress_buffer_size_byte(0), + m_max_src_egress_buffer_size_byte(a_max_src_egress_buffer_size_byte), + m_last_src_egress_pass_time(0), + m_transmission_delay(a_delay), + m_next_ingress(), + m_ingress_buffer_size_byte(0), + m_max_ingress_buffer_size_byte(a_max_ingress_buffer_size) { + if (a_from.expired() || a_to.expired()) { + LOG_WARN("Passed link to device is expired"); + } else if (a_speed_gbps == 0) { + LOG_WARN("Passed zero link speed"); + } +} + +Time EPLink::get_transmission_time(const Packet& packet) const { + if (m_speed_gbps == 0) { + LOG_WARN("Passed zero link speed"); + return 0; + } + const std::uint32_t byte_to_bit_multiplier = 8; + + Size packet_size_bit = packet.size_byte * byte_to_bit_multiplier; + std::uint32_t transmission_speed_bit_ns = m_speed_gbps; + return (packet_size_bit + transmission_speed_bit_ns - 1) / + transmission_speed_bit_ns + + m_transmission_delay; +}; + +void EPLink::update_token_bucket() { + Time passed = Scheduler::get_instance().get_current_time() - m_last_token_update; + Size tokens_income = passed * (double)m_speed_gbps * m_token_getting_speed * 8; + + m_token_bucket_size = std::min(m_token_bucket_size + tokens_income, m_max_token_bucket_size); + + m_last_token_update = Scheduler::get_instance().get_current_time(); +} + +void EPLink::schedule_arrival(Packet packet) { + if (m_to.expired()) { + LOG_WARN("Destination device pointer is expired"); + return; + } + + update_token_bucket(); + + Time total_delay; + + // TODO: rewrite this part with limiting throughput instead of calculating sending time separately + auto src = m_from.lock(); + if (packet.type == PacketType::CREDIT) { + if (m_current_credit_queue_capacity + packet.size_byte > m_max_credit_queue_capacity) { + // LOG_ERROR("Dropped credit: " + packet.to_string() + " Time: " + std::to_string(Scheduler::get_instance().get_current_time())); + return; + } + + if (packet.size_byte > m_token_bucket_size) { + return; + } + m_token_bucket_size -= packet.size_byte; + + m_current_credit_queue_capacity += packet.size_byte; + total_delay = std::max(m_next_credit_can_be_sent, Scheduler::get_instance().get_current_time()); + m_next_credit_can_be_sent = total_delay + (packet.size_byte * 8 + 1538 * 8) / m_speed_gbps; + // LOG_ERROR("Next credit can be sent: " + std::to_string(m_next_credit_can_be_sent)); + } else { + if (m_src_egress_buffer_size_byte + packet.size_byte > + m_max_src_egress_buffer_size_byte) { + LOG_ERROR("Buffer in link overflowed; packet " + packet.to_string() + + " lost"); + return; + } + + LOG_INFO("Packet arrived to link's ingress queue. Packet: " + + packet.to_string()); + Time transmission_time = get_transmission_time(packet); + + m_last_src_egress_pass_time = + std::max(m_last_src_egress_pass_time, + Scheduler::get_instance().get_current_time()) + + transmission_time; + m_src_egress_buffer_size_byte += packet.size_byte; + m_max_buffer_size = std::max(m_max_buffer_size, m_src_egress_buffer_size_byte); + total_delay = m_last_src_egress_pass_time; + } + + m_to.lock()->notify_about_arrival( + Scheduler::get_instance().get_current_time()); + + Scheduler::get_instance().add(std::make_unique( + Arrive(total_delay, weak_from_this(), packet))); +}; + +void EPLink::process_arrival(Packet packet) { + auto src = m_from.lock(); + if (packet.type == PacketType::CREDIT) { + m_current_credit_queue_capacity -= packet.size_byte; + } else { + if (m_ingress_buffer_size_byte + packet.size_byte > + m_max_ingress_buffer_size_byte) { + LOG_ERROR("Ingress buffer on EPLink overflow; packet" + + packet.to_string() + " lost"); + return; + } + m_ingress_buffer_size_byte += packet.size_byte; + m_max_buffer_size = std::max(m_max_buffer_size, m_ingress_buffer_size_byte); + + LOG_INFO("Packet arrived to link's egress queue. Packet: " + + packet.to_string()); + + m_src_egress_buffer_size_byte -= packet.size_byte; + } + m_next_ingress.push(packet); +}; + +std::optional EPLink::get_packet() { + if (m_next_ingress.empty()) { + LOG_INFO("Ingress packet queue is empty"); + return {}; + } + + auto packet = m_next_ingress.front(); + LOG_INFO("Taken packet from link. Packet: " + packet.to_string()); + m_next_ingress.pop(); + m_ingress_buffer_size_byte -= packet.size_byte; + return packet; +}; + +std::shared_ptr EPLink::get_from() const { + if (m_from.expired()) { + LOG_WARN("Source device pointer is expired"); + return nullptr; + } + + return m_from.lock(); +}; + +std::shared_ptr EPLink::get_to() const { + if (m_to.expired()) { + LOG_WARN("Destination device pointer is expired"); + return nullptr; + } + + return m_to.lock(); +}; + +Size EPLink::get_max_from_egress_buffer_size() const { + return m_max_src_egress_buffer_size_byte; +} + +Id EPLink::get_id() const { return m_id; } + +} // namespace sim diff --git a/source/express_pass/ep_link.hpp b/source/express_pass/ep_link.hpp new file mode 100644 index 0000000000..c45a3d8198 --- /dev/null +++ b/source/express_pass/ep_link.hpp @@ -0,0 +1,71 @@ +#pragma once + +#include "link.hpp" +#include + +namespace sim { + +class EPLink : public ILink, public std::enable_shared_from_this { +public: + EPLink(Id a_id, std::weak_ptr a_from, + std::weak_ptr a_to, std::uint32_t a_speed_gbps = 1, + Time a_delay = 0, Size a_max_src_egress_buffer_size_byte = 32768, + Size a_max_ingress_buffer_size_byte = 32768); + ~EPLink() { + // std::cout << "Max buffer: " << m_max_buffer_size << std::endl; + }; + + /** + * Update the source egress delay and schedule the arrival event + * based on the egress queueing and transmission delays. + */ + void schedule_arrival(Packet packet) final; + + /** + * Removes packet from the source egress queue. + */ + void process_arrival(Packet packet) final; + + std::optional get_packet() final; + + std::shared_ptr get_from() const final; + std::shared_ptr get_to() const final; + Size get_max_from_egress_buffer_size() const final; + + Id get_id() const final; + +private: + Time get_transmission_time(const Packet& packet) const; + void update_token_bucket(); + + Id m_id; + std::weak_ptr m_from; + std::weak_ptr m_to; + std::uint32_t m_speed_gbps; + + Size m_src_egress_buffer_size_byte; + Size m_max_src_egress_buffer_size_byte; + Time m_last_src_egress_pass_time; + + Time m_transmission_delay; + + // Queue at the ingress port of the m_next device + std::queue m_next_ingress; + Size m_ingress_buffer_size_byte; + Size m_max_ingress_buffer_size_byte; + + + std::uint32_t m_max_credit_queue_capacity = 8 * 84; + std::uint32_t m_current_credit_queue_capacity = 0; + Time m_next_credit_can_be_sent = 0; + + Size m_max_token_bucket_size = 2 * 84; + Size m_token_bucket_size = 2 * 84; + Time m_last_token_update = 0; + double m_token_getting_speed = 0.0517; + + // Debug + Size m_max_buffer_size = 0; +}; + +} // namespace sim diff --git a/source/link.cpp b/source/link.cpp index d34fc48b06..78c5fcdf16 100644 --- a/source/link.cpp +++ b/source/link.cpp @@ -59,7 +59,10 @@ void Link::schedule_arrival(Packet packet) { return; } - unsigned int transmission_time = get_transmission_time(packet); + LOG_INFO("Packet arrived to link's ingress queue. Packet: " + + packet.to_string()); + + Time transmission_time = get_transmission_time(packet); m_arrival_time = std::max(m_arrival_time, Scheduler::get_instance().get_current_time()) + transmission_time; diff --git a/source/logger/logger.cpp b/source/logger/logger.cpp index 27a03018ac..01c94ab5ce 100644 --- a/source/logger/logger.cpp +++ b/source/logger/logger.cpp @@ -21,11 +21,11 @@ void Logger::disable_logs() { Logger::Logger() { auto console_sink = std::make_shared(); - console_sink->set_level(spdlog::level::warn); + console_sink->set_level(spdlog::level::err); auto file_sink = std::make_shared( "logs/simulator_logs.txt", true); - file_sink->set_level(spdlog::level::trace); + file_sink->set_level(spdlog::level::err); auto logger = std::make_shared( "multi_sink", spdlog::sinks_init_list{console_sink, file_sink}); diff --git a/source/main.cpp b/source/main.cpp index 7323ecd24a..a4057cadf0 100644 --- a/source/main.cpp +++ b/source/main.cpp @@ -1,4 +1,6 @@ #include +#include +#include #include "logger/logger.hpp" #include "metrics/metrics_collector.hpp" @@ -29,6 +31,7 @@ int main(const int argc, char **argv) { Logger::get_instance().disable_logs(); } } + srand(time(0)); try { sim::YamlParser parser; diff --git a/source/packet.cpp b/source/packet.cpp index 16efa40c6e..98952ac13f 100644 --- a/source/packet.cpp +++ b/source/packet.cpp @@ -36,6 +36,15 @@ std::string Packet::to_string() const { case PacketType::DATA: oss << "DATA"; break; + case PacketType::CREDIT: + oss << "CREDIT"; + break; + case PacketType::CREDIT_REQUEST: + oss << "CREQ"; + break; + case PacketType::CREDIT_STOP_P: + oss << "CSTOP"; + break; default: oss << "UNKNOWN"; break; @@ -45,7 +54,7 @@ std::string Packet::to_string() const { oss << ", dest_id: " << dest_id; oss << ", packet_num: " << packet_num; oss << ", size(byte): " << size_byte; - oss << ", flow: " << (flow ? "set" : "null"); + oss << ", flow: " << (flow ? flow->get_id() : "null"); oss << ", send time: " << send_time; oss << "]"; diff --git a/source/packet.hpp b/source/packet.hpp index 8266fda694..c023d8eec6 100644 --- a/source/packet.hpp +++ b/source/packet.hpp @@ -9,7 +9,7 @@ namespace sim { class IFlow; -enum PacketType { ACK, DATA }; +enum PacketType { ACK, DATA, CREDIT_REQUEST, CREDIT, CREDIT_STOP_P }; struct Packet { Packet(PacketType a_type = PacketType::DATA, Size a_size_byte = 0, diff --git a/source/parser/identifiable_parser.hpp b/source/parser/identifiable_parser.hpp index 61b5e7e235..8d0340f342 100644 --- a/source/parser/identifiable_parser.hpp +++ b/source/parser/identifiable_parser.hpp @@ -130,6 +130,73 @@ Id parse_object(const YAML::Node& key_node, return id; } + +template <> +Id parse_object(const YAML::Node& key_node, + const YAML::Node& value_node) { + Id id = key_node.as(); + Id from_id = value_node["from"].as(); + Id to_id = value_node["to"].as(); + auto from_ptr = + IdentifierFactory::get_instance().get_object(from_id); + auto to_ptr = + IdentifierFactory::get_instance().get_object(to_id); + + if (from_ptr == nullptr) { + LOG_ERROR("Failed to find link's source"); + return ""; + } + + if (to_ptr == nullptr) { + LOG_ERROR("Failed to find link's destination"); + return ""; + } + + uint32_t latency = 0; + if (value_node["latency"]) { + latency = parse_latency(value_node["latency"].as()); + } else { + LOG_WARN(fmt::format( + "latency does not set for link {}; use default value {}", id, + latency)); + } + + uint32_t speed = 1; + if (value_node["throughput"]) { + speed = parse_throughput(value_node["throughput"].as()); + } else { + LOG_WARN(fmt::format( + "speed does not set for link {}; use default value {}", id, speed)); + } + + uint32_t ingress_buffer_size = 4096; + if (value_node["ingress_buffer_size"]) { + ingress_buffer_size = parse_buffer_size( + value_node["ingress_buffer_size"].as()); + } else { + LOG_WARN( + fmt::format("ingress buffer size does not set for link {}; use " + "default value {}", + id, ingress_buffer_size)); + } + + uint32_t egress_buffer_size = 4096; + if (value_node["egress_buffer_size"]) { + egress_buffer_size = parse_buffer_size( + value_node["egress_buffer_size"].as()); + } else { + LOG_WARN(fmt::format( + "egress buffer size does not set for link {}; use default value {}", + id, egress_buffer_size)); + } + + parse_object_helper(id, from_ptr, to_ptr, speed, latency, + egress_buffer_size, ingress_buffer_size); + + return id; + +} + struct FlowCommon{ Id id; std::shared_ptr sender_ptr; @@ -181,4 +248,15 @@ Id parse_object(const YAML::Node& key_node, return flow_common.id; } +template <> +Id parse_object(const YAML::Node& key_node, + const YAML::Node& value_node) { + FlowCommon flow_common = parse_flow_common(key_node, value_node); + + parse_object_helper(flow_common.id, flow_common.sender_ptr, flow_common.receiver_ptr, + flow_common.packet_size, flow_common.packet_interval, + flow_common.number_of_packets); + return flow_common.id; +} + } // namespace sim diff --git a/source/simulator.cpp b/source/simulator.cpp index 4eb65b0b2a..b44d5d8cde 100644 --- a/source/simulator.cpp +++ b/source/simulator.cpp @@ -7,6 +7,9 @@ namespace sim { SimulatorVariant create_simulator(std::string_view algorithm) { if (algorithm == "basic") { return BasicSimulator(); + } + if (algorithm == "ep") { + return EPSimulator(); } if (algorithm == "tcp") { return TcpSimulator(); diff --git a/source/simulator.hpp b/source/simulator.hpp index 0b5ad543c9..423a284c2a 100644 --- a/source/simulator.hpp +++ b/source/simulator.hpp @@ -13,9 +13,16 @@ #include "device/receiver.hpp" #include "device/sender.hpp" #include "device/switch.hpp" -#include "event.hpp" +#include "flow/flow.hpp" #include "flow/tcp_flow.hpp" +#include "event.hpp" #include "link.hpp" + +// #include "express_pass/ep_receiver.hpp" +// #include "express_pass/ep_sender.hpp" +#include "express_pass/ep_link.hpp" +#include "express_pass/ep_flow.hpp" + #include "logger/logger.hpp" #include "metrics/metrics_collector.hpp" #include "scheduler.hpp" @@ -127,8 +134,9 @@ class Simulator { using BasicSimulator = Simulator; using TcpSimulator = Simulator; +using EPSimulator = Simulator;//Simulator; -using SimulatorVariant = std::variant; +using SimulatorVariant = std::variant; SimulatorVariant create_simulator(std::string_view algorithm); diff --git a/source/utils/hasher.cpp b/source/utils/hasher.cpp index 9151c07521..4a663de7d0 100644 --- a/source/utils/hasher.cpp +++ b/source/utils/hasher.cpp @@ -1,3 +1,4 @@ +#include #include diff --git a/test/switch/receiver_mock.cpp b/test/switch/receiver_mock.cpp index 0d06d3a88a..45aca71846 100644 --- a/test/switch/receiver_mock.cpp +++ b/test/switch/receiver_mock.cpp @@ -40,4 +40,8 @@ std::set> ReceiverMock::get_outlinks() { Id ReceiverMock::get_id() const { return ""; } +Time ReceiverMock::send_system_packet(sim::Packet packet) { + return 1; +}; + } // namespace test diff --git a/test/switch/receiver_mock.hpp b/test/switch/receiver_mock.hpp index 795702ce4e..8082494742 100644 --- a/test/switch/receiver_mock.hpp +++ b/test/switch/receiver_mock.hpp @@ -22,6 +22,7 @@ class ReceiverMock : public sim::IReceiver { Time process() final; sim::DeviceType get_type() const final; + Time send_system_packet(sim::Packet packet) final; Id get_id() const final; };