November 2014 - tor-commits - lists.torproject.org

[tor/master] Add cmux support for inter-cmux comparisons
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit 9db596d2ef441d9293f5341be28150739baac98d Author: Andrea Shepard <andrea(a)torproject.org> Date: Sat Nov 30 01:25:20 2013 -0800 Add cmux support for inter-cmux comparisons --- src/or/circuitmux.c | 48 ++++++++++++++++++++++++++++++++++++++++++++++++ src/or/circuitmux.h | 6 ++++++ 2 files changed, 54 insertions(+) diff --git a/src/or/circuitmux.c b/src/or/circuitmux.c index 3ca33b0..29c3be1 100644 --- a/src/or/circuitmux.c +++ b/src/or/circuitmux.c @@ -1951,3 +1951,51 @@ circuitmux_count_queued_destroy_cells(const channel_t *chan, return n_destroy_cells; } +/** + * Compare cmuxes to see which is more preferred; return < 0 if + * cmux_1 has higher priority (i.e., cmux_1 < cmux_2 in the scheduler's + * sort order), > 0 if cmux_2 has higher priority, or 0 if they are + * equally preferred. + * + * If the cmuxes have different cmux policies or the policy does not + * support the cmp_cmux method, return 0. + */ + +int +circuitmux_compare_muxes(circuitmux_t *cmux_1, circuitmux_t *cmux_2) +{ + const circuitmux_policy_t *policy; + + tor_assert(cmux_1); + tor_assert(cmux_2); + + if (cmux_1 == cmux_2) { + /* Equivalent because they're the same cmux */ + return 0; + } + + if (cmux_1->policy && cmux_2->policy) { + if (cmux_1->policy == cmux_2->policy) { + policy = cmux_1->policy; + + if (policy->cmp_cmux) { + /* Okay, we can compare! */ + return policy->cmp_cmux(cmux_1, cmux_1->policy_data, + cmux_2, cmux_2->policy_data); + } else { + /* + * Equivalent because the policy doesn't know how to compare between + * muxes. + */ + return 0; + } + } else { + /* Equivalent because they have different policies */ + return 0; + } + } else { + /* Equivalent because one or both are missing a policy */ + return 0; + } +} + diff --git a/src/or/circuitmux.h b/src/or/circuitmux.h index 2b5fb7e..5833ee5 100644 --- a/src/or/circuitmux.h +++ b/src/or/circuitmux.h @@ -57,6 +57,9 @@ struct circuitmux_policy_s { /* Choose a circuit */ circuit_t * (*pick_active_circuit)(circuitmux_t *cmux, circuitmux_policy_data_t *pol_data); + /* Optional: channel comparator for use by the scheduler */ + int (*cmp_cmux)(circuitmux_t *cmux_1, circuitmux_policy_data_t *pol_data_1, + circuitmux_t *cmux_2, circuitmux_policy_data_t *pol_data_2); }; /* @@ -148,5 +151,8 @@ void circuitmux_append_destroy_cell(channel_t *chan, void circuitmux_mark_destroyed_circids_usable(circuitmux_t *cmux, channel_t *chan); +/* Optional interchannel comparisons for scheduling */ +int circuitmux_compare_muxes(circuitmux_t *cmux_1, circuitmux_t *cmux_2); + #endif /* TOR_CIRCUITMUX_H */

1 0

[tor/master] Let the new scheduler handle writes
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit 4f567c8cc8cd68a8ca9bb93fc57d518d7eb55cf0 Author: Andrea Shepard <andrea(a)torproject.org> Date: Thu Nov 14 04:45:16 2013 -0800 Let the new scheduler handle writes --- src/or/relay.c | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/src/or/relay.c b/src/or/relay.c index 2c16e4a..b653616 100644 --- a/src/or/relay.c +++ b/src/or/relay.c @@ -2871,16 +2871,6 @@ append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan, /* New way: mark this as having waiting cells for the scheduler */ scheduler_channel_has_waiting_cells(chan); - - /* TODO remove this once scheduler does it */ - if (!channel_has_queued_writes(chan)) { - /* There is no data at all waiting to be sent on the outbuf. Add a - * cell, so that we can notice when it gets flushed, flushed_some can - * get called, and we can start putting more data onto the buffer then. - */ - log_debug(LD_GENERAL, "Primed a buffer."); - channel_flush_from_first_active_circuit(chan, 1); - } } /** Append an encoded value of <b>addr</b> to <b>payload_out</b>, which must

1 0

[tor/master] Fix return values from channel_flush_some_cells() to correctly count cells directly written by channel_flush_from_first_active_circuit()
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit f314d9509cae7acb9a692895c709e7771c6321f0 Author: Andrea Shepard <andrea(a)torproject.org> Date: Wed Nov 13 16:50:26 2013 -0800 Fix return values from channel_flush_some_cells() to correctly count cells directly written by channel_flush_from_first_active_circuit() --- src/or/channel.c | 38 +++++++++++++++++++++++++++++++++++--- 1 file changed, 35 insertions(+), 3 deletions(-) diff --git a/src/or/channel.c b/src/or/channel.c index e2d102d..dddd7ab 100644 --- a/src/or/channel.c +++ b/src/or/channel.c @@ -175,6 +175,7 @@ static cell_queue_entry_t * cell_queue_entry_new_fixed(cell_t *cell); static cell_queue_entry_t * cell_queue_entry_new_var(var_cell_t *var_cell); +static int chan_cell_queue_len(const chan_cell_queue_t *queue); static int is_destroy_cell(channel_t *chan, const cell_queue_entry_t *q, circid_t *circid_out); @@ -2218,6 +2219,7 @@ channel_flush_some_cells(channel_t *chan, ssize_t num_cells) unsigned int unlimited = 0; ssize_t flushed = 0; int num_cells_from_circs, clamped_num_cells; + int q_len_before, q_len_after; tor_assert(chan); @@ -2243,14 +2245,44 @@ channel_flush_some_cells(channel_t *chan, ssize_t num_cells) clamped_num_cells = (int)(num_cells - flushed); } } + + /* + * Keep track of the change in queue size; we have to count cells + * channel_flush_from_first_active_circuit() writes out directly, + * but not double-count ones we might get later in + * channel_flush_some_cells_from_outgoing_queue() + */ + q_len_before = chan_cell_queue_len(&(chan->outgoing_queue)); + /* Try to get more cells from any active circuits */ num_cells_from_circs = channel_flush_from_first_active_circuit( chan, clamped_num_cells); - /* If it claims we got some, process the queue again */ + q_len_after = chan_cell_queue_len(&(chan->outgoing_queue)); + + /* + * If it claims we got some, adjust the flushed counter and consider + * processing the queue again + */ if (num_cells_from_circs > 0) { - flushed += channel_flush_some_cells_from_outgoing_queue(chan, - (unlimited ? -1 : num_cells - flushed)); + /* + * Adjust flushed by the number of cells counted in + * num_cells_from_circs that didn't go to the cell queue. + */ + + if (q_len_after > q_len_before) { + num_cells_from_circs -= (q_len_after - q_len_before); + if (num_cells_from_circs < 0) num_cells_from_circs = 0; + } + + flushed += num_cells_from_circs; + + /* Now process the queue if necessary */ + + if (q_len_after > q_len_before && num_cells < flushed) { + flushed += channel_flush_some_cells_from_outgoing_queue(chan, + (unlimited ? -1 : num_cells - flushed)); + } } } }

