BUG:TcpFlow

configuration_examples/simulation_examples/basic_simulation.yml

@@ -5,13 +5,11 @@ flows:
     sender_id: "sender1"
     receiver_id: "receiver1"
     packet_size: 1024
-    packet_interval: 100
     number_of_packets: 10000
   flow2:
     sender_id: "sender2"
     receiver_id: "receiver2"
     packet_size: 1024
-    packet_interval: 100
     number_of_packets: 10000
 
 algorithm: basic

configuration_examples/simulation_examples/incast_simulation.yml

@@ -4,31 +4,26 @@ flows:
     sender_id: sender0
     receiver_id: receiver0
     packet_size: 1500
-    packet_interval: 500
     number_of_packets: 100
   flow1:
     sender_id: sender1
     receiver_id: receiver0
     packet_size: 1500
-    packet_interval: 500
     number_of_packets: 100
   flow2:
     sender_id: sender2
     receiver_id: receiver0
     packet_size: 1500
-    packet_interval: 500
     number_of_packets: 100
   flow3:
     sender_id: sender3
     receiver_id: receiver0
     packet_size: 1500
-    packet_interval: 500
     number_of_packets: 100
   flow4:
     sender_id: sender4
     receiver_id: receiver0
     packet_size: 1500
-    packet_interval: 500
     number_of_packets: 100
 algorithm: tcp
 simulation_time: 50000

configuration_examples/simulation_examples/tcp_simulation.yml

@@ -5,13 +5,11 @@ flows:
     sender_id: "sender1"
     receiver_id: "receiver1"
     packet_size: 64
-    packet_interval: 600
     number_of_packets: 100
   flow2:
     sender_id: "sender2"
     receiver_id: "receiver2"
     packet_size: 128
-    packet_interval: 600
     number_of_packets: 100
 
 algorithm: tcp

source/flow/i_flow.hpp

@@ -7,15 +7,13 @@ namespace sim {
 
 class IFlow : public Identifiable {
 public:
+    // Start sending packets
     virtual void start() = 0;
-    // Adds new packet to sending queue
-    // This packet will be send at some time in future (depends on concrete
-    // flow) Used in event Generate
-    virtual Time create_new_data_packet() = 0;
 
     // Update the internal state according to some congestion control algorithm
     // Calls when data available for sending on corresponding device
     virtual void update(Packet packet, DeviceType type) = 0;
+
     virtual std::shared_ptr<IHost> get_sender() const = 0;
     virtual std::shared_ptr<IHost> get_receiver() const = 0;
 };

source/flow/tcp/tahoe/tcp_tahoe_cc.cpp

@@ -2,23 +2,35 @@
 
 #include <spdlog/fmt/fmt.h>
 
+#include "scheduler.hpp"
+
 namespace sim {
 TcpTahoeCC::TcpTahoeCC(Time a_dealay_threshold, double a_sstresh)
     : m_delay_threshold(a_dealay_threshold),
       m_ssthresh(a_sstresh),
-      m_cwnd(1.0) {}
+      m_cwnd(1.0),
+      m_last_congestion_detected(0),
+      m_avg_rtt(0.0) {}
 
 bool TcpTahoeCC::on_ack(Time rtt, bool ecn_flag) {
+    update_avg_rtt(rtt);
+    Time current_time = Scheduler::get_instance().get_current_time();
     if (ecn_flag || rtt >= m_delay_threshold) {
         // trigger_congestion
-        m_ssthresh = m_cwnd / 2;
-        m_cwnd = 1.;
+        if (current_time > m_last_congestion_detected + m_avg_rtt) {
+            // To avoid too frequent cwnd and sstresh decreases
+            m_last_congestion_detected = current_time;
+            m_ssthresh = m_cwnd / 2;
+            m_cwnd = 1.;
+        }
         return true;
     }
     if (m_cwnd < m_ssthresh) {
-        m_cwnd *= 2;
+        // Slow start
+        m_cwnd++;
     } else {
-        m_cwnd += 1.;
+        // Congestion avoidance
+        m_cwnd += 1 / m_cwnd;
     }
     return false;
 }
@@ -32,4 +44,14 @@ std::string TcpTahoeCC::to_string() const {
                        m_delay_threshold, m_cwnd, m_ssthresh);
 }
 
-}  // namespace sim
+void TcpTahoeCC::update_avg_rtt(Time rtt) {
+    if (m_avg_rtt == 0.0) {
+        // If not initialized before
+        m_avg_rtt = rtt;
+    } else {
+        m_avg_rtt = m_avg_rtt * M_RTT_WEIGHT_DECAY_FACTOR +
+                    rtt * (1 - M_RTT_WEIGHT_DECAY_FACTOR);
+    }
+}
+
+}  // namespace sim

