romenskiy2012/curl
1
0
Fork 0
mirror of https://github.com/curl/curl.git synced 2026-08-25 00:13:34 +03:00
Code Issues Projects Releases Packages Wiki Activity

ratelimit: download finetune

Browse source 
When a download size is known and rate limiting is in effect, adjust the
duration of each measurement step and its rate for maximum precision.

Since it is unpredictable how long the last bytes of a download will
take, download speed can be thrown off if the "last bytes" are a
significant amount of the total download. Make the "last bytes" small in
comparision to the rest and "stretch" the rate limit intervals to
accommodate the difference.

Fix ngtcp2 receive data acknowldgements to be based on a local window
size tracking. This allows window updates controlled by rate limits.

Fix ratelimit wait time calculation to accomodate negative tokens.

h3 rate limit, update timeers

Make download rate limits work correctly in ngtcp2. Fix multi handling
of rate limits to set a timer for when limits will update again.

Without running the transfer on limit updates, protocols like h2/h3 may
stall if the server does not send due to stream windows being too small.

scorecard: measure download speedlimits

When running scorecard with --limit-rate=n, show the reported download
speed averages plus percentage deviation from the limit.

Closes #20228
This commit is contained in:
Stefan Eissing 2026-01-15 11:58:56 +01:00 • committed by Daniel Stenberg
parent 9f120d2b50
commit 1a57302d1a
No known key found for this signature in database
GPG key ID: 5CC908FDB71E12C2
10 changed files with 407 additions and 199 deletions
Download patch file Download diff file Expand all files Collapse all files

6
lib/http2.c
View file

@ -216,9 +216,9 @@ static uint32_t cf_h2_initial_win_size(struct Curl_easy *data)
/* If the transfer has a rate-limit lower than the default initial
* stream window size, use that. It needs to be at least 8k or servers
* may be unhappy. */
if(data->progress.dl.rlimit.rate_per_step &&
(data->progress.dl.rlimit.rate_per_step < H2_STREAM_WINDOW_SIZE_INITIAL))
return CURLMAX((uint32_t)data->progress.dl.rlimit.rate_per_step, 8192);
curl_off_t rps = Curl_rlimit_per_step(&data->progress.dl.rlimit);
if((rps > 0) && (rps < H2_STREAM_WINDOW_SIZE_INITIAL))
return CURLMAX((uint32_t)rps, 8192);
#endif
return H2_STREAM_WINDOW_SIZE_INITIAL;
}

47
lib/multi.c
View file