1 0

[tor/master] Implement global queue size query in channel.c
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit 2fc3da3ff51f3fe7fa5338c2d6b52a06ed9c4f19 Author: Andrea Shepard <andrea(a)torproject.org> Date: Mon Nov 11 21:50:16 2013 -0800 Implement global queue size query in channel.c --- src/or/channel.c | 10 ++++++++++ src/or/channel.h | 1 + 2 files changed, 11 insertions(+) diff --git a/src/or/channel.c b/src/or/channel.c index f729a17..e2d102d 100644 --- a/src/or/channel.c +++ b/src/or/channel.c @@ -4036,6 +4036,16 @@ channel_mark_outgoing(channel_t *chan) ***********************/ /* + * Get the latest estimate for the total queue size of all open channels + */ + +uint64_t +channel_get_global_queue_estimate(void) +{ + return estimated_total_queue_size; +} + +/* * Estimate the number of writeable cells * * Ask the lower layer for an estimate of how many cells it can accept, and diff --git a/src/or/channel.h b/src/or/channel.h index 388c729..18f7cfc 100644 --- a/src/or/channel.h +++ b/src/or/channel.h @@ -484,6 +484,7 @@ void channel_listener_dump_transport_statistics(channel_listener_t *chan_l, int severity); /* Flow control queries */ +uint64_t channel_get_global_queue_estimate(void); int channel_num_cells_writeable(channel_t *chan); /* Timestamp queries */

1 0

