FEAT: implement the DCQCN

source/network/connection/mplb/rdma/dcqcn.cpp

@@ -0,0 +1,148 @@
+#include "dcqcn.hpp"
+
+#include "scheduler/scheduler.hpp"
+
+namespace sim {
+DCQCN::DCQCN(const ParamsDQCCN& a_params)
+    : m_params(a_params),
+      m_current_rate(a_params.rpg_min_rate),
+      m_target_rate(m_current_rate),
+      m_alpha(m_params.initial_alpha_value) {}
+
+void DCQCN::start() {
+    Scheduler& sched = Scheduler::get_instance();
+    TimeNs now = sched.get_current_time();
+
+    // enqueue first alpha update event
+    sched.add(now + m_params.dce_tcp_rtt, [this]() { on_alpha_timer(); });
+
+    // enqueue first rate increase timer event
+    sched.add(now + m_params.rpg_time_reset, [this, last_cnp = m_last_cnp]() {
+        on_rate_increase_timer(last_cnp);
+    });
+
+    sched.add(now + m_params.rate_reduce_monitor_period,
+              [this]() { on_rate_reduce_monitor_period(); });
+}
+
+void DCQCN::stop() { m_stop_request = true; }
+
+void DCQCN::on_cnp() {
+    if (m_stop_request) {
+        return;
+    }
+
+    TimeNs now = Scheduler::get_instance().get_current_time();
+    m_last_cnp = now;
+    m_bytes_from_last_byte_reset = SizeByte(0ul);
+    m_time_counter = 0;
+}
+
+void DCQCN::on_data_delivery(SizeByte size) {
+    if (m_stop_request) {
+        return;
+    }
+
+    m_bytes_from_last_byte_reset += size;
+    while (m_bytes_from_last_byte_reset >= m_params.rpg_byte_reset) {
+        m_bytes_from_last_byte_reset -= m_params.rpg_byte_reset;
+        m_byte_counter++;
+        on_rate_increase_event();
+    }
+}
+
+SpeedMbps DCQCN::get_rate() const { return m_current_rate; }
+
+void DCQCN::on_rate_reduce_monitor_period() {
+    if (m_stop_request) {
+        return;
+    }
+
+    Scheduler& sched = Scheduler::get_instance();
+    TimeNs now = sched.get_current_time();
+    if (m_last_cnp &&
+        now <= m_last_cnp.value() + m_params.rate_reduce_monitor_period) {
+        // found CNP over last m_params.rate_reduce_monitor_period => reset
+        // timers
+
+        if (m_params.clamp_tgt_rate && !m_dec_target_rate) {
+            m_target_rate = m_current_rate;
+        }
+        m_dec_target_rate = true;
+
+        // decrement current rate
+        m_current_rate =
+            m_current_rate *
+            std::max(m_params.rpg_min_dec_fac,
+                     (1 - m_alpha / static_cast<double>(1 << m_params.rpg_gd)));
+        m_current_rate = std::max(m_current_rate, m_params.rpg_min_rate);
+        m_bytes_from_last_byte_reset = SizeByte(0ul);
+        m_byte_counter = 0;
+        m_time_counter = 0;
+    }
+
+    sched.add(now + m_params.rpg_time_reset, [this, last_cnp = m_last_cnp]() {
+        on_rate_increase_timer(last_cnp);
+    });
+}
+
+void DCQCN::on_alpha_timer() {
+    if (m_stop_request) {
+        return;
+    }
+
+    Scheduler& sched = Scheduler::get_instance();
+    TimeNs now = sched.get_current_time();
+    static constexpr int two_pow = (1 << 10);
+    if (m_last_cnp && now <= m_last_cnp.value() + m_params.dce_tcp_rtt) {
+        // cnp detected over last m_params.dce_tcp_rtt => increment alpha
+        m_alpha =
+            (m_params.dce_tcp_g / static_cast<double>(two_pow)) * m_alpha +
+            two_pow - m_params.dce_tcp_g;
+    } else {
+        // no cnp over last last m_params.dce_tcp_rtt => decrement alpha
+        m_alpha = (m_params.dce_tcp_g / static_cast<double>(two_pow)) * m_alpha;
+    }
+}
+
+void DCQCN::on_rate_increase_timer(std::optional<TimeNs> last_elapced_cnp) {
+    if (m_stop_request) {
+        return;
+    }
+
+    if (m_last_cnp != last_elapced_cnp) {
+        // there was cnp over m_params.rpg_time_reset => event should be
+        // cancelled
+        return;
+    }
+    // no cnp over last m_params.rpg_time_reset => update time counter &
+    // reschedule event
+    m_time_counter++;
+    on_rate_increase_event();
+    Scheduler& sched = Scheduler::get_instance();
+    TimeNs now = sched.get_current_time();
+    sched.add(now + m_params.rpg_time_reset,
+              [this, new_last_elapced_cnp = m_last_cnp]() {
+                  on_rate_increase_timer(new_last_elapced_cnp);
+              });
+}
+
+void DCQCN::on_rate_increase_event() {
+    if (m_stop_request) {
+        return;
+    }
+
+    if (std::max(m_time_counter, m_byte_counter) < m_params.rpg_threshold) {
+        // fast recovery; no target rate update
+    } else if (std::min(m_time_counter, m_byte_counter) <=
+               m_params.rpg_threshold) {
+        // additive increase
+        m_target_rate += m_params.rpg_ai_rate;
+    } else {
+        // hyper increase
+        m_target_rate += m_params.rpg_hai_rate;
+    }
+    m_current_rate = (m_current_rate + m_target_rate) / 2.0;
+}
+
+}  // namespace sim

source/network/connection/mplb/rdma/dcqcn.hpp

