diff --git a/source/network/connection/mplb/rdma/dcqcn.cpp b/source/network/connection/mplb/rdma/dcqcn.cpp new file mode 100644 index 0000000000..56c24818ba --- /dev/null +++ b/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(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(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(two_pow)) * m_alpha; + } +} + +void DCQCN::on_rate_increase_timer(std::optional 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 \ No newline at end of file diff --git a/source/network/connection/mplb/rdma/dcqcn.hpp b/source/network/connection/mplb/rdma/dcqcn.hpp new file mode 100644 index 0000000000..dd5cf62f13 --- /dev/null +++ b/source/network/connection/mplb/rdma/dcqcn.hpp @@ -0,0 +1,111 @@ +#pragma once +#include + +#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 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 m_last_cnp = std::nullopt; + + int m_alpha; + bool m_stop_request = false; +}; + +} // namespace sim \ No newline at end of file diff --git a/source/scheduler/scheduler.cpp b/source/scheduler/scheduler.cpp index c989f54200..19cd7f5b96 100644 --- a/source/scheduler/scheduler.cpp +++ b/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, diff --git a/source/scheduler/scheduler.hpp b/source/scheduler/scheduler.hpp index 386770fdd7..95b7b268e4 100644 --- a/source/scheduler/scheduler.hpp +++ b/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: diff --git a/source/types.hpp b/source/types.hpp index 7e68cee318..4459da3d9c 100644 --- a/source/types.hpp +++ b/source/types.hpp @@ -8,8 +8,13 @@ #include "units/units.hpp" using TimeNs = Time; +using TimeUs = Time; + using SizeByte = Size; + using SpeedGbps = Speed; +using SpeedMbps = Speed; + using Id = std::string; using OnDeliveryCallback = std::function; diff --git a/source/units/speed.hpp b/source/units/speed.hpp index 7d6a4490d6..41fe07d47f 100644 --- a/source/units/speed.hpp +++ b/source/units/speed.hpp @@ -43,6 +43,10 @@ class Speed { return Speed(m_value_bit_per_ns * mult); } + constexpr ThisSpeed operator/(double mult) const { + return Speed(m_value_bit_per_ns / mult); + } + constexpr double operator/(ThisSpeed speed) const { return m_value_bit_per_ns / speed.value_bit_per_ns(); } diff --git a/test/connection/rdma/dcqcn_test.cpp b/test/connection/rdma/dcqcn_test.cpp new file mode 100644 index 0000000000..12b6421f72 --- /dev/null +++ b/test/connection/rdma/dcqcn_test.cpp @@ -0,0 +1,111 @@ +#include "network/connection/mplb/rdma/dcqcn.hpp" + +#include + +#include "scheduler/scheduler.hpp" +#include "utils/defer.hpp" + +namespace sim { +namespace test2 { + +struct Fixture : public ::testing::Test { + Fixture() : params() { + // задать удобные параметры для детерминированного тестирования + params.rpg_time_reset = TimeUs(100); + params.rpg_byte_reset = SizeByte(1000ul); + params.rpg_threshold = 1; + params.rpg_ai_rate = SpeedMbps(10); + params.rpg_hai_rate = SpeedMbps(100); + params.dce_tcp_rtt = TimeUs(10); + params.rate_reduce_monitor_period = TimeUs(20); + params.initial_alpha_value = 1023; + params.dce_tcp_g = 1019; + params.rpg_min_rate = SpeedMbps(100); + params.rpg_min_dec_fac = 0.5; + } + + void TearDown() override { Scheduler::get_instance().clear(); }; + void SetUp() override { Scheduler::get_instance().clear(); } + + ParamsDQCCN params; +}; + +// Helper: устанавливаем начальный момент времени +TimeNs now_ns(int us) { return TimeNs(us * 1000); } + +TEST_F(Fixture, AdditiveIncreaseByTimer) { + DCQCN cc(params); + cc.start(); + utils::Defer defer([&cc]() { cc.stop(); }); + + SpeedMbps before = cc.get_rate(); + + // one additive increase should be triggered + Scheduler::get_instance().tick_to(params.rpg_time_reset); + + SpeedMbps new_rate = cc.get_rate(); + EXPECT_GT(new_rate, before); +} + +TEST_F(Fixture, AdditiveIncreaseByBytes) { + DCQCN cc(params); + cc.start(); + utils::Defer defer([&cc]() { cc.stop(); }); + + SpeedMbps before = cc.get_rate(); + + cc.on_data_delivery(params.rpg_byte_reset); + SpeedMbps after = cc.get_rate(); + EXPECT_GT(after, before); +} + +TEST_F(Fixture, AdditiveAndHyperIncrease) { + DCQCN cc(params); + cc.start(); + utils::Defer defer([&cc]() { cc.stop(); }); + + SpeedMbps rate_start = cc.get_rate(); + + // one additive increase + cc.on_data_delivery(params.rpg_byte_reset); + + SpeedGbps rate_ai = cc.get_rate(); + EXPECT_GT(rate_ai, rate_start); + + // two more additive increase + Scheduler::get_instance().tick_to(params.rpg_time_reset * 2); + + SpeedMbps rate_before_hai = cc.get_rate(); + + // one hyper increase + cc.on_data_delivery(params.rpg_byte_reset); + + SpeedMbps rate_hai = cc.get_rate(); + + EXPECT_GT(rate_hai, rate_before_hai); + + static constexpr double thresh = 2.0; + + EXPECT_GT(rate_hai - rate_before_hai, thresh * (rate_ai - rate_start)); +} + +TEST_F(Fixture, RateDecreaseOnCnp) { + DCQCN cc(params); + cc.start(); + utils::Defer defer([&cc]() { cc.stop(); }); + + // increase speed + cc.on_data_delivery(params.rpg_byte_reset); + cc.on_data_delivery(params.rpg_byte_reset); + SpeedMbps up = cc.get_rate(); + + cc.on_cnp(); + + Scheduler::get_instance().tick_to(params.rate_reduce_monitor_period); + SpeedMbps down = cc.get_rate(); + + EXPECT_LT(down, up); +} + +} // namespace test2 +} // namespace sim \ No newline at end of file