source/flow/tcp/tahoe/tcp_tahoe_cc.hpp

@@ -13,9 +13,15 @@ class TcpTahoeCC : public ITcpCC {
     std::string to_string() const final;
 
 private:
+    void update_avg_rtt(Time rtt);
+
+    const static inline double M_RTT_WEIGHT_DECAY_FACTOR = 0.8;
     Time m_delay_threshold;  // delay threshold for update
 
     double m_ssthresh;  // Slow start threshold
     double m_cwnd;      // Congestion window
+    Time m_last_congestion_detected;
+
+    double m_avg_rtt;
 };
 }  // namespace sim

source/flow/tcp/tcp_flow.hpp

@@ -3,7 +3,6 @@
 #include <type_traits>
 
 #include "device/interfaces/i_host.hpp"
-#include "event/generate.hpp"
 #include "flow/i_flow.hpp"
 #include "i_tcp_cc.hpp"
 #include "metrics/metrics_collector.hpp"
@@ -20,14 +19,12 @@ class TcpFlow : public IFlow,
 public:
     TcpFlow(Id a_id, std::shared_ptr<IHost> a_src,
             std::shared_ptr<IHost> a_dest, TTcpCC a_cc, Size a_packet_size,
-            Time a_delay_between_packets, std::uint32_t a_packets_to_send,
-            bool a_ecn_capable = true)
+            std::uint32_t a_packets_to_send, bool a_ecn_capable = true)
         : m_id(std::move(a_id)),
           m_src(a_src),
           m_dest(a_dest),
           m_cc(std::move(a_cc)),
           m_packet_size(a_packet_size),
-          m_delay_between_packets(a_delay_between_packets),
           m_packets_to_send(a_packets_to_send),
           m_ecn_capable(a_ecn_capable),
           m_packets_in_flight(0),
@@ -42,22 +39,7 @@ class TcpFlow : public IFlow,
         initialize_flag_manager();
     }
 
-    void start() final {
-        Time curr_time = Scheduler::get_instance().get_current_time();
-        Scheduler::get_instance().add<Generate>(
-            curr_time, this->shared_from_this(), m_packet_size);
-    }
-
-    Time create_new_data_packet() final {
-        if (m_packets_to_send == 0) {
-            return 0;
-        }
-        if (try_to_put_data_to_device()) {
-            --m_packets_to_send;
-        }
-
-        return m_delay_between_packets;
-    }
+    void start() final { send_packets(); }
 
     void update(Packet packet, DeviceType type) final {
         (void)type;
@@ -84,21 +66,17 @@ class TcpFlow : public IFlow,
 
             double old_cwnd = m_cc.get_cwnd();
 
-            if (m_cc.on_ack(rtt, packet.congestion_experienced)) {
-                // Trigger congestion
-                m_packets_in_flight = 0;
-            } else {
-                if (m_packets_in_flight > 0) {
-                    m_packets_in_flight--;
-                }
+            if (m_packets_in_flight > 0) {
+                m_packets_in_flight--;
+            }
+            if (!m_cc.on_ack(rtt, packet.congestion_experienced)) {
+                // No congestion
                 m_packets_acked++;
             }
 
             double cwnd = m_cc.get_cwnd();
 
             if (old_cwnd != cwnd) {
-                MetricsCollector::get_instance().add_cwnd(
-                    m_id, current_time - 1, old_cwnd);
                 MetricsCollector::get_instance().add_cwnd(m_id, current_time,
                                                           cwnd);
             }
@@ -113,6 +91,7 @@ class TcpFlow : public IFlow,
             m_flag_manager.set_flag(packet, packet_type_label, PacketType::ACK);
             m_dest.lock()->enqueue_packet(ack);
         }
+        send_packets();
     }
 
     std::shared_ptr<IHost> get_sender() const final { return m_src.lock(); }