[tor/master] Add global cell/byte counters and per channel byte counters to channel.c
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit 5e0a6d54d0f1341eddc26a6fa6d349e22966c255 Author: Andrea Shepard <andrea(a)torproject.org> Date: Mon Nov 4 19:17:05 2013 -0800 Add global cell/byte counters and per channel byte counters to channel.c --- src/or/channel.c | 165 +++++++++++++++++++++++++++++++++++++++++++++++++++++- src/or/channel.h | 4 +- 2 files changed, 164 insertions(+), 5 deletions(-) diff --git a/src/or/channel.c b/src/or/channel.c index e9451f1..0caebfb 100644 --- a/src/or/channel.c +++ b/src/or/channel.c @@ -77,6 +77,53 @@ static smartlist_t *finished_listeners = NULL; /* Counter for ID numbers */ static uint64_t n_channels_allocated = 0; +/* + * Channel global byte/cell counters, for statistics and for scheduler high + * /low-water marks. + */ + +/* + * Total number of cells ever given to any channel with the + * channel_write_*_cell() functions. + */ + +static uint64_t n_channel_cells_queued = 0; + +/* + * Total number of cells ever passed to a channel lower layer with the + * write_*_cell() methods. + */ + +static uint64_t n_channel_cells_passed_to_lower_layer = 0; + +/* + * Current number of cells in all channel queues; should be + * n_channel_cells_queued - n_channel_cells_passed_to_lower_layer. + */ + +static uint64_t n_channel_cells_in_queues = 0; + +/* + * Total number of bytes for all cells ever queued to a channel and + * counted in n_channel_cells_queued. + */ + +static uint64_t n_channel_bytes_queued = 0; + +/* + * Total number of bytes for all cells ever passed to a channel lower layer + * and counted in n_channel_cells_passed_to_lower_layer. + */ + +static uint64_t n_channel_bytes_passed_to_lower_layer = 0; + +/* + * Current number of bytes in all channel queues; should be + * n_channel_bytes_queued - n_channel_bytes_passed_to_lower_layer. + */ + +static uint64_t n_channel_bytes_in_queues = 0; + /* Digest->channel map * * Similar to the one used in connection_or.c, this maps from the identity @@ -124,6 +171,8 @@ cell_queue_entry_new_var(var_cell_t *var_cell); static int is_destroy_cell(channel_t *chan, const cell_queue_entry_t *q, circid_t *circid_out); +static void channel_assert_counter_consistency(void); + /* Functions to maintain the digest map */ static void channel_add_to_digest_map(channel_t *chan); static void channel_remove_from_digest_map(channel_t *chan); @@ -141,6 +190,8 @@ channel_free_list(smartlist_t *channels, int mark_for_close); static void channel_listener_free_list(smartlist_t *channels, int mark_for_close); static void channel_listener_force_free(channel_listener_t *chan_l); +static size_t channel_get_cell_queue_entry_size(channel_t *chan, + cell_queue_entry_t *q); static void channel_write_cell_queue_entry(channel_t *chan, cell_queue_entry_t *q); @@ -1673,6 +1724,34 @@ cell_queue_entry_new_var(var_cell_t *var_cell) } /** + * Ask how big the cell contained in a cell_queue_entry_t is + */ + +static size_t +channel_get_cell_queue_entry_size(channel_t *chan, cell_queue_entry_t *q) +{ + tor_assert(chan); + tor_assert(q); + + switch (q->type) { + case CELL_QUEUE_FIXED: + return get_cell_network_size(chan->wide_circ_ids); + break; + case CELL_QUEUE_VAR: + tor_assert(q->u.var.var_cell); + return get_var_cell_header_size(chan->wide_circ_ids) + + q->u.var.var_cell->payload_len; + break; + case CELL_QUEUE_PACKED: + return get_cell_network_size(chan->wide_circ_ids); + break; + default: + tor_assert(1); + return 0; + } +} + +/** * Write to a channel based on a cell_queue_entry_t * * Given a cell_queue_entry_t filled out by the caller, try to send the cell @@ -1684,6 +1763,7 @@ channel_write_cell_queue_entry(channel_t *chan, cell_queue_entry_t *q) { int result = 0, sent = 0; cell_queue_entry_t *tmp = NULL; + size_t cell_bytes; tor_assert(chan); tor_assert(q); @@ -1700,6 +1780,9 @@ channel_write_cell_queue_entry(channel_t *chan, cell_queue_entry_t *q) } } + /* For statistical purposes, figure out how big this cell is */ + cell_bytes = channel_get_cell_queue_entry_size(chan, q); + /* Can we send it right out? If so, try */ if (TOR_SIMPLEQ_EMPTY(&chan->outgoing_queue) && chan->state == CHANNEL_STATE_OPEN) { @@ -1733,6 +1816,13 @@ channel_write_cell_queue_entry(channel_t *chan, cell_queue_entry_t *q) channel_timestamp_drained(chan); /* Update the counter */ ++(chan->n_cells_xmitted); + chan->n_bytes_xmitted += cell_bytes; + /* Update global counters */ + ++n_channel_cells_queued; + ++n_channel_cells_passed_to_lower_layer; + n_channel_bytes_queued += cell_bytes; + n_channel_bytes_passed_to_lower_layer += cell_bytes; + channel_assert_counter_consistency(); } } @@ -1744,6 +1834,12 @@ channel_write_cell_queue_entry(channel_t *chan, cell_queue_entry_t *q) */ tmp = cell_queue_entry_dup(q); TOR_SIMPLEQ_INSERT_TAIL(&chan->outgoing_queue, tmp, next); + /* Update global counters */ + ++n_channel_cells_queued; + ++n_channel_cells_in_queues; + n_channel_bytes_queued += cell_bytes; + n_channel_bytes_in_queues += cell_bytes; + channel_assert_counter_consistency(); /* Try to process the queue? */ if (chan->state == CHANNEL_STATE_OPEN) channel_flush_cells(chan); } @@ -2136,6 +2232,7 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan, unsigned int unlimited = 0; ssize_t flushed = 0; cell_queue_entry_t *q = NULL; + size_t cell_size; tor_assert(chan); tor_assert(chan->write_cell); @@ -2151,6 +2248,8 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan, NULL != (q = TOR_SIMPLEQ_FIRST(&chan->outgoing_queue))) { if (1) { + /* Figure out how big it is for statistical purposes */ + cell_size = channel_get_cell_queue_entry_size(chan, q); /* * Okay, we have a good queue entry, try to give it to the lower * layer. @@ -2163,6 +2262,7 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan, ++flushed; channel_timestamp_xmit(chan); ++(chan->n_cells_xmitted); + chan->n_bytes_xmitted += cell_size; cell_queue_entry_free(q, 1); q = NULL; } @@ -2186,6 +2286,7 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan, ++flushed; channel_timestamp_xmit(chan); ++(chan->n_cells_xmitted); + chan->n_bytes_xmitted += cell_size; cell_queue_entry_free(q, 1); q = NULL; } @@ -2209,6 +2310,7 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan, ++flushed; channel_timestamp_xmit(chan); ++(chan->n_cells_xmitted); + chan->n_bytes_xmitted += cell_size; cell_queue_entry_free(q, 1); q = NULL; } @@ -2237,7 +2339,18 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan, } /* if q got NULLed out, we used it and should remove the queue entry */ - if (!q) TOR_SIMPLEQ_REMOVE_HEAD(&chan->outgoing_queue, next); + if (!q) { + TOR_SIMPLEQ_REMOVE_HEAD(&chan->outgoing_queue, next); + /* + * ...and we handed a cell off to the lower layer, so we should + * update the counters. + */ + ++n_channel_cells_passed_to_lower_layer; + --n_channel_cells_in_queues; + n_channel_bytes_passed_to_lower_layer += cell_size; + n_channel_bytes_in_queues -= cell_size; + channel_assert_counter_consistency(); + } /* No cell removed from list, so we can't go on any further */ else break; } @@ -2560,8 +2673,9 @@ channel_queue_cell(channel_t *chan, cell_t *cell) /* Timestamp for receiving */ channel_timestamp_recv(chan); - /* Update the counter */ + /* Update the counters */ ++(chan->n_cells_recved); + chan->n_bytes_recved += get_cell_network_size(chan->wide_circ_ids); /* If we don't need to queue we can just call cell_handler */ if (!need_to_queue) { @@ -2615,6 +2729,8 @@ channel_queue_var_cell(channel_t *chan, var_cell_t *var_cell) /* Update the counter */ ++(chan->n_cells_recved); + chan->n_bytes_recved += get_var_cell_header_size(chan->wide_circ_ids) + + var_cell->payload_len; /* If we don't need to queue we can just call cell_handler */ if (!need_to_queue) { @@ -2664,6 +2780,19 @@ packed_cell_is_destroy(channel_t *chan, return 0; } +/** + * Assert that the global channel stats counters are internally consistent + */ + +static void +channel_assert_counter_consistency(void) +{ + tor_assert(n_channel_cells_queued == + (n_channel_cells_in_queues + n_channel_cells_passed_to_lower_layer)); + tor_assert(n_channel_bytes_queued == + (n_channel_bytes_in_queues + n_channel_bytes_passed_to_lower_layer)); +} + /** DOCDOC */ static int is_destroy_cell(channel_t *chan, @@ -2746,6 +2875,19 @@ channel_dumpstats(int severity) { if (all_channels && smartlist_len(all_channels) > 0) { tor_log(severity, LD_GENERAL, + "Channels have queued " U64_FORMAT " bytes in " U64_FORMAT " cells, " + "and handed " U64_FORMAT " bytes in " U64_FORMAT " cells to the lower" + " layer.", + U64_PRINTF_ARG(n_channel_bytes_queued), + U64_PRINTF_ARG(n_channel_cells_queued), + U64_PRINTF_ARG(n_channel_bytes_passed_to_lower_layer), + U64_PRINTF_ARG(n_channel_cells_passed_to_lower_layer)); + tor_log(severity, LD_GENERAL, + "There are currently " U64_FORMAT " bytes in " U64_FORMAT " cells " + "in channel queues.", + U64_PRINTF_ARG(n_channel_bytes_in_queues), + U64_PRINTF_ARG(n_channel_cells_in_queues)); + tor_log(severity, LD_GENERAL, "Dumping statistics about %d channels:", smartlist_len(all_channels)); tor_log(severity, LD_GENERAL, @@ -3388,12 +3530,22 @@ channel_dump_statistics(channel_t *chan, int severity) /* Describe counters and rates */ tor_log(severity, LD_GENERAL, " * Channel " U64_FORMAT " has received " - U64_FORMAT " cells and transmitted " U64_FORMAT, + U64_FORMAT " bytes in " U64_FORMAT " cells and transmitted " + U64_FORMAT " bytes in " U64_FORMAT " cells", U64_PRINTF_ARG(chan->global_identifier), + U64_PRINTF_ARG(chan->n_bytes_recved), U64_PRINTF_ARG(chan->n_cells_recved), + U64_PRINTF_ARG(chan->n_bytes_xmitted), U64_PRINTF_ARG(chan->n_cells_xmitted)); if (now > chan->timestamp_created && chan->timestamp_created > 0) { + if (chan->n_bytes_recved > 0) { + avg = (double)(chan->n_bytes_recved) / age; + tor_log(severity, LD_GENERAL, + " * Channel " U64_FORMAT " has averaged %f " + "bytes received per second", + U64_PRINTF_ARG(chan->global_identifier), avg); + } if (chan->n_cells_recved > 0) { avg = (double)(chan->n_cells_recved) / age; if (avg >= 1.0) { @@ -3409,6 +3561,13 @@ channel_dump_statistics(channel_t *chan, int severity) U64_PRINTF_ARG(chan->global_identifier), interval); } } + if (chan->n_bytes_xmitted > 0) { + avg = (double)(chan->n_bytes_xmitted) / age; + tor_log(severity, LD_GENERAL, + " * Channel " U64_FORMAT " has averaged %f " + "bytes transmitted per second", + U64_PRINTF_ARG(chan->global_identifier), avg); + } if (chan->n_cells_xmitted > 0) { avg = (double)(chan->n_cells_xmitted) / age; if (avg >= 1.0) { diff --git a/src/or/channel.h b/src/or/channel.h index 28b5ab0..f35cb27 100644 --- a/src/or/channel.h +++ b/src/or/channel.h @@ -200,8 +200,8 @@ struct channel_s { uint64_t dirreq_id; /** Channel counters for cell channels */ - uint64_t n_cells_recved; - uint64_t n_cells_xmitted; + uint64_t n_cells_recved, n_bytes_recved; + uint64_t n_cells_xmitted, n_bytes_xmitted; }; struct channel_listener_s {

1 0

[tor/master] Provide generic mechanism for scheduler to query writeable cells on a channel
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit 2efbab2aaf98d8f8f0df504efd4fdd0fac77d354 Author: Andrea Shepard <andrea(a)torproject.org> Date: Tue Oct 29 02:13:53 2013 -0700 Provide generic mechanism for scheduler to query writeable cells on a channel --- src/or/channel.c | 34 ++++++++++++++++++++++++++++++++++ src/or/channel.h | 7 ++++++- src/or/channeltls.c | 30 ++++++++++++++++++++++++++++++ src/or/connection_or.c | 8 -------- src/or/or.h | 12 ++++++++++++ src/or/scheduler.c | 16 +++++++++++++--- 6 files changed, 95 insertions(+), 12 deletions(-) diff --git a/src/or/channel.c b/src/or/channel.c index da2493a..e9451f1 100644 --- a/src/or/channel.c +++ b/src/or/channel.c @@ -3826,6 +3826,40 @@ channel_mark_outgoing(channel_t *chan) chan->is_incoming = 0; } +/************************ + * Flow control queries * + ***********************/ + +/* + * Estimate the number of writeable cells + * + * Ask the lower layer for an estimate of how many cells it can accept, and + * then subtract the length of our outgoing_queue, if any, to produce an + * estimate of the number of cells this channel can accept for writes. + */ + +int +channel_num_cells_writeable(channel_t *chan) +{ + int result; + + tor_assert(chan); + tor_assert(chan->num_cells_writeable); + + if (chan->state == CHANNEL_STATE_OPEN) { + /* Query lower layer */ + result = chan->num_cells_writeable(chan); + /* Subtract cell queue length, if any */ + result -= chan_cell_queue_len(&chan->outgoing_queue); + if (result < 0) result = 0; + } else { + /* No cells are writeable in any other state */ + result = 0; + } + + return result; +} + /********************* * Timestamp updates * ********************/ diff --git a/src/or/channel.h b/src/or/channel.h index 1481992..28b5ab0 100644 --- a/src/or/channel.h +++ b/src/or/channel.h @@ -110,7 +110,9 @@ struct channel_s { int (*matches_extend_info)(channel_t *, extend_info_t *); /** Check if this channel matches a target address when extending */ int (*matches_target)(channel_t *, const tor_addr_t *); - /** Write a cell to an open channel */ + /* Ask the lower layer how many cells can be written */ + int (*num_cells_writeable)(channel_t *); + /* Write a cell to an open channel */ int (*write_cell)(channel_t *, cell_t *); /** Write a packed cell to an open channel */ int (*write_packed_cell)(channel_t *, packed_cell_t *); @@ -465,6 +467,9 @@ void channel_listener_dump_statistics(channel_listener_t *chan_l, void channel_listener_dump_transport_statistics(channel_listener_t *chan_l, int severity); +/* Flow control queries */ +int channel_num_cells_writeable(channel_t *chan); + /* Timestamp queries */ time_t channel_when_created(channel_t *chan); time_t channel_when_last_active(channel_t *chan); diff --git a/src/or/channeltls.c b/src/or/channeltls.c index af1aab3..b828b15 100644 --- a/src/or/channeltls.c +++ b/src/or/channeltls.c @@ -68,6 +68,7 @@ channel_tls_matches_extend_info_method(channel_t *chan, extend_info_t *extend_info); static int channel_tls_matches_target_method(channel_t *chan, const tor_addr_t *target); +static int channel_tls_num_cells_writeable_method(channel_t *chan); static int channel_tls_write_cell_method(channel_t *chan, cell_t *cell); static int channel_tls_write_packed_cell_method(channel_t *chan, @@ -124,6 +125,7 @@ channel_tls_common_init(channel_tls_t *tlschan) chan->is_canonical = channel_tls_is_canonical_method; chan->matches_extend_info = channel_tls_matches_extend_info_method; chan->matches_target = channel_tls_matches_target_method; + chan->num_cells_writeable = channel_tls_num_cells_writeable_method; chan->write_cell = channel_tls_write_cell_method; chan->write_packed_cell = channel_tls_write_packed_cell_method; chan->write_var_cell = channel_tls_write_var_cell_method; @@ -674,6 +676,34 @@ channel_tls_matches_target_method(channel_t *chan, } /** + * Tell the upper layer how many cells we can accept to write + * + * This implements the num_cells_writeable method for channel_tls_t; it + * returns an estimate of the number of cells we can accept with + * channel_tls_write_*_cell(). + */ + +static int +channel_tls_num_cells_writeable_method(channel_t *chan) +{ + size_t outbuf_len; + int n; + channel_tls_t *tlschan = BASE_CHAN_TO_TLS(chan); + size_t cell_network_size; + + tor_assert(tlschan); + tor_assert(tlschan->conn); + + cell_network_size = get_cell_network_size(tlschan->conn->wide_circ_ids); + outbuf_len = connection_get_outbuf_len(TO_CONN(tlschan->conn)); + /* Get the number of cells */ + n = CEIL_DIV(OR_CONN_HIGHWATER - outbuf_len, cell_network_size); + if (n < 0) n = 0; + + return n; +} + +/** * Write a cell to a channel_tls_t * * This implements the write_cell method for channel_tls_t; given a diff --git a/src/or/connection_or.c b/src/or/connection_or.c index 9074c0a..a8f9d41 100644 --- a/src/or/connection_or.c +++ b/src/or/connection_or.c @@ -576,14 +576,6 @@ connection_or_process_inbuf(or_connection_t *conn) return ret; } -/** When adding cells to an OR connection's outbuf, keep adding until the - * outbuf is at least this long, or we run out of cells. */ -#define OR_CONN_HIGHWATER (32*1024) - -/** Add cells to an OR connection's outbuf whenever the outbuf's data length - * drops below this size. */ -#define OR_CONN_LOWWATER (16*1024) - /** Called whenever we have flushed some data on an or_conn: add more data * from active circuits. */ int diff --git a/src/or/or.h b/src/or/or.h index b2b0d5f..bdcb29e 100644 --- a/src/or/or.h +++ b/src/or/or.h @@ -1426,6 +1426,18 @@ typedef struct or_handshake_state_t { /** Length of Extended ORPort connection identifier. */ #define EXT_OR_CONN_ID_LEN DIGEST_LEN /* 20 */ +/* + * OR_CONN_HIGHWATER and OR_CONN_LOWWATER moved from connection_or.c so + * channeltls.c can see them too. + */ + +/** When adding cells to an OR connection's outbuf, keep adding until the + * outbuf is at least this long, or we run out of cells. */ +#define OR_CONN_HIGHWATER (32*1024) + +/** Add cells to an OR connection's outbuf whenever the outbuf's data length + * drops below this size. */ +#define OR_CONN_LOWWATER (16*1024) /** Subtype of connection_t for an "OR connection" -- that is, one that speaks * cells over TLS. */ diff --git a/src/or/scheduler.c b/src/or/scheduler.c index e2dcdb5..7023eaa 100644 --- a/src/or/scheduler.c +++ b/src/or/scheduler.c @@ -324,6 +324,7 @@ void scheduler_run(void) { smartlist_t *tmp = NULL; + int n_cells; log_debug(LD_SCHED, "We have a chance to run the scheduler"); @@ -337,9 +338,18 @@ scheduler_run(void) channels_pending = smartlist_new(); SMARTLIST_FOREACH_BEGIN(tmp, channel_t *, chan) { - log_debug(LD_SCHED, - "Scheduler saw pending channel " U64_FORMAT " at %p", - U64_PRINTF_ARG(chan->global_identifier), chan); + n_cells = channel_num_cells_writeable(chan); + if (n_cells > 0) { + log_debug(LD_SCHED, + "Scheduler saw pending channel " U64_FORMAT " at %p with " + "%d cells writeable", + U64_PRINTF_ARG(chan->global_identifier), chan, n_cells); + } else { + log_info(LD_SCHED, + "Scheduler saw pending channel " U64_FORMAT " at %p with " + "no cells writeable", + U64_PRINTF_ARG(chan->global_identifier), chan); + } } SMARTLIST_FOREACH_END(chan); smartlist_free(tmp);

1 0

[tor/master] Uglify scheduler init logic to avoid crash on startup.
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit 472b62bfe4edb2f5e332c997be2ec69bdf590660 Author: Nick Mathewson <nickm(a)torproject.org> Date: Mon Oct 14 13:33:36 2013 -0400 Uglify scheduler init logic to avoid crash on startup. Otherwise, when we authority try to do a self-test because of init-keys, if that self-test can't be launched for whatever reason and so we close the channel immediately, we crash. Yes, this a silly way for initialization to work. --- src/or/config.c | 9 +++++++++ src/or/main.c | 8 -------- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/src/or/config.c b/src/or/config.c index 16acec7..4ae9fad 100644 --- a/src/or/config.c +++ b/src/or/config.c @@ -43,6 +43,7 @@ #include "util.h" #include "routerlist.h" #include "routerset.h" +#include "scheduler.h" #include "statefile.h" #include "transports.h" #include "ext_orport.h" @@ -1040,6 +1041,14 @@ options_act_reversible(const or_options_t *old_options, char **msg) if (running_tor && !libevent_initialized) { init_libevent(options); libevent_initialized = 1; + + /* + * Initialize the scheduler - this has to come after + * options_init_from_torrc() sets up libevent - why yes, that seems + * completely sensible to hide the libevent setup in the option parsing + * code! It also needs to happen before init_keys(), so it needs to + * happen here too. How yucky. */ + scheduler_init(); } /* Adjust the port configuration so we can launch listeners. */ diff --git a/src/or/main.c b/src/or/main.c index 1a2cfad..7b38f45 100644 --- a/src/or/main.c +++ b/src/or/main.c @@ -2456,14 +2456,6 @@ tor_init(int argc, char *argv[]) log_warn(LD_NET, "Problem initializing libevent RNG."); } - /* - * Initialize the scheduler - this has to come after - * options_init_from_torrc() sets up libevent - why yes, that seems - * completely sensible to hide the libevent setup in the option parsing - * code! - */ - scheduler_init(); - return 0; }

1 0

[tor/master] Make 'make check-spaces' not complain about function pointers returning size_t or double
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit 7674308f622904bc5a4cff5e7cb4d0f22fbf84d7 Author: Andrea Shepard <andrea(a)torproject.org> Date: Mon Nov 4 22:40:24 2013 -0800 Make 'make check-spaces' not complain about function pointers returning size_t or double --- scripts/maint/checkSpace.pl | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/scripts/maint/checkSpace.pl b/scripts/maint/checkSpace.pl index b529103..c785d89 100755 --- a/scripts/maint/checkSpace.pl +++ b/scripts/maint/checkSpace.pl @@ -128,7 +128,8 @@ for $fn (@ARGV) { if ($1 ne "if" and $1 ne "while" and $1 ne "for" and $1 ne "switch" and $1 ne "return" and $1 ne "int" and $1 ne "elsif" and $1 ne "WINAPI" and $2 ne "WINAPI" and - $1 ne "void" and $1 ne "__attribute__" and $1 ne "op") { + $1 ne "void" and $1 ne "__attribute__" and $1 ne "op" and + $1 ne "size_t" and $1 ne "double") { print " fn ():$fn:$.\n"; } }

1 0

[tor/master] Actually call channel_flush_some_cells() from the scheduler
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit b09f41424c7e0da63733c2fdda5332d63a7f4e0f Author: Andrea Shepard <andrea(a)torproject.org> Date: Thu Oct 31 19:37:57 2013 -0700 Actually call channel_flush_some_cells() from the scheduler --- src/or/connection_or.c | 29 +---------------------------- src/or/scheduler.c | 16 ++++++++++------ 2 files changed, 11 insertions(+), 34 deletions(-) diff --git a/src/or/connection_or.c b/src/or/connection_or.c index a8f9d41..6b276cc 100644 --- a/src/or/connection_or.c +++ b/src/or/connection_or.c @@ -581,9 +581,7 @@ connection_or_process_inbuf(or_connection_t *conn) int connection_or_flushed_some(or_connection_t *conn) { - size_t datalen, temp; - ssize_t n, flushed; - size_t cell_network_size = get_cell_network_size(conn->wide_circ_ids); + size_t datalen; /* If we're under the low water mark, add cells until we're just over the * high water mark. */ @@ -591,31 +589,6 @@ connection_or_flushed_some(or_connection_t *conn) if (datalen < OR_CONN_LOWWATER) { /* Let the scheduler know */ scheduler_channel_wants_writes(TLS_CHAN_TO_BASE(conn->chan)); - - /* - * TODO this will be done from the scheduler, so it will - * need a generic way to ask how many cells a channel can - * accept and if it still wants writes or not to know how - * to account for it in the case that it runs out of cells - * to send first. - */ - - while ((conn->chan) && channel_tls_more_to_flush(conn->chan)) { - /* Compute how many more cells we want at most */ - n = CEIL_DIV(OR_CONN_HIGHWATER - datalen, cell_network_size); - /* Bail out if we don't want any more */ - if (n <= 0) break; - /* We're still here; try to flush some more cells */ - flushed = channel_tls_flush_some_cells(conn->chan, n); - /* Bail out if it says it didn't flush anything */ - if (flushed <= 0) break; - /* How much in the outbuf now? */ - temp = connection_get_outbuf_len(TO_CONN(conn)); - /* Bail out if we didn't actually increase the outbuf size */ - if (temp <= datalen) break; - /* Update datalen for the next iteration */ - datalen = temp; - } } return 0; diff --git a/src/or/scheduler.c b/src/or/scheduler.c index 7023eaa..1c6a2fd 100644 --- a/src/or/scheduler.c +++ b/src/or/scheduler.c @@ -325,18 +325,15 @@ scheduler_run(void) { smartlist_t *tmp = NULL; int n_cells; + ssize_t flushed, flushed_this_time; log_debug(LD_SCHED, "We have a chance to run the scheduler"); - /* - * TODO make this work properly - * - * For now, just empty the pending list and log that we saw stuff in it - */ - tmp = channels_pending; channels_pending = smartlist_new(); + /* For now, just run the old scheduler on all the chans in the list */ + SMARTLIST_FOREACH_BEGIN(tmp, channel_t *, chan) { n_cells = channel_num_cells_writeable(chan); if (n_cells > 0) { @@ -344,6 +341,13 @@ scheduler_run(void) "Scheduler saw pending channel " U64_FORMAT " at %p with " "%d cells writeable", U64_PRINTF_ARG(chan->global_identifier), chan, n_cells); + + flushed = 0; + while (flushed < n_cells) { + flushed_this_time = channel_flush_some_cells(chan, n_cells - flushed); + if (flushed_this_time <= 0) break; + flushed += flushed_this_time; + } } else { log_info(LD_SCHED, "Scheduler saw pending channel " U64_FORMAT " at %p with "

1 0

[tor/master] Track total queue size per channel, with overhead estimates, and global queue total
by nickm＠torproject.org 27 Nov '14

27 Nov '14

commit 8852a1794cfa9eb5dae494f5d85242d8fd6955fc Author: Andrea Shepard <andrea(a)torproject.org> Date: Tue Nov 5 00:30:02 2013 -0800 Track total queue size per channel, with overhead estimates, and global queue total --- src/or/channel.c | 126 ++++++++++++++++++++++++++++++++++++++++++++++++ src/or/channel.h | 16 ++++++ src/or/channeltls.c | 54 +++++++++++++++++++++ src/or/connection.c | 2 + src/or/connection_or.c | 3 ++ src/or/or.h | 6 +++ 6 files changed, 207 insertions(+) diff --git a/src/or/channel.c b/src/or/channel.c index 0caebfb..f729a17 100644 --- a/src/or/channel.c +++ b/src/or/channel.c @@ -124,6 +124,13 @@ static uint64_t n_channel_bytes_passed_to_lower_layer = 0; static uint64_t n_channel_bytes_in_queues = 0; +/* + * Current total estimated queue size *including lower layer queues and + * transmit overhead* + */ + +static uint64_t estimated_total_queue_size = 0; + /* Digest->channel map * * Similar to the one used in connection_or.c, this maps from the identity @@ -1840,6 +1847,8 @@ channel_write_cell_queue_entry(channel_t *chan, cell_queue_entry_t *q) n_channel_bytes_queued += cell_bytes; n_channel_bytes_in_queues += cell_bytes; channel_assert_counter_consistency(); + /* Update channel queue size */ + chan->bytes_in_queue += cell_bytes; /* Try to process the queue? */ if (chan->state == CHANNEL_STATE_OPEN) channel_flush_cells(chan); } @@ -1878,6 +1887,9 @@ channel_write_cell(channel_t *chan, cell_t *cell) q.type = CELL_QUEUE_FIXED; q.u.fixed.cell = cell; channel_write_cell_queue_entry(chan, &q); + + /* Update the queue size estimate */ + channel_update_xmit_queue_size(chan); } /** @@ -1913,6 +1925,9 @@ channel_write_packed_cell(channel_t *chan, packed_cell_t *packed_cell) q.type = CELL_QUEUE_PACKED; q.u.packed.packed_cell = packed_cell; channel_write_cell_queue_entry(chan, &q); + + /* Update the queue size estimate */ + channel_update_xmit_queue_size(chan); } /** @@ -1949,6 +1964,9 @@ channel_write_var_cell(channel_t *chan, var_cell_t *var_cell) q.type = CELL_QUEUE_VAR; q.u.var.var_cell = var_cell; channel_write_cell_queue_entry(chan, &q); + + /* Update the queue size estimate */ + channel_update_xmit_queue_size(chan); } /** @@ -2056,6 +2074,29 @@ channel_change_state(channel_t *chan, channel_state_t to_state) scheduler_channel_doesnt_want_writes(chan); } + /* + * If we're closing, this channel no longer counts toward the global + * estimated queue size; if we're open, it now does. + */ + if ((to_state == CHANNEL_STATE_CLOSING || + to_state == CHANNEL_STATE_CLOSED || + to_state == CHANNEL_STATE_ERROR) && + (from_state == CHANNEL_STATE_OPEN || + from_state == CHANNEL_STATE_MAINT)) { + estimated_total_queue_size -= chan->bytes_in_queue; + } + + /* + * If we're opening, this channel now does count toward the global + * estimated queue size. + */ + if ((to_state == CHANNEL_STATE_OPEN || + to_state == CHANNEL_STATE_MAINT) && + !(from_state == CHANNEL_STATE_OPEN || + from_state == CHANNEL_STATE_MAINT)) { + estimated_total_queue_size += chan->bytes_in_queue; + } + /* Tell circuits if we opened and stuff */ if (to_state == CHANNEL_STATE_OPEN) { channel_do_open_actions(chan); @@ -2350,6 +2391,8 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan, n_channel_bytes_passed_to_lower_layer += cell_size; n_channel_bytes_in_queues -= cell_size; channel_assert_counter_consistency(); + /* Update the channel's queue size too */ + chan->bytes_in_queue -= cell_size; } /* No cell removed from list, so we can't go on any further */ else break; @@ -2362,6 +2405,9 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan, channel_timestamp_drained(chan); } + /* Update the estimate queue size */ + channel_update_xmit_queue_size(chan); + return flushed; } @@ -4421,3 +4467,83 @@ channel_set_circid_type(channel_t *chan, } } +/** + * Update the estimated number of bytes queued to transmit for this channel, + * and notify the scheduler. The estimate includes both the channel queue and + * the queue size reported by the lower layer, and an overhead estimate + * optionally provided by the lower layer. + */ + +void +channel_update_xmit_queue_size(channel_t *chan) +{ + uint64_t queued, adj; + double overhead; + + tor_assert(chan); + tor_assert(chan->num_bytes_queued); + + /* + * First, get the number of bytes we have queued without factoring in + * lower-layer overhead. + */ + queued = chan->num_bytes_queued(chan) + chan->bytes_in_queue; + /* Next, adjust by the overhead factor, if any is available */ + if (chan->get_overhead_estimate) { + overhead = chan->get_overhead_estimate(chan); + if (overhead >= 1.0f) { + queued *= overhead; + } else { + /* Ignore silly overhead factors */ + log_notice(LD_CHANNEL, "Ignoring silly overhead factor %f", overhead); + } + } + + /* Now, compare to the previous estimate */ + if (queued > chan->bytes_queued_for_xmit) { + adj = queued - chan->bytes_queued_for_xmit; + log_debug(LD_CHANNEL, + "Increasing queue size for channel " U64_FORMAT " by " U64_FORMAT + " from " U64_FORMAT " to " U64_FORMAT, + U64_PRINTF_ARG(chan->global_identifier), + U64_PRINTF_ARG(adj), + U64_PRINTF_ARG(chan->bytes_queued_for_xmit), + U64_PRINTF_ARG(queued)); + /* Update the channel's estimate */ + chan->bytes_queued_for_xmit = queued; + + /* Update the global queue size estimate if appropriate */ + if (chan->state == CHANNEL_STATE_OPEN || + chan->state == CHANNEL_STATE_MAINT) { + estimated_total_queue_size += adj; + log_debug(LD_CHANNEL, + "Increasing global queue size by " U64_FORMAT " for channel " + U64_FORMAT ", new size is " U64_FORMAT, + U64_PRINTF_ARG(adj), U64_PRINTF_ARG(chan->global_identifier), + U64_PRINTF_ARG(estimated_total_queue_size)); + } + } else if (queued < chan->bytes_queued_for_xmit) { + adj = chan->bytes_queued_for_xmit - queued; + log_debug(LD_CHANNEL, + "Decreasing queue size for channel " U64_FORMAT " by " U64_FORMAT + " from " U64_FORMAT " to " U64_FORMAT, + U64_PRINTF_ARG(chan->global_identifier), + U64_PRINTF_ARG(adj), + U64_PRINTF_ARG(chan->bytes_queued_for_xmit), + U64_PRINTF_ARG(queued)); + /* Update the channel's estimate */ + chan->bytes_queued_for_xmit = queued; + + /* Update the global queue size estimate if appropriate */ + if (chan->state == CHANNEL_STATE_OPEN || + chan->state == CHANNEL_STATE_MAINT) { + estimated_total_queue_size -= adj; + log_debug(LD_CHANNEL, + "Decreasing global queue size by " U64_FORMAT " for channel " + U64_FORMAT ", new size is " U64_FORMAT, + U64_PRINTF_ARG(adj), U64_PRINTF_ARG(chan->global_identifier), + U64_PRINTF_ARG(estimated_total_queue_size)); + } + } +} + diff --git a/src/or/channel.h b/src/or/channel.h index f35cb27..388c729 100644 --- a/src/or/channel.h +++ b/src/or/channel.h @@ -79,6 +79,11 @@ struct channel_s { /* Methods implemented by the lower layer */ /** + * Ask the lower layer for an estimate of the average overhead for + * transmissions on this channel. + */ + double (*get_overhead_estimate)(channel_t *); + /* * Ask the underlying transport what the remote endpoint address is, in * a tor_addr_t. This is optional and subclasses may leave this NULL. * If they implement it, they should write the address out to the @@ -110,6 +115,8 @@ struct channel_s { int (*matches_extend_info)(channel_t *, extend_info_t *); /** Check if this channel matches a target address when extending */ int (*matches_target)(channel_t *, const tor_addr_t *); + /* Ask the lower layer how many bytes it has queued but not yet sent */ + size_t (*num_bytes_queued)(channel_t *); /* Ask the lower layer how many cells can be written */ int (*num_cells_writeable)(channel_t *); /* Write a cell to an open channel */ @@ -202,6 +209,14 @@ struct channel_s { /** Channel counters for cell channels */ uint64_t n_cells_recved, n_bytes_recved; uint64_t n_cells_xmitted, n_bytes_xmitted; + + /** Our current contribution to the scheduler's total xmit queue */ + uint64_t bytes_queued_for_xmit; + + /** Number of bytes in this channel's cell queue; does not include + * lower-layer queueing. + */ + uint64_t bytes_in_queue; }; struct channel_listener_s { @@ -460,6 +475,7 @@ unsigned int channel_num_circuits(channel_t *chan); void channel_set_circid_type(channel_t *chan, crypto_pk_t *identity_rcvd, int consider_identity); void channel_timestamp_client(channel_t *chan); +void channel_update_xmit_queue_size(channel_t *chan); const char * channel_listener_describe_transport(channel_listener_t *chan_l); void channel_listener_dump_statistics(channel_listener_t *chan_l, diff --git a/src/or/channeltls.c b/src/or/channeltls.c index 7df2d35..a8222de 100644 --- a/src/or/channeltls.c +++ b/src/or/channeltls.c @@ -55,6 +55,7 @@ static void channel_tls_common_init(channel_tls_t *tlschan); static void channel_tls_close_method(channel_t *chan); static const char * channel_tls_describe_transport_method(channel_t *chan); static void channel_tls_free_method(channel_t *chan); +static double channel_tls_get_overhead_estimate_method(channel_t *chan); static int channel_tls_get_remote_addr_method(channel_t *chan, tor_addr_t *addr_out); static int @@ -69,6 +70,7 @@ channel_tls_matches_extend_info_method(channel_t *chan, static int channel_tls_matches_target_method(channel_t *chan, const tor_addr_t *target); static int channel_tls_num_cells_writeable_method(channel_t *chan); +static size_t channel_tls_num_bytes_queued_method(channel_t *chan); static int channel_tls_write_cell_method(channel_t *chan, cell_t *cell); static int channel_tls_write_packed_cell_method(channel_t *chan, @@ -118,6 +120,7 @@ channel_tls_common_init(channel_tls_t *tlschan) chan->close = channel_tls_close_method; chan->describe_transport = channel_tls_describe_transport_method; chan->free = channel_tls_free_method; + chan->get_overhead_estimate = channel_tls_get_overhead_estimate_method; chan->get_remote_addr = channel_tls_get_remote_addr_method; chan->get_remote_descr = channel_tls_get_remote_descr_method; chan->get_transport_name = channel_tls_get_transport_name_method; @@ -125,6 +128,7 @@ channel_tls_common_init(channel_tls_t *tlschan) chan->is_canonical = channel_tls_is_canonical_method; chan->matches_extend_info = channel_tls_matches_extend_info_method; chan->matches_target = channel_tls_matches_target_method; + chan->num_bytes_queued = channel_tls_num_bytes_queued_method; chan->num_cells_writeable = channel_tls_num_cells_writeable_method; chan->write_cell = channel_tls_write_cell_method; chan->write_packed_cell = channel_tls_write_packed_cell_method; @@ -438,6 +442,40 @@ channel_tls_free_method(channel_t *chan) } /** + * Get an estimate of the average TLS overhead for the upper layer + */ + +static double +channel_tls_get_overhead_estimate_method(channel_t *chan) +{ + double overhead = 1.0f; + channel_tls_t *tlschan = BASE_CHAN_TO_TLS(chan); + + tor_assert(tlschan); + tor_assert(tlschan->conn); + + /* Just return 1.0f if we don't have sensible data */ + if (tlschan->conn->bytes_xmitted > 0 && + tlschan->conn->bytes_xmitted_by_tls >= + tlschan->conn->bytes_xmitted) { + overhead = ((double)(tlschan->conn->bytes_xmitted_by_tls)) / + ((double)(tlschan->conn->bytes_xmitted)); + + /* + * Never estimate more than 2.0; otherwise we get silly large estimates + * at the very start of a new TLS connection. + */ + if (overhead > 2.0f) overhead = 2.0f; + } + + log_debug(LD_CHANNEL, + "Estimated overhead ratio for TLS chan " U64_FORMAT " is %f", + U64_PRINTF_ARG(chan->global_identifier), overhead); + + return overhead; +} + +/** * Get the remote address of a channel_tls_t * * This implements the get_remote_addr method for channel_tls_t; copy the @@ -676,6 +714,22 @@ channel_tls_matches_target_method(channel_t *chan, } /** + * Tell the upper layer how many bytes we have queued and not yet + * sent. + */ + +static size_t +channel_tls_num_bytes_queued_method(channel_t *chan) +{ + channel_tls_t *tlschan = BASE_CHAN_TO_TLS(chan); + + tor_assert(tlschan); + tor_assert(tlschan->conn); + + return connection_get_outbuf_len(TO_CONN(tlschan->conn)); +} + +/** * Tell the upper layer how many cells we can accept to write * * This implements the num_cells_writeable method for channel_tls_t; it diff --git a/src/or/connection.c b/src/or/connection.c index 4a3bd2c..525f4b5 100644 --- a/src/or/connection.c +++ b/src/or/connection.c @@ -3839,6 +3839,8 @@ connection_handle_write_impl(connection_t *conn, int force) tor_tls_get_n_raw_bytes(or_conn->tls, &n_read, &n_written); log_debug(LD_GENERAL, "After TLS write of %d: %ld read, %ld written", result, (long)n_read, (long)n_written); + or_conn->bytes_xmitted += result; + or_conn->bytes_xmitted_by_tls += n_written; /* So we notice bytes were written even on error */ /* XXXX024 This cast is safe since we can never write INT_MAX bytes in a * single set of TLS operations. But it looks kinda ugly. If we refactor diff --git a/src/or/connection_or.c b/src/or/connection_or.c index 6b276cc..445335b 100644 --- a/src/or/connection_or.c +++ b/src/or/connection_or.c @@ -583,6 +583,9 @@ connection_or_flushed_some(or_connection_t *conn) { size_t datalen; + /* The channel will want to update its estimated queue size */ + channel_update_xmit_queue_size(TLS_CHAN_TO_BASE(conn->chan)); + /* If we're under the low water mark, add cells until we're just over the * high water mark. */ datalen = connection_get_outbuf_len(TO_CONN(conn)); diff --git a/src/or/or.h b/src/or/or.h index bdcb29e..320931d 100644 --- a/src/or/or.h +++ b/src/or/or.h @@ -1529,6 +1529,12 @@ typedef struct or_connection_t { /** Last emptied write token bucket in msec since midnight; only used if * TB_EMPTY events are enabled. */ uint32_t write_emptied_time; + + /* + * Count the number of bytes flushed out on this orconn, and the number of + * bytes TLS actually sent - used for overhead estimation for scheduling. + */ + uint64_t bytes_xmitted, bytes_xmitted_by_tls; } or_connection_t; /** Subtype of connection_t for an "edge connection" -- that is, an entry (ap)

1 0