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

[tor/master] Make scheduler_release_channel() mockable
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 85ee07085281a1fa47d0b44b0addbb54fcfa6061 Author: Andrea Shepard <andrea(a)torproject.org> Date: Fri Dec 13 06:27:00 2013 -0800 Make scheduler_release_channel() mockable --- src/or/scheduler.c | 4 ++-- src/or/scheduler.h | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/or/scheduler.c b/src/or/scheduler.c index 140ff2f..450eb02 100644 --- a/src/or/scheduler.c +++ b/src/or/scheduler.c @@ -360,8 +360,8 @@ scheduler_retrigger(void) /** Notify the scheduler of a channel being closed */ -void -scheduler_release_channel(channel_t *chan) +MOCK_IMPL(void, +scheduler_release_channel,(channel_t *chan)) { tor_assert(chan); tor_assert(channels_pending); diff --git a/src/or/scheduler.h b/src/or/scheduler.h index e2d2eb5..9cdf6c1 100644 --- a/src/or/scheduler.h +++ b/src/or/scheduler.h @@ -11,6 +11,7 @@ #include "or.h" #include "channel.h" +#include "testsupport.h" /* Global-visibility scheduler functions */ @@ -25,7 +26,7 @@ void scheduler_channel_has_waiting_cells(channel_t *chan); void scheduler_channel_wants_writes(channel_t *chan); /* Notify the scheduler of a channel being closed */ -void scheduler_release_channel(channel_t *chan); +MOCK_DECL(void,scheduler_release_channel,(channel_t *chan)); /* Notify scheduler of queue size adjustments */ void scheduler_adjust_queue_size(channel_t *chan, char dir, uint64_t adj);

1 0

[tor/master] Use a non-stupid data structure in the scheduler
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit ed1927d6bf8b9d60d40f6fbc20f9e1575a35e59d Author: Andrea Shepard <andrea(a)torproject.org> Date: Thu Dec 12 04:22:53 2013 -0800 Use a non-stupid data structure in the scheduler --- src/or/channel.h | 3 + src/or/scheduler.c | 190 ++++++++++++++++++++++++++++++++++++---------------- 2 files changed, 137 insertions(+), 56 deletions(-) diff --git a/src/or/channel.h b/src/or/channel.h index ced717a..023c39d 100644 --- a/src/or/channel.h +++ b/src/or/channel.h @@ -80,6 +80,9 @@ struct channel_s { SCHED_CHAN_PENDING } scheduler_state; + /** Heap index for use by the scheduler */ + int sched_heap_idx; + /** Timestamps for both cell channels and listeners */ time_t timestamp_created; /* Channel created */ time_t timestamp_active; /* Any activity */ diff --git a/src/or/scheduler.c b/src/or/scheduler.c index c1b64df..4f12696 100644 --- a/src/or/scheduler.c +++ b/src/or/scheduler.c @@ -24,6 +24,14 @@ #define SCHED_Q_HIGH_WATER (2 * SCHED_Q_LOW_WATER) /* + * Maximum cells to flush in a single call to channel_flush_some_cells(); + * setting this low means more calls, but too high and we could overshoot + * SCHED_Q_HIGH_WATER. + */ + +#define SCHED_MAX_FLUSH_CELLS 16 + +/* * Write scheduling works by keeping track of which channels can * accept cells, and have cells to write. From the scheduler's perspective, * a channel can be in four possible states: @@ -100,7 +108,7 @@ * is reserved for our use. */ -/* List of channels that can write and have cells (pending work) */ +/* Pqueue of channels that can write and have cells (pending work) */ static smartlist_t *channels_pending = NULL; /* @@ -125,7 +133,7 @@ static time_t queue_heuristic_timestamp = 0; /* Scheduler static function declarations */ -static int scheduler_compare_channels(const void **c1_v, const void **c2_v); +static int scheduler_compare_channels(const void *c1_v, const void *c2_v); static void scheduler_evt_callback(evutil_socket_t fd, short events, void *arg); static int scheduler_more_work(void); @@ -162,7 +170,7 @@ scheduler_free_all(void) */ static int -scheduler_compare_channels(const void **c1_v, const void **c2_v) +scheduler_compare_channels(const void *c1_v, const void *c2_v) { channel_t *c1 = NULL, *c2 = NULL; /* These are a workaround for -Wbad-function-cast throwing a fit */ @@ -172,8 +180,8 @@ scheduler_compare_channels(const void **c1_v, const void **c2_v) tor_assert(c1_v); tor_assert(c2_v); - c1 = (channel_t *)(*c1_v); - c2 = (channel_t *)(*c2_v); + c1 = (channel_t *)(c1_v); + c2 = (channel_t *)(c2_v); tor_assert(c1); tor_assert(c2); @@ -241,7 +249,10 @@ scheduler_channel_doesnt_want_writes(channel_t *chan) * the other lists. It can't write any more, so it goes to * channels_waiting_to_write. */ - smartlist_remove(channels_pending, chan); + smartlist_pqueue_remove(channels_pending, + scheduler_compare_channels, + STRUCT_OFFSET(channel_t, sched_heap_idx), + chan); chan->scheduler_state = SCHED_CHAN_WAITING_TO_WRITE; log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p went from pending " @@ -280,7 +291,10 @@ scheduler_channel_has_waiting_cells(channel_t *chan) * channels_pending. */ chan->scheduler_state = SCHED_CHAN_PENDING; - smartlist_add(channels_pending, chan); + smartlist_pqueue_add(channels_pending, + scheduler_compare_channels, + STRUCT_OFFSET(channel_t, sched_heap_idx), + chan); log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p went from waiting_for_cells " "to pending", @@ -353,7 +367,10 @@ scheduler_release_channel(channel_t *chan) tor_assert(channels_pending); if (chan->scheduler_state == SCHED_CHAN_PENDING) { - smartlist_remove(channels_pending, chan); + smartlist_pqueue_remove(channels_pending, + scheduler_compare_channels, + STRUCT_OFFSET(channel_t, sched_heap_idx), + chan); } chan->scheduler_state = SCHED_CHAN_IDLE; @@ -364,10 +381,11 @@ scheduler_release_channel(channel_t *chan) void scheduler_run(void) { - smartlist_t *tmp = NULL; int n_cells, n_chans_before, n_chans_after; uint64_t q_len_before, q_heur_before, q_len_after, q_heur_after; ssize_t flushed, flushed_this_time; + smartlist_t *to_readd = NULL; + channel_t *chan = NULL; log_debug(LD_SCHED, "We have a chance to run the scheduler"); @@ -375,61 +393,118 @@ scheduler_run(void) n_chans_before = smartlist_len(channels_pending); q_len_before = channel_get_global_queue_estimate(); q_heur_before = scheduler_get_queue_heuristic(); - tmp = channels_pending; - channels_pending = smartlist_new(); - - /* - * UGLY HACK: sort the list on each invocation - * - * TODO smarter data structures - */ - smartlist_sort(tmp, scheduler_compare_channels); - - SMARTLIST_FOREACH_BEGIN(tmp, channel_t *, chan) { - if (scheduler_get_queue_heuristic() <= SCHED_Q_HIGH_WATER) { - 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); - - 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; - } - if (flushed < n_cells) { - /* We ran out of cells to flush */ - chan->scheduler_state = SCHED_CHAN_WAITING_FOR_CELLS; - } else { - /* TODO get this right */ - } + while (scheduler_get_queue_heuristic() <= SCHED_Q_HIGH_WATER && + smartlist_len(channels_pending) > 0) { + /* Pop off a channel */ + chan = smartlist_pqueue_pop(channels_pending, + scheduler_compare_channels, + STRUCT_OFFSET(channel_t, sched_heap_idx)); + tor_assert(chan); + + /* Figure out how many cells we can write */ + 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); + + flushed = 0; + while (flushed < n_cells && + scheduler_get_queue_heuristic() <= SCHED_Q_HIGH_WATER) { + flushed_this_time = + channel_flush_some_cells(chan, + MIN(SCHED_MAX_FLUSH_CELLS, + n_cells - flushed)); + if (flushed_this_time <= 0) break; + flushed += flushed_this_time; + } + if (flushed < n_cells) { + /* We ran out of cells to flush */ + chan->scheduler_state = SCHED_CHAN_WAITING_FOR_CELLS; log_debug(LD_SCHED, - "Scheduler flushed %d cells onto pending channel " - U64_FORMAT " at %p", - (int)flushed, U64_PRINTF_ARG(chan->global_identifier), + "Channel " U64_FORMAT " at %p " + "entered waiting_for_cells from pending", + U64_PRINTF_ARG(chan->global_identifier), chan); } else { - log_info(LD_SCHED, - "Scheduler saw pending channel " U64_FORMAT " at %p with " - "no cells writeable", - U64_PRINTF_ARG(chan->global_identifier), chan); - /* Put it back to WAITING_TO_WRITE */ - chan->scheduler_state = SCHED_CHAN_WAITING_TO_WRITE; + /* The channel may still have some cells */ + if (channel_more_to_flush(chan)) { + /* The channel goes to either pending or waiting_to_write */ + if (channel_num_cells_writeable(chan) > 0) { + /* Add it back to pending later */ + if (!to_readd) to_readd = smartlist_new(); + smartlist_add(to_readd, chan); + log_debug(LD_SCHED, + "Channel " U64_FORMAT " at %p " + "is still pending", + U64_PRINTF_ARG(chan->global_identifier), + chan); + } else { + /* It's waiting to be able to write more */ + chan->scheduler_state = SCHED_CHAN_WAITING_TO_WRITE; + log_debug(LD_SCHED, + "Channel " U64_FORMAT " at %p " + "entered waiting_to_write from pending", + U64_PRINTF_ARG(chan->global_identifier), + chan); + } + } else { + /* No cells left; it can go to idle or waiting_for_cells */ + if (channel_num_cells_writeable(chan) > 0) { + /* + * It can still accept writes, so it goes to + * waiting_for_cells + */ + chan->scheduler_state = SCHED_CHAN_WAITING_FOR_CELLS; + log_debug(LD_SCHED, + "Channel " U64_FORMAT " at %p " + "entered waiting_for_cells from pending", + U64_PRINTF_ARG(chan->global_identifier), + chan); + } else { + /* + * We exactly filled up the output queue with all available + * cells; go to idle. + */ + chan->scheduler_state = SCHED_CHAN_IDLE; + log_debug(LD_SCHED, + "Channel " U64_FORMAT " at %p " + "become idle from pending", + U64_PRINTF_ARG(chan->global_identifier), + chan); + } + } } + + log_debug(LD_SCHED, + "Scheduler flushed %d cells onto pending channel " + U64_FORMAT " at %p", + (int)flushed, U64_PRINTF_ARG(chan->global_identifier), + chan); } else { - /* Not getting it this round; put it back on the list */ - smartlist_add(channels_pending, chan); - /* It states in SCHED_CHAN_PENDING */ + log_info(LD_SCHED, + "Scheduler saw pending channel " U64_FORMAT " at %p with " + "no cells writeable", + U64_PRINTF_ARG(chan->global_identifier), chan); + /* Put it back to WAITING_TO_WRITE */ + chan->scheduler_state = SCHED_CHAN_WAITING_TO_WRITE; } - } SMARTLIST_FOREACH_END(chan); + } - smartlist_free(tmp); + /* Readd any channels we need to */ + if (to_readd) { + SMARTLIST_FOREACH_BEGIN(to_readd, channel_t *, chan) { + chan->scheduler_state = SCHED_CHAN_PENDING; + smartlist_pqueue_add(channels_pending, + scheduler_compare_channels, + STRUCT_OFFSET(channel_t, sched_heap_idx), + chan); + } SMARTLIST_FOREACH_END(chan); + smartlist_free(to_readd); + } n_chans_after = smartlist_len(channels_pending); q_len_after = channel_get_global_queue_estimate(); @@ -473,7 +548,10 @@ scheduler_channel_wants_writes(channel_t *chan) /* * It can write now, so it goes to channels_pending. */ - smartlist_add(channels_pending, chan); + smartlist_pqueue_add(channels_pending, + scheduler_compare_channels, + STRUCT_OFFSET(channel_t, sched_heap_idx), + chan); chan->scheduler_state = SCHED_CHAN_PENDING; log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p went from waiting_to_write "

1 0

[tor/master] Eliminate some unnecessary smartlists in scheduler.c
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 3530825c53b7a28cf894b64e6d97aa90d0acccae Author: Andrea Shepard <andrea(a)torproject.org> Date: Fri Dec 6 00:04:12 2013 -0800 Eliminate some unnecessary smartlists in scheduler.c --- src/or/channel.c | 3 ++ src/or/channel.h | 23 +++++++++++ src/or/scheduler.c | 112 +++++++++++++++++++++------------------------------- 3 files changed, 70 insertions(+), 68 deletions(-) diff --git a/src/or/channel.c b/src/or/channel.c index 7ed3894..60e59c7 100644 --- a/src/or/channel.c +++ b/src/or/channel.c @@ -806,6 +806,9 @@ channel_init(channel_t *chan) /* It hasn't been open yet. */ chan->has_been_open = 0; + + /* Scheduler state is idle */ + chan->scheduler_state = SCHED_CHAN_IDLE; } /** diff --git a/src/or/channel.h b/src/or/channel.h index 18f7cfc..ced717a 100644 --- a/src/or/channel.h +++ b/src/or/channel.h @@ -57,6 +57,29 @@ struct channel_s { CHANNEL_CLOSE_FOR_ERROR } reason_for_closing; + /** State variable for use by the scheduler */ + enum { + /* + * The channel is not open, or it has a full output buffer but no queued + * cells. + */ + SCHED_CHAN_IDLE = 0, + /* + * The channel has space on its output buffer to write, but no queued + * cells. + */ + SCHED_CHAN_WAITING_FOR_CELLS, + /* + * The scheduler has queued cells but no output buffer space to write. + */ + SCHED_CHAN_WAITING_TO_WRITE, + /* + * The scheduler has both queued cells and output buffer space, and is + * eligible for the scheduler loop. + */ + SCHED_CHAN_PENDING + } scheduler_state; + /** Timestamps for both cell channels and listeners */ time_t timestamp_created; /* Channel created */ time_t timestamp_active; /* Any activity */ diff --git a/src/or/scheduler.c b/src/or/scheduler.c index c4c2632..c1b64df 100644 --- a/src/or/scheduler.c +++ b/src/or/scheduler.c @@ -24,12 +24,13 @@ #define SCHED_Q_HIGH_WATER (2 * SCHED_Q_LOW_WATER) /* - * Write scheduling works by keeping track of lists of channels that can + * Write scheduling works by keeping track of which channels can * accept cells, and have cells to write. From the scheduler's perspective, * a channel can be in four possible states: * * 1.) Not open for writes, no cells to send - * - Not much to do here, and the channel will appear in neither list. + * - Not much to do here, and the channel will have scheduler_state == + * SCHED_CHAN_IDLE * - Transitions from: * - Open for writes/has cells by simultaneously draining all circuit * queues and filling the output buffer. @@ -41,7 +42,8 @@ * * 2.) Open for writes, no cells to send * - Not much here either; this will be the state an idle but open channel - * can be expected to settle in. + * can be expected to settle in. It will have scheduler_state == + * SCHED_CHAN_WAITING_FOR_CELLS * - Transitions from: * - Not open for writes/no cells by flushing some of the output * buffer. @@ -55,6 +57,7 @@ * 3.) Not open for writes, cells to send * - This is the state of a busy circuit limited by output bandwidth; * cells have piled up in the circuit queues waiting to be relayed. + * The channel will have scheduler_state == SCHED_CHAN_WAITING_TO_WRITE. * - Transitions from: * - Not open for writes/no cells by arrival of cells on an attached * circuit @@ -66,7 +69,8 @@ * * 4.) Open for writes, cells to send * - This connection is ready to relay some cells and waiting for - * the scheduler to choose it + * the scheduler to choose it. The channel will have scheduler_state == + * SCHED_CHAN_PENDING. * - Transitions from: * - Not open for writes/has cells by the connection_or_flushed_some() * path @@ -91,29 +95,11 @@ /* Scheduler global data structures */ /* - * We keep lists of channels that either have cells queued, can accept - * writes, or both (states 2, 3 and 4 above) - no explicit list of state - * 1 channels is kept, so we don't have to worry about registering new - * channels here or anything. The scheduler will learn about them when - * it needs to. We can check how many channels in state 4 in O(1), so - * the test whether we have anything to do in scheduler_run() is fast - * and there's no harm in calling it opportunistically whenever we get - * the chance. - * - * Note that it takes time O(n) to search for a channel in these smartlists - * or move one; I don't think the number of channels on a relay will be large - * enough for this to be a severe problem, but this would benefit from using - * a doubly-linked list rather than smartlist_t, together with a hash map from - * channel identifiers to pointers to list entries, so we can perform those - * operations in O(log(n)). + * We keep a list of channels that are pending - i.e, have cells to write + * and can accept them to send. The enum scheduler_state in channel_t + * is reserved for our use. */ -/* List of channels that can write but have no cells (state 2 above) */ -static smartlist_t *channels_waiting_for_cells = NULL; - -/* List of channels with cells waiting to write (state 3 above) */ -static smartlist_t *channels_waiting_to_write = NULL; - /* List of channels that can write and have cells (pending work) */ static smartlist_t *channels_pending = NULL; @@ -165,16 +151,6 @@ scheduler_free_all(void) run_sched_ev = NULL; } - if (channels_waiting_for_cells) { - smartlist_free(channels_waiting_for_cells); - channels_waiting_for_cells = NULL; - } - - if (channels_waiting_to_write) { - smartlist_free(channels_waiting_to_write); - channels_waiting_to_write = NULL; - } - if (channels_pending) { smartlist_free(channels_pending); channels_pending = NULL; @@ -255,19 +231,18 @@ void scheduler_channel_doesnt_want_writes(channel_t *chan) { tor_assert(chan); - tor_assert(channels_waiting_for_cells); - tor_assert(channels_waiting_to_write); + tor_assert(channels_pending); /* If it's already in pending, we can put it in waiting_to_write */ - if (smartlist_contains(channels_pending, chan)) { + if (chan->scheduler_state == SCHED_CHAN_PENDING) { /* * It's in channels_pending, so it shouldn't be in any of * the other lists. It can't write any more, so it goes to * channels_waiting_to_write. */ smartlist_remove(channels_pending, chan); - smartlist_add(channels_waiting_to_write, chan); + chan->scheduler_state = SCHED_CHAN_WAITING_TO_WRITE; log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p went from pending " "to waiting_to_write", @@ -278,8 +253,8 @@ scheduler_channel_doesnt_want_writes(channel_t *chan) * either not in any of the lists (nothing to do) or it's already in * waiting_for_cells (remove it, can't write any more). */ - if (smartlist_contains(channels_waiting_for_cells, chan)) { - smartlist_remove(channels_waiting_for_cells, chan); + if (chan->scheduler_state == SCHED_CHAN_WAITING_FOR_CELLS) { + chan->scheduler_state = SCHED_CHAN_IDLE; log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p left waiting_for_cells", U64_PRINTF_ARG(chan->global_identifier), chan); @@ -295,18 +270,16 @@ scheduler_channel_has_waiting_cells(channel_t *chan) int became_pending = 0; tor_assert(chan); - tor_assert(channels_waiting_for_cells); - tor_assert(channels_waiting_to_write); tor_assert(channels_pending); /* First, check if this one also writeable */ - if (smartlist_contains(channels_waiting_for_cells, chan)) { + if (chan->scheduler_state == SCHED_CHAN_WAITING_FOR_CELLS) { /* * It's in channels_waiting_for_cells, so it shouldn't be in any of * the other lists. It has waiting cells now, so it goes to * channels_pending. */ - smartlist_remove(channels_waiting_for_cells, chan); + chan->scheduler_state = SCHED_CHAN_PENDING; smartlist_add(channels_pending, chan); log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p went from waiting_for_cells " @@ -319,9 +292,9 @@ scheduler_channel_has_waiting_cells(channel_t *chan) * either not in any of the lists (we add it to waiting_to_write) * or it's already in waiting_to_write or pending (we do nothing) */ - if (!(smartlist_contains(channels_waiting_to_write, chan) || - smartlist_contains(channels_pending, chan))) { - smartlist_add(channels_waiting_to_write, chan); + if (!(chan->scheduler_state == SCHED_CHAN_WAITING_TO_WRITE || + chan->scheduler_state == SCHED_CHAN_PENDING)) { + chan->scheduler_state = SCHED_CHAN_WAITING_TO_WRITE; log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p entered waiting_to_write", U64_PRINTF_ARG(chan->global_identifier), chan); @@ -346,8 +319,6 @@ scheduler_init(void) run_sched_ev = tor_event_new(tor_libevent_get_base(), -1, 0, scheduler_evt_callback, NULL); - channels_waiting_for_cells = smartlist_new(); - channels_waiting_to_write = smartlist_new(); channels_pending = smartlist_new(); queue_heuristic = 0; queue_heuristic_timestamp = approx_time(); @@ -379,14 +350,13 @@ void scheduler_release_channel(channel_t *chan) { tor_assert(chan); - - tor_assert(channels_waiting_for_cells); - tor_assert(channels_waiting_to_write); tor_assert(channels_pending); - smartlist_remove(channels_waiting_for_cells, chan); - smartlist_remove(channels_waiting_to_write, chan); - smartlist_remove(channels_pending, chan); + if (chan->scheduler_state == SCHED_CHAN_PENDING) { + smartlist_remove(channels_pending, chan); + } + + chan->scheduler_state = SCHED_CHAN_IDLE; } /** Run the scheduling algorithm if necessary */ @@ -432,6 +402,13 @@ scheduler_run(void) flushed += flushed_this_time; } + if (flushed < n_cells) { + /* We ran out of cells to flush */ + chan->scheduler_state = SCHED_CHAN_WAITING_FOR_CELLS; + } else { + /* TODO get this right */ + } + log_debug(LD_SCHED, "Scheduler flushed %d cells onto pending channel " U64_FORMAT " at %p", @@ -442,10 +419,13 @@ scheduler_run(void) "Scheduler saw pending channel " U64_FORMAT " at %p with " "no cells writeable", U64_PRINTF_ARG(chan->global_identifier), chan); + /* Put it back to WAITING_TO_WRITE */ + chan->scheduler_state = SCHED_CHAN_WAITING_TO_WRITE; } } else { /* Not getting it this round; put it back on the list */ smartlist_add(channels_pending, chan); + /* It states in SCHED_CHAN_PENDING */ } } SMARTLIST_FOREACH_END(chan); @@ -486,18 +466,15 @@ scheduler_channel_wants_writes(channel_t *chan) int became_pending = 0; tor_assert(chan); - tor_assert(channels_waiting_for_cells); - tor_assert(channels_waiting_to_write); tor_assert(channels_pending); /* If it's already in waiting_to_write, we can put it in pending */ - if (smartlist_contains(channels_waiting_to_write, chan)) { + if (chan->scheduler_state == SCHED_CHAN_WAITING_TO_WRITE) { /* - * It's in channels_waiting_to_write, so it shouldn't be in any of - * the other lists. It can write now, so it goes to channels_pending. + * It can write now, so it goes to channels_pending. */ - smartlist_remove(channels_waiting_to_write, chan); smartlist_add(channels_pending, chan); + chan->scheduler_state = SCHED_CHAN_PENDING; log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p went from waiting_to_write " "to pending", @@ -505,13 +482,12 @@ scheduler_channel_wants_writes(channel_t *chan) became_pending = 1; } else { /* - * It's not in waiting_to_write, so it can't become pending; it's - * either not in any of the lists (we add it to waiting_for_cells) - * or it's already in waiting_for_cells or pending (we do nothing) + * It's not in SCHED_CHAN_WAITING_TO_WRITE, so it can't become pending; + * it's either idle and goes to WAITING_FOR_CELLS, or it's a no-op. */ - if (!(smartlist_contains(channels_waiting_for_cells, chan) || - smartlist_contains(channels_pending, chan))) { - smartlist_add(channels_waiting_for_cells, chan); + if (!(chan->scheduler_state == SCHED_CHAN_WAITING_FOR_CELLS || + chan->scheduler_state == SCHED_CHAN_PENDING)) { + chan->scheduler_state = SCHED_CHAN_WAITING_FOR_CELLS; log_debug(LD_SCHED, "Channel " U64_FORMAT " at %p entered waiting_for_cells", U64_PRINTF_ARG(chan->global_identifier), chan);

