Skip to content

Commit 8c6105a

Browse files
haproxyFredhaproxyFred
committed
WIP: implement delivery rate sampling algorithm
1 parent 4d7d3b5 commit 8c6105a

File tree

6 files changed

+215
-1
lines changed

6 files changed

+215
-1
lines changed

‎Makefile‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -652,7 +652,7 @@ OPTIONS_OBJS += src/quic_rx.o src/mux_quic.o src/h3.o src/quic_tx.o \
652652
src/cfgparse-quic.o src/qmux_trace.o src/qpack-enc.o \
653653
src/qpack-tbl.o src/h3_stats.o src/quic_stats.o \
654654
src/quic_fctl.o src/cbuf.o src/quic_rules.o \
655-
src/quic_token.o src/quic_pacing.o
655+
src/quic_token.o src/quic_pacing.o src/quic_cc_drs.o
656656
endif
657657
658658
ifneq ($(USE_QUIC_OPENSSL_COMPAT:0=),)

‎include/haproxy/quic_cc-t.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,7 @@ struct quic_cc_path {
115115
uint64_t in_flight;
116116
/* Number of in flight ack-eliciting packets. */
117117
uint64_t ifae_pkts;
118+
uint64_t delivery_rate; /* bytes per second */
118119
};
119120

120121
struct quic_cc_algo {

‎include/haproxy/quic_cc.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,7 @@ static inline void quic_cc_path_init(struct quic_cc_path *path, int ipv4, unsign
9494
path->in_flight = 0;
9595
path->ifae_pkts = 0;
9696
quic_cc_init(&path->cc, algo, qc);
97+
path->delivery_rate = 0;
9798
}
9899

99100
/* Return the remaining <room> available on <path> QUIC path for prepared data

‎include/haproxy/quic_cc_drs.h‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
#include <inttypes.h>
2+
3+
#include <haproxy/window_filter.h>
4+
5+
/* Rate sample */
6+
struct quic_cc_rs {
7+
uint64_t delivered;
8+
uint64_t prior_delivered;
9+
uint64_t tx_in_flight;
10+
uint64_t lost;
11+
uint64_t prior_lost;
12+
int64_t last_end_seq;
13+
uint32_t interval;
14+
uint32_t prior_time;
15+
uint32_t send_elapsed;
16+
uint32_t ack_elapsed;
17+
uint32_t is_app_limited;
18+
};
19+
20+
/* Delivery rate sampling */
21+
struct quic_cc_drs {
22+
struct quic_cc_rs rs;
23+
struct wf wf;
24+
uint64_t round_count;
25+
uint64_t next_round_delivered;
26+
uint64_t delivered;
27+
uint64_t lost;
28+
int64_t last_seq;
29+
uint32_t delivered_time;
30+
uint32_t first_sent_time;
31+
int is_cwnd_limited; /* boolean */
32+
int app_limited; /* boolean */
33+
};
34+
35+
extern struct pool_head *pool_head_quic_cc_drs;
36+
37+
void quic_cc_drs_init(struct quic_cc_drs *drs);
38+
void quic_cc_drs_on_pkt_sent(struct quic_cc_path *path,
39+
struct quic_tx_packet *pkt, struct quic_cc_drs *drs);
40+
void quic_cc_drs_update_rate_sample(struct quic_cc_drs *drs,
41+
struct quic_tx_packet *pkt);
42+
void quic_cc_drs_on_ack_recv(struct quic_cc_drs *drs, struct quic_cc_path *path,
43+
uint64_t pkt_delivered);

‎include/haproxy/quic_tx-t.h‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,16 @@ struct quic_tx_packet {
5353
struct quic_tx_packet *prev;
5454
/* Largest acknowledged packet number if this packet contains an ACK frame */
5555
int64_t largest_acked_pn;
56+
/* Delivery rate sampling information */
57+
struct {
58+
uint64_t delivered;
59+
uint64_t tx_in_flight;
60+
uint64_t lost;
61+
int64_t end_seq;
62+
uint32_t delivered_time;
63+
uint32_t first_sent_time;
64+
int is_app_limited;
65+
} rs;
5666
unsigned char type;
5767
};
5868

‎src/quic_cc_drs.c‎

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
/* Delivery Rate Sampling */
2+
3+
#include <haproxy/pool.h>
4+
#include <haproxy/quic_cc-t.h>
5+
#include <haproxy/quic_cc_drs.h>
6+
#include <haproxy/quic_tx-t.h>
7+
#include <haproxy/ticks.h>
8+
#include <haproxy/window_filter.h>
9+
10+
DECLARE_POOL(pool_head_quic_cc_drs, "quic_cc_drs", sizeof(struct quic_cc_drs));
11+
12+
static void quic_cc_rs_init(struct quic_cc_rs *rs)
13+
{
14+
rs->interval = UINT32_MAX;
15+
rs->delivered = 0;
16+
rs->prior_delivered = 0;
17+
rs->prior_time = TICK_ETERNITY;
18+
rs->tx_in_flight = 0;
19+
rs->lost = 0;
20+
rs->prior_lost = 0;
21+
rs->send_elapsed = 0;
22+
rs->ack_elapsed = 0;
23+
rs->last_end_seq = -1;
24+
rs->is_app_limited = 0;
25+
}
26+
27+
void quic_cc_drs_init(struct quic_cc_drs *drs)
28+
{
29+
quic_cc_rs_init(&drs->rs);
30+
wf_init(&drs->wf, 12, 0, ~0U);
31+
drs->round_count = 0;
32+
drs->next_round_delivered = 0;
33+
drs->delivered = 0;
34+
drs->lost = 0;
35+
drs->last_seq = -1;
36+
drs->delivered_time = TICK_ETERNITY;
37+
drs->first_sent_time = TICK_ETERNITY;
38+
drs->app_limited = 0;
39+
drs->is_cwnd_limited = 0;
40+
}
41+
42+
/* Update <pkt> TX packet rate sampling information.
43+
* Must be called after <pkt> has just been sent.
44+
*/
45+
void quic_cc_drs_on_pkt_sent(struct quic_cc_path *path,
46+
struct quic_tx_packet *pkt, struct quic_cc_drs *drs)
47+
{
48+
if (!path->in_flight)
49+
drs->first_sent_time = drs->delivered_time = pkt->time_sent;
50+
51+
pkt->rs.first_sent_time = drs->first_sent_time;
52+
pkt->rs.delivered_time = drs->delivered_time;
53+
pkt->rs.delivered = drs->delivered;
54+
pkt->rs.is_app_limited = drs->app_limited != 0;
55+
56+
pkt->rs.tx_in_flight = path->in_flight + pkt->len;
57+
pkt->rs.lost = drs->lost;
58+
pkt->rs.end_seq = ++drs->last_seq;
59+
}
60+
61+
/* Return 1 if <pkt> TX packet is the most recently sent packet
62+
* that has been delivered, 0 if not.
63+
*/
64+
static inline int quic_cc_drs_is_newest_packet(struct quic_cc_drs *drs,
65+
struct quic_tx_packet *pkt)
66+
{
67+
return tick_is_lt(drs->first_sent_time, pkt->time_sent) ||
68+
(pkt->time_sent == drs->first_sent_time &&
69+
pkt->rs.end_seq > drs->rs.last_end_seq);
70+
}
71+
72+
/* RFC https://datatracker.ietf.org/doc/draft-ietf-ccwg-bbr/
73+
* 4.5.2.3.3. Upon receiving an ACK
74+
*
75+
* When an ACK arrives, the sender invokes GenerateRateSample() to fill
76+
* in a rate sample. For each packet that was newly SACKed or ACKed,
77+
* UpdateRateSample() updates the rate sample based on a snapshot of
78+
* connection delivery information from the time at which the packet was
79+
* last transmitted. UpdateRateSample() is invoked multiple times when
80+
* a stretched ACK acknowledges multiple data packets. In this case we
81+
* use the information from the most recently sent packet, i.e., the
82+
* packet with the highest "P.delivered" value.
83+
*
84+
* haproxy implementation: quic_cc_drs_update_rate_sample() matches with
85+
* RFC UpdateRateSample() called from first part of GenerateRateSample().
86+
*/
87+
void quic_cc_drs_update_rate_sample(struct quic_cc_drs *drs,
88+
struct quic_tx_packet *pkt)
89+
{
90+
struct quic_cc_rs *rs = &drs->rs;
91+
92+
if (!tick_isset(pkt->rs.delivered_time))
93+
return;
94+
95+
drs->delivered += pkt->len;
96+
drs->delivered_time = now_ms;
97+
/* Update info using the newest packet. */
98+
if (tick_isset(rs->prior_time) && !quic_cc_drs_is_newest_packet(drs, pkt))
99+
return;
100+
101+
rs->prior_delivered = pkt->rs.delivered;
102+
rs->prior_time = pkt->rs.delivered_time;
103+
rs->is_app_limited = pkt->rs.is_app_limited;
104+
rs->send_elapsed = pkt->time_sent - pkt->rs.first_sent_time;
105+
rs->ack_elapsed = drs->delivered_time - pkt->rs.delivered_time;
106+
rs->tx_in_flight = pkt->rs.tx_in_flight;
107+
rs->prior_lost = pkt->rs.lost;
108+
rs->last_end_seq = pkt->rs.end_seq;
109+
drs->first_sent_time = pkt->time_sent;
110+
/* Mark the packet as delivered once it's SACKed to
111+
* avoid being used again when it's cumulatively acked.
112+
*/
113+
pkt->rs.delivered_time = TICK_ETERNITY;
114+
}
115+
116+
/* RFC https://datatracker.ietf.org/doc/draft-ietf-ccwg-bbr/
117+
* 4.5.2.3.3. Upon receiving an ACK
118+
*
119+
* haproxy implementation: second part of GenerateRateSample(). Follows the
120+
* first one above.
121+
*/
122+
void quic_cc_drs_on_ack_recv(struct quic_cc_drs *drs, struct quic_cc_path *path,
123+
uint64_t pkt_delivered)
124+
{
125+
struct quic_cc_rs *rs = &drs->rs;
126+
uint64_t rate;
127+
128+
if (drs->app_limited && drs->delivered > drs->app_limited)
129+
drs->app_limited = 0;
130+
131+
if (pkt_delivered >= drs->next_round_delivered) {
132+
drs->next_round_delivered = pkt_delivered;
133+
++drs->round_count;
134+
}
135+
136+
if (!tick_isset(rs->prior_time))
137+
return;
138+
139+
rs->interval = MAX(rs->send_elapsed, rs->ack_elapsed);
140+
141+
BUG_ON(drs->delivered <= rs->prior_delivered);
142+
rs->delivered = drs->delivered - rs->prior_delivered;
143+
BUG_ON(drs->lost < rs->prior_lost);
144+
rs->lost = drs->lost - rs->prior_lost;
145+
146+
if (rs->interval < path->loss.rtt_min) {
147+
rs->interval = UINT32_MAX;
148+
return;
149+
}
150+
151+
if (!rs->interval)
152+
return;
153+
154+
rate = rs->delivered * 1000 / rs->interval;
155+
if (rate >= wf_get_max(&drs->wf) || !drs->app_limited) {
156+
wf_max_update(&drs->wf, rate, drs->round_count);
157+
path->delivery_rate = wf_get_max(&drs->wf);
158+
}
159+
}

0 commit comments

Comments
 (0)