nihilist/tor
1
0
Fork 0
mirror of https://gitlab.torproject.org/tpo/core/tor.git synced 2024-11-14 07:03:44 +01:00
Code Issues Packages Projects Releases Wiki Activity

Replace the global buckets with token_bucket_t

Browse Source
This commit is contained in:
Nick Mathewson 2018-04-10 13:19:25 -04:00
parent 9fced56ef1
commit a38fd9bc5b
4 changed files with 68 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
src/or/connection.c
View File

@ -2846,7 +2846,7 @@ connection_counts_as_relayed_traffic(connection_t *conn, time_t now)
* non-negative) provides an upper limit for our answer. */ * non-negative) provides an upper limit for our answer. */
static ssize_t static ssize_t
connection_bucket_round_robin(int base, int priority, connection_bucket_round_robin(int base, int priority,
ssize_t global_bucket, ssize_t conn_bucket) ssize_t global_bucket_val, ssize_t conn_bucket)
{ {
ssize_t at_most; ssize_t at_most;
ssize_t num_bytes_high = (priority ? 32 : 16) * base; ssize_t num_bytes_high = (priority ? 32 : 16) * base;
@ -2855,15 +2855,15 @@ connection_bucket_round_robin(int base, int priority,
/* Do a rudimentary round-robin so one circuit can't hog a connection. /* Do a rudimentary round-robin so one circuit can't hog a connection.
* Pick at most 32 cells, at least 4 cells if possible, and if we're in * Pick at most 32 cells, at least 4 cells if possible, and if we're in
* the middle pick 1/8 of the available bandwidth. */ * the middle pick 1/8 of the available bandwidth. */
at_most = global_bucket / 8; at_most = global_bucket_val / 8;
at_most -= (at_most % base); /* round down */ at_most -= (at_most % base); /* round down */
if (at_most > num_bytes_high) /* 16 KB, or 8 KB for low-priority */ if (at_most > num_bytes_high) /* 16 KB, or 8 KB for low-priority */
at_most = num_bytes_high; at_most = num_bytes_high;
else if (at_most < num_bytes_low) /* 2 KB, or 1 KB for low-priority */ else if (at_most < num_bytes_low) /* 2 KB, or 1 KB for low-priority */
at_most = num_bytes_low; at_most = num_bytes_low;
if (at_most > global_bucket) if (at_most > global_bucket_val)
at_most = global_bucket; at_most = global_bucket_val;
if (conn_bucket >= 0 && at_most > conn_bucket) if (conn_bucket >= 0 && at_most > conn_bucket)
at_most = conn_bucket; at_most = conn_bucket;
@ -2880,7 +2880,7 @@ connection_bucket_read_limit(connection_t *conn, time_t now)
int base = RELAY_PAYLOAD_SIZE; int base = RELAY_PAYLOAD_SIZE;
int priority = conn->type != CONN_TYPE_DIR; int priority = conn->type != CONN_TYPE_DIR;
int conn_bucket = -1; int conn_bucket = -1;
int global_bucket = global_read_bucket; int global_bucket_val = (int) token_bucket_get_read(&global_bucket);
if (connection_speaks_cells(conn)) { if (connection_speaks_cells(conn)) {
or_connection_t *or_conn = TO_OR_CONN(conn); or_connection_t *or_conn = TO_OR_CONN(conn);
@ -2894,12 +2894,13 @@ connection_bucket_read_limit(connection_t *conn, time_t now)
return conn_bucket>=0 ? conn_bucket : 1<<14; return conn_bucket>=0 ? conn_bucket : 1<<14;
} }
if (connection_counts_as_relayed_traffic(conn, now) && if (connection_counts_as_relayed_traffic(conn, now)) {
global_relayed_read_bucket <= global_read_bucket) int relayed = token_bucket_get_read(&global_relayed_bucket);
global_bucket = global_relayed_read_bucket; global_bucket_val = MIN(global_bucket_val, relayed);
}
return connection_bucket_round_robin(base, priority, return connection_bucket_round_robin(base, priority,
global_bucket, conn_bucket); global_bucket_val, conn_bucket);
} }
/** How many bytes at most can we write onto this connection? */ /** How many bytes at most can we write onto this connection? */
@ -2909,7 +2910,7 @@ connection_bucket_write_limit(connection_t *conn, time_t now)
int base = RELAY_PAYLOAD_SIZE; int base = RELAY_PAYLOAD_SIZE;
int priority = conn->type != CONN_TYPE_DIR; int priority = conn->type != CONN_TYPE_DIR;
int conn_bucket = (int)conn->outbuf_flushlen; int conn_bucket = (int)conn->outbuf_flushlen;
int global_bucket = global_write_bucket; int global_bucket_val = (int) token_bucket_get_write(&global_bucket);
if (!connection_is_rate_limited(conn)) { if (!connection_is_rate_limited(conn)) {
/* be willing to write to local conns even if our buckets are empty */ /* be willing to write to local conns even if our buckets are empty */
@ -2924,12 +2925,13 @@ connection_bucket_write_limit(connection_t *conn, time_t now)
base = get_cell_network_size(or_conn->wide_circ_ids); base = get_cell_network_size(or_conn->wide_circ_ids);
} }
if (connection_counts_as_relayed_traffic(conn, now) && if (connection_counts_as_relayed_traffic(conn, now)) {
global_relayed_write_bucket <= global_write_bucket) int relayed = token_bucket_get_write(&global_relayed_bucket);
global_bucket = global_relayed_write_bucket; global_bucket_val = MIN(global_bucket_val, relayed);
}
return connection_bucket_round_robin(base, priority, return connection_bucket_round_robin(base, priority,
global_bucket, conn_bucket); global_bucket_val, conn_bucket);
} }
/** Return 1 if the global write buckets are low enough that we /** Return 1 if the global write buckets are low enough that we
@ -2954,8 +2956,8 @@ connection_bucket_write_limit(connection_t *conn, time_t now)
int int
global_write_bucket_low(connection_t *conn, size_t attempt, int priority) global_write_bucket_low(connection_t *conn, size_t attempt, int priority)
{ {
int smaller_bucket = global_write_bucket < global_relayed_write_bucket ? int smaller_bucket = MIN(token_bucket_get_write(&global_bucket),
global_write_bucket : global_relayed_write_bucket; token_bucket_get_write(&global_relayed_bucket));
if (authdir_mode(get_options()) && priority>1) if (authdir_mode(get_options()) && priority>1)
return 0; /* there's always room to answer v2 if we're an auth dir */ return 0; /* there's always room to answer v2 if we're an auth dir */
@ -3039,11 +3041,9 @@ connection_buckets_decrement(connection_t *conn, time_t now,
return; /* local IPs are free */ return; /* local IPs are free */
if (connection_counts_as_relayed_traffic(conn, now)) { if (connection_counts_as_relayed_traffic(conn, now)) {
global_relayed_read_bucket -= (int)num_read; token_bucket_dec(&global_relayed_bucket, num_read, num_written);
global_relayed_write_bucket -= (int)num_written;
} }
global_read_bucket -= (int)num_read; token_bucket_dec(&global_bucket, num_read, num_written);
global_write_bucket -= (int)num_written;
if (connection_speaks_cells(conn) && conn->state == OR_CONN_STATE_OPEN) { if (connection_speaks_cells(conn) && conn->state == OR_CONN_STATE_OPEN) {
or_connection_t *or_conn = TO_OR_CONN(conn); or_connection_t *or_conn = TO_OR_CONN(conn);
token_bucket_dec(&or_conn->bucket, num_read, num_written); token_bucket_dec(&or_conn->bucket, num_read, num_written);
@ -3060,10 +3060,10 @@ connection_consider_empty_read_buckets(connection_t *conn)
if (!connection_is_rate_limited(conn)) if (!connection_is_rate_limited(conn))
return; /* Always okay. */ return; /* Always okay. */
if (global_read_bucket <= 0) { if (token_bucket_get_read(&global_bucket) <= 0) {
reason = "global read bucket exhausted. Pausing."; reason = "global read bucket exhausted. Pausing.";
} else if (connection_counts_as_relayed_traffic(conn, approx_time()) && } else if (connection_counts_as_relayed_traffic(conn, approx_time()) &&
global_relayed_read_bucket <= 0) { token_bucket_get_read(&global_relayed_bucket) <= 0) {
reason = "global relayed read bucket exhausted. Pausing."; reason = "global relayed read bucket exhausted. Pausing.";
} else if (connection_speaks_cells(conn) && } else if (connection_speaks_cells(conn) &&
conn->state == OR_CONN_STATE_OPEN && conn->state == OR_CONN_STATE_OPEN &&
@ -3087,10 +3087,10 @@ connection_consider_empty_write_buckets(connection_t *conn)
if (!connection_is_rate_limited(conn)) if (!connection_is_rate_limited(conn))
return; /* Always okay. */ return; /* Always okay. */
if (global_write_bucket <= 0) { if (token_bucket_get_write(&global_bucket) <= 0) {
reason = "global write bucket exhausted. Pausing."; reason = "global write bucket exhausted. Pausing.";
} else if (connection_counts_as_relayed_traffic(conn, approx_time()) && } else if (connection_counts_as_relayed_traffic(conn, approx_time()) &&
global_relayed_write_bucket <= 0) { token_bucket_get_write(&global_relayed_bucket) <= 0) {
reason = "global relayed write bucket exhausted. Pausing."; reason = "global relayed write bucket exhausted. Pausing.";
} else if (connection_speaks_cells(conn) && } else if (connection_speaks_cells(conn) &&
conn->state == OR_CONN_STATE_OPEN && conn->state == OR_CONN_STATE_OPEN &&
@ -3109,88 +3109,37 @@ void
connection_bucket_init(void) connection_bucket_init(void)
{ {
const or_options_t *options = get_options(); const or_options_t *options = get_options();
/* start it at max traffic */ const uint32_t now_ts = monotime_coarse_get_stamp();
global_read_bucket = (int)options->BandwidthBurst; token_bucket_init(&global_bucket,
global_write_bucket = (int)options->BandwidthBurst; (int32_t)options->BandwidthRate,
(int32_t)options->BandwidthBurst,
now_ts);
if (options->RelayBandwidthRate) { if (options->RelayBandwidthRate) {
global_relayed_read_bucket = (int)options->RelayBandwidthBurst; token_bucket_init(&global_relayed_bucket,
global_relayed_write_bucket = (int)options->RelayBandwidthBurst; (int32_t)options->RelayBandwidthRate,
(int32_t)options->RelayBandwidthBurst,
now_ts);
} else { } else {
global_relayed_read_bucket = (int)options->BandwidthBurst; token_bucket_init(&global_relayed_bucket,
global_relayed_write_bucket = (int)options->BandwidthBurst; (int32_t)options->BandwidthRate,
} (int32_t)options->BandwidthBurst,
} now_ts);
/** Refill a single <b>bucket</b> called <b>name</b> with bandwidth rate per
* second <b>rate</b> and bandwidth burst <b>burst</b>, assuming that
* <b>milliseconds_elapsed</b> milliseconds have passed since the last
* call. */
static void
connection_bucket_refill_helper(int *bucket, int rate, int burst,
int milliseconds_elapsed,
const char *name)
{
int starting_bucket = *bucket;
if (starting_bucket < burst && milliseconds_elapsed > 0) {
int64_t incr = (((int64_t)rate) * milliseconds_elapsed) / 1000;
if ((burst - starting_bucket) < incr) {
*bucket = burst; /* We would overflow the bucket; just set it to
* the maximum. */
} else {
*bucket += (int)incr;
if (*bucket > burst || *bucket < starting_bucket) {
/* If we overflow the burst, or underflow our starting bucket,
* cap the bucket value to burst. */
/* XXXX this might be redundant now, but it doesn't show up
* in profiles. Remove it after analysis. */
*bucket = burst;
}
}
log_debug(LD_NET,"%s now %d.", name, *bucket);
} }
} }
/** Time has passed; increment buckets appropriately. */ /** Time has passed; increment buckets appropriately. */
void void
connection_bucket_refill(int milliseconds_elapsed, time_t now, uint32_t now_ts) connection_bucket_refill(time_t now, uint32_t now_ts)
{ {
const or_options_t *options = get_options();
smartlist_t *conns = get_connection_array(); smartlist_t *conns = get_connection_array();
int bandwidthrate, bandwidthburst, relayrate, relayburst;
bandwidthrate = (int)options->BandwidthRate;
bandwidthburst = (int)options->BandwidthBurst;
if (options->RelayBandwidthRate) {
relayrate = (int)options->RelayBandwidthRate;
relayburst = (int)options->RelayBandwidthBurst;
} else {
relayrate = bandwidthrate;
relayburst = bandwidthburst;
}
tor_assert(milliseconds_elapsed >= 0);
write_buckets_empty_last_second = write_buckets_empty_last_second =
global_relayed_write_bucket <= 0 || global_write_bucket <= 0; token_bucket_get_write(&global_bucket) <= 0 ||
token_bucket_get_write(&global_relayed_bucket) <= 0;
/* refill the global buckets */ /* refill the global buckets */
connection_bucket_refill_helper(&global_read_bucket, token_bucket_refill(&global_bucket, now_ts);
bandwidthrate, bandwidthburst, token_bucket_refill(&global_relayed_bucket, now_ts);
milliseconds_elapsed,
"global_read_bucket");
connection_bucket_refill_helper(&global_write_bucket,
bandwidthrate, bandwidthburst,
milliseconds_elapsed,
"global_write_bucket");
connection_bucket_refill_helper(&global_relayed_read_bucket,
relayrate, relayburst,
milliseconds_elapsed,
"global_relayed_read_bucket");
connection_bucket_refill_helper(&global_relayed_write_bucket,
relayrate, relayburst,
milliseconds_elapsed,
"global_relayed_write_bucket");
/* refill the per-connection buckets */ /* refill the per-connection buckets */
SMARTLIST_FOREACH_BEGIN(conns, connection_t *, conn) { SMARTLIST_FOREACH_BEGIN(conns, connection_t *, conn) {
@ -3203,9 +3152,9 @@ connection_bucket_refill(int milliseconds_elapsed, time_t now, uint32_t now_ts)
} }
if (conn->read_blocked_on_bw == 1 /* marked to turn reading back on now */ if (conn->read_blocked_on_bw == 1 /* marked to turn reading back on now */
&& global_read_bucket > 0 /* and we're allowed to read */ && token_bucket_get_read(&global_bucket) > 0 /* and we can read */
&& (!connection_counts_as_relayed_traffic(conn, now) || && (!connection_counts_as_relayed_traffic(conn, now) ||
global_relayed_read_bucket > 0) /* even if we're relayed traffic */ token_bucket_get_read(&global_relayed_bucket) > 0)
&& (!connection_speaks_cells(conn) || && (!connection_speaks_cells(conn) ||
conn->state != OR_CONN_STATE_OPEN || conn->state != OR_CONN_STATE_OPEN ||
token_bucket_get_read(&TO_OR_CONN(conn)->bucket) > 0)) { token_bucket_get_read(&TO_OR_CONN(conn)->bucket) > 0)) {
@ -3217,9 +3166,9 @@ connection_bucket_refill(int milliseconds_elapsed, time_t now, uint32_t now_ts)
} }
if (conn->write_blocked_on_bw == 1 if (conn->write_blocked_on_bw == 1
&& global_write_bucket > 0 /* and we're allowed to write */ && token_bucket_get_write(&global_bucket) > 0 /* and we can write */
&& (!connection_counts_as_relayed_traffic(conn, now) || && (!connection_counts_as_relayed_traffic(conn, now) ||
global_relayed_write_bucket > 0) /* even if it's relayed traffic */ token_bucket_get_write(&global_relayed_bucket) > 0)
&& (!connection_speaks_cells(conn) || && (!connection_speaks_cells(conn) ||
conn->state != OR_CONN_STATE_OPEN || conn->state != OR_CONN_STATE_OPEN ||
token_bucket_get_write(&TO_OR_CONN(conn)->bucket) > 0)) { token_bucket_get_write(&TO_OR_CONN(conn)->bucket) > 0)) {

2
src/or/connection.h
View File

@ -122,7 +122,7 @@ void connection_mark_all_noncontrol_connections(void);
ssize_t connection_bucket_write_limit(connection_t *conn, time_t now); ssize_t connection_bucket_write_limit(connection_t *conn, time_t now);
int global_write_bucket_low(connection_t *conn, size_t attempt, int priority); int global_write_bucket_low(connection_t *conn, size_t attempt, int priority);
void connection_bucket_init(void); void connection_bucket_init(void);
void connection_bucket_refill(int seconds_elapsed, time_t now, void connection_bucket_refill(time_t now,
uint32_t now_ts); uint32_t now_ts);
int connection_handle_read(connection_t *conn); int connection_handle_read(connection_t *conn);

36
src/or/main.c
View File

@ -152,15 +152,15 @@ static void shutdown_did_not_work_callback(evutil_socket_t fd, short event,
void *arg) ATTR_NORETURN; void *arg) ATTR_NORETURN;
/********* START VARIABLES **********/ /********* START VARIABLES **********/
int global_read_bucket; /**< Max number of bytes I can read this second. */
int global_write_bucket; /**< Max number of bytes I can write this second. */
/** Max number of relayed (bandwidth class 1) bytes I can read this second. */ /* Token bucket for all traffic. */
int global_relayed_read_bucket; token_bucket_t global_bucket;
/** Max number of relayed (bandwidth class 1) bytes I can write this second. */
int global_relayed_write_bucket; /* Token bucket for relayed traffic. */
/** What was the read bucket before the last second_elapsed_callback() call? token_bucket_t global_relayed_bucket;
* (used to determine how many bytes we've read). */
/** What was the read/write bucket before the last second_elapsed_callback()
* call? (used to determine how many bytes we've read). */
static int stats_prev_global_read_bucket; static int stats_prev_global_read_bucket;
/** What was the write bucket before the last second_elapsed_callback() call? /** What was the write bucket before the last second_elapsed_callback() call?
* (used to determine how many bytes we've written). */ * (used to determine how many bytes we've written). */
@ -2418,8 +2418,10 @@ refill_callback(periodic_timer_t *timer, void *arg)
refill_timer_current_millisecond.tv_sec); refill_timer_current_millisecond.tv_sec);
} }
bytes_written = stats_prev_global_write_bucket - global_write_bucket; bytes_written = stats_prev_global_write_bucket -
bytes_read = stats_prev_global_read_bucket - global_read_bucket; token_bucket_get_write(&global_bucket);
bytes_read = stats_prev_global_read_bucket -
token_bucket_get_read(&global_bucket);
stats_n_bytes_read += bytes_read; stats_n_bytes_read += bytes_read;
stats_n_bytes_written += bytes_written; stats_n_bytes_written += bytes_written;
@ -2427,12 +2429,12 @@ refill_callback(periodic_timer_t *timer, void *arg)
accounting_add_bytes(bytes_read, bytes_written, seconds_rolled_over); accounting_add_bytes(bytes_read, bytes_written, seconds_rolled_over);
if (milliseconds_elapsed > 0) { if (milliseconds_elapsed > 0) {
connection_bucket_refill(milliseconds_elapsed, (time_t)now.tv_sec, connection_bucket_refill((time_t)now.tv_sec,
monotime_coarse_get_stamp()); monotime_coarse_get_stamp());
} }
stats_prev_global_read_bucket = global_read_bucket; stats_prev_global_read_bucket = token_bucket_get_read(&global_bucket);
stats_prev_global_write_bucket = global_write_bucket; stats_prev_global_write_bucket = token_bucket_get_write(&global_bucket);
/* remember what time it is, for next time */ /* remember what time it is, for next time */
refill_timer_current_millisecond = now; refill_timer_current_millisecond = now;
@ -2636,8 +2638,8 @@ do_main_loop(void)
/* Set up our buckets */ /* Set up our buckets */
connection_bucket_init(); connection_bucket_init();
stats_prev_global_read_bucket = global_read_bucket; stats_prev_global_read_bucket = token_bucket_get_read(&global_bucket);
stats_prev_global_write_bucket = global_write_bucket; stats_prev_global_write_bucket = token_bucket_get_write(&global_bucket);
/* initialize the bootstrap status events to know we're starting up */ /* initialize the bootstrap status events to know we're starting up */
control_event_bootstrap(BOOTSTRAP_STATUS_STARTING, 0); control_event_bootstrap(BOOTSTRAP_STATUS_STARTING, 0);
@ -3532,8 +3534,8 @@ tor_free_all(int postfork)
periodic_timer_free(systemd_watchdog_timer); periodic_timer_free(systemd_watchdog_timer);
#endif #endif
global_read_bucket = global_write_bucket = 0; memset(&global_bucket, 0, sizeof(global_bucket));
global_relayed_read_bucket = global_relayed_write_bucket = 0; memset(&global_relayed_bucket, 0, sizeof(global_relayed_bucket));
stats_prev_global_read_bucket = stats_prev_global_write_bucket = 0; stats_prev_global_read_bucket = stats_prev_global_write_bucket = 0;
stats_prev_n_read = stats_prev_n_written = 0; stats_prev_n_read = stats_prev_n_written = 0;
stats_n_bytes_read = stats_n_bytes_written = 0; stats_n_bytes_read = stats_n_bytes_written = 0;

6
src/or/main.h
View File

@ -89,10 +89,8 @@ uint64_t get_main_loop_idle_count(void);
extern time_t time_of_process_start; extern time_t time_of_process_start;
extern int quiet_level; extern int quiet_level;
extern int global_read_bucket; extern token_bucket_t global_bucket;
extern int global_write_bucket; extern token_bucket_t global_relayed_bucket;
extern int global_relayed_read_bucket;
extern int global_relayed_write_bucket;
#ifdef MAIN_PRIVATE #ifdef MAIN_PRIVATE
STATIC void init_connection_lists(void); STATIC void init_connection_lists(void);