@ -1922,25 +1922,42 @@ static CURLcode multi_follow(struct Curl_easy *data,
static CURLcode mspeed_check(struct Curl_easy *data)
{
const struct curltime *pnow = Curl_pgrs_now(data);
timediff_t recv_wait_ms = 0;
timediff_t send_wait_ms = 0;
if(Curl_rlimit_active(&data->progress.dl.rlimit) ||
Curl_rlimit_active(&data->progress.ul.rlimit)) {
/* check if our send/recv limits require idle waits */
const struct curltime *pnow = Curl_pgrs_now(data);
timediff_t recv_ms, send_ms;
/* check if our send/recv limits require idle waits */
send_wait_ms = Curl_rlimit_wait_ms(&data->progress.ul.rlimit, pnow);
recv_wait_ms = Curl_rlimit_wait_ms(&data->progress.dl.rlimit, pnow);
send_ms = Curl_rlimit_wait_ms(&data->progress.ul.rlimit, pnow);
recv_ms = Curl_rlimit_wait_ms(&data->progress.dl.rlimit, pnow);
if(send_wait_ms || recv_wait_ms) {
if(data->mstate != MSTATE_RATELIMITING) {
multistate(data, MSTATE_RATELIMITING);
if(send_ms || recv_ms) {
if(data->mstate != MSTATE_RATELIMITING) {
multistate(data, MSTATE_RATELIMITING);
}
Curl_expire(data, CURLMAX(send_ms, recv_ms), EXPIRE_TOOFAST);
Curl_multi_clear_dirty(data);
CURL_TRC_M(data, "[RLIMIT] waiting %" FMT_TIMEDIFF_T "ms",
CURLMAX(send_ms, recv_ms));
return CURLE_AGAIN;
}
else {
/* when will the rate limits increase next? The transfer needs
* to run again at that time or it may stall. */
send_ms = Curl_rlimit_next_step_ms(&data->progress.ul.rlimit, pnow);
recv_ms = Curl_rlimit_next_step_ms(&data->progress.dl.rlimit, pnow);
if(send_ms || recv_ms) {
timediff_t next_ms = CURLMIN(send_ms, recv_ms);
if(!next_ms)
next_ms = CURLMAX(send_ms, recv_ms);
Curl_expire(data, next_ms, EXPIRE_TOOFAST);
CURL_TRC_M(data, "[RLIMIT] next token update in %" FMT_TIMEDIFF_T "ms",
next_ms);
}
}
Curl_expire(data, CURLMAX(send_wait_ms, recv_wait_ms), EXPIRE_TOOFAST);
Curl_multi_clear_dirty(data);
CURL_TRC_M(data, "[RLIMIT] waiting %" FMT_TIMEDIFF_T "ms",
CURLMAX(send_wait_ms, recv_wait_ms));
return CURLE_AGAIN;
}
else if(data->mstate != MSTATE_PERFORMING) {
if(data->mstate != MSTATE_PERFORMING) {
CURL_TRC_M(data, "[RLIMIT] wait over, continue");
multistate(data, MSTATE_PERFORMING);
}

18
lib/progress.c
View file

@ -415,7 +415,7 @@ static bool progress_calc(struct Curl_easy *data,
{
struct Progress * const p = &data->progress;
int i_next, i_oldest, i_latest;
timediff_t duration_ms;
timediff_t duration_us;
curl_off_t amount;
/* The time spent so far (from the start) in microseconds */
@ -466,21 +466,19 @@ static bool progress_calc(struct Curl_easy *data,
/* How much we transferred between oldest and current records */
amount = p->speed_amount[i_latest] - p->speed_amount[i_oldest];
/* How long this took */
duration_ms = curlx_ptimediff_ms(&p->speed_time[i_latest],
duration_us = curlx_ptimediff_us(&p->speed_time[i_latest],
&p->speed_time[i_oldest]);
if(duration_ms <= 0)
duration_ms = 1;
if(duration_us <= 0)
duration_us = 1;
if(amount > (CURL_OFF_T_MAX / 1000)) {
if(amount > (CURL_OFF_T_MAX / 1000000)) {
/* the 'amount' value is bigger than would fit in 64 bits if
multiplied with 1000, so we use the double math for this */
multiplied with 1000000, so we use the double math for this */
p->current_speed =
(curl_off_t)(((double)amount * 1000.0) / (double)duration_ms);
(curl_off_t)(((double)amount * 1000000.0) / (double)duration_us);
}
else {
/* the 'amount' value is small enough to fit within 32 bits even
when multiplied with 1000 */
p->current_speed = amount * 1000 / duration_ms;
p->current_speed = amount * 1000000 / duration_us;
}
if((p->lastshow == pnow->tv_sec) && !data->req.done)

240
lib/ratelimit.c
View file

@ -23,64 +23,20 @@
***************************************************************************/
#include "curl_setup.h"
#include "urldata.h"
#include "curl_trc.h"
#include "progress.h"
#include "ratelimit.h"
#define CURL_US_PER_SEC 1000000
#define CURL_RLIMIT_MIN_CHUNK (16 * 1024)
#define CURL_RLIMIT_MAX_STEPS 2 /* 500ms interval */
#define CURL_US_PER_SEC 1000000
#define CURL_RLIMIT_MIN_RATE (4 * 1024) /* minimum step rate */
#define CURL_RLIMIT_STEP_MIN_MS 2 /* minimum step duration */
void Curl_rlimit_init(struct Curl_rlimit *r,
curl_off_t rate_per_s,
curl_off_t burst_per_s,
const struct curltime *pts)
{
curl_off_t rate_steps;
DEBUGASSERT(rate_per_s >= 0);
DEBUGASSERT(burst_per_s >= rate_per_s || !burst_per_s);
DEBUGASSERT(pts);
r->step_us = CURL_US_PER_SEC;
r->rate_per_step = rate_per_s;
r->burst_per_step = burst_per_s;
/* On rates that are multiples of CURL_RLIMIT_MIN_CHUNK, we reduce
* the interval `step_us` from 1 second to smaller steps with at
* most CURL_RLIMIT_MAX_STEPS.
* Smaller means more CPU, but also more precision. */
rate_steps = rate_per_s / CURL_RLIMIT_MIN_CHUNK;
rate_steps = CURLMIN(rate_steps, CURL_RLIMIT_MAX_STEPS);
if(rate_steps >= 2) {
r->step_us /= rate_steps;
r->rate_per_step /= rate_steps;
r->burst_per_step /= rate_steps;
}
r->tokens = r->rate_per_step;
r->spare_us = 0;
r->ts = *pts;
r->blocked = FALSE;
}
void Curl_rlimit_start(struct Curl_rlimit *r, const struct curltime *pts)
{
r->tokens = r->rate_per_step;
r->spare_us = 0;
r->ts = *pts;
}
bool Curl_rlimit_active(struct Curl_rlimit *r)
{
return (r->rate_per_step > 0) || r->blocked;
}
bool Curl_rlimit_is_blocked(struct Curl_rlimit *r)
{
return r->blocked;
}
static void ratelimit_update(struct Curl_rlimit *r,
const struct curltime *pts)
static void rlimit_update(struct Curl_rlimit *r,
const struct curltime *pts)
{
timediff_t elapsed_us, elapsed_steps;
curl_off_t token_gain;
int64_t token_gain;
DEBUGASSERT(r->rate_per_step);
if((r->ts.tv_sec == pts->tv_sec) && (r->ts.tv_usec == pts->tv_usec))
@ -102,31 +58,151 @@ static void ratelimit_update(struct Curl_rlimit *r,
r->spare_us = elapsed_us % r->step_us;
/* How many tokens did we gain since the last update? */
if(r->rate_per_step > (CURL_OFF_T_MAX / elapsed_steps))
token_gain = CURL_OFF_T_MAX;
if(r->rate_per_step > (INT64_MAX / elapsed_steps))
token_gain = INT64_MAX;
else {
token_gain = r->rate_per_step * elapsed_steps;
}
/* Limit the token again by the burst rate per second (if set), so we
if((INT64_MAX - token_gain) > r->tokens)
r->tokens += token_gain;
else
r->tokens = INT64_MAX;
/* Limit the token again by the burst rate (if set), so we
* do not suddenly have a huge number of tokens after inactivity. */
r->tokens += token_gain;
if(r->burst_per_step && (r->tokens > r->burst_per_step)) {
r->tokens = r->burst_per_step;
}
}
curl_off_t Curl_rlimit_avail(struct Curl_rlimit *r,
const struct curltime *pts)
static void rlimit_tune_steps(struct Curl_rlimit *r,
int64_t tokens_total)
{
int64_t tokens_last, tokens_main, msteps;
/* Tune the ratelimit at the start *if* we know how many tokens
* are expected to be consumed in total.
* The reason for tuning is that rlimit provides tokens to be consumed
* per "step" which starts out to be a second. The tokens may be consumed
* in full at the beginning of a step. The remainder of the second will
* have no tokens available, effectively blocking the consumption and
* so keeping the "step average" in line.
* This works will up to the last step. When no more tokens are needed,
* no wait will happen and the last step would be too fast. This is
* especially noticeable when only a few steps are needed.
*
* Example: downloading 1.5kb with a ratelimit of 1k could be done in
* roughly 1 second (1k in the first second and the 0.5 at the start of
* the second one).
*
* The tuning tries to make the last step small, using only
* 1 percent of the total tokens (at least 1). The rest of the tokens
* are to be consumed in the steps before by adjusting the duration of
* the step and the amount of tokens it provides. */
if(!r->rate_per_step ||
(tokens_total <= 1) ||
(tokens_total > (INT64_MAX / 1000)))
return;
/* Calculate tokens for the last step and the ones before. */
tokens_last = tokens_total / 100;
if(!tokens_last) /* less than 100 total, just use 1 */
tokens_last = 1;
else if(tokens_last > CURL_RLIMIT_MIN_RATE)
tokens_last = CURL_RLIMIT_MIN_RATE;
DEBUGASSERT(tokens_last);
tokens_main = tokens_total - tokens_last;
DEBUGASSERT(tokens_main);
/* how many milli-steps will it take to consume those, give the
* original rate limit per second? */
DEBUGASSERT(r->step_us == CURL_US_PER_SEC);
msteps = (tokens_main * 1000 / r->rate_per_step);
if(msteps < CURL_RLIMIT_STEP_MIN_MS) {
/* Steps this small will not work. Do not tune. */
return;
}
else if(msteps < 1000) {
/* It needs less than one step to provide the needed tokens.
* Make it exactly that long and with exactly those tokens. */
r->step_us = (timediff_t)msteps * 1000;
r->rate_per_step = tokens_main;
r->tokens = r->rate_per_step;
}
else {
/* More than 1 step. Spread the remainder milli steps and
* the tokens they need to provide across all steps. If integer
* arithmetic can do it. */
curl_off_t ms_unaccounted = (msteps % 1000);
curl_off_t mstep_inc = (ms_unaccounted / (msteps / 1000));
if(mstep_inc) {
curl_off_t rate_inc = ((r->rate_per_step * mstep_inc) / 1000);
if(rate_inc) {
r->step_us = CURL_US_PER_SEC + ((timediff_t)mstep_inc * 1000);
r->rate_per_step += rate_inc;
r->tokens = r->rate_per_step;
}
}
}
if(r->burst_per_step)
r->burst_per_step = r->rate_per_step;
}
void Curl_rlimit_init(struct Curl_rlimit *r,
int64_t rate_per_sec,
int64_t burst_per_sec,
const struct curltime *pts)
{
DEBUGASSERT(rate_per_sec >= 0);
DEBUGASSERT(burst_per_sec >= rate_per_sec || !burst_per_sec);
DEBUGASSERT(pts);
r->rate_per_step = rate_per_sec;
r->burst_per_step = burst_per_sec;
r->step_us = CURL_US_PER_SEC;
r->spare_us = 0;
r->tokens = r->rate_per_step;
r->ts = *pts;
r->blocked = FALSE;
}
void Curl_rlimit_start(struct Curl_rlimit *r, const struct curltime *pts,
int64_t total_tokens)
{
r->tokens = r->rate_per_step;
r->spare_us = 0;
r->ts = *pts;
rlimit_tune_steps(r, total_tokens);
}
int64_t Curl_rlimit_per_step(struct Curl_rlimit *r)
{
return r->rate_per_step;
}
bool Curl_rlimit_active(struct Curl_rlimit *r)
{
return (r->rate_per_step > 0) || r->blocked;
}
bool Curl_rlimit_is_blocked(struct Curl_rlimit *r)
{
return r->blocked;
}
int64_t Curl_rlimit_avail(struct Curl_rlimit *r,
const struct curltime *pts)
{
if(r->blocked)
return 0;
else if(r->rate_per_step) {
ratelimit_update(r, pts);
rlimit_update(r, pts);
return r->tokens;
}
else
return CURL_OFF_T_MAX;
return INT64_MAX;
}
void Curl_rlimit_drain(struct Curl_rlimit *r,
@ -136,20 +212,19 @@ void Curl_rlimit_drain(struct Curl_rlimit *r,
if(r->blocked || !r->rate_per_step)
return;
ratelimit_update(r, pts);
#if SIZEOF_CURL_OFF_T <= SIZEOF_SIZE_T
if(tokens > CURL_OFF_T_MAX) {
r->tokens = CURL_OFF_T_MIN;
return;
rlimit_update(r, pts);
#if 8 <= SIZEOF_SIZE_T
if(tokens > INT64_MAX) {
r->tokens = INT64_MAX;
}
else
#endif
{
curl_off_t val = (curl_off_t)tokens;
if((CURL_OFF_T_MIN + val) < r->tokens)
int64_t val = (int64_t)tokens;
if((INT64_MIN + val) < r->tokens)
r->tokens -= val;
else
r->tokens = CURL_OFF_T_MIN;
r->tokens = INT64_MIN;
}
}
@ -160,14 +235,18 @@ timediff_t Curl_rlimit_wait_ms(struct Curl_rlimit *r,
if(r->blocked || !r->rate_per_step)
return 0;
ratelimit_update(r, pts);
rlimit_update(r, pts);
if(r->tokens > 0)
return 0;
/* How much time will it take tokens to become positive again?
* Deduct `spare_us` and check against already elapsed time */
wait_us = (1 + (-r->tokens / r->rate_per_step)) * r->step_us;
wait_us -= r->spare_us;
wait_us = r->step_us - r->spare_us;
if(r->tokens < 0) {
curl_off_t debt_pct = ((-r->tokens) * 100 / r->rate_per_step);
if(debt_pct)
wait_us += (r->step_us * debt_pct / 100);
}
elapsed_us = curlx_ptimediff_us(pts, &r->ts);
if(elapsed_us >= wait_us)
@ -176,6 +255,21 @@ timediff_t Curl_rlimit_wait_ms(struct Curl_rlimit *r,
return (wait_us + 999) / 1000; /* in milliseconds */
}
timediff_t Curl_rlimit_next_step_ms(struct Curl_rlimit *r,
const struct curltime *pts)
{
if(!r->blocked && r->rate_per_step) {
timediff_t elapsed_us, next_us;
elapsed_us = curlx_ptimediff_us(pts, &r->ts) + r->spare_us;
if(r->step_us > elapsed_us) {
next_us = r->step_us - elapsed_us;
return (next_us + 999) / 1000; /* in milliseconds */
}
}
return 0;
}
void Curl_rlimit_block(struct Curl_rlimit *r,
bool activate,
const struct curltime *pts)
@ -188,7 +282,7 @@ void Curl_rlimit_block(struct Curl_rlimit *r,
if(!r->blocked) {
/* Start rate limiting fresh. The amount of time this was blocked
* does not generate extra tokens. */
Curl_rlimit_start(r, pts);
Curl_rlimit_start(r, pts, -1);
}
else {
r->tokens = 0;

37
lib/ratelimit.h
View file

@ -28,8 +28,10 @@
struct Curl_easy;
/* This is a rate limiter that provides "tokens" to be consumed
* per second with a "burst" rate limitation. Example:
* A rate limit of 1 megabyte per second with a burst rate of 1.5MB.
* per second. In the literature, this is referred to as a
* "token bucket" (https://en.wikipedia.org/wiki/Token_bucket).
* Example:
* A rate limit of 1 megabyte per second.
* - initially 1 million tokens are available.
* - these are drained in the first second.
* - checking available tokens before the 2nd second will return 0.
@ -43,44 +45,55 @@ struct Curl_easy;
* - setting "burst" to the same value as "rate" would make a
* download always try to stay *at/below* the rate and slow times will
* not generate extra tokens.
*
* A rate limit can be blocked, causing the available tokens to become
* always 0 until unblocked. After unblocking, the rate limiting starts
* again with no history of the past.
*
* Finally, a rate limiter with rate 0 will always have CURL_OFF_T_MAX
* tokens available, unless blocked.
*/
struct Curl_rlimit {
curl_off_t rate_per_step; /* rate tokens are generated per step us */
curl_off_t burst_per_step; /* burst rate of tokens per step us */
int64_t rate_per_step; /* rate tokens generated per step us */
int64_t burst_per_step; /* burst rate of tokens per step us */
timediff_t step_us; /* microseconds between token increases */
curl_off_t tokens; /* tokens available in the next second */
int64_t tokens; /* tokens available in the next second */
timediff_t spare_us; /* microseconds unaffecting tokens */
struct curltime ts; /* time of the last update */
BIT(blocked); /* blocking sets available tokens to 0 */
};
void Curl_rlimit_init(struct Curl_rlimit *r,
curl_off_t rate_per_s,
curl_off_t burst_per_s,
int64_t rate_per_sec,
int64_t burst_per_sec,
const struct curltime *pts);
/* Start ratelimiting with the given timestamp. Resets available tokens. */
void Curl_rlimit_start(struct Curl_rlimit *r, const struct curltime *pts);
/* Start ratelimiting with the given timestamp. Resets available tokens.
* `total_tokens` is either -1 or the number of total tokens expected
* to be consumed. */
void Curl_rlimit_start(struct Curl_rlimit *r, const struct curltime *pts,
int64_t total_tokens);
/* How many milliseconds to wait until token are available again. */
timediff_t Curl_rlimit_wait_ms(struct Curl_rlimit *r,
const struct curltime *pts);
/* When the rate limit will update its tokens again */
timediff_t Curl_rlimit_next_step_ms(struct Curl_rlimit *r,
const struct curltime *pts);
/* Return if rate limiting of tokens is active */
bool Curl_rlimit_active(struct Curl_rlimit *r);
bool Curl_rlimit_is_blocked(struct Curl_rlimit *r);
int64_t Curl_rlimit_per_step(struct Curl_rlimit *r);
/* Return how many tokens are available to spend, may be negative */
curl_off_t Curl_rlimit_avail(struct Curl_rlimit *r,
const struct curltime *pts);
int64_t Curl_rlimit_avail(struct Curl_rlimit *r,
const struct curltime *pts);
/* Drain tokens from the ratelimit, return how many are now available. */
/* Drain tokens from the ratelimit, give an estimate of how many tokens
* remain to be drained in the future (-1 for unknown). */
void Curl_rlimit_drain(struct Curl_rlimit *r,
size_t tokens,
const struct curltime *pts);

11
lib/sendf.c
View file

@ -169,6 +169,7 @@ static size_t get_max_body_write_len(struct Curl_easy *data, curl_off_t limit)
struct cw_download_ctx {
struct Curl_cwriter super;
BIT(started_response);
BIT(started_body);
};
/* Download client writer in phase CURL_CW_PROTOCOL that
@ -185,7 +186,6 @@ static CURLcode cw_download_write(struct Curl_easy *data,
if(!ctx->started_response &&
!(type & (CLIENTWRITE_INFO | CLIENTWRITE_CONNECT))) {
Curl_pgrsTime(data, TIMER_STARTTRANSFER);
Curl_rlimit_start(&data->progress.dl.rlimit, Curl_pgrs_now(data));
ctx->started_response = TRUE;
}
@ -198,6 +198,13 @@ static CURLcode cw_download_write(struct Curl_easy *data,
return result;
}
if(!ctx->started_body &&
!(type & (CLIENTWRITE_INFO | CLIENTWRITE_CONNECT))) {
Curl_rlimit_start(&data->progress.dl.rlimit, Curl_pgrs_now(data),
data->req.size);
ctx->started_body = TRUE;
}
/* Here, we deal with REAL BODY bytes. All filtering and transfer
* encodings have been applied and only the true content, e.g. BODY,
* bytes are passed here.
@ -1189,7 +1196,7 @@ CURLcode Curl_client_read(struct Curl_easy *data, char *buf, size_t blen,
DEBUGASSERT(data->req.reader_stack);
}
if(!data->req.reader_started) {
Curl_rlimit_start(&data->progress.ul.rlimit, Curl_pgrs_now(data));
Curl_rlimit_start(&data->progress.ul.rlimit, Curl_pgrs_now(data), -1);
data->req.reader_started = TRUE;
}

138
lib/vquic/curl_ngtcp2.c
View file

@ -231,7 +231,8 @@ struct h3_stream_ctx {
size_t sendbuf_len_in_flight; /* sendbuf amount "in flight" */
uint64_t error3; /* HTTP/3 stream error code */
curl_off_t upload_left; /* number of request bytes left to upload */
uint64_t download_unacked; /* bytes not acknowledged yet */
uint64_t rx_offset; /* current receive offset */
uint64_t rx_offset_max; /* allowed receive offset */
uint64_t window_size_max; /* max flow control window set for stream */
int status_code; /* HTTP status code */
CURLcode xfer_result; /* result from xfer_resp_write(_hd) */
@ -273,6 +274,9 @@ static CURLcode h3_data_setup(struct Curl_cfilter *cf,
return CURLE_OUT_OF_MEMORY;
stream->id = -1;
stream->rx_offset = 0;
stream->rx_offset_max = H3_STREAM_WINDOW_SIZE_INITIAL;
/* on send, we control how much we put into the buffer */
Curl_bufq_initp(&stream->sendbuf, &ctx->stream_bufcp,
H3_STREAM_SEND_CHUNKS, BUFQ_OPT_NONE);
@ -610,20 +614,15 @@ static int cb_recv_stream_data(ngtcp2_conn *tconn, uint32_t flags,
{
struct Curl_cfilter *cf = user_data;
struct cf_ngtcp2_ctx *ctx = cf->ctx;
nghttp3_ssize nconsumed;
nghttp3_ssize rc;
uint64_t nconsumed;
int fin = (flags & NGTCP2_STREAM_DATA_FLAG_FIN) ? 1 : 0;
struct Curl_easy *data = stream_user_data;
struct h3_stream_ctx *stream = H3_STREAM_CTX(ctx, data);
(void)offset;
nconsumed =
nghttp3_conn_read_stream(ctx->h3conn, stream_id, buf, buflen, fin);
if(!data)
data = CF_DATA_CURRENT(cf);
if(data)
CURL_TRC_CF(data, cf, "[%" PRId64 "] read_stream(len=%zu) -> %zd",
stream_id, buflen, nconsumed);
if(nconsumed < 0) {
struct h3_stream_ctx *stream = H3_STREAM_CTX(ctx, data);
rc = nghttp3_conn_read_stream(ctx->h3conn, stream_id, buf, buflen, fin);
if(rc < 0) {
if(data && stream) {
CURL_TRC_CF(data, cf, "[%" PRId64 "] error on known stream, "
"reset=%d, closed=%d",
@ -631,13 +630,18 @@ static int cb_recv_stream_data(ngtcp2_conn *tconn, uint32_t flags,
}
return NGTCP2_ERR_CALLBACK_FAILURE;
}
/* number of bytes inside buflen which consists of framing overhead
* including QPACK HEADERS. In other words, it does not consume payload of
* DATA frame. */
ngtcp2_conn_extend_max_stream_offset(tconn, stream_id, (uint64_t)nconsumed);
ngtcp2_conn_extend_max_offset(tconn, (uint64_t)nconsumed);
nconsumed = (uint64_t)rc;
if(nconsumed) {
/* number of bytes inside buflen which consists of framing overhead
* including QPACK HEADERS. In other words, it does not consume payload of
* DATA frame. */
ngtcp2_conn_extend_max_stream_offset(tconn, stream_id, nconsumed);
ngtcp2_conn_extend_max_offset(tconn, nconsumed);
if(stream) {
stream->rx_offset += nconsumed;
stream->rx_offset_max += nconsumed;
}
}
return 0;
}
@ -1044,57 +1048,50 @@ static void h3_xfer_write_resp(struct Curl_cfilter *cf,
}
}
static void cf_ngtcp2_stream_update_window(struct Curl_cfilter *cf,
struct Curl_easy *data,
struct h3_stream_ctx *stream)
{
/* stream receive max window size for flow control. We can only
* grow it from the initial window size */
uint64_t swin_max = data->progress.dl.rlimit.rate_per_step ?
data->progress.dl.rlimit.rate_per_step : H3_STREAM_WINDOW_SIZE_MAX;
if(swin_max > stream->window_size_max) {
struct cf_ngtcp2_ctx *ctx = cf->ctx;
int rc = ngtcp2_conn_extend_max_stream_offset(ctx->qconn, stream->id,
swin_max - stream->window_size_max);
if(rc) {
CURL_TRC_CF(data, cf, "[%" PRId64 "] extend_max_stream_offset to %"
PRIu64 " -> %s (%d)",
stream->id, swin_max, ngtcp2_strerror(rc), rc);
DEBUGASSERT(0);
}
stream->window_size_max = swin_max;
}
}
static void cf_ngtcp2_ack_stream(struct Curl_cfilter *cf,
static void cf_ngtcp2_upd_rx_win(struct Curl_cfilter *cf,
struct Curl_easy *data,
struct h3_stream_ctx *stream)
{
struct cf_ngtcp2_ctx *ctx = cf->ctx;
curl_off_t avail;
uint64_t ack_len = 0;
/* How many byte to ack on the stream? */
uint64_t cur_win, wanted_win = H3_STREAM_WINDOW_SIZE_MAX;
/* how much does rate limiting allow us to acknowledge? */
avail = Curl_rlimit_avail(&data->progress.dl.rlimit,
Curl_pgrs_now(data));
if(avail == CURL_OFF_T_MAX) { /* no rate limit, ack all */
ack_len = stream->download_unacked;
}
else if(avail > 0) {
ack_len = CURLMIN(stream->download_unacked, (uint64_t)avail);
if(Curl_rlimit_active(&data->progress.dl.rlimit)) {
int64_t avail;
/* start rate limit updates only after first bytes arrived */
if(!stream->rx_offset)
return;
avail = Curl_rlimit_avail(&data->progress.dl.rlimit,
Curl_pgrs_now(data));
if(avail <= 0) {
/* nothing available, do not extend the rx offset */
CURL_TRC_CF(data, cf, "[%" PRId64 "] dl rate limit exhausted (%" PRId64
" tokens)", stream->id, avail);
return;
}
wanted_win = CURLMIN((uint64_t)avail, H3_STREAM_WINDOW_SIZE_MAX);
}
if(ack_len) {
CURL_TRC_CF(data, cf, "[%" PRId64 "] ACK %" PRIu64
"/%" PRIu64 " bytes of DATA", stream->id,
ack_len, stream->download_unacked);
ngtcp2_conn_extend_max_stream_offset(ctx->qconn, stream->id, ack_len);
stream->download_unacked -= ack_len;
if(stream->rx_offset_max < stream->rx_offset) {
DEBUGASSERT(0);
return;
}
cur_win = stream->rx_offset_max - stream->rx_offset;
cf_ngtcp2_stream_update_window(cf, data, stream);
if(wanted_win > cur_win) {
uint64_t delta = wanted_win - cur_win;
if(UINT64_MAX - delta < stream->rx_offset_max)
delta = UINT64_MAX - stream->rx_offset_max;
if(delta) {
CURL_TRC_CF(data, cf, "[%" PRId64 "] rx window, extend by %" PRIu64
" bytes", stream->id, delta);
stream->rx_offset_max += delta;
ngtcp2_conn_extend_max_stream_offset(ctx->qconn, stream->id, delta);
}
}
}
static int cb_h3_recv_data(nghttp3_conn *conn, int64_t stream3_id,
@ -1113,15 +1110,15 @@ static int cb_h3_recv_data(nghttp3_conn *conn, int64_t stream3_id,
return NGHTTP3_ERR_CALLBACK_FAILURE;
h3_xfer_write_resp(cf, data, stream, (const char *)buf, blen, FALSE);
CURL_TRC_CF(data, cf, "[%" PRId64 "] DATA len=%zu", stream->id, blen);
ngtcp2_conn_extend_max_offset(ctx->qconn, blen);
if(UINT64_MAX - blen < stream->download_unacked)
stream->download_unacked = UINT64_MAX; /* unlikely */
else
stream->download_unacked += blen;
stream->rx_offset += blen;
if(stream->rx_offset_max < stream->rx_offset)
stream->rx_offset_max = stream->rx_offset;
cf_ngtcp2_ack_stream(cf, data, stream);
CURL_TRC_CF(data, cf, "[%" PRId64 "] DATA len=%zu, rx win=%" PRId64,
stream->id, blen, stream->rx_offset_max - stream->rx_offset);
cf_ngtcp2_upd_rx_win(cf, data, stream);
return 0;
}
@ -1131,13 +1128,18 @@ static int cb_h3_deferred_consume(nghttp3_conn *conn, int64_t stream3_id,
{
struct Curl_cfilter *cf = user_data;
struct cf_ngtcp2_ctx *ctx = cf->ctx;
struct Curl_easy *data = stream_user_data;
struct h3_stream_ctx *stream = H3_STREAM_CTX(ctx, data);
(void)conn;
(void)stream_user_data;
/* nghttp3 has consumed bytes on the QUIC stream and we need to
* tell the QUIC connection to increase its flow control */
ngtcp2_conn_extend_max_stream_offset(ctx->qconn, stream3_id, consumed);
ngtcp2_conn_extend_max_offset(ctx->qconn, consumed);
if(stream) {
stream->rx_offset += consumed;
stream->rx_offset_max += consumed;
}
return 0;
}
@ -1418,7 +1420,7 @@ static CURLcode cf_ngtcp2_recv(struct Curl_cfilter *cf, struct Curl_easy *data,
goto out;
}
cf_ngtcp2_ack_stream(cf, data, stream);
cf_ngtcp2_upd_rx_win(cf, data, stream);
/* first check for results/closed already known without touching
* the connection. For an already failed/closed stream, errors on
@ -1669,7 +1671,7 @@ static CURLcode h3_stream_open(struct Curl_cfilter *cf,
goto out;
}
cf_ngtcp2_stream_update_window(cf, data, stream);
cf_ngtcp2_upd_rx_win(cf, data, stream);
if(Curl_trc_is_verbose(data)) {
infof(data, "[HTTP/3] [%" PRId64 "] OPENED stream for %s",

95
tests/http/scorecard.py
View file

@ -64,11 +64,34 @@ class Card:
if val is None or val < 0:
return '--'
if val >= (1024*1024):
return f'{val/(1024*1024):0.000f} MB/s'
return f'{val/(1024*1024):.3g} MB/s'
elif val >= 1024:
return f'{val / 1024:0.000f} KB/s'
return f'{val / 1024:.3g} KB/s'
else:
return f'{val:0.000f} B/s'
return f'{val:.3g} B/s'
@classmethod
def fmt_speed(cls, val):
if val is None or val < 0:
return '--'
if val >= (10*1024*1024):
return f'{(val/(1024*1024)):.3f} MB/s'
elif val >= (10*1024):
return f'{val/1024:.3f} KB/s'
else:
return f'{val:.3f} B/s'
@classmethod
def fmt_speed_result(cls, val, limit):
if val is None or val < 0:
return '--'
pct = ((val / limit) * 100) - 100
if val >= (10*1024*1024):
return f'{(val/(1024*1024)):.3f} MB/s, {pct:+.1f}%'
elif val >= (10*1024):
return f'{val/1024:.3f} KB/s, {pct:+.1f}%'
else:
return f'{val:.3f} B/s, {pct:+.1f}%'
@classmethod
def fmt_reqs(cls, val):
@ -87,6 +110,19 @@ class Card:
cell['errors'] = errors
return cell
@classmethod
def mk_speed_cell(cls, samples, profiles, errors, limit):
val = mean(samples) if len(samples) else -1
cell = {
'val': val,
'sval': Card.fmt_speed_result(val, limit) if val >= 0 else '--',
}
if len(profiles):
cell['stats'] = RunProfile.AverageStats(profiles)
if len(errors):
cell['errors'] = errors
return cell
@classmethod
def mk_reqs_cell(cls, samples, profiles, errors):
val = mean(samples) if len(samples) else -1
@ -211,7 +247,23 @@ class ScoreRunner:
self._upload_parallel = upload_parallel
self._with_flame = with_flame
self._socks_args = socks_args
self._limit_rate_num = 0
self._limit_rate = limit_rate
if self._limit_rate:
m = re.match(r'(\d+(\.\d+)?)([gmkb])?', self._limit_rate.lower())
if not m:
raise Exception(f'unrecognised limit-rate: {self._limit_rate}')
self._limit_rate_num = float(m.group(1))
if m.group(3) == 'g':
self._limit_rate_num *= (1024*1024*1024)
elif m.group(3) == 'm':
self._limit_rate_num *= (1024*1024)
elif m.group(3) == 'k':
self._limit_rate_num *= (1024)
elif m.group(3) == 'b':
pass
else:
raise Exception(f'unrecognised limit-rate: {self._limit_rate}')
self.suppress_cl = suppress_cl
def info(self, msg):
@ -306,11 +358,18 @@ class ScoreRunner:
err = self._check_downloads(r, count)
if err:
errors.append(err)
elif self._limit_rate:
total_speed = sum([s['speed_download'] for s in r.stats])
samples.append(total_speed / len(r.stats))
profiles.append(r.profile)
else:
total_size = sum([s['size_download'] for s in r.stats])
samples.append(total_size / r.duration.total_seconds())
profiles.append(r.profile)
return Card.mk_mbs_cell(samples, profiles, errors)
if self._limit_rate:
return Card.mk_speed_cell(samples, profiles, errors, self._limit_rate_num)
else:
return Card.mk_mbs_cell(samples, profiles, errors)
def dl_serial(self, url: str, count: int, nsamples: int = 1):
samples = []
@ -328,11 +387,18 @@ class ScoreRunner:
err = self._check_downloads(r, count)
if err:
errors.append(err)
elif self._limit_rate:
total_speed = sum([s['speed_download'] for s in r.stats])
samples.append(total_speed / len(r.stats))
profiles.append(r.profile)
else:
total_size = sum([s['size_download'] for s in r.stats])
samples.append(total_size / r.duration.total_seconds())
profiles.append(r.profile)
return Card.mk_mbs_cell(samples, profiles, errors)
if self._limit_rate:
return Card.mk_speed_cell(samples, profiles, errors, self._limit_rate_num)
else:
return Card.mk_mbs_cell(samples, profiles, errors)
def dl_parallel(self, url: str, count: int, nsamples: int = 1):
samples = []
@ -355,11 +421,18 @@ class ScoreRunner:
err = self._check_downloads(r, count)
if err:
errors.append(err)
elif self._limit_rate:
total_speed = sum([s['speed_download'] for s in r.stats])
samples.append(total_speed / len(r.stats))
profiles.append(r.profile)
else:
total_size = sum([s['size_download'] for s in r.stats])
samples.append(total_size / r.duration.total_seconds())
profiles.append(r.profile)
return Card.mk_mbs_cell(samples, profiles, errors)
if self._limit_rate:
return Card.mk_speed_cell(samples, profiles, errors, self._limit_rate_num)
else:
return Card.mk_mbs_cell(samples, profiles, errors)
def downloads(self, count: int, fsizes: List[int], meta: Dict[str, Any]) -> Dict[str, Any]:
nsamples = meta['samples']
@ -370,7 +443,10 @@ class ScoreRunner:
if count > 1:
cols.append(f'serial({count})')
if count > 1:
cols.append(f'parallel({count}x{max_parallel})')
if max_parallel == 1:
cols.append(f'serial({count})')
else:
cols.append(f'parallel({count}x{max_parallel})')
rows = []
for fsize in fsizes:
row = [{
@ -387,7 +463,10 @@ class ScoreRunner:
row.append(self.dl_parallel(url=url, count=count, nsamples=nsamples))
rows.append(row)
self.info('done.\n')
title = f'Downloads from {meta["server"]}'
if self._limit_rate:
title = f'Download Speed ({self.protocol}), limit={Card.fmt_speed(self._limit_rate_num)}, from {meta["server"]}'
else:
title = f'Downloads ({self.protocol})from {meta["server"]}'
if self._socks_args:
title += f' via {self._socks_args}'
return {

10
tests/http/test_02_download.py
View file

@ -395,20 +395,18 @@ class TestDownload:
# speed limited download
@pytest.mark.parametrize("proto", Env.http_protos())
def test_02_24_speed_limit(self, env: Env, httpd, nghttpx, proto):
if proto == 'h3' and not env.curl_uses_lib('ngtcp2'):
pytest.skip("precise h3 rate limits only with ngtcp2")
count = 1
url = f'https://{env.authority_for(env.domain1, proto)}/data-1m'
curl = CurlClient(env=env)
speed_limit = 256 * 1024
speed_limit = 512 * 1024
r = curl.http_download(urls=[url], alpn_proto=proto, extra_args=[
'--limit-rate', f'{speed_limit}'
])
r.check_response(count=count, http_status=200)
dl_speed = r.stats[0]['speed_download']
# speed limit is only exact on long durations. Ideally this transfer
# would take 4 seconds, but it may end just after 3 because then
# we have downloaded the rest and will not wait for the rate
# limit to increase again.
assert dl_speed <= ((1024*1024)/3), f'{r.stats[0]}'
assert dl_speed <= (speed_limit * 1.1), f'{r.stats[0]}'
# make extreme parallel h2 upgrades, check invalid conn reuse
# before protocol switch has happened

4
tests/unit/unit3216.c
View file

@ -58,9 +58,9 @@ static CURLcode test_unit3216(const char *arg)
ts.tv_usec += 1000; /* 1ms */
Curl_rlimit_drain(&r, 3, &ts);
fail_unless(Curl_rlimit_avail(&r, &ts) == -1, "drain to -1");
fail_unless(Curl_rlimit_wait_ms(&r, &ts) == 999, "wait 999ms");
fail_unless(Curl_rlimit_wait_ms(&r, &ts) == 1099, "wait 1099ms");
ts.tv_usec += 1000; /* 1ms */
fail_unless(Curl_rlimit_wait_ms(&r, &ts) == 998, "wait 998ms");
fail_unless(Curl_rlimit_wait_ms(&r, &ts) == 1098, "wait 1098ms");
ts.tv_sec += 1;
fail_unless(Curl_rlimit_avail(&r, &ts) == 9, "10 inc per sec");
ts.tv_sec += 1;