commit a38fd9bc5bf508d029ff2557311caeb487638968 Author: Nick Mathewson <nickm@torproject.org> Date: Tue Apr 10 13:19:25 2018 -0400 Replace the global buckets with token_bucket_t --- src/or/connection.c | 143 +++++++++++++++++----------------------------------- src/or/connection.h | 2 +- src/or/main.c | 38 +++++++------- src/or/main.h | 6 +-- 4 files changed, 69 insertions(+), 120 deletions(-) diff --git a/src/or/connection.c b/src/or/connection.c index 4013e0538..b2ef28776 100644 --- a/src/or/connection.c +++ b/src/or/connection.c @@ -2846,7 +2846,7 @@ connection_counts_as_relayed_traffic(connection_t *conn, time_t now) * non-negative) provides an upper limit for our answer. */ static ssize_t 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 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. * 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. */ - at_most = global_bucket / 8; + at_most = global_bucket_val / 8; at_most -= (at_most % base); /* round down */ if (at_most > num_bytes_high) /* 16 KB, or 8 KB for low-priority */ at_most = num_bytes_high; else if (at_most < num_bytes_low) /* 2 KB, or 1 KB for low-priority */ at_most = num_bytes_low; - if (at_most > global_bucket) - at_most = global_bucket; + if (at_most > global_bucket_val) + at_most = global_bucket_val; if (conn_bucket >= 0 && 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 priority = conn->type != CONN_TYPE_DIR; 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)) { 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; } - if (connection_counts_as_relayed_traffic(conn, now) && - global_relayed_read_bucket <= global_read_bucket) - global_bucket = global_relayed_read_bucket; + if (connection_counts_as_relayed_traffic(conn, now)) { + int relayed = token_bucket_get_read(&global_relayed_bucket); + global_bucket_val = MIN(global_bucket_val, relayed); + } 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? */ @@ -2909,7 +2910,7 @@ connection_bucket_write_limit(connection_t *conn, time_t now) int base = RELAY_PAYLOAD_SIZE; int priority = conn->type != CONN_TYPE_DIR; 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)) { /* 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); } - if (connection_counts_as_relayed_traffic(conn, now) && - global_relayed_write_bucket <= global_write_bucket) - global_bucket = global_relayed_write_bucket; + if (connection_counts_as_relayed_traffic(conn, now)) { + int relayed = token_bucket_get_write(&global_relayed_bucket); + global_bucket_val = MIN(global_bucket_val, relayed); + } 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 @@ -2954,8 +2956,8 @@ connection_bucket_write_limit(connection_t *conn, time_t now) int global_write_bucket_low(connection_t *conn, size_t attempt, int priority) { - int smaller_bucket = global_write_bucket < global_relayed_write_bucket ? - global_write_bucket : global_relayed_write_bucket; + int smaller_bucket = MIN(token_bucket_get_write(&global_bucket), + token_bucket_get_write(&global_relayed_bucket)); if (authdir_mode(get_options()) && priority>1) 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 */ if (connection_counts_as_relayed_traffic(conn, now)) { - global_relayed_read_bucket -= (int)num_read; - global_relayed_write_bucket -= (int)num_written; + token_bucket_dec(&global_relayed_bucket, num_read, num_written); } - global_read_bucket -= (int)num_read; - global_write_bucket -= (int)num_written; + token_bucket_dec(&global_bucket, num_read, num_written); if (connection_speaks_cells(conn) && conn->state == OR_CONN_STATE_OPEN) { or_connection_t *or_conn = TO_OR_CONN(conn); 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)) return; /* Always okay. */ - if (global_read_bucket <= 0) { + if (token_bucket_get_read(&global_bucket) <= 0) { reason = "global read bucket exhausted. Pausing."; } 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."; } else if (connection_speaks_cells(conn) && conn->state == OR_CONN_STATE_OPEN && @@ -3087,10 +3087,10 @@ connection_consider_empty_write_buckets(connection_t *conn) if (!connection_is_rate_limited(conn)) return; /* Always okay. */ - if (global_write_bucket <= 0) { + if (token_bucket_get_write(&global_bucket) <= 0) { reason = "global write bucket exhausted. Pausing."; } 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."; } else if (connection_speaks_cells(conn) && conn->state == OR_CONN_STATE_OPEN && @@ -3109,88 +3109,37 @@ void connection_bucket_init(void) { const or_options_t *options = get_options(); - /* start it at max traffic */ - global_read_bucket = (int)options->BandwidthBurst; - global_write_bucket = (int)options->BandwidthBurst; + const uint32_t now_ts = monotime_coarse_get_stamp(); + token_bucket_init(&global_bucket, + (int32_t)options->BandwidthRate, + (int32_t)options->BandwidthBurst, + now_ts); if (options->RelayBandwidthRate) { - global_relayed_read_bucket = (int)options->RelayBandwidthBurst; - global_relayed_write_bucket = (int)options->RelayBandwidthBurst; + token_bucket_init(&global_relayed_bucket, + (int32_t)options->RelayBandwidthRate, + (int32_t)options->RelayBandwidthBurst, + now_ts); } else { - global_relayed_read_bucket = (int)options->BandwidthBurst; - global_relayed_write_bucket = (int)options->BandwidthBurst; - } -} - -/** 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); + token_bucket_init(&global_relayed_bucket, + (int32_t)options->BandwidthRate, + (int32_t)options->BandwidthBurst, + now_ts); } } /** Time has passed; increment buckets appropriately. */ 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(); - 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 = - 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 */ - connection_bucket_refill_helper(&global_read_bucket, - bandwidthrate, bandwidthburst, - 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"); + token_bucket_refill(&global_bucket, now_ts); + token_bucket_refill(&global_relayed_bucket, now_ts); /* refill the per-connection buckets */ 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 */ - && 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) || - global_relayed_read_bucket > 0) /* even if we're relayed traffic */ + token_bucket_get_read(&global_relayed_bucket) > 0) && (!connection_speaks_cells(conn) || conn->state != OR_CONN_STATE_OPEN || 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 - && 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) || - global_relayed_write_bucket > 0) /* even if it's relayed traffic */ + token_bucket_get_write(&global_relayed_bucket) > 0) && (!connection_speaks_cells(conn) || conn->state != OR_CONN_STATE_OPEN || token_bucket_get_write(&TO_OR_CONN(conn)->bucket) > 0)) { diff --git a/src/or/connection.h b/src/or/connection.h index acc63640d..4a57bd311 100644 --- a/src/or/connection.h +++ b/src/or/connection.h @@ -122,7 +122,7 @@ void connection_mark_all_noncontrol_connections(void); 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); 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); int connection_handle_read(connection_t *conn); diff --git a/src/or/main.c b/src/or/main.c index b12effce1..f59ee24a7 100644 --- a/src/or/main.c +++ b/src/or/main.c @@ -152,15 +152,15 @@ static void shutdown_did_not_work_callback(evutil_socket_t fd, short event, void *arg) ATTR_NORETURN; /********* 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. */ -int global_relayed_read_bucket; -/** Max number of relayed (bandwidth class 1) bytes I can write this second. */ -int global_relayed_write_bucket; -/** What was the read bucket before the last second_elapsed_callback() call? - * (used to determine how many bytes we've read). */ + +/* Token bucket for all traffic. */ +token_bucket_t global_bucket; + +/* Token bucket for relayed traffic. */ +token_bucket_t global_relayed_bucket; + +/** 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; /** What was the write bucket before the last second_elapsed_callback() call? * (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); } - bytes_written = stats_prev_global_write_bucket - global_write_bucket; - bytes_read = stats_prev_global_read_bucket - global_read_bucket; + bytes_written = stats_prev_global_write_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_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); 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()); } - stats_prev_global_read_bucket = global_read_bucket; - stats_prev_global_write_bucket = global_write_bucket; + stats_prev_global_read_bucket = token_bucket_get_read(&global_bucket); + stats_prev_global_write_bucket = token_bucket_get_write(&global_bucket); /* remember what time it is, for next time */ refill_timer_current_millisecond = now; @@ -2636,8 +2638,8 @@ do_main_loop(void) /* Set up our buckets */ connection_bucket_init(); - stats_prev_global_read_bucket = global_read_bucket; - stats_prev_global_write_bucket = global_write_bucket; + stats_prev_global_read_bucket = token_bucket_get_read(&global_bucket); + stats_prev_global_write_bucket = token_bucket_get_write(&global_bucket); /* initialize the bootstrap status events to know we're starting up */ control_event_bootstrap(BOOTSTRAP_STATUS_STARTING, 0); @@ -3532,8 +3534,8 @@ tor_free_all(int postfork) periodic_timer_free(systemd_watchdog_timer); #endif - global_read_bucket = global_write_bucket = 0; - global_relayed_read_bucket = global_relayed_write_bucket = 0; + memset(&global_bucket, 0, sizeof(global_bucket)); + memset(&global_relayed_bucket, 0, sizeof(global_relayed_bucket)); stats_prev_global_read_bucket = stats_prev_global_write_bucket = 0; stats_prev_n_read = stats_prev_n_written = 0; stats_n_bytes_read = stats_n_bytes_written = 0; diff --git a/src/or/main.h b/src/or/main.h index f01506fce..9ef5b9472 100644 --- a/src/or/main.h +++ b/src/or/main.h @@ -89,10 +89,8 @@ uint64_t get_main_loop_idle_count(void); extern time_t time_of_process_start; extern int quiet_level; -extern int global_read_bucket; -extern int global_write_bucket; -extern int global_relayed_read_bucket; -extern int global_relayed_write_bucket; +extern token_bucket_t global_bucket; +extern token_bucket_t global_relayed_bucket; #ifdef MAIN_PRIVATE STATIC void init_connection_lists(void);