@@ -0,0 +1,111 @@
+#pragma once
+#include <optional>
+
+#include "types.hpp"
+
+namespace sim {
+
+// DQCCN congestion control realization
+// Based on NVIDIA documentation:
+// https://enterprise-support.nvidia.com/s/article/DCQCN-CC-algorithm
+// https://enterprise-support.nvidia.com/s/article/dcqcn-parameters
+
+struct ParamsDQCCN {
+    // -------------Rate increment-------------
+
+    // The time period between rate increase events.
+    TimeUs rpg_time_reset = TimeUs(300);
+
+    // 	The sent bytes counter between rate increase events.
+    SizeByte rpg_byte_reset = SizeByte(64 * 32767ul);
+
+    // The threshold of rate increase events for moving to next rate increase
+    // phase.
+    std::size_t rpg_threshold = 1;
+
+    // The rate increase value in the Additive Increase phase.
+    SpeedMbps rpg_ai_rate = SpeedMbps(5);
+
+    // The rate increase value in the Hyper Increase phase.
+    SpeedMbps rpg_hai_rate = SpeedMbps(50);
+
+    // -------------Alpha update-------------
+
+    // This parameter sets the initial value of alpha that should be used when
+    // receiving the first CNP for a flow.
+    int initial_alpha_value = 1023;
+
+    // Controls aggressiveness of alpha's updates
+    // The lower G is, the more aggressive are the changes.
+    int dce_tcp_g = 1019;
+
+    // The Time period between alpha updates.
+    TimeUs dce_tcp_rtt = TimeUs(1);
+
+    // -------------Rate decrement-------------
+
+    // The time period between rate reductions.
+    TimeUs rate_reduce_monitor_period = TimeUs(4);
+
+    // Rates (current, target) on first CNP (0 – 85% of line rate).
+    SpeedMbps rate_to_set_on_first_cnp = SpeedMbps(0);
+
+    // If true, every rate decreases. The target rate is updated to the current
+    // rate.
+    // Otherwise, the target rate is updated to the current rate only on the
+    // first decrement after the increment event.
+    bool clamp_tgt_rate = false;
+
+    // The coefficient between alpha and the rate reduction factor.
+    // Log2 of value in fixed point with 10 in the fraction part
+    int rpg_gd = 11;
+
+    // Minimal rate limit of the QP.
+    SpeedMbps rpg_min_rate = SpeedMbps(1);
+    // Maximal rate limit of the QP.
+    double rpg_min_dec_fac = 0.5;
+};
+
+class DCQCN {
+public:
+    explicit DCQCN(const ParamsDQCCN& a_params);
+
+    // enqueue initial events
+    void start();
+
+    // stop creating new events
+    void stop();
+
+    // Calls when sender got congestion notification
+    void on_cnp();
+
+    // Calls when sender got asknowledge of receiving data_size data
+    void on_data_delivery(SizeByte data_size);
+
+    SpeedMbps get_rate() const;
+
+private:
+    void on_rate_reduce_monitor_period();
+    void on_alpha_timer();
+
+    void on_rate_increase_timer(std::optional<TimeNs> last_elapced_cnp);
+    void on_rate_increase_event();
+
+    ParamsDQCCN m_params;
+    SpeedMbps m_current_rate;
+    SpeedMbps m_target_rate;
+    bool m_dec_target_rate = false;
+
+    // Size & time counters (T & BC on Increment scheme part of NVIDIA docs)
+    SizeByte m_bytes_from_last_byte_reset = SizeByte(0ul);
+    std::uint32_t m_time_counter = 0;
+    std::uint32_t m_byte_counter = 0;
+
+    // Last time CNP was got
+    std::optional<TimeNs> m_last_cnp = std::nullopt;
+
+    int m_alpha;
+    bool m_stop_request = false;
+};
+
+}  // namespace sim

source/scheduler/scheduler.cpp

@@ -35,6 +35,23 @@ bool Scheduler::tick() {
     return true;
 }
 
+uint32_t Scheduler::tick_to(TimeNs time_point) {
+    time_point += TimeNs(1);
+    if (time_point <= m_current_event_local_time) {
+        return 0;
+    }
+    bool arrived = false;
+    uint32_t events_count = 0;
+    add(time_point, [&arrived]() { arrived = true; });
+    while (!arrived && tick()) {
+        events_count++;
+    }
+    if (arrived) {
+        events_count--;
+    }
+    return events_count;
+}
+
 void Scheduler::clear() {
     m_near_events.clear();
     std::priority_queue<NewEvent, std::vector<NewEvent>,

source/scheduler/scheduler.hpp

@@ -36,6 +36,7 @@ class Scheduler {
 
     void clear();  // Clear all events
     bool tick();
+    uint32_t tick_to(TimeNs time_point);
     TimeNs get_current_time();
 
 private:

source/types.hpp

@@ -8,8 +8,13 @@
 #include "units/units.hpp"
 
 using TimeNs = Time<Nanosecond>;
+using TimeUs = Time<Microsecond>;
+
 using SizeByte = Size<Byte>;
+
 using SpeedGbps = Speed<GBit, Second>;
+using SpeedMbps = Speed<MBit, Second>;
+
 using Id = std::string;
 using OnDeliveryCallback = std::function<void()>;
 

source/units/speed.hpp

@@ -43,6 +43,10 @@ class Speed {
         return Speed<Bit, Nanosecond>(m_value_bit_per_ns * mult);
     }
 
+    constexpr ThisSpeed operator/(double mult) const {
+        return Speed<Bit, Nanosecond>(m_value_bit_per_ns / mult);
+    }
+
     constexpr double operator/(ThisSpeed speed) const {
         return m_value_bit_per_ns / speed.value_bit_per_ns();
     }