1 0

[tor/master] Make channel_note_destroy_not_pending() mockable
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 8907554cf3e4e87515fecde40ae2088fdded523f Author: Andrea Shepard <andrea(a)torproject.org> Date: Fri Dec 13 06:13:05 2013 -0800 Make channel_note_destroy_not_pending() mockable --- src/or/circuitlist.c | 4 ++-- src/or/circuitlist.h | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/or/circuitlist.c b/src/or/circuitlist.c index 9d72ea1..7b4eda3 100644 --- a/src/or/circuitlist.c +++ b/src/or/circuitlist.c @@ -302,8 +302,8 @@ channel_note_destroy_pending(channel_t *chan, circid_t id) /** Called to indicate that a DESTROY is no longer pending on <b>chan</b> with * circuit ID <b>id</b> -- typically, because it has been sent. */ -void -channel_note_destroy_not_pending(channel_t *chan, circid_t id) +MOCK_IMPL(void, channel_note_destroy_not_pending, + (channel_t *chan, circid_t id)) { circuit_t *circ = circuit_get_by_circid_channel_even_if_marked(id,chan); if (circ) { diff --git a/src/or/circuitlist.h b/src/or/circuitlist.h index 03934f9..a8d746b 100644 --- a/src/or/circuitlist.h +++ b/src/or/circuitlist.h @@ -72,7 +72,8 @@ void circuit_free_all(void); void circuits_handle_oom(size_t current_allocation); void channel_note_destroy_pending(channel_t *chan, circid_t id); -void channel_note_destroy_not_pending(channel_t *chan, circid_t id); +MOCK_DECL(void, channel_note_destroy_not_pending, + (channel_t *chan, circid_t id)); #ifdef CIRCUITLIST_PRIVATE STATIC void circuit_free(circuit_t *circ);

1 0

[tor/master] Add changes file for global scheduler/cmux refactor
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 52cfaa84b7b2f1098a15c431433f486a150cc854 Author: Andrea Shepard <andrea(a)torproject.org> Date: Thu Dec 12 04:36:15 2013 -0800 Add changes file for global scheduler/cmux refactor --- changes/global_scheduler | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/changes/global_scheduler b/changes/global_scheduler new file mode 100644 index 0000000..d0b8305 --- /dev/null +++ b/changes/global_scheduler @@ -0,0 +1,4 @@ + o Major changes: + - Implement a new inter-cmux comparison API, a global high/low watermark + mechanism and a global scheduler loop for transmission prioritization + across all channels as well as among circuits on one channel.

1 0

[tor/master] Fix scheduler assertion in circuitmux/destroy_cell_queue unit test
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 283646fd9089a337003dcd7c020e0820c7270662 Author: Andrea Shepard <andrea(a)torproject.org> Date: Wed Dec 4 21:43:13 2013 -0800 Fix scheduler assertion in circuitmux/destroy_cell_queue unit test --- src/test/test_circuitmux.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/test/test_circuitmux.c b/src/test/test_circuitmux.c index a3cacc4..c1c8f77 100644 --- a/src/test/test_circuitmux.c +++ b/src/test/test_circuitmux.c @@ -8,6 +8,7 @@ #include "channel.h" #include "circuitmux.h" #include "relay.h" +#include "scheduler.h" #include "test.h" /* XXXX duplicated function from test_circuitlist.c */ @@ -36,6 +37,8 @@ test_cmux_destroy_cell_queue(void *arg) cell_queue_t *cq = NULL; packed_cell_t *pc = NULL; + scheduler_init(); + #ifdef ENABLE_MEMPOOLS init_cell_pool(); #endif /* ENABLE_MEMPOOLS */

1 0

[tor/master] Fix compiler warning
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 63bb9a795e3a05620831512fe7d70690e5f5fb5a Author: Andrea Shepard <andrea(a)torproject.org> Date: Mon Dec 2 01:05:53 2013 -0800 Fix compiler warning --- src/or/scheduler.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/or/scheduler.c b/src/or/scheduler.c index b45df3e..c4c2632 100644 --- a/src/or/scheduler.c +++ b/src/or/scheduler.c @@ -435,7 +435,8 @@ scheduler_run(void) log_debug(LD_SCHED, "Scheduler flushed %d cells onto pending channel " U64_FORMAT " at %p", - flushed, U64_PRINTF_ARG(chan->global_identifier), chan); + (int)flushed, U64_PRINTF_ARG(chan->global_identifier), + chan); } else { log_info(LD_SCHED, "Scheduler saw pending channel " U64_FORMAT " at %p with "

1 0

[tor/master] Sort the scheduler's channel list by cmux comparisons
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 55907da28dd933a7724935a8a4676aec1b54059d Author: Andrea Shepard <andrea(a)torproject.org> Date: Mon Dec 2 01:05:28 2013 -0800 Sort the scheduler's channel list by cmux comparisons --- src/or/scheduler.c | 53 ++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 51 insertions(+), 2 deletions(-) diff --git a/src/or/scheduler.c b/src/or/scheduler.c index 1baf375..b45df3e 100644 --- a/src/or/scheduler.c +++ b/src/or/scheduler.c @@ -139,6 +139,7 @@ static time_t queue_heuristic_timestamp = 0; /* Scheduler static function declarations */ +static int scheduler_compare_channels(const void **c1_v, const void **c2_v); static void scheduler_evt_callback(evutil_socket_t fd, short events, void *arg); static int scheduler_more_work(void); @@ -180,6 +181,52 @@ scheduler_free_all(void) } } +/** + * Comparison function to use when sorting pending channels + */ + +static int +scheduler_compare_channels(const void **c1_v, const void **c2_v) +{ + channel_t *c1 = NULL, *c2 = NULL; + /* These are a workaround for -Wbad-function-cast throwing a fit */ + const circuitmux_policy_t *p1, *p2; + uintptr_t p1_i, p2_i; + + tor_assert(c1_v); + tor_assert(c2_v); + + c1 = (channel_t *)(*c1_v); + c2 = (channel_t *)(*c2_v); + + tor_assert(c1); + tor_assert(c2); + + if (c1 != c2) { + if (circuitmux_get_policy(c1->cmux) == + circuitmux_get_policy(c2->cmux)) { + /* Same cmux policy, so use the mux comparison */ + return circuitmux_compare_muxes(c1->cmux, c2->cmux); + } else { + /* + * Different policies; not important to get this edge case perfect + * because the current code never actually gives different channels + * different cmux policies anyway. Just use this arbitrary but + * definite choice. + */ + p1 = circuitmux_get_policy(c1->cmux); + p2 = circuitmux_get_policy(c2->cmux); + p1_i = (uintptr_t)p1; + p2_i = (uintptr_t)p2; + + return (p1_i < p2_i) ? -1 : 1; + } + } else { + /* c1 == c2, so always equal */ + return 0; + } +} + /* * Scheduler event callback; this should get triggered once per event loop * if any scheduling work was created during the event loop. @@ -362,9 +409,11 @@ scheduler_run(void) channels_pending = smartlist_new(); /* - * For now, just run the old scheduler on all the chans in the list, until - * we hit the high-water mark. TODO real channel priority API + * UGLY HACK: sort the list on each invocation + * + * TODO smarter data structures */ + smartlist_sort(tmp, scheduler_compare_channels); SMARTLIST_FOREACH_BEGIN(tmp, channel_t *, chan) { if (scheduler_get_queue_heuristic() <= SCHED_Q_HIGH_WATER) {

1 0

[tor/master] Add inter-cmux comparison support to circuitmux_ewma.c
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 700d6e75258fb94819df3da686482dd8565e3691 Author: Andrea Shepard <andrea(a)torproject.org> Date: Sat Nov 30 03:04:36 2013 -0800 Add inter-cmux comparison support to circuitmux_ewma.c --- src/or/circuitmux_ewma.c | 58 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 57 insertions(+), 1 deletion(-) diff --git a/src/or/circuitmux_ewma.c b/src/or/circuitmux_ewma.c index 3f37d7b..3a2df26 100644 --- a/src/or/circuitmux_ewma.c +++ b/src/or/circuitmux_ewma.c @@ -187,6 +187,9 @@ ewma_notify_xmit_cells(circuitmux_t *cmux, static circuit_t * ewma_pick_active_circuit(circuitmux_t *cmux, circuitmux_policy_data_t *pol_data); +static int +ewma_cmp_cmux(circuitmux_t *cmux_1, circuitmux_policy_data_t *pol_data_1, + circuitmux_t *cmux_2, circuitmux_policy_data_t *pol_data_2); /*** EWMA global variables ***/ @@ -209,7 +212,8 @@ circuitmux_policy_t ewma_policy = { /*.notify_circ_inactive =*/ ewma_notify_circ_inactive, /*.notify_set_n_cells =*/ NULL, /* EWMA doesn't need this */ /*.notify_xmit_cells =*/ ewma_notify_xmit_cells, - /*.pick_active_circuit =*/ ewma_pick_active_circuit + /*.pick_active_circuit =*/ ewma_pick_active_circuit, + /*.cmp_cmux =*/ ewma_cmp_cmux }; /*** EWMA method implementations using the below EWMA helper functions ***/ @@ -453,6 +457,58 @@ ewma_pick_active_circuit(circuitmux_t *cmux, return circ; } +/** + * Compare two EWMA cmuxes, and return -1, 0 or 1 to indicate which should + * be more preferred - see circuitmux_compare_muxes() of circuitmux.c. + */ + +static int +ewma_cmp_cmux(circuitmux_t *cmux_1, circuitmux_policy_data_t *pol_data_1, + circuitmux_t *cmux_2, circuitmux_policy_data_t *pol_data_2) +{ + ewma_policy_data_t *p1 = NULL, *p2 = NULL; + cell_ewma_t *ce1 = NULL, *ce2 = NULL; + + tor_assert(cmux_1); + tor_assert(pol_data_1); + tor_assert(cmux_2); + tor_assert(pol_data_2); + + p1 = TO_EWMA_POL_DATA(pol_data_1); + p2 = TO_EWMA_POL_DATA(pol_data_1); + + if (p1 != p2) { + /* Get the head cell_ewma_t from each queue */ + if (smartlist_len(p1->active_circuit_pqueue) > 0) { + ce1 = smartlist_get(p1->active_circuit_pqueue, 0); + } + + if (smartlist_len(p2->active_circuit_pqueue) > 0) { + ce2 = smartlist_get(p2->active_circuit_pqueue, 0); + } + + /* Got both of them? */ + if (ce1 != NULL && ce2 != NULL) { + /* Pick whichever one has the better best circuit */ + return compare_cell_ewma_counts(ce1, ce2); + } else { + if (ce1 != NULL ) { + /* We only have a circuit on cmux_1, so prefer it */ + return -1; + } else if (ce2 != NULL) { + /* We only have a circuit on cmux_2, so prefer it */ + return 1; + } else { + /* No circuits at all; no preference */ + return 0; + } + } + } else { + /* We got identical params */ + return 0; + } +} + /** Helper for sorting cell_ewma_t values in their priority queue. */ static int compare_cell_ewma_counts(const void *p1, const void *p2)

1 0

[tor/master] Schedule according to a queue size heuristic
by nickm＠torproject.org 28 Nov '14

28 Nov '14

commit 1275002a46dfb131f6db5c0fe28bc1828db327e2 Author: Andrea Shepard <andrea(a)torproject.org> Date: Thu Nov 14 04:45:47 2013 -0800 Schedule according to a queue size heuristic --- src/or/channel.c | 4 ++ src/or/scheduler.c | 192 +++++++++++++++++++++++++++++++++++++++++++++------- src/or/scheduler.h | 3 + 3 files changed, 173 insertions(+), 26 deletions(-) diff --git a/src/or/channel.c b/src/or/channel.c index dddd7ab..7ed3894 100644 --- a/src/or/channel.c +++ b/src/or/channel.c @@ -4563,6 +4563,8 @@ channel_update_xmit_queue_size(channel_t *chan) U64_FORMAT ", new size is " U64_FORMAT, U64_PRINTF_ARG(adj), U64_PRINTF_ARG(chan->global_identifier), U64_PRINTF_ARG(estimated_total_queue_size)); + /* Tell the scheduler we're increasing the queue size */ + scheduler_adjust_queue_size(chan, 1, adj); } } else if (queued < chan->bytes_queued_for_xmit) { adj = chan->bytes_queued_for_xmit - queued; @@ -4585,6 +4587,8 @@ channel_update_xmit_queue_size(channel_t *chan) U64_FORMAT ", new size is " U64_FORMAT, U64_PRINTF_ARG(adj), U64_PRINTF_ARG(chan->global_identifier), U64_PRINTF_ARG(estimated_total_queue_size)); + /* Tell the scheduler we're decreasing the queue size */ + scheduler_adjust_queue_size(chan, -1, adj); } } } diff --git a/src/or/scheduler.c b/src/or/scheduler.c index 1c6a2fd..1baf375 100644 --- a/src/or/scheduler.c +++ b/src/or/scheduler.c @@ -7,7 +7,10 @@ **/ #include "or.h" + +#define TOR_CHANNEL_INTERNAL_ /* For channel_flush_some_cells() */ #include "channel.h" + #include "compat_libevent.h" #include "scheduler.h" @@ -17,6 +20,9 @@ #include <event.h> #endif +#define SCHED_Q_LOW_WATER 16384 +#define SCHED_Q_HIGH_WATER (2 * SCHED_Q_LOW_WATER) + /* * Write scheduling works by keeping track of lists of channels that can * accept cells, and have cells to write. From the scheduler's perspective, @@ -118,6 +124,19 @@ static smartlist_t *channels_pending = NULL; static struct event *run_sched_ev = NULL; +/* + * Queue heuristic; this is not the queue size, but an 'effective queuesize' + * that ages out contributions from stalled channels. + */ + +static uint64_t queue_heuristic = 0; + +/* + * Timestamp for last queue heuristic update + */ + +static time_t queue_heuristic_timestamp = 0; + /* Scheduler static function declarations */ static void scheduler_evt_callback(evutil_socket_t fd, @@ -127,6 +146,8 @@ static void scheduler_retrigger(void); #if 0 static void scheduler_trigger(void); #endif +static uint64_t scheduler_get_queue_heuristic(void); +static void scheduler_update_queue_heuristic(time_t now); /* Scheduler function implementations */ @@ -281,6 +302,8 @@ scheduler_init(void) channels_waiting_for_cells = smartlist_new(); channels_waiting_to_write = smartlist_new(); channels_pending = smartlist_new(); + queue_heuristic = 0; + queue_heuristic_timestamp = approx_time(); } /** Check if there's more scheduling work */ @@ -290,7 +313,8 @@ scheduler_more_work(void) { tor_assert(channels_pending); - return (smartlist_len(channels_pending) > 0) ? 1 : 0; + return ((scheduler_get_queue_heuristic() < SCHED_Q_LOW_WATER) && + ((smartlist_len(channels_pending) > 0))) ? 1 : 0; } /** Retrigger the scheduler in a way safe to use from the callback */ @@ -324,39 +348,70 @@ void scheduler_run(void) { smartlist_t *tmp = NULL; - int n_cells; + int n_cells, n_chans_before, n_chans_after; + uint64_t q_len_before, q_heur_before, q_len_after, q_heur_after; ssize_t flushed, flushed_this_time; log_debug(LD_SCHED, "We have a chance to run the scheduler"); - tmp = channels_pending; - channels_pending = smartlist_new(); + if (scheduler_get_queue_heuristic() < SCHED_Q_LOW_WATER) { + n_chans_before = smartlist_len(channels_pending); + q_len_before = channel_get_global_queue_estimate(); + q_heur_before = scheduler_get_queue_heuristic(); + tmp = channels_pending; + channels_pending = smartlist_new(); - /* For now, just run the old scheduler on all the chans in the list */ + /* + * For now, just run the old scheduler on all the chans in the list, until + * we hit the high-water mark. TODO real channel priority API + */ - SMARTLIST_FOREACH_BEGIN(tmp, channel_t *, 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); - - 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; + SMARTLIST_FOREACH_BEGIN(tmp, channel_t *, chan) { + if (scheduler_get_queue_heuristic() <= SCHED_Q_HIGH_WATER) { + 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); + + 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; + } + + log_debug(LD_SCHED, + "Scheduler flushed %d cells onto pending channel " + U64_FORMAT " at %p", + flushed, U64_PRINTF_ARG(chan->global_identifier), chan); + } else { + log_info(LD_SCHED, + "Scheduler saw pending channel " U64_FORMAT " at %p with " + "no cells writeable", + U64_PRINTF_ARG(chan->global_identifier), chan); + } + } else { + /* Not getting it this round; put it back on the list */ + smartlist_add(channels_pending, chan); } - } 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_FOREACH_END(chan); - smartlist_free(tmp); + smartlist_free(tmp); + + n_chans_after = smartlist_len(channels_pending); + q_len_after = channel_get_global_queue_estimate(); + q_heur_after = scheduler_get_queue_heuristic(); + log_debug(LD_SCHED, + "Scheduler handled %d of %d pending channels, queue size from " + U64_FORMAT " to " U64_FORMAT ", queue heuristic from " + U64_FORMAT " to " U64_FORMAT, + n_chans_before - n_chans_after, n_chans_before, + U64_PRINTF_ARG(q_len_before), U64_PRINTF_ARG(q_len_after), + U64_PRINTF_ARG(q_heur_before), U64_PRINTF_ARG(q_heur_after)); + } } /** Trigger the scheduling event so we run the scheduler later */ @@ -420,3 +475,88 @@ scheduler_channel_wants_writes(channel_t *chan) if (became_pending) scheduler_retrigger(); } +/** + * Notify the scheduler of a queue size adjustment, to recalculate the + * queue heuristic. + */ + +void +scheduler_adjust_queue_size(channel_t *chan, char dir, uint64_t adj) +{ + time_t now = approx_time(); + + log_debug(LD_SCHED, + "Queue size adjustment by %s" U64_FORMAT " for channel " + U64_FORMAT, + (dir >= 0) ? "+" : "-", + U64_PRINTF_ARG(adj), + U64_PRINTF_ARG(chan->global_identifier)); + + /* Get the queue heuristic up to date */ + scheduler_update_queue_heuristic(now); + + /* Adjust as appropriate */ + if (dir >= 0) { + /* Increasing it */ + queue_heuristic += adj; + } else { + /* Decreasing it */ + if (queue_heuristic > adj) queue_heuristic -= adj; + else queue_heuristic = 0; + } + + log_debug(LD_SCHED, + "Queue heuristic is now " U64_FORMAT, + U64_PRINTF_ARG(queue_heuristic)); +} + +/** + * Query the current value of the queue heuristic + */ + +static uint64_t +scheduler_get_queue_heuristic(void) +{ + time_t now = approx_time(); + + scheduler_update_queue_heuristic(now); + + return queue_heuristic; +} + +/** + * Adjust the queue heuristic value to the present time + */ + +static void +scheduler_update_queue_heuristic(time_t now) +{ + time_t diff; + + if (queue_heuristic_timestamp == 0) { + /* + * Nothing we can sensibly do; must not have been initted properly. + * Oh well. + */ + queue_heuristic_timestamp = now; + } else if (queue_heuristic_timestamp < now) { + diff = now - queue_heuristic_timestamp; + /* + * This is a simple exponential age-out; the other proposed alternative + * was a linear age-out using the bandwidth history in rephist.c; I'm + * going with this out of concern that if an adversary can jam the + * scheduler long enough, it would cause the bandwidth to drop to + * zero and render the aging mechanism ineffective thereafter. + */ + if (0 <= diff && diff < 64) queue_heuristic >>= diff; + else queue_heuristic = 0; + + queue_heuristic_timestamp = now; + + log_debug(LD_SCHED, + "Queue heuristic is now " U64_FORMAT, + U64_PRINTF_ARG(queue_heuristic)); + } + /* else no update needed, or time went backward */ +} + diff --git a/src/or/scheduler.h b/src/or/scheduler.h index b25e36e..8fe59cb 100644 --- a/src/or/scheduler.h +++ b/src/or/scheduler.h @@ -27,5 +27,8 @@ void scheduler_channel_wants_writes(channel_t *chan); /* Notify the scheduler of a channel being closed */ void scheduler_release_channel(channel_t *chan); +/* Notify scheduler of queue size adjustments */ +void scheduler_adjust_queue_size(channel_t *chan, char dir, uint64_t adj); + #endif /* !defined(TOR_SCHEDULER_H) */

1 0