1a57302d1a

lib/http2.c+3 -3

@@ -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;
	}

lib/multi.c+35 -18

@@ -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;
-
-	/* 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);
-
-	if(send_wait_ms \|\| recv_wait_ms) {
-	if(data->mstate != MSTATE_RATELIMITING) {
-	multistate(data, MSTATE_RATELIMITING);
-	}
-	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;
+	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;
+
+	send_ms = Curl_rlimit_wait_ms(&data->progress.ul.rlimit, pnow);
+	recv_ms = Curl_rlimit_wait_ms(&data->progress.dl.rlimit, pnow);
+
+	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);
+	}
+	}
	}
-	else if(data->mstate != MSTATE_PERFORMING) {
+
+	if(data->mstate != MSTATE_PERFORMING) {
	CURL_TRC_M(data, "[RLIMIT] wait over, continue");
	multistate(data, MSTATE_PERFORMING);
	}

lib/progress.c+8 -10

@@ -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)

lib/ratelimit.c+167 -73

@@ -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;

lib/ratelimit.h+25 -12

@@ -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);

lib/sendf.c+9 -2

@@ -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;
	}

lib/vquic/curl_ngtcp2.c+70 -68

@@ -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",

tests/http/scorecard.py+87 -8

@@ -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 >= (1010241024):
+	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 >= (1010241024):
+	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 = (10241024*1024)
+	elif m.group(3) == 'm':
+	self._limit_rate_num = (10241024)
+	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 {

tests/http/test_02_download.py+4 -6

@@ -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

tests/unit/unit3216.c+2 -2

@@ -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;

Subject

Binary

Architecture & Compiler Flags

Build Log

Commit Diff GitHub →