FEAT: add connections & flows summary (#529)

closes #528
closes #520

---------

Co-authored-by: Alexey Zharkov <62723911+Az3git@users.noreply.github.com>

Author: PaulRalnikov

configs/fat-tree/network.yml

@@ -4,31 +4,36 @@ presets:
   default-flow:
     type: tcp
     metrics_filters:
-      rtt: true
+      rtt: false
       reordering: false
       delivery_rate: false
-  
+
   default-connection:
     type: mplb-connection
     mplb:
       type: single-cc
       packet-size: 1500B
-      
+
+      metrics_filters:
+        fairness: false
+
       cc:
         type: swift
-        base_target: 36000ns
+        base_target: 20000ns
+        metrics_filters:
+          cwnd: false
 
       path-chooser:
         type: round-robin
         flows:
           flow-1:
             type: tcp
             preset-name: default-flow
-          
+
           flow-2:
             type: tcp
             preset-name: default-flow
-          
+
           flow-3:
             type: tcp
             preset-name: default-flow

configs/fat-tree/scenario.yml

@@ -7,4 +7,3 @@ scenario:
   size: 5MB
   repeat_count: 1
   connections: ^connection.*
-  repeat_interval: 500000ns

configs/fat-tree/topology.yml

@@ -1,7 +1,7 @@
 presets:
   edge-host-link:
     latency: 1000ns
-    throughput: 1Gbps
+    throughput: 10Gbps
     ingress_buffer_size: 4096KB
     egress_buffer_size: 4096KB
     metrics_filters:

source/connection/flow/endpoint_ports.hpp

@@ -0,0 +1,15 @@
+#pragma once
+#include <cstdint>
+namespace sim {
+
+using Port = std::uint32_t;
+
+struct EndpointPorts {
+    Port sender_port = 0;
+    Port receriver_port = 0;
+
+    EndpointPorts(Port a_sender_port = 0, Port a_receiver_port = 0)
+        : sender_port(a_sender_port), receriver_port(a_receiver_port) {}
+};
+
+}  // namespace sim

source/connection/flow/flow_four_tuple.hpp

@@ -0,0 +1,12 @@
+#pragma once
+#include "endpoint_ports.hpp"
+#include "endpoints.hpp"
+
+namespace sim {
+
+struct FlowFourTuple : Endpoints, EndpointPorts {
+    FlowFourTuple(Endpoints endpoints, EndpointPorts ports = {})
+        : Endpoints(endpoints), EndpointPorts(ports) {}
+};
+
+}  // namespace sim

source/connection/flow/i_new_flow.hpp

@@ -1,7 +1,7 @@
 #pragma once
 
 #include "device/interfaces/i_host.hpp"
-#include "endpoints.hpp"
+#include "flow_four_tuple.hpp"
 #include "metrics/metrics_storage.hpp"
 #include "metrics/metrics_table/i_metricable.hpp"
 #include "packet_ack_info.hpp"
@@ -10,8 +10,7 @@
 
 namespace sim {
 
-struct FlowContext : Endpoints {
-    using Endpoints::Endpoints;
+struct FlowContext : FlowFourTuple {
     SizeByte sent_size = SizeByte(0);
     SizeByte delivered_size = SizeByte(0);
     SizeByte retransmit_size = SizeByte(0);
@@ -20,6 +19,8 @@ struct FlowContext : Endpoints {
     std::optional<TimeNs> last_ack_receive_time = std::nullopt;
     utils::Statistics<TimeNs> rtt_statistics;
     utils::Statistics<SpeedGbps> delivery_rate_statistics;
+
+    FlowContext(FlowFourTuple four_tuple) : FlowFourTuple(four_tuple) {}
 };
 
 // Transport layer interface for reliable data delivery along single physical

source/connection/flow/new_flows_summary/new_flows_summary.cpp

@@ -0,0 +1,59 @@
+#include "new_flows_summary.hpp"
+
+#include "utils/filesystem.hpp"
+
+namespace sim {
+NewFlowsSummary::NewFlowsSummary(const utils::IdTable<INewFlow>& flows_table) {
+    for (const auto& [id, flow] : flows_table) {
+        m_flows_contexts.emplace(id, flow->get_context());
+    }
+}
+
+void NewFlowsSummary::write_to_csv(std::filesystem::path output_path) const {
+    utils::create_all_directories(output_path);
+    std::ofstream out(output_path);
+    if (!out) {
+        throw std::runtime_error("Failed to create file for summary");
+    }
+    out << "Flow id"
+        << ", Sent size (bytes)"
+        << ", Delivered size (bytes)"
+        << ", Overhead (%)"
+        << ", Retransmit size (bytes)"
+        << ", Effective delivery rate (Gbps)"
+        << ", Delivery rate (Gbps)"
+        << ", FCT (ns)\n";
+    for (const auto& [flow_id, ctx] : m_flows_contexts) {
+        TimeNs fct =
+            (ctx.last_ack_receive_time.has_value() && ctx.start_time.has_value()
+                 ? ctx.last_ack_receive_time.value() - ctx.start_time.value()
+                 : TimeNs(0));
+
+        double overhead = ctx.delivered_size != SizeByte(0)
+                              ? (ctx.sent_size / ctx.delivered_size - 1) * 100
+                              : std::nan("");
+        // Effective delivery rate is average delivery rate calculated on the
+        // spot (when connection is active) so does not include incative time
+        // segments
+        SpeedGbps effective_delivery_rate =
+            ctx.delivery_rate_statistics.get_mean().value_or(SpeedGbps(0.0));
+
+        constexpr static double EPS = 1e-8;
+
+        SpeedGbps delivery_rate =
+            (std::fabs(fct.value()) > EPS ? SpeedGbps(ctx.delivered_size / fct)
+                                          : SpeedGbps(0.0));
+
+        out << flow_id;
+        out << ", " << ctx.sent_size;
+        out << ", " << ctx.delivered_size;
+        out << ", " << overhead;
+        out << ", " << ctx.retransmit_size;
+        out << ", " << effective_delivery_rate;
+        out << ", " << delivery_rate;
+        out << ", " << fct;
+        out << "\n";
+    }
+}
+
+}  // namespace sim

source/connection/flow/new_flows_summary/new_flows_summary.hpp

@@ -0,0 +1,18 @@
+#pragma once
+#include "../i_new_flow.hpp"
+#include "utils/id_table.hpp"
+#include <filesystem>
+
+namespace sim {
+
+class NewFlowsSummary {
+public:
+    explicit NewFlowsSummary(const utils::IdTable<INewFlow>& flows_table);
+
+    void write_to_csv(std::filesystem::path output_path) const;
+
+private:
+    std::unordered_map<Id, FlowContext> m_flows_contexts;
+};
+
+}  // namespace sim

source/connection/flow/tcp/new_tcp_flow.cpp

@@ -9,17 +9,20 @@ namespace sim {
 std::string NewTcpFlow::m_packet_type_label = "type";
 
 std::shared_ptr<NewTcpFlow> NewTcpFlow::create_shared(
-    Id a_id, std::shared_ptr<IHost> a_sender, std::shared_ptr<IHost> a_receiver,
-    bool a_ecn_capable, RTO a_rto, TcpFlowMetricsFilters a_metrics_flags) {
+    Id a_id, FlowFourTuple a_four_tuple, bool a_ecn_capable, RTO a_rto,
+    TcpFlowMetricsFilters a_metrics_flags) {
     return std::shared_ptr<NewTcpFlow>(
-        new NewTcpFlow(std::move(a_id), a_sender, a_receiver, a_ecn_capable,
+        new NewTcpFlow(std::move(a_id), std::move(a_four_tuple), a_ecn_capable,
                        std::move(a_rto), std::move(a_metrics_flags)));
 }
 
 void NewTcpFlow::send(std::vector<PacketInfo> packets_info) {
     if (packets_info.empty()) {
         return;
     }
+    if (!m_context.start_time.has_value()) {
+        m_context.start_time = Scheduler::get_instance().get_current_time();
+    }
     auto sender = m_context.sender;
     auto receiver = m_context.receiver;
     for (auto info : packets_info) {
@@ -52,11 +55,10 @@ MetricsTable NewTcpFlow::get_metrics_table() const {
 void NewTcpFlow::write_inner_metrics(
     [[maybe_unused]] std::filesystem::path output_dir) const {};
 
-NewTcpFlow::NewTcpFlow(Id a_id, std::shared_ptr<IHost> a_sender,
-                       std::shared_ptr<IHost> a_receiver, bool a_ecn_capable,
+NewTcpFlow::NewTcpFlow(Id a_id, FlowFourTuple a_four_tuple, bool a_ecn_capable,
                        RTO a_rto, TcpFlowMetricsFilters a_metrics_flags)
     : m_id(std::move(a_id)),
-      m_context(a_sender, a_receiver),
+      m_context(a_four_tuple),
       m_ecn_capable(a_ecn_capable),
       m_rto(std::move(a_rto)),
       m_metrics({std::make_shared<MetricsStorage>(),
@@ -86,6 +88,8 @@ Packet NewTcpFlow::create_data_packet(PacketInfo info,
 
     packet.data_id = std::move(info.id);
     packet.source_id = sender->get_id();
+    packet.sender_port = m_context.sender_port;
+    packet.receriver_port = m_context.receriver_port;
     packet.dest_id = receiver->get_id();
     packet.size = info.packet_size;
 
@@ -134,7 +138,9 @@ void NewTcpFlow::process_data_packet(const Packet& data,
         m_packet_reordering.value());
     Packet ack = data;
     ack.source_id = m_context.receiver->get_id();
+    ack.sender_port = m_context.receriver_port;
     ack.dest_id = m_context.sender->get_id();
+    ack.receriver_port = m_context.sender_port;
     ack.size = SizeByte(1);
     ack.ttl = M_MAX_TTL;
     ack.flags.set_flag(m_packet_type_label, PacketType::ACK)
@@ -171,6 +177,7 @@ void NewTcpFlow::process_ack(const Packet& ack, SizeByte data_packet_size,
                         m_id, ack.to_string()));
         return;
     }
+    m_context.last_ack_receive_time = now;
 
     m_context.delivered_size += data_packet_size;
 

source/connection/flow/tcp/new_tcp_flow.hpp

@@ -23,9 +23,8 @@ class NewTcpFlow : public INewFlow,
     constexpr static inline TcpFlowMetricsFilters DEFAULT_METRICS_FLAGS = {};
 
     static std::shared_ptr<NewTcpFlow> create_shared(
-        Id a_id, std::shared_ptr<IHost> a_sender,
-        std::shared_ptr<IHost> a_receiver,
-        bool a_ecn_capable = DEFAULT_ECN_CAPABLE, RTO a_rto = DEFAULT_START_RTO,
+        Id a_id, FlowFourTuple a_four_tuple, bool a_ecn_capable = true,
+        RTO a_rto = DEFAULT_START_RTO,
         TcpFlowMetricsFilters a_metrics_flags = DEFAULT_METRICS_FLAGS);
 
     virtual void send(std::vector<PacketInfo> packets_info) final;
@@ -40,9 +39,8 @@ class NewTcpFlow : public INewFlow,
         std::filesystem::path output_dir) const final;
 
 private:
-    NewTcpFlow(Id a_id, std::shared_ptr<IHost> a_sender,
-               std::shared_ptr<IHost> a_receiver, bool a_ecn_capable, RTO a_rto,
-               TcpFlowMetricsFilters a_metrics_flags);
+    NewTcpFlow(Id a_id, FlowFourTuple a_four_tuple, bool a_ecn_capable,
+               RTO a_rto, TcpFlowMetricsFilters a_metrics_flags);
 
     Packet create_data_packet(PacketInfo info, std::shared_ptr<IHost> sender,
                               std::shared_ptr<IHost> receiver);