From 16b8fc0f1ace45980b4df59c0e4c7aa9c998895f Mon Sep 17 00:00:00 2001 From: Alan Jowett Date: Fri, 5 Dec 2025 11:03:44 -0800 Subject: [PATCH 1/5] Add pluggable congestion controllers (null, BBR); client --cc option; docs --- README.md | 20 +++++- src/client/main.cpp | 37 ++++++---- src/common/arg_parser.hpp | 2 + src/common/bbr.hpp | 137 ++++++++++++++++++++++++++++++++++++++ src/common/null_cc.hpp | 15 +++++ src/common/pacer.hpp | 69 +++++++++++++++---- 6 files changed, 254 insertions(+), 26 deletions(-) create mode 100644 src/common/bbr.hpp create mode 100644 src/common/null_cc.hpp diff --git a/README.md b/README.md index 85aec1a..a4bb141 100644 --- a/README.md +++ b/README.md @@ -197,4 +197,22 @@ Choose based on your workload and goals: Recommendation: keep the default overlapped IO for production and high-throughput testing. Use `--sync-reply` for small experiments, micro-benchmarks, or when you explicitly want the simpler -blocking send path for diagnosis. \ No newline at end of file +blocking send path for diagnosis. + +## Congestion Control (client) + +The client supports selectable congestion-control policies via the `--cc` option. + +- `--cc null` (default): No congestion controller — the client will attempt to send at the + exact rate specified by `--rate` (subject to OS and NIC limits). +- `--cc bbr`: A lightweight BBR-style controller that estimates bandwidth and minimum RTT and + adjusts a target pacing rate to achieve high throughput while attempting to avoid excessive RTT + inflation. This is experimental and provided for evaluation. + +Example: + +```bash +echo_client --server 127.0.0.1 --port 5000 --rate 100000 --cc bbr --duration 30 +``` + +The available controllers are also listed in the client's `--help` output. \ No newline at end of file diff --git a/src/client/main.cpp b/src/client/main.cpp index 7e9ffc3..7862b2b 100644 --- a/src/client/main.cpp +++ b/src/client/main.cpp @@ -96,7 +96,7 @@ struct client_worker_context { std::unique_ptr current_pacing_tdigest; // Last send timestamp (ns) used to compute inter-packet interval uint64_t last_send_timestamp_ns{0}; - std::unique_ptr pacer; + std::unique_ptr pacer; }; std::mutex g_worker_contexts_mutex; @@ -399,17 +399,20 @@ void worker_thread_func(client_worker_context* ctx, size_t payload_size) try { ctx->packets_received.fetch_add(1); ctx->bytes_received.fetch_add(bytes_transferred); - if (bytes_transferred >= HEADER_SIZE) { - packet_header* header = reinterpret_cast(io_ctx->buffer.data()); - uint64_t recv_time = get_timestamp_ns(); - uint64_t rtt = recv_time - header->timestamp_ns; - - ctx->total_rtt_ns.fetch_add(rtt); - update_min(ctx->min_rtt_ns, rtt); - update_max(ctx->max_rtt_ns, rtt); - // Rotate approximately once a second based on rate. - post_rtt(ctx->curren_rtt_tdigest, ctx->per_worker_rate, rtt); - ctx->outstanding_sequences.erase(header->sequence_number); + if (bytes_transferred >= HEADER_SIZE) { + packet_header* header = reinterpret_cast(io_ctx->buffer.data()); + uint64_t recv_time = get_timestamp_ns(); + uint64_t rtt = recv_time - header->timestamp_ns; + + ctx->total_rtt_ns.fetch_add(rtt); + update_min(ctx->min_rtt_ns, rtt); + update_max(ctx->max_rtt_ns, rtt); + // Rotate approximately once a second based on rate. + post_rtt(ctx->curren_rtt_tdigest, ctx->per_worker_rate, rtt); + ctx->outstanding_sequences.erase(header->sequence_number); + // Feed acknowledgement into pacer congestion controller so it can + // update bandwidth/RTT estimates. Provide sequence number from header. + if (ctx->pacer) ctx->pacer->on_ack(recv_time, header->sequence_number, rtt); } // Re-post receive on the socket that completed @@ -483,6 +486,7 @@ int main(int argc, char* argv[]) try { parser.add_option("duration", 'd', "10", true, "Test duration in seconds (default: 10)"); parser.add_option("rate", 'r', "10000", true, "Total packet rate limit (packets/sec, 0=unlimited)"); + parser.add_option("cc", 'C', "null", true, "Congestion controller to use: null|bbr (default: null)"); parser.add_option("recvbuf", 'b', "4194304", true, "Socket receive buffer size in bytes (default: 4194304)"); parser.add_option("sockets", 'k', "16", true, "Number of sockets per worker (default: 16)"); @@ -504,6 +508,7 @@ int main(int argc, char* argv[]) try { const std::string rate_str = parser.get("rate"); const std::string recvbuf_str = parser.get("recvbuf"); const std::string sockets_str = parser.get("sockets"); + const std::string cc_choice = parser.get("cc"); const std::string stats_file = parser.get("stats-file"); const std::string verbose_str = parser.get("verbose"); size_t payload_size = 0; @@ -561,6 +566,7 @@ int main(int argc, char* argv[]) try { std::cout << std::format("Duration: {} seconds\n", duration_sec); std::cout << std::format("Rate limit: {} packets/sec total ({} per worker)\n", g_rate_limit, per_worker_display); + std::cout << std::format("Congestion controller: {}\n", cc_choice.empty() ? "null" : cc_choice); // Initialize Winsock initialize_winsock(); @@ -704,7 +710,12 @@ int main(int argc, char* argv[]) try { // Start worker threads for (auto& ctx : workers) { // create per-worker pacer now so it doesn't accumulate tokens before start - ctx->pacer = std::make_unique(static_cast(ctx->per_worker_rate)); + if (cc_choice == "bbr") { + ctx->pacer = std::make_unique>(static_cast(ctx->per_worker_rate)); + } else { + // default: null controller (allow requested rate) + ctx->pacer = std::make_unique>(static_cast(ctx->per_worker_rate)); + } ctx->worker_thread = std::thread(worker_thread_func, ctx.get(), payload_size); } diff --git a/src/common/arg_parser.hpp b/src/common/arg_parser.hpp index e38a3b7..b7507b0 100644 --- a/src/common/arg_parser.hpp +++ b/src/common/arg_parser.hpp @@ -193,6 +193,8 @@ class ArgParser { } std::cout << "\n"; } + // Add custom help note for congestion controllers available + std::cout << "\nAvailable congestion controllers: null, bbr\n"; } /** diff --git a/src/common/bbr.hpp b/src/common/bbr.hpp new file mode 100644 index 0000000..f5eb9de --- /dev/null +++ b/src/common/bbr.hpp @@ -0,0 +1,137 @@ +// bbr.hpp +// Simple header-only BBR-like congestion controller (lightweight, approximate) +// Purpose: provide a pluggable congestion-control policy for the client pacer. +#pragma once + +#include +#include +#include + +// Windows headers sometimes define macros named `min` and `max` which +// conflict with `std::min`/`std::max` usage. Undefine them to avoid +// compilation errors when this header is included after Windows headers. +#ifdef max +#undef max +#endif +#ifdef min +#undef min +#endif + +// Very small, self-contained BBR-like controller suitable for experiments. +// Not a full BBR implementation — provides the basic elements: track +// minimum RTT, estimate bottleneck bandwidth (packets/sec), and expose a +// target pacing rate. The goal is to find a high-throughput rate while +// reacting to RTT increases. +class bbr_congestion_controller { +public: + bbr_congestion_controller() = default; + + void set_initial_rate(double pps) { + if (pps <= 0.0) { + unlimited_ = true; + } else { + unlimited_ = false; + pacing_rate_pps_ = pps; + max_bw_pps_ = pps; + ewma_bw_pps_ = pps; + pacing_gain_ = 1.0; + state_ = state_t::STARTUP; + } + } + + // Called when a packet is sent. We record send timestamps by sequence. + void on_send(uint64_t now_ns, uint64_t seq) { + if (unlimited_) return; + send_times_[seq] = now_ns; + } + + // Called when an ack/echo is received. rtt_ns is the measured RTT for that packet. + void on_ack(uint64_t now_ns, uint64_t seq, uint64_t rtt_ns) { + if (unlimited_) return; + + // Update min RTT (simple minimum over window) + if (rtt_ns > 0) { + if (min_rtt_ns_ == 0 || rtt_ns < min_rtt_ns_) min_rtt_ns_ = rtt_ns; + } + + // Estimate sample bandwidth (packets/sec) using send timestamp if available. + auto it = send_times_.find(seq); + if (it != send_times_.end()) { + uint64_t send_ns = it->second; + if (now_ns > send_ns) { + uint64_t interval_ns = now_ns - send_ns; + double sample_pps = 1e9 / static_cast(interval_ns); + // Update EWMA bandwidth estimate (smooth samples) + if (ewma_bw_pps_ <= 0.0) ewma_bw_pps_ = sample_pps; + const double bw_alpha = 0.85; // favor historical, not too noisy + ewma_bw_pps_ = bw_alpha * ewma_bw_pps_ + (1.0 - bw_alpha) * sample_pps; + + // Keep a separate max estimate for occasional probes + if (ewma_bw_pps_ > max_bw_pps_) max_bw_pps_ = ewma_bw_pps_; + } + } + + // Remove tracking to bound memory + if (it != send_times_.end()) send_times_.erase(it); + + // State machine: STARTUP tries to ramp quickly, PROBE_BW probes conservatively + if (state_ == state_t::STARTUP) { + // If RTT inflates, exit startup into probe + if (min_rtt_ns_ > 0 && rtt_ns > min_rtt_ns_ * rtt_inflation_threshold_) { + state_ = state_t::PROBE_BW; + pacing_gain_ = 1.0; // be conservative after inflation + } else { + // Ramp pacing gain multiplicatively (fast ramp). Limit to a cap. + pacing_gain_ = std::min(max_startup_gain_, pacing_gain_ * startup_gain_mul_); + } + } else { + // PROBE_BW: gentle increases when bandwidth appears to grow, back off on RTT spikes + if (min_rtt_ns_ > 0 && rtt_ns > min_rtt_ns_ * rtt_inflation_threshold_) { + // RTT inflated -> back off quickly + pacing_gain_ = std::max(0.6, pacing_gain_ * 0.7); + } else { + // Gentle additive increase toward probe gain + pacing_gain_ = std::min(max_probe_gain_, pacing_gain_ + probe_gain_add_); + } + } + + // Update pacing rate exposed to pacer using EWMA bandwidth estimate + pacing_rate_pps_ = ewma_bw_pps_ * pacing_gain_; + // Clamp to reasonable bounds + if (pacing_rate_pps_ < 1.0) pacing_rate_pps_ = 1.0; + } + + // Periodic poll (called from pacer.poll). Can be used for decay/timers. + void on_poll(uint64_t now_ns) { + if (unlimited_) return; + // slowly decay max_bw_pps_ to forget very old samples + const double decay = 0.9995; + max_bw_pps_ = std::max(1.0, max_bw_pps_ * decay); + // refresh pacing_rate + pacing_rate_pps_ = max_bw_pps_ * pacing_gain_; + } + + double target_rate_pps() const { + if (unlimited_) return 0.0; + return pacing_rate_pps_; + } + +private: + enum class state_t { STARTUP, PROBE_BW }; + + bool unlimited_{false}; + uint64_t min_rtt_ns_{0}; + double max_bw_pps_{1000.0}; + double ewma_bw_pps_{0.0}; + double pacing_gain_{1.0}; + double pacing_rate_pps_{1000.0}; + std::unordered_map send_times_; + state_t state_{state_t::PROBE_BW}; + + // Tuning parameters + const double rtt_inflation_threshold_{1.6}; + const double startup_gain_mul_{1.5}; + const double max_startup_gain_{8.0}; + const double probe_gain_add_{0.05}; + const double max_probe_gain_{3.0}; +}; diff --git a/src/common/null_cc.hpp b/src/common/null_cc.hpp new file mode 100644 index 0000000..a407ef3 --- /dev/null +++ b/src/common/null_cc.hpp @@ -0,0 +1,15 @@ +// null_cc.hpp +// Null congestion controller: no control, allow pacer to use requested rate. +#pragma once + +#include + +class null_congestion_controller { +public: + null_congestion_controller() = default; + void set_initial_rate(double) {} + void on_send(uint64_t, uint64_t) {} + void on_ack(uint64_t, uint64_t, uint64_t) {} + void on_poll(uint64_t) {} + double target_rate_pps() const { return 0.0; } +}; diff --git a/src/common/pacer.hpp b/src/common/pacer.hpp index cc350bb..e1964a2 100644 --- a/src/common/pacer.hpp +++ b/src/common/pacer.hpp @@ -15,26 +15,50 @@ #include #include #include +#include "null_cc.hpp" +// bbr.hpp remains available to include by callers if desired +#include "bbr.hpp" + +/** + * @brief Abstract base for client send pacer. Provides a stable polymorphic + * interface so different templated pacer implementations can be stored + * behind a single pointer type. + */ +class client_send_pacer_base { +public: + virtual ~client_send_pacer_base() = default; + virtual bool can_send() = 0; + virtual void record_send() = 0; + virtual void poll() = 0; + virtual uint64_t get_next_send_time_ns() const = 0; + virtual void reset_to_now() = 0; + virtual void on_ack(uint64_t now_ns, uint64_t seq, uint64_t rtt_ns) = 0; + virtual double get_target_rate_pps() const = 0; +}; /** * @brief Client send pacer using token bucket algorithm. * @note: Not thread-safe; caller must ensure synchronization if used from multiple threads. */ -class client_send_pacer { +template +class client_send_pacer : public client_send_pacer_base { public: /** * @brief Construct a new client send pacer object. * * @param[in] pps The target packet rate in packets per second (0 = unlimited). */ - explicit client_send_pacer(double pps) - : rate_pps_(pps), - unlimited_(pps == 0.0), - // Allow a small burst window (fraction of a second) to tolerate - // scheduling jitter. Default burst window = 0.005s (5 ms). - capacity_((std::max)(1.0, pps * 0.005)), + explicit client_send_pacer(double pps) + : rate_pps_(pps), + unlimited_(pps == 0.0), + // Allow a small burst window (fraction of a second) to tolerate + // scheduling jitter. Default burst window = 0.005s (5 ms). + capacity_((std::max)(1.0, pps * 0.005)), tokens_(0.0), - last_refill_ns_(now_ns()) {} + last_refill_ns_(now_ns()) { + // Initialize congestion controller with initial rate + cc_.set_initial_rate(pps); + } ~client_send_pacer() = default; @@ -47,8 +71,15 @@ class client_send_pacer { * @return true A packet may be sent immediately. * @return false A packet must be delayed to satisfy the rate. */ - bool can_send() { + bool can_send() override { if (unlimited_) return true; + // Allow congestion controller to influence effective rate + double target = cc_.target_rate_pps(); + if (target > 0.0 && target != rate_pps_) { + rate_pps_ = target; + // recompute capacity when rate changes + capacity_ = (std::max)(1.0, rate_pps_ * 0.005); + } uint64_t now = now_ns(); refill(now); return tokens_ >= 1.0 - 1e-12; @@ -60,21 +91,24 @@ class client_send_pacer { * Decrements the token count; if called when tokens are unavailable, a * bounded negative debt is permitted to represent transient overshoot. */ - void record_send() { + void record_send() override { if (unlimited_) return; uint64_t now = now_ns(); refill(now); // consume one token; do not allow negative debt tokens_ = (std::max)(0.0, tokens_ - 1.0); + // inform congestion controller about the send (no sequence here) + cc_.on_send(now, last_sequence_++); } /** * @brief Perform periodic polling to update internal state. */ - void poll() { + void poll() override { if (unlimited_) return; uint64_t now = now_ns(); refill(now); + cc_.on_poll(now); } /** @@ -85,7 +119,7 @@ class client_send_pacer { * * @return uint64_t wait time in nanoseconds (0 == send now) */ - uint64_t get_next_send_time_ns() const { + uint64_t get_next_send_time_ns() const override { if (unlimited_) return 0; uint64_t now = now_ns(); // compute tokens as if we refilled now (without mutating state) @@ -114,6 +148,13 @@ class client_send_pacer { last_refill_ns_ = now; } + // Let caller pass RTT/ack info to congestion controller + void on_ack(uint64_t now_ns, uint64_t seq, uint64_t rtt_ns) override { + if (unlimited_) return; + cc_.on_ack(now_ns, seq, rtt_ns); + } + double get_target_rate_pps() const override { return cc_.target_rate_pps(); } + private: static uint64_t now_ns() { return static_cast( @@ -135,4 +176,8 @@ class client_send_pacer { double capacity_{1.0}; double tokens_{0.0}; uint64_t last_refill_ns_{0}; + // simple send sequence counter used for CC bookkeeping when no external + // sequence is provided by the caller. + uint64_t last_sequence_{1}; + CongestionController cc_; }; \ No newline at end of file From 9b3f1f161723f56a04be17e1bd5d1f31336dbe0d Mon Sep 17 00:00:00 2001 From: Alan Jowett Date: Fri, 5 Dec 2025 11:05:12 -0800 Subject: [PATCH 2/5] CI: add loopback test using BBR controller --- .github/workflows/reusable-test.yml | 62 +++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/.github/workflows/reusable-test.yml b/.github/workflows/reusable-test.yml index 1d4a186..6d0bcb9 100644 --- a/.github/workflows/reusable-test.yml +++ b/.github/workflows/reusable-test.yml @@ -102,6 +102,68 @@ jobs: exit 1 } + - name: Run loopback test (IOCP mode, BBR) + shell: pwsh + run: | + Write-Host "=== Testing IOCP Mode (BBR) ===" + # Start the server in background on port 5000 (use explicit flags) + $serverPath = Resolve-Path "./downloaded_build/${{ inputs.build_dir }}/${{ inputs.config }}/echo_server.exe" + $clientPath = Resolve-Path "./downloaded_build/${{ inputs.build_dir }}/${{ inputs.config }}/echo_client.exe" + $server = Start-Process -FilePath $serverPath -ArgumentList "--port", "5000", "--cores", "1" -PassThru -NoNewWindow + + # Wait for server to start + Start-Sleep -Seconds 2 + + try { + # Run client test for 60 seconds with 1 worker using BBR controller + $clientOutput = & $clientPath "--server" "127.0.0.1" "--port" "5000" "--payload" "64" "--cores" "1" "--duration" "60" "--cc" "bbr" 2>&1 + $clientExitCode = $LASTEXITCODE + + Write-Host "Client output (BBR):" + Write-Host $clientOutput + + # Parse results to verify packets were sent and received + $output = $clientOutput -join "`n" + + if ($output -match "Packets sent: (\d+)") { + $sent = [int]$matches[1] + Write-Host "Packets sent: $sent" + } + + if ($output -match "Packets received: (\d+)") { + $received = [int]$matches[1] + Write-Host "Packets received: $received" + } + + # Verify we sent and received packets + if ($sent -gt 0 -and $received -gt 0) { + Write-Host "Test PASSED (IOCP BBR): Successfully sent $sent packets and received $received responses" + $testPassed = $true + } else { + Write-Host "Test FAILED (IOCP BBR): No packets exchanged (sent=$sent, received=$received)" + $testPassed = $false + } + + # Check drop rate is acceptable (< 50% for loopback) + if ($sent -gt 0) { + $dropRate = (($sent - $received) / $sent) * 100 + Write-Host "Drop rate: $dropRate%" + if ($dropRate -gt 50) { + Write-Host "Warning: High drop rate detected" + } + } + + } finally { + # Stop the server + if ($server -and !$server.HasExited) { + Stop-Process -Id $server.Id -Force -ErrorAction SilentlyContinue + } + } + + if (-not $testPassed) { + exit 1 + } + - name: Archive build artifacts if: failure() uses: actions/upload-artifact@b4b15b8c7c6ac21ea08fcf65892d2ee8f75cf882 # v4.4.3 From 81714dbf0d00d9a404e0b95cdf39db08aad4f326 Mon Sep 17 00:00:00 2001 From: Alan Jowett Date: Fri, 5 Dec 2025 11:13:32 -0800 Subject: [PATCH 3/5] Add 'reno' congestion controller and wire selection; update docs/help --- README.md | 4 ++ src/client/main.cpp | 2 + src/common/arg_parser.hpp | 2 +- src/common/reno.hpp | 102 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 src/common/reno.hpp diff --git a/README.md b/README.md index a4bb141..cbfec70 100644 --- a/README.md +++ b/README.md @@ -209,6 +209,10 @@ The client supports selectable congestion-control policies via the `--cc` option adjusts a target pacing rate to achieve high throughput while attempting to avoid excessive RTT inflation. This is experimental and provided for evaluation. +- `--cc reno`: A simple Reno-like controller (window-based). It maintains a congestion window + in packets and computes a pacing rate as cwnd / min_rtt. This is an experimental, classic + TCP-style controller provided for comparison. + Example: ```bash diff --git a/src/client/main.cpp b/src/client/main.cpp index 7862b2b..bdc22a5 100644 --- a/src/client/main.cpp +++ b/src/client/main.cpp @@ -712,6 +712,8 @@ int main(int argc, char* argv[]) try { // create per-worker pacer now so it doesn't accumulate tokens before start if (cc_choice == "bbr") { ctx->pacer = std::make_unique>(static_cast(ctx->per_worker_rate)); + } else if (cc_choice == "reno") { + ctx->pacer = std::make_unique>(static_cast(ctx->per_worker_rate)); } else { // default: null controller (allow requested rate) ctx->pacer = std::make_unique>(static_cast(ctx->per_worker_rate)); diff --git a/src/common/arg_parser.hpp b/src/common/arg_parser.hpp index b7507b0..92396f5 100644 --- a/src/common/arg_parser.hpp +++ b/src/common/arg_parser.hpp @@ -194,7 +194,7 @@ class ArgParser { std::cout << "\n"; } // Add custom help note for congestion controllers available - std::cout << "\nAvailable congestion controllers: null, bbr\n"; + std::cout << "\nAvailable congestion controllers: null, bbr, reno\n"; } /** diff --git a/src/common/reno.hpp b/src/common/reno.hpp new file mode 100644 index 0000000..3771b43 --- /dev/null +++ b/src/common/reno.hpp @@ -0,0 +1,102 @@ +// reno.hpp +// Simple Reno-like congestion controller (window-based) for experiments. +#pragma once + +#include +#include +#include + +class reno_congestion_controller { +public: + reno_congestion_controller() = default; + + void set_initial_rate(double pps) { + if (pps <= 0.0) { + unlimited_ = true; + return; + } + unlimited_ = false; + // initialize min RTT unknown; cwnd start small + cwnd_packets_ = 10.0; // start with modest cwnd + ssthresh_ = 1000.0; + // approximate initial window -> target pps when min RTT known + // leave ewma and min_rtt unset until acks arrive + } + + void on_send(uint64_t now_ns, uint64_t seq) { + if (unlimited_) return; + send_times_[seq] = now_ns; + } + + void on_ack(uint64_t now_ns, uint64_t seq, uint64_t rtt_ns) { + if (unlimited_) return; + + // update min RTT observed + if (rtt_ns > 0) { + if (min_rtt_ns_ == 0 || rtt_ns < min_rtt_ns_) min_rtt_ns_ = rtt_ns; + } + + // Remove send tracking + auto it = send_times_.find(seq); + if (it != send_times_.end()) send_times_.erase(it); + + // Count ack and adjust cwnd according to Reno rules + if (cwnd_packets_ < ssthresh_) { + // slow start: increase by 1 packet per ACK (approx) + cwnd_packets_ += 1.0; + } else { + // congestion avoidance: increase by ~1 packet per RTT -> 1/cwnd per ACK + cwnd_packets_ += 1.0 / std::max(1.0, cwnd_packets_); + } + + // clamp cwnd + if (cwnd_packets_ < 1.0) cwnd_packets_ = 1.0; + if (cwnd_packets_ > 1e6) cwnd_packets_ = 1e6; + + update_target_rate(); + } + + void on_poll(uint64_t now_ns) { + if (unlimited_) return; + // detect RTT inflation: if last rtt sample > min_rtt * threshold, treat as congestion + if (last_rtt_ns_ > 0 && min_rtt_ns_ > 0) { + if (last_rtt_ns_ > min_rtt_ns_ * rtt_inflation_threshold_) { + // multiplicative decrease + ssthresh_ = std::max(2.0, cwnd_packets_ / 2.0); + cwnd_packets_ = ssthresh_; + update_target_rate(); + } + } + // no-op otherwise + } + + double target_rate_pps() const { + if (unlimited_) return 0.0; + return pacing_rate_pps_; + } + +private: + void update_target_rate() { + if (min_rtt_ns_ == 0) { + pacing_rate_pps_ = 0.0; + return; + } + double min_rtt_s = static_cast(min_rtt_ns_) / 1e9; + if (min_rtt_s <= 0.0) { + pacing_rate_pps_ = 0.0; + return; + } + pacing_rate_pps_ = cwnd_packets_ / min_rtt_s; + if (pacing_rate_pps_ < 1.0) pacing_rate_pps_ = 1.0; + } + + bool unlimited_{false}; + uint64_t last_rtt_ns_{0}; + uint64_t min_rtt_ns_{0}; + double cwnd_packets_{10.0}; + double ssthresh_{1000.0}; + double pacing_rate_pps_{0.0}; + std::unordered_map send_times_; + + const double rtt_inflation_threshold_{1.6}; +}; From f5361afd5f557bc1704851446176bef1c93efd94 Mon Sep 17 00:00:00 2001 From: Alan Jowett Date: Fri, 5 Dec 2025 11:30:57 -0800 Subject: [PATCH 4/5] Stop reno runaway Signed-off-by: Alan Jowett --- .github/workflows/reusable-test.yml | 62 +++++++++++++++++++++++++++++ src/client/main.cpp | 13 ++++++ src/common/reno.hpp | 20 +++++++--- 3 files changed, 89 insertions(+), 6 deletions(-) diff --git a/.github/workflows/reusable-test.yml b/.github/workflows/reusable-test.yml index 6d0bcb9..e748e7b 100644 --- a/.github/workflows/reusable-test.yml +++ b/.github/workflows/reusable-test.yml @@ -164,6 +164,68 @@ jobs: exit 1 } + - name: Run loopback test (IOCP mode, RENO) + shell: pwsh + run: | + Write-Host "=== Testing IOCP Mode (RENO) ===" + # Start the server in background on port 5000 (use explicit flags) + $serverPath = Resolve-Path "./downloaded_build/${{ inputs.build_dir }}/${{ inputs.config }}/echo_server.exe" + $clientPath = Resolve-Path "./downloaded_build/${{ inputs.build_dir }}/${{ inputs.config }}/echo_client.exe" + $server = Start-Process -FilePath $serverPath -ArgumentList "--port", "5000", "--cores", "1" -PassThru -NoNewWindow + + # Wait for server to start + Start-Sleep -Seconds 2 + + try { + # Run client test for 60 seconds with 1 worker using RENO controller + $clientOutput = & $clientPath "--server" "127.0.0.1" "--port" "5000" "--payload" "64" "--cores" "1" "--duration" "60" "--cc" "reno" 2>&1 + $clientExitCode = $LASTEXITCODE + + Write-Host "Client output (RENO):" + Write-Host $clientOutput + + # Parse results to verify packets were sent and received + $output = $clientOutput -join "`n" + + if ($output -match "Packets sent: (\d+)") { + $sent = [int]$matches[1] + Write-Host "Packets sent: $sent" + } + + if ($output -match "Packets received: (\d+)") { + $received = [int]$matches[1] + Write-Host "Packets received: $received" + } + + # Verify we sent and received packets + if ($sent -gt 0 -and $received -gt 0) { + Write-Host "Test PASSED (IOCP RENO): Successfully sent $sent packets and received $received responses" + $testPassed = $true + } else { + Write-Host "Test FAILED (IOCP RENO): No packets exchanged (sent=$sent, received=$received)" + $testPassed = $false + } + + # Check drop rate is acceptable (< 50% for loopback) + if ($sent -gt 0) { + $dropRate = (($sent - $received) / $sent) * 100 + Write-Host "Drop rate: $dropRate%" + if ($dropRate -gt 50) { + Write-Host "Warning: High drop rate detected" + } + } + + } finally { + # Stop the server + if ($server -and !$server.HasExited) { + Stop-Process -Id $server.Id -Force -ErrorAction SilentlyContinue + } + } + + if (-not $testPassed) { + exit 1 + } + - name: Archive build artifacts if: failure() uses: actions/upload-artifact@b4b15b8c7c6ac21ea08fcf65892d2ee8f75cf882 # v4.4.3 diff --git a/src/client/main.cpp b/src/client/main.cpp index bdc22a5..75a72b6 100644 --- a/src/client/main.cpp +++ b/src/client/main.cpp @@ -33,6 +33,7 @@ #include "common/socket_utils.hpp" #include "common/tdigest.hpp" #include "common/pacer.hpp" +#include "common/reno.hpp" // Global flag for shutdown; set to true to request orderly termination. std::atomic g_shutdown{false}; @@ -40,6 +41,8 @@ std::atomic g_shutdown{false}; std::atomic g_verbose{false}; // Start signal to synchronize when workers begin sending std::atomic g_start_sending{false}; +// Signal to stop sending but allow processing of outstanding replies +std::atomic g_stop_sending{false}; /** * @brief Signal handler that requests shutdown. @@ -257,6 +260,10 @@ void worker_thread_func(client_worker_context* ctx, size_t payload_size) try { while (!g_shutdown.load()) { uint64_t sent_so_far = ctx->packets_sent.load(); + // Stop initiating new sends when ordered to stop; this allows in-flight + // replies to be processed and not counted as dropped. + if (g_stop_sending.load()) break; + while (!available_send_contexts.empty() && ctx->pacer->can_send()) { auto* send_ctx = available_send_contexts.back(); available_send_contexts.pop_back(); @@ -755,6 +762,12 @@ int main(int argc, char* argv[]) try { total_recv, total_sent - total_recv); } + // Signal workers to stop sending new packets, allow in-flight replies to arrive + g_stop_sending.store(true); + // Wait 1 second for in-flight replies to be processed + std::this_thread::sleep_for(std::chrono::seconds(1)); + + // Now request shutdown to terminate worker loops g_shutdown.store(true); std::cout << "\nStopping workers...\n"; diff --git a/src/common/reno.hpp b/src/common/reno.hpp index 3771b43..f55bc9a 100644 --- a/src/common/reno.hpp +++ b/src/common/reno.hpp @@ -19,6 +19,9 @@ class reno_congestion_controller { // initialize min RTT unknown; cwnd start small cwnd_packets_ = 10.0; // start with modest cwnd ssthresh_ = 1000.0; + // store initial request rate and seed pacing rate + initial_rate_pps_ = pps; + pacing_rate_pps_ = pps; // approximate initial window -> target pps when min RTT known // leave ewma and min_rtt unset until acks arrive } @@ -31,9 +34,10 @@ class reno_congestion_controller { void on_ack(uint64_t now_ns, uint64_t seq, uint64_t rtt_ns) { if (unlimited_) return; - // update min RTT observed + // update min RTT observed and remember last RTT if (rtt_ns > 0) { if (min_rtt_ns_ == 0 || rtt_ns < min_rtt_ns_) min_rtt_ns_ = rtt_ns; + last_rtt_ns_ = rtt_ns; } // Remove send tracking @@ -78,15 +82,18 @@ class reno_congestion_controller { private: void update_target_rate() { if (min_rtt_ns_ == 0) { - pacing_rate_pps_ = 0.0; + // keep existing pacing rate if we don't yet have a min RTT return; } double min_rtt_s = static_cast(min_rtt_ns_) / 1e9; - if (min_rtt_s <= 0.0) { - pacing_rate_pps_ = 0.0; - return; - } + if (min_rtt_s <= 0.0) return; pacing_rate_pps_ = cwnd_packets_ / min_rtt_s; + // Clamp to avoid runaway: limit to a multiplier of initial requested rate + const double max_mult = 1.0; // do not exceed initial requested rate + if (initial_rate_pps_ > 0.0) { + double cap = initial_rate_pps_ * max_mult; + if (pacing_rate_pps_ > cap) pacing_rate_pps_ = cap; + } if (pacing_rate_pps_ < 1.0) pacing_rate_pps_ = 1.0; } @@ -97,6 +104,7 @@ class reno_congestion_controller { double ssthresh_{1000.0}; double pacing_rate_pps_{0.0}; std::unordered_map send_times_; + double initial_rate_pps_{0.0}; const double rtt_inflation_threshold_{1.6}; }; From b7e14a20e2625671d8ae4dbb6533690a1acb087f Mon Sep 17 00:00:00 2001 From: Alan Jowett Date: Fri, 5 Dec 2025 12:05:38 -0800 Subject: [PATCH 5/5] Updated congestion controllers Signed-off-by: Alan Jowett --- README.md | 21 +- src/client/main.cpp | 101 +++++---- src/common/arg_parser.hpp | 2 - src/common/bbr.hpp | 311 +++++++++++++++++---------- src/common/congestion_controller.hpp | 26 +++ src/common/null_cc.hpp | 62 +++++- src/common/pacer.hpp | 62 +++--- src/common/reno.hpp | 124 ++++++++++- 8 files changed, 510 insertions(+), 199 deletions(-) create mode 100644 src/common/congestion_controller.hpp diff --git a/README.md b/README.md index cbfec70..e1c3d10 100644 --- a/README.md +++ b/README.md @@ -219,4 +219,23 @@ Example: echo_client --server 127.0.0.1 --port 5000 --rate 100000 --cc bbr --duration 30 ``` -The available controllers are also listed in the client's `--help` output. \ No newline at end of file +The available controllers are also listed in the client's `--help` output. + +## Developer Notes: Documentation and Doxygen + +- The congestion controller implementations live under `src/common` and are + documented with Doxygen-style comments to aid automated API documentation. + Key files: + - `src/common/null_cc.hpp` - no-op controller (placeholder / tests) + - `src/common/bbr.hpp` - lightweight BBR-like experimental controller + - `src/common/reno.hpp` - simple Reno-like windowed controller + +- To generate HTML documentation with Doxygen (if you have Doxygen installed): + +```powershell +doxygen Doxyfile +``` + +- The Doxygen comments are intentionally concise; please open the header + files in `src/common` for per-method and member documentation used by the + generated docs. \ No newline at end of file diff --git a/src/client/main.cpp b/src/client/main.cpp index 75a72b6..d6ecbf5 100644 --- a/src/client/main.cpp +++ b/src/client/main.cpp @@ -30,10 +30,12 @@ #include #include "common/arg_parser.hpp" -#include "common/socket_utils.hpp" -#include "common/tdigest.hpp" +#include "common/bbr.hpp" +#include "common/null_cc.hpp" #include "common/pacer.hpp" #include "common/reno.hpp" +#include "common/socket_utils.hpp" +#include "common/tdigest.hpp" // Global flag for shutdown; set to true to request orderly termination. std::atomic g_shutdown{false}; @@ -112,7 +114,7 @@ std::vector> g_worker_pacing_tdigests; // Each worker will be assigned an equal share (plus remainder distribution). uint64_t g_rate_limit = 10000; // default total -TDigest g_overall_rtt_tdigest(100.0); ///< Global RTT TDigest for percentile estimation +TDigest g_overall_rtt_tdigest(100.0); ///< Global RTT TDigest for percentile estimation TDigest g_overall_pacing_tdigest(100.0); ///< Global pacing TDigest (ms) /** @@ -257,7 +259,7 @@ void worker_thread_func(client_worker_context* ctx, size_t payload_size) try { while (!g_start_sending.load() && !g_shutdown.load()) { std::this_thread::yield(); } - + while (!g_shutdown.load()) { uint64_t sent_so_far = ctx->packets_sent.load(); // Stop initiating new sends when ordered to stop; this allows in-flight @@ -296,7 +298,10 @@ void worker_thread_func(client_worker_context* ctx, size_t payload_size) try { ctx->bytes_sent.fetch_add(total_size); sent_so_far++; - ctx->pacer->record_send(); + // Tell pacer the actual sequence number so congestion controllers + // that index by sequence (e.g., bandwidth estimators) can match + // sends to ACKs. + if (ctx->pacer) ctx->pacer->record_send(header->sequence_number); } // Check for completions (use GetQueuedCompletionStatusEx to batch completions) @@ -312,8 +317,8 @@ void worker_thread_func(client_worker_context* ctx, size_t payload_size) try { // - If wait <= BUSY_SPIN_NS, poll GQCSEx non-blocking and busy-wait (yield) // until the next send time to achieve sub-ms spacing. // thresholds - constexpr uint64_t TIGHT_SPIN_NS = 200'000ULL; // 200 us - constexpr uint64_t SHORT_SLEEP_NS = 2'000'000ULL; // 2 ms + constexpr uint64_t TIGHT_SPIN_NS = 200'000ULL; // 200 us + constexpr uint64_t SHORT_SLEEP_NS = 2'000'000ULL; // 2 ms DWORD timeout = 0; BOOL ex_result = FALSE; @@ -321,23 +326,24 @@ void worker_thread_func(client_worker_context* ctx, size_t payload_size) try { // can send now: non-blocking poll timeout = 0; ex_result = GetQueuedCompletionStatusEx(ctx->iocp.get(), entries.data(), max_entries, - &num_removed, timeout, FALSE); + &num_removed, timeout, FALSE); } else if (wait_ns > SHORT_SLEEP_NS) { // longer waits: block in kernel for up to IOCP_TIMEOUT_MS uint64_t wait_ms = (wait_ns + 999999ULL) / 1000000ULL; - timeout = static_cast((std::min)(static_cast(IOCP_TIMEOUT_MS), wait_ms)); + timeout = + static_cast((std::min)(static_cast(IOCP_TIMEOUT_MS), wait_ms)); ex_result = GetQueuedCompletionStatusEx(ctx->iocp.get(), entries.data(), max_entries, - &num_removed, timeout, FALSE); + &num_removed, timeout, FALSE); } else if (wait_ns > TIGHT_SPIN_NS) { // moderate short wait: do a quick non-blocking poll then Sleep(0) until target timeout = 0; ex_result = GetQueuedCompletionStatusEx(ctx->iocp.get(), entries.data(), max_entries, - &num_removed, timeout, FALSE); + &num_removed, timeout, FALSE); if (num_removed == 0) { - uint64_t now = static_cast( - std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count()); + uint64_t now = + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count()); uint64_t target = now + wait_ns; while (!g_shutdown.load()) { now = static_cast( @@ -352,12 +358,12 @@ void worker_thread_func(client_worker_context* ctx, size_t payload_size) try { // very short wait: non-blocking poll and tight yield-based spin timeout = 0; ex_result = GetQueuedCompletionStatusEx(ctx->iocp.get(), entries.data(), max_entries, - &num_removed, timeout, FALSE); + &num_removed, timeout, FALSE); if (num_removed == 0) { - uint64_t now = static_cast( - std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count()); + uint64_t now = + static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count()); uint64_t target = now + wait_ns; while (!g_shutdown.load()) { now = static_cast( @@ -406,20 +412,20 @@ void worker_thread_func(client_worker_context* ctx, size_t payload_size) try { ctx->packets_received.fetch_add(1); ctx->bytes_received.fetch_add(bytes_transferred); - if (bytes_transferred >= HEADER_SIZE) { - packet_header* header = reinterpret_cast(io_ctx->buffer.data()); - uint64_t recv_time = get_timestamp_ns(); - uint64_t rtt = recv_time - header->timestamp_ns; - - ctx->total_rtt_ns.fetch_add(rtt); - update_min(ctx->min_rtt_ns, rtt); - update_max(ctx->max_rtt_ns, rtt); - // Rotate approximately once a second based on rate. - post_rtt(ctx->curren_rtt_tdigest, ctx->per_worker_rate, rtt); - ctx->outstanding_sequences.erase(header->sequence_number); - // Feed acknowledgement into pacer congestion controller so it can - // update bandwidth/RTT estimates. Provide sequence number from header. - if (ctx->pacer) ctx->pacer->on_ack(recv_time, header->sequence_number, rtt); + if (bytes_transferred >= HEADER_SIZE) { + packet_header* header = reinterpret_cast(io_ctx->buffer.data()); + uint64_t recv_time = get_timestamp_ns(); + uint64_t rtt = recv_time - header->timestamp_ns; + + ctx->total_rtt_ns.fetch_add(rtt); + update_min(ctx->min_rtt_ns, rtt); + update_max(ctx->max_rtt_ns, rtt); + // Rotate approximately once a second based on rate. + post_rtt(ctx->curren_rtt_tdigest, ctx->per_worker_rate, rtt); + ctx->outstanding_sequences.erase(header->sequence_number); + // Feed acknowledgement into pacer congestion controller so it can + // update bandwidth/RTT estimates. Provide sequence number from header. + if (ctx->pacer) ctx->pacer->on_ack(recv_time, header->sequence_number, rtt); } // Re-post receive on the socket that completed @@ -493,7 +499,8 @@ int main(int argc, char* argv[]) try { parser.add_option("duration", 'd', "10", true, "Test duration in seconds (default: 10)"); parser.add_option("rate", 'r', "10000", true, "Total packet rate limit (packets/sec, 0=unlimited)"); - parser.add_option("cc", 'C', "null", true, "Congestion controller to use: null|bbr (default: null)"); + parser.add_option("cc", 'C', "null", true, + "Congestion controller to use: null|bbr|reno (default: null)"); parser.add_option("recvbuf", 'b', "4194304", true, "Socket receive buffer size in bytes (default: 4194304)"); parser.add_option("sockets", 'k', "16", true, "Number of sockets per worker (default: 16)"); @@ -575,6 +582,21 @@ int main(int argc, char* argv[]) try { per_worker_display); std::cout << std::format("Congestion controller: {}\n", cc_choice.empty() ? "null" : cc_choice); + // Validate congestion controller choice + const std::vector valid_cc = {"null", "bbr", "reno"}; + if (!cc_choice.empty()) { + bool ok = false; + if (std::any_of(valid_cc.begin(), valid_cc.end(), + [&](const auto& v) { return v == cc_choice; })) { + ok = true; + } + if (!ok) { + std::cerr << std::format("Unknown congestion controller: {}\n", cc_choice); + std::cerr << "Valid options: null|bbr|reno\n"; + return 1; + } + } + // Initialize Winsock initialize_winsock(); @@ -718,12 +740,15 @@ int main(int argc, char* argv[]) try { for (auto& ctx : workers) { // create per-worker pacer now so it doesn't accumulate tokens before start if (cc_choice == "bbr") { - ctx->pacer = std::make_unique>(static_cast(ctx->per_worker_rate)); + ctx->pacer = std::make_unique>( + static_cast(ctx->per_worker_rate)); } else if (cc_choice == "reno") { - ctx->pacer = std::make_unique>(static_cast(ctx->per_worker_rate)); + ctx->pacer = std::make_unique>( + static_cast(ctx->per_worker_rate)); } else { // default: null controller (allow requested rate) - ctx->pacer = std::make_unique>(static_cast(ctx->per_worker_rate)); + ctx->pacer = + std::make_unique>(static_cast(ctx->per_worker_rate)); } ctx->worker_thread = std::thread(worker_thread_func, ctx.get(), payload_size); } diff --git a/src/common/arg_parser.hpp b/src/common/arg_parser.hpp index 92396f5..dd69f79 100644 --- a/src/common/arg_parser.hpp +++ b/src/common/arg_parser.hpp @@ -182,8 +182,6 @@ class ArgParser { std::cout << " "; std::cout << "Default:"; // pad so that defaults align in a column - size_t after_default = 0; // length of label "Default:" - after_default = 8; std::cout << " "; if (r.default_val.size() < def_col_width) std::cout << std::string(def_col_width - r.default_val.size(), ' '); diff --git a/src/common/bbr.hpp b/src/common/bbr.hpp index f5eb9de..7a8b389 100644 --- a/src/common/bbr.hpp +++ b/src/common/bbr.hpp @@ -1,137 +1,218 @@ -// bbr.hpp -// Simple header-only BBR-like congestion controller (lightweight, approximate) -// Purpose: provide a pluggable congestion-control policy for the client pacer. + +/** + * @file bbr.hpp + * @brief Simple header-only BBR-like congestion controller (lightweight, approximate) + * + * This file defines a basic BBR-like congestion controller that can be plugged + * into the client pacer. It estimates bandwidth and RTT to adjust the sending rate. + * The implementation is simplified for demonstration purposes and may not + * include all features of a full BBR implementation. + * + * @copyright Copyright (c) 2025 WinUDPShardedEcho Contributors + * SPDX-License-Identifier: MIT + */ + #pragma once +#include +#include +#include #include -#include -#include - -// Windows headers sometimes define macros named `min` and `max` which -// conflict with `std::min`/`std::max` usage. Undefine them to avoid -// compilation errors when this header is included after Windows headers. -#ifdef max -#undef max -#endif -#ifdef min -#undef min -#endif - -// Very small, self-contained BBR-like controller suitable for experiments. -// Not a full BBR implementation — provides the basic elements: track -// minimum RTT, estimate bottleneck bandwidth (packets/sec), and expose a -// target pacing rate. The goal is to find a high-throughput rate while -// reacting to RTT increases. +#include + +#include "congestion_controller.hpp" + +/** + * @class bbr_congestion_controller + * @brief Lightweight, header-only BBR-like congestion controller. + * + * This class provides a simplified BBR-inspired rate estimator intended for + * use by the client pacer. It maintains a short history of sent sequence + * numbers and ACK timestamps to estimate delivered bandwidth (packets per + * second) and a smoothed RTT estimate. The controller exposes a target + * sending rate in packets-per-second which callers should respect when + * scheduling sends. + */ class bbr_congestion_controller { -public: + public: + /** + * @brief Default-construct a controller with default parameters. + * + * The controller starts with a conservative target rate (1000 pps by + * default in the atomic `target_rate_`). Call `set_initial_rate` to + * constrain the controller to a target ceiling derived from the caller's + * configured rate. + */ bbr_congestion_controller() = default; + /** + * @brief Set an initial upper-bound rate for the controller. + * + * This method records the externally requested initial rate (in + * packets-per-second) and updates the internal target and bandwidth + * estimates to match that value. The controller will not increase its + * computed target above this initial value. + * + * @param pps Initial packets-per-second rate to use as an upper bound. + */ void set_initial_rate(double pps) { - if (pps <= 0.0) { - unlimited_ = true; - } else { - unlimited_ = false; - pacing_rate_pps_ = pps; - max_bw_pps_ = pps; - ewma_bw_pps_ = pps; - pacing_gain_ = 1.0; - state_ = state_t::STARTUP; - } + initial_rate_pps_ = pps; + target_rate_.store(pps); + bandwidth_pps_ = pps; } - // Called when a packet is sent. We record send timestamps by sequence. + /** + * @brief Notify the controller that a packet was sent. + * + * The controller records the sequence number and send timestamp so that + * delivered rate can later be estimated when ACKs arrive. Implementations + * should call this for each transmitted packet that they expect to be + * ACKed. + * + * @param now_ns Current time in nanoseconds. + * @param seq Packet sequence number associated with the send. + */ void on_send(uint64_t now_ns, uint64_t seq) { - if (unlimited_) return; - send_times_[seq] = now_ns; + sent_history_.emplace_back(seq, now_ns); + while (sent_history_.size() > max_history_) sent_history_.pop_front(); } - // Called when an ack/echo is received. rtt_ns is the measured RTT for that packet. + /** + * @brief Handle an incoming ACK and update bandwidth / RTT estimates. + * + * Updates a smoothed RTT estimate, appends the ACK timestamp into a + * sliding window used to compute delivered packets-per-second, and + * computes a new target sending rate. The target is smoothed and never + * exceeds the value set by `set_initial_rate`. + * + * @param now_ns Current time in nanoseconds (ACK receive time). + * @param seq Sequence number acknowledged by this ACK. + * @param rtt_ns Measured RTT for the acknowledged packet, in + * nanoseconds. + */ void on_ack(uint64_t now_ns, uint64_t seq, uint64_t rtt_ns) { - if (unlimited_) return; - - // Update min RTT (simple minimum over window) - if (rtt_ns > 0) { - if (min_rtt_ns_ == 0 || rtt_ns < min_rtt_ns_) min_rtt_ns_ = rtt_ns; + if (rtt_est_ns_ == 0) + rtt_est_ns_ = rtt_ns; + else + rtt_est_ns_ = static_cast(rtt_est_ns_ * 0.875 + rtt_ns * 0.125); + + // Add this ACK timestamp to sliding window and evict old samples + ack_history_.push_back(now_ns); + const uint64_t window_ns = ack_window_ns_; + while (!ack_history_.empty() && ack_history_.front() + window_ns < now_ns) + ack_history_.pop_front(); + + // Compute sample delivered packets/sec over the window + double sample_pps = 0.0; + if (!ack_history_.empty()) { + double window_s = + static_cast(std::max(1, now_ns - ack_history_.front())) * 1e-9; + if (window_s > 0.0) sample_pps = static_cast(ack_history_.size()) / window_s; } - // Estimate sample bandwidth (packets/sec) using send timestamp if available. - auto it = send_times_.find(seq); - if (it != send_times_.end()) { - uint64_t send_ns = it->second; - if (now_ns > send_ns) { - uint64_t interval_ns = now_ns - send_ns; - double sample_pps = 1e9 / static_cast(interval_ns); - // Update EWMA bandwidth estimate (smooth samples) - if (ewma_bw_pps_ <= 0.0) ewma_bw_pps_ = sample_pps; - const double bw_alpha = 0.85; // favor historical, not too noisy - ewma_bw_pps_ = bw_alpha * ewma_bw_pps_ + (1.0 - bw_alpha) * sample_pps; - - // Keep a separate max estimate for occasional probes - if (ewma_bw_pps_ > max_bw_pps_) max_bw_pps_ = ewma_bw_pps_; - } - } + if (bandwidth_pps_ == 0.0) + bandwidth_pps_ = sample_pps; + else + bandwidth_pps_ = bandwidth_pps_ * 0.9 + sample_pps * 0.1; - // Remove tracking to bound memory - if (it != send_times_.end()) send_times_.erase(it); - - // State machine: STARTUP tries to ramp quickly, PROBE_BW probes conservatively - if (state_ == state_t::STARTUP) { - // If RTT inflates, exit startup into probe - if (min_rtt_ns_ > 0 && rtt_ns > min_rtt_ns_ * rtt_inflation_threshold_) { - state_ = state_t::PROBE_BW; - pacing_gain_ = 1.0; // be conservative after inflation - } else { - // Ramp pacing gain multiplicatively (fast ramp). Limit to a cap. - pacing_gain_ = std::min(max_startup_gain_, pacing_gain_ * startup_gain_mul_); - } - } else { - // PROBE_BW: gentle increases when bandwidth appears to grow, back off on RTT spikes - if (min_rtt_ns_ > 0 && rtt_ns > min_rtt_ns_ * rtt_inflation_threshold_) { - // RTT inflated -> back off quickly - pacing_gain_ = std::max(0.6, pacing_gain_ * 0.7); - } else { - // Gentle additive increase toward probe gain - pacing_gain_ = std::min(max_probe_gain_, pacing_gain_ + probe_gain_add_); - } - } - - // Update pacing rate exposed to pacer using EWMA bandwidth estimate - pacing_rate_pps_ = ewma_bw_pps_ * pacing_gain_; - // Clamp to reasonable bounds - if (pacing_rate_pps_ < 1.0) pacing_rate_pps_ = 1.0; + // Compute new target but never exceed the initially requested rate + double new_target = std::max(bandwidth_pps_ * pacing_gain_, min_pps_); + if (initial_rate_pps_ > 0.0) new_target = std::min(new_target, initial_rate_pps_); + target_rate_.store(new_target); } - // Periodic poll (called from pacer.poll). Can be used for decay/timers. + /** + * @brief Periodic poll hook called by the pacer/driver. + * + * This allows the controller to apply time-based decay to its bandwidth + * estimate when no recent ACKs have been seen. Callers should call this + * with a regularly advancing time value (nanoseconds) from their main + * loop. + * + * @param now_ns Current time in nanoseconds. + */ void on_poll(uint64_t now_ns) { - if (unlimited_) return; - // slowly decay max_bw_pps_ to forget very old samples - const double decay = 0.9995; - max_bw_pps_ = std::max(1.0, max_bw_pps_ * decay); - // refresh pacing_rate - pacing_rate_pps_ = max_bw_pps_ * pacing_gain_; - } - - double target_rate_pps() const { - if (unlimited_) return 0.0; - return pacing_rate_pps_; + if (last_poll_ns_ != 0 && now_ns > last_poll_ns_) { + uint64_t delta_ns = now_ns - last_poll_ns_; + if (delta_ns > decay_interval_ns_ && bandwidth_pps_ > 0.0) { + bandwidth_pps_ *= 0.95; + double new_target = std::max(bandwidth_pps_ * pacing_gain_, min_pps_); + target_rate_.store(new_target); + } + } + last_poll_ns_ = now_ns; } -private: - enum class state_t { STARTUP, PROBE_BW }; - - bool unlimited_{false}; - uint64_t min_rtt_ns_{0}; - double max_bw_pps_{1000.0}; - double ewma_bw_pps_{0.0}; - double pacing_gain_{1.0}; - double pacing_rate_pps_{1000.0}; - std::unordered_map send_times_; - state_t state_{state_t::PROBE_BW}; - - // Tuning parameters - const double rtt_inflation_threshold_{1.6}; - const double startup_gain_mul_{1.5}; - const double max_startup_gain_{8.0}; - const double probe_gain_add_{0.05}; - const double max_probe_gain_{3.0}; + /** + * @brief Get the current target sending rate in packets-per-second. + * + * This returns the controller's smoothed target rate which callers should + * respect when scheduling packet transmissions. + * + * @returns Target packets-per-second. + */ + double target_rate_pps() const { return target_rate_.load(); } + + private: + /** + * @brief Atomic target rate (packets per second) exposed to callers. + */ + std::atomic target_rate_{1000.0}; + + /** + * @brief Smoothed estimated delivered bandwidth in packets-per-second. + */ + double bandwidth_pps_ = 0.0; + + /** + * @brief Smoothed RTT estimate in nanoseconds. + */ + uint64_t rtt_est_ns_ = 0; + + /** + * @brief Timestamp of the last `on_poll` call (nanoseconds). + */ + uint64_t last_poll_ns_ = 0; + + /** + * @brief Multiplicative pacing gain applied to the bandwidth estimate. + */ + const double pacing_gain_ = 1.25; + + /** + * @brief Minimum allowed target sending rate (pps). + */ + const double min_pps_ = 100.0; + + /** + * @brief Interval (ns) after which the bandwidth estimate decays. + */ + const uint64_t decay_interval_ns_ = 200000000; // 200ms + + /** + * @brief Maximum number of recent sends to remember. + */ + const size_t max_history_ = 1024; + + /** + * @brief Recent sent packet records: pair. + */ + std::deque> sent_history_; // pair + + /** + * @brief ACK receive timestamps used to compute delivered rate over a + * sliding window. + */ + const uint64_t ack_window_ns_ = 200000000; // 200ms + std::deque ack_history_; + + /** + * @brief Optional initial rate (pps) supplied by the caller; acts as an + * upper bound for target rate adjustments when > 0. + */ + double initial_rate_pps_ = 0.0; }; + +static_assert(CongestionControllerConcept, + "bbr_congestion_controller does not meet " + "CongestionControllerConcept requirements"); \ No newline at end of file diff --git a/src/common/congestion_controller.hpp b/src/common/congestion_controller.hpp new file mode 100644 index 0000000..9ef028d --- /dev/null +++ b/src/common/congestion_controller.hpp @@ -0,0 +1,26 @@ +/** + * @file congetion_controller.hpp + * @brief Congestion controller concept definition for client pacers. + * + * @copyright Copyright (c) 2025 WinUDPShardedEcho Contributors + * SPDX-License-Identifier: MIT + */ + +#pragma once +#include +#include + +/** + * @brief A congestion controller concept used by client_send_pacer. + * + * @tparam T The congestion controller type to check. + */ +template +concept CongestionControllerConcept = + requires(T cc, double pps, uint64_t now_ns, uint64_t seq, uint64_t rtt_ns) { + { cc.set_initial_rate(pps) } -> std::same_as; + { cc.on_send(now_ns, seq) } -> std::same_as; + { cc.on_ack(now_ns, seq, rtt_ns) } -> std::same_as; + { cc.on_poll(now_ns) } -> std::same_as; + { cc.target_rate_pps() } -> std::same_as; + }; diff --git a/src/common/null_cc.hpp b/src/common/null_cc.hpp index a407ef3..58ffa1d 100644 --- a/src/common/null_cc.hpp +++ b/src/common/null_cc.hpp @@ -1,15 +1,71 @@ -// null_cc.hpp -// Null congestion controller: no control, allow pacer to use requested rate. +/** + * @file null_cc.hpp + * @brief Null congestion controller (no control). + * + * @copyright Copyright (c) 2025 WinUDPShardedEcho Contributors + * SPDX-License-Identifier: MIT + */ #pragma once #include +/** + * @class null_congestion_controller + * @brief A no-op congestion controller used for testing or disabling pacing. + * + * This controller implements the required congestion controller concept but + * performs no rate estimation or state updates. It is useful as a placeholder + * when congestion control is intentionally disabled or for unit tests. + */ class null_congestion_controller { -public: + public: + /** + * @brief Default constructor. + */ null_congestion_controller() = default; + + /** + * @brief No-op: accept an initial rate but ignore it. + * + * @param pps Initial packets-per-second (ignored). + */ void set_initial_rate(double) {} + + /** + * @brief No-op hook for packet send events. + * + * @param now_ns Timestamp of send in nanoseconds (ignored). + * @param seq Sequence number of the sent packet (ignored). + */ void on_send(uint64_t, uint64_t) {} + + /** + * @brief No-op hook for ACK events. + * + * @param now_ns ACK receive time in nanoseconds (ignored). + * @param seq Sequence number acknowledged (ignored). + * @param rtt_ns Measured RTT in nanoseconds (ignored). + */ void on_ack(uint64_t, uint64_t, uint64_t) {} + + /** + * @brief No-op periodic poll hook. + * + * @param now_ns Current time in nanoseconds (ignored). + */ void on_poll(uint64_t) {} + + /** + * @brief Returns the controller's target sending rate (pps). + * + * For the null controller this is always 0.0 which indicates no pacing + * guidance to the caller. + * + * @returns Target packets-per-second (always 0.0). + */ double target_rate_pps() const { return 0.0; } }; + +static_assert(CongestionControllerConcept, + "null_congestion_controller does not meet " + "CongestionControllerConcept requirements"); \ No newline at end of file diff --git a/src/common/pacer.hpp b/src/common/pacer.hpp index e1964a2..4bcd75f 100644 --- a/src/common/pacer.hpp +++ b/src/common/pacer.hpp @@ -12,12 +12,11 @@ */ #pragma once -#include #include #include -#include "null_cc.hpp" -// bbr.hpp remains available to include by callers if desired -#include "bbr.hpp" +#include + +#include "congestion_controller.hpp" /** * @brief Abstract base for client send pacer. Provides a stable polymorphic @@ -25,10 +24,10 @@ * behind a single pointer type. */ class client_send_pacer_base { -public: + public: virtual ~client_send_pacer_base() = default; virtual bool can_send() = 0; - virtual void record_send() = 0; + virtual void record_send(uint64_t seq) = 0; virtual void poll() = 0; virtual uint64_t get_next_send_time_ns() const = 0; virtual void reset_to_now() = 0; @@ -42,25 +41,25 @@ class client_send_pacer_base { */ template class client_send_pacer : public client_send_pacer_base { -public: + public: /** * @brief Construct a new client send pacer object. - * + * * @param[in] pps The target packet rate in packets per second (0 = unlimited). */ - explicit client_send_pacer(double pps) - : rate_pps_(pps), - unlimited_(pps == 0.0), - // Allow a small burst window (fraction of a second) to tolerate - // scheduling jitter. Default burst window = 0.005s (5 ms). - capacity_((std::max)(1.0, pps * 0.005)), - tokens_(0.0), - last_refill_ns_(now_ns()) { - // Initialize congestion controller with initial rate - cc_.set_initial_rate(pps); - } + explicit client_send_pacer(double pps) + : rate_pps_(pps), + unlimited_(pps == 0.0), + // Allow a small burst window (fraction of a second) to tolerate + // scheduling jitter. Default burst window = 0.005s (5 ms). + capacity_((std::max)(1.0, pps * 0.005)), + tokens_(0.0), + last_refill_ns_(now_ns()) { + // Initialize congestion controller with initial rate + cc_.set_initial_rate(pps); + } - ~client_send_pacer() = default; + ~client_send_pacer() override = default; /** * @brief Query whether a packet can be sent now. @@ -91,14 +90,14 @@ class client_send_pacer : public client_send_pacer_base { * Decrements the token count; if called when tokens are unavailable, a * bounded negative debt is permitted to represent transient overshoot. */ - void record_send() override { + void record_send(uint64_t seq) override { if (unlimited_) return; uint64_t now = now_ns(); refill(now); // consume one token; do not allow negative debt tokens_ = (std::max)(0.0, tokens_ - 1.0); - // inform congestion controller about the send (no sequence here) - cc_.on_send(now, last_sequence_++); + // inform congestion controller about the send (caller-provided sequence) + cc_.on_send(now, seq); } /** @@ -129,7 +128,7 @@ class client_send_pacer : public client_send_pacer_base { tokens_at_now = (std::min)(capacity_, tokens_at_now + rate_pps_ * delta_s); } if (tokens_at_now >= 1.0) return 0; - double deficit = 1.0 - tokens_at_now; // tokens needed + double deficit = 1.0 - tokens_at_now; // tokens needed double wait_s = deficit / rate_pps_; uint64_t wait_ns = static_cast(wait_s * 1e9); return wait_ns; @@ -141,7 +140,7 @@ class client_send_pacer : public client_send_pacer_base { * Use this to align pacer state to a synchronized start time so that * no tokens accumulate before measurement begins. */ - void reset_to_now() { + void reset_to_now() override { if (unlimited_) return; uint64_t now = now_ns(); tokens_ = 0.0; @@ -155,12 +154,11 @@ class client_send_pacer : public client_send_pacer_base { } double get_target_rate_pps() const override { return cc_.target_rate_pps(); } -private: + private: static uint64_t now_ns() { - return static_cast( - std::chrono::duration_cast( - std::chrono::steady_clock::now().time_since_epoch()) - .count()); + return static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count()); } void refill(uint64_t now_ns) { @@ -176,8 +174,6 @@ class client_send_pacer : public client_send_pacer_base { double capacity_{1.0}; double tokens_{0.0}; uint64_t last_refill_ns_{0}; - // simple send sequence counter used for CC bookkeeping when no external - // sequence is provided by the caller. - uint64_t last_sequence_{1}; + // (removed unused last_sequence_ member) CongestionController cc_; }; \ No newline at end of file diff --git a/src/common/reno.hpp b/src/common/reno.hpp index f55bc9a..a040e9b 100644 --- a/src/common/reno.hpp +++ b/src/common/reno.hpp @@ -1,15 +1,54 @@ -// reno.hpp -// Simple Reno-like congestion controller (window-based) for experiments. +/** + * reno.hpp + * + * Simple Reno-like congestion controller (window-based) for experiments. + * + * @file reno.hpp + * @brief Reno-like congestion controller + * + * @copyright Copyright (c) 2025 LinuxUDPShardedEcho Contributors + * SPDX-License-Identifier: MIT + * + */ + #pragma once +#include #include #include -#include +#include "congestion_controller.hpp" + +#undef max +#undef min + +/** + * @class reno_congestion_controller + * @brief Simple Reno-like window-based congestion controller for experiments. + * + * This implementation provides a minimal Reno-style controller that tracks + * a congestion window (`cwnd_packets_`) and adapts it on ACKs and RTT + * inflation. It exposes a pacing rate in packets-per-second derived from the + * congestion window and the observed minimum RTT. The controller is + * intentionally lightweight and aimed at experimentation rather than + * production use. + */ class reno_congestion_controller { -public: + public: + /** + * @brief Default constructor. + */ reno_congestion_controller() = default; + /** + * @brief Configure an initial target rate (packets-per-second). + * + * If `pps` is <= 0 the controller is placed into an `unlimited_` mode and + * will not perform window-based control. Otherwise the controller seeds + * its pacing rate and initial window from the provided value. + * + * @param pps Initial requested packets-per-second (<=0 disables control). + */ void set_initial_rate(double pps) { if (pps <= 0.0) { unlimited_ = true; @@ -17,7 +56,7 @@ class reno_congestion_controller { } unlimited_ = false; // initialize min RTT unknown; cwnd start small - cwnd_packets_ = 10.0; // start with modest cwnd + cwnd_packets_ = 10.0; // start with modest cwnd ssthresh_ = 1000.0; // store initial request rate and seed pacing rate initial_rate_pps_ = pps; @@ -26,11 +65,32 @@ class reno_congestion_controller { // leave ewma and min_rtt unset until acks arrive } + /** + * @brief Record a send event for sequence tracking. + * + * The send timestamp is recorded so that late ACKs and RTTs can be + * correlated with send events. Ignored when `unlimited_` is true. + * + * @param now_ns Send time in nanoseconds. + * @param seq Sequence number of the sent packet. + */ void on_send(uint64_t now_ns, uint64_t seq) { if (unlimited_) return; send_times_[seq] = now_ns; } + /** + * @brief Process an ACK: update RTT, cwnd and pacing rate. + * + * Updates the observed minimum RTT, removes send-tracking state for the + * acknowledged packet, and performs Reno slow-start or congestion + * avoidance updates to the congestion window. The pacing rate is then + * recomputed from the window and minimum RTT. + * + * @param now_ns ACK receive time in nanoseconds. + * @param seq Sequence number acknowledged. + * @param rtt_ns RTT sample in nanoseconds for this ACK (0 if unknown). + */ void on_ack(uint64_t now_ns, uint64_t seq, uint64_t rtt_ns) { if (unlimited_) return; @@ -60,6 +120,15 @@ class reno_congestion_controller { update_target_rate(); } + /** + * @brief Periodic maintenance hook. + * + * Detects RTT inflation (a sign of congestion) and applies a + * multiplicative decrease to the congestion window and recomputes the + * pacing rate when necessary. Ignored when `unlimited_` is true. + * + * @param now_ns Current time in nanoseconds. + */ void on_poll(uint64_t now_ns) { if (unlimited_) return; // detect RTT inflation: if last rtt sample > min_rtt * threshold, treat as congestion @@ -74,12 +143,22 @@ class reno_congestion_controller { // no-op otherwise } + /** + * @brief Return the controller's target sending rate (packets-per-second). + * + * When `unlimited_` is true returns 0.0 to indicate no pacing guidance. + */ double target_rate_pps() const { if (unlimited_) return 0.0; return pacing_rate_pps_; } -private: + private: + /** + * @brief Recompute the pacing rate from `cwnd_packets_` and `min_rtt_ns_`. + * + * If `min_rtt_ns_` is unknown the pacing rate is left unchanged. + */ void update_target_rate() { if (min_rtt_ns_ == 0) { // keep existing pacing rate if we don't yet have a min RTT @@ -89,7 +168,7 @@ class reno_congestion_controller { if (min_rtt_s <= 0.0) return; pacing_rate_pps_ = cwnd_packets_ / min_rtt_s; // Clamp to avoid runaway: limit to a multiplier of initial requested rate - const double max_mult = 1.0; // do not exceed initial requested rate + const double max_mult = 1.0; // do not exceed initial requested rate if (initial_rate_pps_ > 0.0) { double cap = initial_rate_pps_ * max_mult; if (pacing_rate_pps_ > cap) pacing_rate_pps_ = cap; @@ -97,14 +176,45 @@ class reno_congestion_controller { if (pacing_rate_pps_ < 1.0) pacing_rate_pps_ = 1.0; } + /** + * @brief Whether the controller is in unlimited (disabled) mode. + */ bool unlimited_{false}; + /** + * @brief Most recent RTT sample (nanoseconds). + */ uint64_t last_rtt_ns_{0}; + /** + * @brief Minimum observed RTT (nanoseconds), used to compute pacing. + */ uint64_t min_rtt_ns_{0}; + /** + * @brief Congestion window in packets. + */ double cwnd_packets_{10.0}; + /** + * @brief Slow-start threshold (packets). + */ double ssthresh_{1000.0}; + /** + * @brief Computed pacing rate in packets-per-second. + */ double pacing_rate_pps_{0.0}; + /** + * @brief Map of outstanding send times by sequence number. + */ std::unordered_map send_times_; + /** + * @brief Initial requested rate (pps) provided via `set_initial_rate`. + */ double initial_rate_pps_{0.0}; + /** + * @brief Threshold multiplier for RTT inflation detection. + */ const double rtt_inflation_threshold_{1.6}; }; + +static_assert(CongestionControllerConcept, + "reno_congestion_controller does not meet " + "CongestionControllerConcept requirements"); \ No newline at end of file