@@ -130,7 +109,6 @@ class TcpFlow : public IFlow,
         oss << ", CC module: " << m_cc.to_string();
         oss << ", packet size: " << m_packet_size;
         oss << ", to send packets: " << m_packets_to_send;
-        oss << ", delay: " << m_delay_between_packets;
         oss << ", packets_in_flight: " << m_packets_in_flight;
         oss << ", acked packets: " << m_packets_acked;
         oss << "]";
@@ -152,7 +130,7 @@ class TcpFlow : public IFlow,
                 return;
             }
             auto flow = m_flow.lock();
-            flow->send_packet_now(m_packet);
+            flow->send_packet_now(std::move(m_packet));
         }
 
     private:
@@ -174,26 +152,34 @@ class TcpFlow : public IFlow,
     }
 
     void send_packet_now(Packet packet) {
-        m_packets_in_flight++;
+        // TODO: think about this place(should be here or in send_packets)
         m_sent_bytes += packet.size_byte;
         packet.sent_time = Scheduler::get_instance().get_current_time();
-        m_src.lock()->enqueue_packet(packet);
+        m_src.lock()->enqueue_packet(std::move(packet));
     }
 
-    bool try_to_put_data_to_device() {
-        if (m_packets_in_flight < m_cc.get_cwnd()) {
+    // Send (ot plan sending) as many packets as possible
+    void send_packets() {
+        constexpr double EPS = 1e-6;
+
+        Time total_delay = 0;
+        Time pacing_delay = m_cc.get_pacing_delay();
+        Time curr_time = Scheduler::get_instance().get_current_time();
+
+        while (m_packets_to_send > 0 &&
+               m_packets_in_flight < m_cc.get_cwnd() + EPS) {
             Packet packet = generate_packet();
-            Time pacing_delay = m_cc.get_pacing_delay();
+            total_delay += pacing_delay;
             if (pacing_delay == 0) {
-                send_packet_now(packet);
+                send_packet_now(std::move(packet));
             } else {
-                Time curr_time = Scheduler::get_instance().get_current_time();
                 Scheduler::get_instance().add<SendAtTime>(
-                    curr_time + pacing_delay, this->shared_from_this(), packet);
+                    curr_time + total_delay, this->shared_from_this(),
+                    std::move(packet));
             }
-            return true;
+            m_packets_in_flight++;
+            m_packets_to_send--;
         }
-        return false;
     }
 
     static void initialize_flag_manager() {
@@ -217,7 +203,6 @@ class TcpFlow : public IFlow,
     TTcpCC m_cc;
 
     Size m_packet_size;
-    Time m_delay_between_packets;
     std::uint32_t m_packets_to_send;
     bool m_ecn_capable;
 

source/parser/identifiable_parser/flow/parse_tcp_flow.hpp

@@ -9,8 +9,8 @@ template <typename TTcpCC>
 requires std::derived_from<TTcpCC, ITcpCC>
 class Parser<TcpFlow<TTcpCC>> {
 public:
-    static std::shared_ptr<TcpFlow<TTcpCC>> parse_object(const YAML::Node& key_node,
-                                           const YAML::Node& value_node) {
+    static std::shared_ptr<TcpFlow<TTcpCC>> parse_object(
+        const YAML::Node& key_node, const YAML::Node& value_node) {
         TTcpCC cc = parse_tcp_cc(key_node, value_node);
         Id id = key_node.as<Id>();
 
@@ -25,11 +25,9 @@ class Parser<TcpFlow<TTcpCC>> {
         Size packet_size = value_node["packet_size"].as<Size>();
         std::uint32_t number_of_packets =
             value_node["number_of_packets"].as<std::uint32_t>();
-        Time packet_interval = value_node["packet_interval"].as<Time>();
 
-        return std::make_shared<TcpFlow<TTcpCC>>(id, sender_ptr, receiver_ptr, cc,
-                                              packet_size, packet_interval,
-                                              number_of_packets);
+        return std::make_shared<TcpFlow<TTcpCC>>(
+            id, sender_ptr, receiver_ptr, cc, packet_size, number_of_packets);
     }
 
 private: