Merge remote-tracking branch 'andrea/cmux_refactor_configurable_threshold'

Conflicts:
	src/or/or.h
	src/test/Makefile.nmake

25 files changed, 1603 insertions, 169 deletions

diff --git a/src/or/Makefile.nmake b/src/or/Makefile.nmake
index 523bf3306b..2ac98cd372 100644
--- a/src/or/Makefile.nmake
+++ b/src/or/Makefile.nmake
@@ -63,6 +63,7 @@ LIBTOR_OBJECTS = \
   routerlist.obj \
   routerparse.obj \
   routerset.obj \
+  scheduler.obj \
   statefile.obj \
   status.obj \
   transports.obj
diff --git a/src/or/buffers.c b/src/or/buffers.c
index bd33fe451d..4cdc03bc03 100644
--- a/src/or/buffers.c
+++ b/src/or/buffers.c
@@ -562,8 +562,8 @@ buf_clear(buf_t *buf)
 }
 
 /** Return the number of bytes stored in <b>buf</b> */
-size_t
-buf_datalen(const buf_t *buf)
+MOCK_IMPL(size_t,
+buf_datalen, (const buf_t *buf))
 {
   return buf->datalen;
 }
diff --git a/src/or/buffers.h b/src/or/buffers.h
index 9c2c7d0e9d..4687fbefd7 100644
--- a/src/or/buffers.h
+++ b/src/or/buffers.h
@@ -24,7 +24,7 @@ void buf_shrink(buf_t *buf);
 size_t buf_shrink_freelists(int free_all);
 void buf_dump_freelist_sizes(int severity);
 
-size_t buf_datalen(const buf_t *buf);
+MOCK_DECL(size_t, buf_datalen, (const buf_t *buf));
 size_t buf_allocation(const buf_t *buf);
 size_t buf_slack(const buf_t *buf);
 
diff --git a/src/or/channel.c b/src/or/channel.c
index 13a122662a..5ee0748802 100644
--- a/src/or/channel.c
+++ b/src/or/channel.c
@@ -13,6 +13,9 @@
 
 #define TOR_CHANNEL_INTERNAL_
 
+/* This one's for stuff only channel.c and the test suite should see */
+#define CHANNEL_PRIVATE_
+
 #include "or.h"
 #include "channel.h"
 #include "channeltls.h"
@@ -29,29 +32,7 @@
 #include "rephist.h"
 #include "router.h"
 #include "routerlist.h"
-
-/* Cell queue structure */
-
-typedef struct cell_queue_entry_s cell_queue_entry_t;
-struct cell_queue_entry_s {
-  TOR_SIMPLEQ_ENTRY(cell_queue_entry_s) next;
-  enum {
-    CELL_QUEUE_FIXED,
-    CELL_QUEUE_VAR,
-    CELL_QUEUE_PACKED
-  } type;
-  union {
-    struct {
-      cell_t *cell;
-    } fixed;
-    struct {
-      var_cell_t *var_cell;
-    } var;
-    struct {
-      packed_cell_t *packed_cell;
-    } packed;
-  } u;
-};
+#include "scheduler.h"
 
 /* Global lists of channels */
 
@@ -76,6 +57,60 @@ 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;
+
+/*
+ * 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
@@ -123,6 +158,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);
@@ -140,6 +177,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);
 
@@ -746,6 +785,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;
 }
 
 /**
@@ -788,6 +830,9 @@ channel_free(channel_t *chan)
             "Freeing channel " U64_FORMAT " at %p",
             U64_PRINTF_ARG(chan->global_identifier), chan);
 
+  /* Get this one out of the scheduler */
+  scheduler_release_channel(chan);
+
   /*
    * Get rid of cmux policy before we do anything, so cmux policies don't
    * see channels in weird half-freed states.
@@ -863,6 +908,9 @@ channel_force_free(channel_t *chan)
             "Force-freeing channel " U64_FORMAT " at %p",
             U64_PRINTF_ARG(chan->global_identifier), chan);
 
+  /* Get this one out of the scheduler */
+  scheduler_release_channel(chan);
+
   /*
    * Get rid of cmux policy before we do anything, so cmux policies don't
    * see channels in weird half-freed states.
@@ -1666,6 +1714,36 @@ 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)
+{
+  size_t rv = 0;
+
+  tor_assert(chan);
+  tor_assert(q);
+
+  switch (q->type) {
+    case CELL_QUEUE_FIXED:
+      rv = get_cell_network_size(chan->wide_circ_ids);
+      break;
+    case CELL_QUEUE_VAR:
+      rv = get_var_cell_header_size(chan->wide_circ_ids) +
+           (q->u.var.var_cell ? q->u.var.var_cell->payload_len : 0);
+      break;
+    case CELL_QUEUE_PACKED:
+      rv = get_cell_network_size(chan->wide_circ_ids);
+      break;
+    default:
+      tor_assert(1);
+  }
+
+  return rv;
+}
+
+/**
  * 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
@@ -1677,6 +1755,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);
@@ -1693,6 +1772,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) {
@@ -1726,6 +1808,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();
     }
   }
 
@@ -1737,6 +1826,14 @@ 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();
+    /* 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);
   }
@@ -1775,6 +1872,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);
 }
 
 /**
@@ -1810,6 +1910,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);
 }
 
 /**
@@ -1846,6 +1949,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);
 }
 
 /**
@@ -1941,6 +2047,41 @@ channel_change_state(channel_t *chan, channel_state_t to_state)
     }
   }
 
+  /*
+   * If we're going to a closed/closing state, we don't need scheduling any
+   * more; in CHANNEL_STATE_MAINT we can't accept writes.
+   */
+  if (to_state == CHANNEL_STATE_CLOSING ||
+      to_state == CHANNEL_STATE_CLOSED ||
+      to_state == CHANNEL_STATE_ERROR) {
+    scheduler_release_channel(chan);
+  } else if (to_state == CHANNEL_STATE_MAINT) {
+    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);
@@ -2056,12 +2197,13 @@ channel_listener_change_state(channel_listener_t *chan_l,
 
 #define MAX_CELLS_TO_GET_FROM_CIRCUITS_FOR_UNLIMITED 256
 
-ssize_t
-channel_flush_some_cells(channel_t *chan, ssize_t num_cells)
+MOCK_IMPL(ssize_t,
+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);
 
@@ -2087,14 +2229,45 @@ 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) &&
+            (unlimited || (flushed < num_cells))) {
+          flushed += channel_flush_some_cells_from_outgoing_queue(chan,
+            (unlimited ? -1 : num_cells - flushed));
+        }
       }
     }
   }
@@ -2117,6 +2290,8 @@ 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;
+  int free_q = 0, handed_off = 0;
 
   tor_assert(chan);
   tor_assert(chan->write_cell);
@@ -2130,8 +2305,12 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan,
   if (chan->state == CHANNEL_STATE_OPEN) {
     while ((unlimited || num_cells > flushed) &&
            NULL != (q = TOR_SIMPLEQ_FIRST(&chan->outgoing_queue))) {
+      free_q = 0;
+      handed_off = 0;
 
       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.
@@ -2144,8 +2323,9 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan,
                 ++flushed;
                 channel_timestamp_xmit(chan);
                 ++(chan->n_cells_xmitted);
-                cell_queue_entry_free(q, 1);
-                q = NULL;
+                chan->n_bytes_xmitted += cell_size;
+                free_q = 1;
+                handed_off = 1;
               }
               /* Else couldn't write it; leave it on the queue */
             } else {
@@ -2156,8 +2336,8 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan,
                        "(global ID " U64_FORMAT ").",
                        chan, U64_PRINTF_ARG(chan->global_identifier));
               /* Throw it away */
-              cell_queue_entry_free(q, 0);
-              q = NULL;
+              free_q = 1;
+              handed_off = 0;
             }
             break;
          case CELL_QUEUE_PACKED:
@@ -2167,8 +2347,9 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan,
                 ++flushed;
                 channel_timestamp_xmit(chan);
                 ++(chan->n_cells_xmitted);
-                cell_queue_entry_free(q, 1);
-                q = NULL;
+                chan->n_bytes_xmitted += cell_size;
+                free_q = 1;
+                handed_off = 1;
               }
               /* Else couldn't write it; leave it on the queue */
             } else {
@@ -2179,8 +2360,8 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan,
                        "(global ID " U64_FORMAT ").",
                        chan, U64_PRINTF_ARG(chan->global_identifier));
               /* Throw it away */
-              cell_queue_entry_free(q, 0);
-              q = NULL;
+              free_q = 1;
+              handed_off = 0;
             }
             break;
          case CELL_QUEUE_VAR:
@@ -2190,8 +2371,9 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan,
                 ++flushed;
                 channel_timestamp_xmit(chan);
                 ++(chan->n_cells_xmitted);
-                cell_queue_entry_free(q, 1);
-                q = NULL;
+                chan->n_bytes_xmitted += cell_size;
+                free_q = 1;
+                handed_off = 1;
               }
               /* Else couldn't write it; leave it on the queue */
             } else {
@@ -2202,8 +2384,8 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan,
                        "(global ID " U64_FORMAT ").",
                        chan, U64_PRINTF_ARG(chan->global_identifier));
               /* Throw it away */
-              cell_queue_entry_free(q, 0);
-              q = NULL;
+              free_q = 1;
+              handed_off = 0;
             }
             break;
           default:
@@ -2213,12 +2395,32 @@ channel_flush_some_cells_from_outgoing_queue(channel_t *chan,
                      "(global ID " U64_FORMAT "; ignoring it."
                      "  Someone should fix this.",
                      q->type, chan, U64_PRINTF_ARG(chan->global_identifier));
-            cell_queue_entry_free(q, 0);
-            q = NULL;
+            free_q = 1;
+            handed_off = 0;
         }
 
-        /* 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 free_q is set, we used it and should remove the queue entry;
+         * we have to do the free down here so TOR_SIMPLEQ_REMOVE_HEAD isn't
+         * accessing freed memory
+         */
+        if (free_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();
+          /* Update the channel's queue size too */
+          chan->bytes_in_queue -= cell_size;
+          /* Finally, free q */
+          cell_queue_entry_free(q, handed_off);
+          q = NULL;
+        }
         /* No cell removed from list, so we can't go on any further */
         else break;
       }
@@ -2230,6 +2432,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;
 }
 
@@ -2541,8 +2746,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) {
@@ -2596,6 +2802,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) {
@@ -2645,6 +2853,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,
@@ -2727,6 +2948,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,
@@ -3200,7 +3434,7 @@ channel_listener_describe_transport(channel_listener_t *chan_l)
 /**
  * Return the number of entries in <b>queue</b>
  */
-static int
+STATIC int
 chan_cell_queue_len(const chan_cell_queue_t *queue)
 {
   int r = 0;
@@ -3216,8 +3450,8 @@ chan_cell_queue_len(const chan_cell_queue_t *queue)
  * Dump statistics for one channel to the log
  */
 
-void
-channel_dump_statistics(channel_t *chan, int severity)
+MOCK_IMPL(void,
+channel_dump_statistics, (channel_t *chan, int severity))
 {
   double avg, interval, age;
   time_t now = time(NULL);
@@ -3369,12 +3603,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) {
@@ -3390,6 +3634,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) {
@@ -3807,6 +4058,50 @@ channel_mark_outgoing(channel_t *chan)
   chan->is_incoming = 0;
 }
 
+/************************
+ * Flow control queries *
+ ***********************/
+
+/*
+ * 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
+ * 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 *
  ********************/
@@ -4209,3 +4504,87 @@ 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));
+      /* 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;
+    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));
+      /* Tell the scheduler we're decreasing the queue size */
+      scheduler_adjust_queue_size(chan, -1, adj);
+    }
+  }
+}
+
diff --git a/src/or/channel.h b/src/or/channel.h
index 4cd8f4391e..5c2a1a35e2 100644
--- a/src/or/channel.h
+++ b/src/or/channel.h
@@ -57,6 +57,32 @@ 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;
+
+  /** 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 */
@@ -79,6 +105,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,7 +141,11 @@ 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 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 */
   int (*write_cell)(channel_t *, cell_t *);
   /** Write a packed cell to an open channel */
   int (*write_packed_cell)(channel_t *, packed_cell_t *);
@@ -198,8 +233,16 @@ 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;
+
+  /** 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 {
@@ -311,6 +354,34 @@ void channel_set_cmux_policy_everywhere(circuitmux_policy_t *pol);
 
 #ifdef TOR_CHANNEL_INTERNAL_
 
+#ifdef CHANNEL_PRIVATE_
+/* Cell queue structure (here rather than channel.c for test suite use) */
+
+typedef struct cell_queue_entry_s cell_queue_entry_t;
+struct cell_queue_entry_s {
+  TOR_SIMPLEQ_ENTRY(cell_queue_entry_s) next;
+  enum {
+    CELL_QUEUE_FIXED,
+    CELL_QUEUE_VAR,
+    CELL_QUEUE_PACKED
+  } type;
+  union {
+    struct {
+      cell_t *cell;
+    } fixed;
+    struct {
+      var_cell_t *var_cell;
+    } var;
+    struct {
+      packed_cell_t *packed_cell;
+    } packed;
+  } u;
+};
+
+/* Cell queue functions for benefit of test suite */
+STATIC int chan_cell_queue_len(const chan_cell_queue_t *queue);
+#endif
+
 /* Channel operations for subclasses and internal use only */
 
 /* Initialize a newly allocated channel - do this first in subclass
@@ -384,7 +455,8 @@ void channel_queue_var_cell(channel_t *chan, var_cell_t *var_cell);
 void channel_flush_cells(channel_t *chan);
 
 /* Request from lower layer for more cells if available */
-ssize_t channel_flush_some_cells(channel_t *chan, ssize_t num_cells);
+MOCK_DECL(ssize_t, channel_flush_some_cells,
+          (channel_t *chan, ssize_t num_cells));
 
 /* Query if data available on this channel */
 int channel_more_to_flush(channel_t *chan);
@@ -435,7 +507,7 @@ channel_t * channel_next_with_digest(channel_t *chan);
  */
 
 const char * channel_describe_transport(channel_t *chan);
-void channel_dump_statistics(channel_t *chan, int severity);
+MOCK_DECL(void, channel_dump_statistics, (channel_t *chan, int severity));
 void channel_dump_transport_statistics(channel_t *chan, int severity);
 const char * channel_get_actual_remote_descr(channel_t *chan);
 const char * channel_get_actual_remote_address(channel_t *chan);
@@ -458,6 +530,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,
@@ -465,6 +538,10 @@ 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 */
+uint64_t channel_get_global_queue_estimate(void);
+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 db044aee56..719a153dd6 100644
--- a/src/or/channeltls.c
+++ b/src/or/channeltls.c
@@ -25,6 +25,7 @@
 #include "relay.h"
 #include "router.h"
 #include "routerlist.h"
+#include "scheduler.h"
 
 /** How many CELL_PADDING cells have we received, ever? */
 uint64_t stats_n_padding_cells_processed = 0;
@@ -54,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
@@ -67,6 +69,8 @@ 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 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,
@@ -116,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;
@@ -123,6 +128,8 @@ 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;
   chan->write_var_cell = channel_tls_write_var_cell_method;
@@ -435,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
@@ -673,6 +714,50 @@ 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
+ * 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
@@ -867,6 +952,10 @@ channel_tls_handle_state_change_on_orconn(channel_tls_t *chan,
      * CHANNEL_STATE_MAINT on this.
      */
     channel_change_state(base_chan, CHANNEL_STATE_OPEN);
+    /* We might have just become writeable; check and tell the scheduler */
+    if (connection_or_num_cells_writeable(conn) > 0) {
+      scheduler_channel_wants_writes(base_chan);
+    }
   } else {
     /*
      * Not open, so from CHANNEL_STATE_OPEN we go to CHANNEL_STATE_MAINT,
@@ -878,58 +967,6 @@ channel_tls_handle_state_change_on_orconn(channel_tls_t *chan,
   }
 }
 
-/**
- * Flush cells from a channel_tls_t
- *
- * Try to flush up to about num_cells cells, and return how many we flushed.
- */
-
-ssize_t
-channel_tls_flush_some_cells(channel_tls_t *chan, ssize_t num_cells)
-{
-  ssize_t flushed = 0;
-
-  tor_assert(chan);
-
-  if (flushed >= num_cells) goto done;
-
-  /*
-   * If channel_tls_t ever buffers anything below the channel_t layer, flush
-   * that first here.
-   */
-
-  flushed += channel_flush_some_cells(TLS_CHAN_TO_BASE(chan),
-                                      num_cells - flushed);
-
-  /*
-   * If channel_tls_t ever buffers anything below the channel_t layer, check
-   * how much we actually got and push it on down here.
-   */
-
- done:
-  return flushed;
-}
-
-/**
- * Check if a channel_tls_t has anything to flush
- *
- * Return true if there is any more to flush on this channel (cells in queue
- * or active circuits).
- */
-
-int
-channel_tls_more_to_flush(channel_tls_t *chan)
-{
-  tor_assert(chan);
-
-  /*
-   * If channel_tls_t ever buffers anything below channel_t, the
-   * check for that should go here first.
-   */
-
-  return channel_more_to_flush(TLS_CHAN_TO_BASE(chan));
-}
-
 #ifdef KEEP_TIMING_STATS
 
 /**
diff --git a/src/or/channeltls.h b/src/or/channeltls.h
index 06be6fa555..133ad43bb4 100644
--- a/src/or/channeltls.h
+++ b/src/or/channeltls.h
@@ -40,8 +40,6 @@ channel_t * channel_tls_to_base(channel_tls_t *tlschan);
 channel_tls_t * channel_tls_from_base(channel_t *chan);
 
 /* Things for connection_or.c to call back into */
-ssize_t channel_tls_flush_some_cells(channel_tls_t *chan, ssize_t num_cells);
-int channel_tls_more_to_flush(channel_tls_t *chan);
 void channel_tls_handle_cell(cell_t *cell, or_connection_t *conn);
 void channel_tls_handle_state_change_on_orconn(channel_tls_t *chan,
                                                or_connection_t *conn,
diff --git a/src/or/circuitbuild.c b/src/or/circuitbuild.c
index 34934dc519..ddb186bea4 100644
--- a/src/or/circuitbuild.c
+++ b/src/or/circuitbuild.c
@@ -14,6 +14,7 @@
 #include "or.h"
 #include "channel.h"
 #include "circpathbias.h"
+#define CIRCUITBUILD_PRIVATE
 #include "circuitbuild.h"
 #include "circuitlist.h"
 #include "circuitstats.h"
diff --git a/src/or/circuitlist.c b/src/or/circuitlist.c
index c7287f921d..0e88b47676 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 addaa725d4..ea1076d53f 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);
diff --git a/src/or/circuitmux.c b/src/or/circuitmux.c
index 663711c6c0..443dad0a54 100644
--- a/src/or/circuitmux.c
+++ b/src/or/circuitmux.c
@@ -621,8 +621,8 @@ circuitmux_clear_policy(circuitmux_t *cmux)
  * Return the policy currently installed on a circuitmux_t
  */
 
-const circuitmux_policy_t *
-circuitmux_get_policy(circuitmux_t *cmux)
+MOCK_IMPL(const circuitmux_policy_t *,
+circuitmux_get_policy, (circuitmux_t *cmux))
 {
   tor_assert(cmux);
 
@@ -896,8 +896,8 @@ circuitmux_num_cells_for_circuit(circuitmux_t *cmux, circuit_t *circ)
  * Query total number of available cells on a circuitmux
  */
 
-unsigned int
-circuitmux_num_cells(circuitmux_t *cmux)
+MOCK_IMPL(unsigned int,
+circuitmux_num_cells, (circuitmux_t *cmux))
 {
   tor_assert(cmux);
 
@@ -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.
+ */
+
+MOCK_IMPL(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 eade2486a2..53092cd66c 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);
 };
 
 /*
@@ -105,7 +108,8 @@ void circuitmux_free(circuitmux_t *cmux);
 
 /* Policy control */
 void circuitmux_clear_policy(circuitmux_t *cmux);
-const circuitmux_policy_t * circuitmux_get_policy(circuitmux_t *cmux);
+MOCK_DECL(const circuitmux_policy_t *,
+          circuitmux_get_policy, (circuitmux_t *cmux));
 void circuitmux_set_policy(circuitmux_t *cmux,
                            const circuitmux_policy_t *pol);
 
@@ -117,7 +121,7 @@ int circuitmux_is_circuit_attached(circuitmux_t *cmux, circuit_t *circ);
 int circuitmux_is_circuit_active(circuitmux_t *cmux, circuit_t *circ);
 unsigned int circuitmux_num_cells_for_circuit(circuitmux_t *cmux,
                                               circuit_t *circ);
-unsigned int circuitmux_num_cells(circuitmux_t *cmux);
+MOCK_DECL(unsigned int, circuitmux_num_cells, (circuitmux_t *cmux));
 unsigned int circuitmux_num_circuits(circuitmux_t *cmux);
 unsigned int circuitmux_num_active_circuits(circuitmux_t *cmux);
 
@@ -148,5 +152,9 @@ 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 */
+MOCK_DECL(int, circuitmux_compare_muxes,
+          (circuitmux_t *cmux_1, circuitmux_t *cmux_2));
+
 #endif /* TOR_CIRCUITMUX_H */
 
diff --git a/src/or/circuitmux_ewma.c b/src/or/circuitmux_ewma.c
index 49d899e5e7..0d7d6ef197 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)
diff --git a/src/or/config.c b/src/or/config.c
index 46c090e65e..ced4288f00 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"
@@ -368,6 +369,9 @@ static config_var_t option_vars_[] = {
   V(ServerDNSSearchDomains,      BOOL,     "0"),
   V(ServerDNSTestAddresses,      CSV,
       "www.google.com,www.mit.edu,www.yahoo.com,www.slashdot.org"),
+  V(SchedulerLowWaterMark,       MEMUNIT,  "16 kB"),
+  V(SchedulerHighWaterMark,      MEMUNIT,  "32 kB"),
+  V(SchedulerMaxFlushCells,      UINT,     "16"),
   V(ShutdownWaitLength,          INTERVAL, "30 seconds"),
   V(SocksListenAddress,          LINELIST, NULL),
   V(SocksPolicy,                 LINELIST, NULL),
@@ -1045,6 +1049,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. */
@@ -1526,6 +1538,25 @@ options_act(const or_options_t *old_options)
     return -1;
   }
 
+  /* Set up scheduler thresholds */
+  if (options->SchedulerLowWaterMark > 0 &&
+      options->SchedulerHighWaterMark > options->SchedulerLowWaterMark) {
+    scheduler_set_watermarks(options->SchedulerLowWaterMark,
+                             options->SchedulerHighWaterMark,
+                             (options->SchedulerMaxFlushCells > 0) ?
+                              options->SchedulerMaxFlushCells : 16);
+  } else {
+    if (options->SchedulerLowWaterMark == 0) {
+      log_warn(LD_GENERAL, "Bad SchedulerLowWaterMark option");
+    }
+
+    if (options->SchedulerHighWaterMark <= options->SchedulerLowWaterMark) {
+      log_warn(LD_GENERAL, "Bad SchedulerHighWaterMark option");
+    }
+
+    return -1;
+  }
+
   /* Set up accounting */
   if (accounting_parse_options(options, 0)<0) {
     log_warn(LD_CONFIG,"Error in accounting options");
@@ -2269,8 +2300,8 @@ resolve_my_address(int warn_severity, const or_options_t *options,
 /** Return true iff <b>addr</b> is judged to be on the same network as us, or
  * on a private network.
  */
-int
-is_local_addr(const tor_addr_t *addr)
+MOCK_IMPL(int,
+is_local_addr, (const tor_addr_t *addr))
 {
   if (tor_addr_is_internal(addr, 0))
     return 1;
diff --git a/src/or/config.h b/src/or/config.h
index 6cc81ab948..133b472eb2 100644
--- a/src/or/config.h
+++ b/src/or/config.h
@@ -33,7 +33,7 @@ void reset_last_resolved_addr(void);
 int resolve_my_address(int warn_severity, const or_options_t *options,
                        uint32_t *addr_out,
                        const char **method_out, char **hostname_out);
-int is_local_addr(const tor_addr_t *addr);
+MOCK_DECL(int, is_local_addr, (const tor_addr_t *addr));
 void options_init(or_options_t *options);
 
 #define OPTIONS_DUMP_MINIMAL 1
diff --git a/src/or/connection.c b/src/or/connection.c
index c9c371c001..ce3fda89c1 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 29b88041b7..f228450723 100644
--- a/src/or/connection_or.c
+++ b/src/or/connection_or.c
@@ -38,6 +38,8 @@
 #include "router.h"
 #include "routerlist.h"
 #include "ext_orport.h"
+#include "scheduler.h"
+
 #ifdef USE_BUFFEREVENTS
 #include <event2/bufferevent_ssl.h>
 #endif
@@ -574,48 +576,51 @@ 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
 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;
+
+  /* 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));
   if (datalen < OR_CONN_LOWWATER) {
-    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;
-    }
+    /* Let the scheduler know */
+    scheduler_channel_wants_writes(TLS_CHAN_TO_BASE(conn->chan));
   }
 
   return 0;
 }
 
+/** This is for channeltls.c to ask how many cells we could accept if
+ * they were available. */
+ssize_t
+connection_or_num_cells_writeable(or_connection_t *conn)
+{
+  size_t datalen, cell_network_size;
+  ssize_t n = 0;
+
+  tor_assert(conn);
+
+  /*
+   * If we're under the high water mark, we're potentially
+   * writeable; note this is different from the calculation above
+   * used to trigger when to start writing after we've stopped.
+   */
+  datalen = connection_get_outbuf_len(TO_CONN(conn));
+  if (datalen < OR_CONN_HIGHWATER) {
+    cell_network_size = get_cell_network_size(conn->wide_circ_ids);
+    n = CEIL_DIV(OR_CONN_HIGHWATER - datalen, cell_network_size);
+  }
+
+  return n;
+}
+
 /** Connection <b>conn</b> has finished writing and has no bytes left on
  * its outbuf.
  *
@@ -1169,10 +1174,10 @@ connection_or_notify_error(or_connection_t *conn,
  *
  * Return the launched conn, or NULL if it failed.
  */
-or_connection_t *
-connection_or_connect(const tor_addr_t *_addr, uint16_t port,
-                      const char *id_digest,
-                      channel_tls_t *chan)
+
+MOCK_IMPL(or_connection_t *,
+connection_or_connect, (const tor_addr_t *_addr, uint16_t port,
+                        const char *id_digest, channel_tls_t *chan))
 {
   or_connection_t *conn;
   const or_options_t *options = get_options();
diff --git a/src/or/connection_or.h b/src/or/connection_or.h
index 1ceabdaa4b..b82896e26d 100644
--- a/src/or/connection_or.h
+++ b/src/or/connection_or.h
@@ -24,6 +24,7 @@ void connection_or_set_bad_connections(const char *digest, int force);
 void connection_or_block_renegotiation(or_connection_t *conn);
 int connection_or_reached_eof(or_connection_t *conn);
 int connection_or_process_inbuf(or_connection_t *conn);
+ssize_t connection_or_num_cells_writeable(or_connection_t *conn);
 int connection_or_flushed_some(or_connection_t *conn);
 int connection_or_finished_flushing(or_connection_t *conn);
 int connection_or_finished_connecting(or_connection_t *conn);
@@ -36,9 +37,10 @@ void connection_or_connect_failed(or_connection_t *conn,
                                   int reason, const char *msg);
 void connection_or_notify_error(or_connection_t *conn,
                                 int reason, const char *msg);
-or_connection_t *connection_or_connect(const tor_addr_t *addr, uint16_t port,
-                                       const char *id_digest,
-                                       channel_tls_t *chan);
+MOCK_DECL(or_connection_t *,
+          connection_or_connect,
+          (const tor_addr_t *addr, uint16_t port,
+           const char *id_digest, channel_tls_t *chan));
 
 void connection_or_close_normally(or_connection_t *orconn, int flush);
 void connection_or_close_for_error(or_connection_t *orconn, int flush);
diff --git a/src/or/include.am b/src/or/include.am
index 0f53f007f0..643f7ce001 100644
--- a/src/or/include.am
+++ b/src/or/include.am
@@ -74,6 +74,7 @@ LIBTOR_A_SOURCES = \
 	src/or/routerlist.c				\
 	src/or/routerparse.c				\
 	src/or/routerset.c				\
+	src/or/scheduler.c				\
 	src/or/statefile.c				\
 	src/or/status.c					\
 	src/or/onion_ntor.c				\
@@ -179,6 +180,7 @@ ORHEADERS = \
 	src/or/routerlist.h				\
 	src/or/routerset.h				\
 	src/or/routerparse.h				\
+	src/or/scheduler.h				\
 	src/or/statefile.h				\
 	src/or/status.h
 
diff --git a/src/or/main.c b/src/or/main.c
index 54c739c4fd..64ccbd496c 100644
--- a/src/or/main.c
+++ b/src/or/main.c
@@ -53,6 +53,7 @@
 #include "router.h"
 #include "routerlist.h"
 #include "routerparse.h"
+#include "scheduler.h"
 #include "statefile.h"
 #include "status.h"
 #include "util_process.h"
@@ -2583,6 +2584,7 @@ tor_free_all(int postfork)
   channel_tls_free_all();
   channel_free_all();
   connection_free_all();
+  scheduler_free_all();
   buf_shrink_freelists(1);
   memarea_clear_freelist();
   nodelist_free_all();
diff --git a/src/or/or.h b/src/or/or.h
index bff7267d37..ccb29ee7df 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -1431,6 +1431,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. */
@@ -1522,6 +1534,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)
@@ -4230,6 +4248,18 @@ typedef struct {
   /** How long (seconds) do we keep a guard before picking a new one? */
   int GuardLifetime;
 
+  /** Low-water mark for global scheduler - start sending when estimated
+   * queued size falls below this threshold.
+   */
+  uint32_t SchedulerLowWaterMark;
+  /** High-water mark for global scheduler - stop sending when estimated
+   * queued size exceeds this threshold.
+   */
+  uint32_t SchedulerHighWaterMark;
+  /** Flush size for global scheduler - flush this many cells at a time
+   * when sending.
+   */
+  unsigned int SchedulerMaxFlushCells;
 } or_options_t;
 
 /** Persistent state for an onion router, as saved to disk. */
diff --git a/src/or/relay.c b/src/or/relay.c
index 05c7b3c955..b95e5841e7 100644
--- a/src/or/relay.c
+++ b/src/or/relay.c
@@ -39,6 +39,7 @@
 #include "router.h"
 #include "routerlist.h"
 #include "routerparse.h"
+#include "scheduler.h"
 
 static edge_connection_t *relay_lookup_conn(circuit_t *circ, cell_t *cell,
                                             cell_direction_t cell_direction,
@@ -2591,8 +2592,8 @@ packed_cell_get_circid(const packed_cell_t *cell, int wide_circ_ids)
  * queue of the first active circuit on <b>chan</b>, and write them to
  * <b>chan</b>-&gt;outbuf.  Return the number of cells written.  Advance
  * the active circuit pointer to the next active circuit in the ring. */
-int
-channel_flush_from_first_active_circuit(channel_t *chan, int max)
+MOCK_IMPL(int,
+channel_flush_from_first_active_circuit, (channel_t *chan, int max))
 {
   circuitmux_t *cmux = NULL;
   int n_flushed = 0;
@@ -2868,14 +2869,8 @@ append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan,
     log_debug(LD_GENERAL, "Made a circuit active.");
   }
 
-  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);
-  }
+  /* New way: mark this as having waiting cells for the scheduler */
+  scheduler_channel_has_waiting_cells(chan);
 }
 
 /** Append an encoded value of <b>addr</b> to <b>payload_out</b>, which must
diff --git a/src/or/relay.h b/src/or/relay.h
index 73c399154d..351516aada 100644
--- a/src/or/relay.h
+++ b/src/or/relay.h
@@ -64,7 +64,8 @@ void append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan,
                                   cell_t *cell, cell_direction_t direction,
                                   streamid_t fromstream);
 void channel_unlink_all_circuits(channel_t *chan, smartlist_t *detached_out);
-int channel_flush_from_first_active_circuit(channel_t *chan, int max);
+MOCK_DECL(int, channel_flush_from_first_active_circuit,
+          (channel_t *chan, int max));
 void assert_circuit_mux_okay(channel_t *chan);
 void update_circuit_on_cmux_(circuit_t *circ, cell_direction_t direction,
                              const char *file, int lineno);
diff --git a/src/or/scheduler.c b/src/or/scheduler.c
new file mode 100644
index 0000000000..c2ede846bf
--- /dev/null
+++ b/src/or/scheduler.c
@@ -0,0 +1,708 @@
+/* * Copyright (c) 2013, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * \file scheduler.c
+ * \brief Relay scheduling system
+ **/
+
+#include "or.h"
+
+#define TOR_CHANNEL_INTERNAL_ /* For channel_flush_some_cells() */
+#include "channel.h"
+
+#include "compat_libevent.h"
+#define SCHEDULER_PRIVATE_
+#include "scheduler.h"
+
+#ifdef HAVE_EVENT2_EVENT_H
+#include <event2/event.h>
+#else
+#include <event.h>
+#endif
+
+/*
+ * Scheduler high/low watermarks
+ */
+
+static uint32_t sched_q_low_water = 16384;
+static uint32_t sched_q_high_water = 32768;
+
+/*
+ * 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.
+ */
+
+static uint32_t 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:
+ *
+ * 1.) Not open for writes, no cells to send
+ *     - 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.
+ *     - Transitions to:
+ *       - Not open for writes/has cells by arrival of cells on an attached
+ *         circuit (this would be driven from append_cell_to_circuit_queue())
+ *       - Open for writes/no cells by a channel type specific path;
+ *         driven from connection_or_flushed_some() for channel_tls_t.
+ *
+ * 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.  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.
+ *       - Open for writes/has cells by the scheduler moving cells from
+ *         circuit queues to channel output queue, but not having enough
+ *         to fill the output queue.
+ *     - Transitions to:
+ *       - Open for writes/has cells by arrival of new cells on an attached
+ *         circuit, in append_cell_to_circuit_queue()
+ *
+ * 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
+ *       - Open for writes/has cells by filling an output buffer without
+ *         draining all cells from attached circuits
+ *    - Transitions to:
+ *       - Opens for writes/has cells by draining some of the output buffer
+ *         via the connection_or_flushed_some() path (for channel_tls_t).
+ *
+ * 4.) Open for writes, cells to send
+ *     - This connection is ready to relay some cells and waiting for
+ *       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
+ *       - Open for writes/no cells by the append_cell_to_circuit_queue()
+ *         path
+ *     - Transitions to:
+ *       - Not open for writes/no cells by draining all circuit queues and
+ *         simultaneously filling the output buffer.
+ *       - Not open for writes/has cells by writing enough cells to fill the
+ *         output buffer
+ *       - Open for writes/no cells by draining all attached circuit queues
+ *         without also filling the output buffer
+ *
+ * Other event-driven parts of the code move channels between these scheduling
+ * states by calling scheduler functions; the scheduler only runs on open-for-
+ * writes/has-cells channels and is the only path for those to transition to
+ * other states.  The scheduler_run() function gives us the opportunity to do
+ * scheduling work, and is called from other scheduler functions whenever a
+ * state transition occurs, and periodically from the main event loop.
+ */
+
+/* Scheduler global data structures */
+
+/*
+ * 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.
+ */
+
+/* Pqueue of channels that can write and have cells (pending work) */
+STATIC smartlist_t *channels_pending = NULL;
+
+/*
+ * This event runs the scheduler from its callback, and is manually
+ * activated whenever a channel enters open for writes/cells to send.
+ */
+
+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,
+                                   short events, void *arg);
+static int scheduler_more_work(void);
+static void scheduler_retrigger(void);
+#if 0
+static void scheduler_trigger(void);
+#endif
+
+/* Scheduler function implementations */
+
+/** Free everything and shut down the scheduling system */
+
+void
+scheduler_free_all(void)
+{
+  log_debug(LD_SCHED, "Shutting down scheduler");
+
+  if (run_sched_ev) {
+    event_del(run_sched_ev);
+    tor_event_free(run_sched_ev);
+    run_sched_ev = NULL;
+  }
+
+  if (channels_pending) {
+    smartlist_free(channels_pending);
+    channels_pending = NULL;
+  }
+}
+
+/**
+ * Comparison function to use when sorting pending channels
+ */
+
+MOCK_IMPL(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.
+ */
+
+static void
+scheduler_evt_callback(evutil_socket_t fd, short events, void *arg)
+{
+  (void)fd;
+  (void)events;
+  (void)arg;
+  log_debug(LD_SCHED, "Scheduler event callback called");
+
+  tor_assert(run_sched_ev);
+
+  /* Run the scheduler */
+  scheduler_run();
+
+  /* Do we have more work to do? */
+  if (scheduler_more_work()) scheduler_retrigger();
+}
+
+/** Mark a channel as no longer ready to accept writes */
+
+MOCK_IMPL(void,
+scheduler_channel_doesnt_want_writes,(channel_t *chan))
+{
+  tor_assert(chan);
+
+  tor_assert(channels_pending);
+
+  /* If it's already in pending, we can put it in waiting_to_write */
+  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_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 "
+              "to waiting_to_write",
+              U64_PRINTF_ARG(chan->global_identifier), chan);
+  } else {
+    /*
+     * It's not in pending, so it can't become waiting_to_write; it's
+     * 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 (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);
+    }
+  }
+}
+
+/** Mark a channel as having waiting cells */
+
+MOCK_IMPL(void,
+scheduler_channel_has_waiting_cells,(channel_t *chan))
+{
+  int became_pending = 0;
+
+  tor_assert(chan);
+  tor_assert(channels_pending);
+
+  /* First, check if this one also writeable */
+  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.
+     */
+    chan->scheduler_state = SCHED_CHAN_PENDING;
+    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",
+              U64_PRINTF_ARG(chan->global_identifier), chan);
+    became_pending = 1;
+  } else {
+    /*
+     * It's not in waiting_for_cells, so it can't become pending; it's
+     * 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 (!(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);
+    }
+  }
+
+  /*
+   * If we made a channel pending, we potentially have scheduling work
+   * to do.
+   */
+  if (became_pending) scheduler_retrigger();
+}
+
+/** Set up the scheduling system */
+
+void
+scheduler_init(void)
+{
+  log_debug(LD_SCHED, "Initting scheduler");
+
+  tor_assert(!run_sched_ev);
+  run_sched_ev = tor_event_new(tor_libevent_get_base(), -1,
+                               0, scheduler_evt_callback, NULL);
+
+  channels_pending = smartlist_new();
+  queue_heuristic = 0;
+  queue_heuristic_timestamp = approx_time();
+}
+
+/** Check if there's more scheduling work */
+
+static int
+scheduler_more_work(void)
+{
+  tor_assert(channels_pending);
+
+  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 */
+
+static void
+scheduler_retrigger(void)
+{
+  tor_assert(run_sched_ev);
+  event_active(run_sched_ev, EV_TIMEOUT, 1);
+}
+
+/** Notify the scheduler of a channel being closed */
+
+MOCK_IMPL(void,
+scheduler_release_channel,(channel_t *chan))
+{
+  tor_assert(chan);
+  tor_assert(channels_pending);
+
+  if (chan->scheduler_state == SCHED_CHAN_PENDING) {
+    smartlist_pqueue_remove(channels_pending,
+                            scheduler_compare_channels,
+                            STRUCT_OFFSET(channel_t, sched_heap_idx),
+                            chan);
+  }
+
+  chan->scheduler_state = SCHED_CHAN_IDLE;
+}
+
+/** Run the scheduling algorithm if necessary */
+
+MOCK_IMPL(void,
+scheduler_run, (void))
+{
+  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");
+
+  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();
+
+    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,
+                    "Channel " U64_FORMAT " at %p "
+                    "entered waiting_for_cells from pending",
+                    U64_PRINTF_ARG(chan->global_identifier),
+                    chan);
+        } else {
+          /* 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 {
+        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;
+      }
+    }
+
+    /* 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();
+    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 */
+
+#if 0
+static void
+scheduler_trigger(void)
+{
+  log_debug(LD_SCHED, "Triggering scheduler event");
+
+  tor_assert(run_sched_ev);
+
+  event_add(run_sched_ev, EV_TIMEOUT, 1);
+}
+#endif
+
+/** Mark a channel as ready to accept writes */
+
+void
+scheduler_channel_wants_writes(channel_t *chan)
+{
+  int became_pending = 0;
+
+  tor_assert(chan);
+  tor_assert(channels_pending);
+
+  /* If it's already in waiting_to_write, we can put it in pending */
+  if (chan->scheduler_state == SCHED_CHAN_WAITING_TO_WRITE) {
+    /*
+     * It can write now, so it goes to channels_pending.
+     */
+    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 "
+              "to pending",
+              U64_PRINTF_ARG(chan->global_identifier), chan);
+    became_pending = 1;
+  } else {
+    /*
+     * 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 (!(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);
+    }
+  }
+
+  /*
+   * If we made a channel pending, we potentially have scheduling work
+   * to do.
+   */
+  if (became_pending) scheduler_retrigger();
+}
+
+/**
+ * Notify the scheduler that a channel's position in the pqueue may have
+ * changed
+ */
+
+void
+scheduler_touch_channel(channel_t *chan)
+{
+  tor_assert(chan);
+
+  if (chan->scheduler_state == SCHED_CHAN_PENDING) {
+    /* Remove and re-add it */
+    smartlist_pqueue_remove(channels_pending,
+                            scheduler_compare_channels,
+                            STRUCT_OFFSET(channel_t, sched_heap_idx),
+                            chan);
+    smartlist_pqueue_add(channels_pending,
+                         scheduler_compare_channels,
+                         STRUCT_OFFSET(channel_t, sched_heap_idx),
+                         chan);
+  }
+  /* else no-op, since it isn't in the queue */
+}
+
+/**
+ * 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 */
+}
+
+/**
+ * Set scheduler watermarks and flush size
+ */
+
+void
+scheduler_set_watermarks(uint32_t lo, uint32_t hi, uint32_t max_flush)
+{
+  /* Sanity assertions - caller should ensure these are true */
+  tor_assert(lo > 0);
+  tor_assert(hi > lo);
+  tor_assert(max_flush > 0);
+
+  sched_q_low_water = lo;
+  sched_q_high_water = hi;
+  sched_max_flush_cells = max_flush;
+}
diff --git a/src/or/scheduler.h b/src/or/scheduler.h
new file mode 100644
index 0000000000..404776b18b
--- /dev/null
+++ b/src/or/scheduler.h
@@ -0,0 +1,50 @@
+/* * Copyright (c) 2013, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * \file scheduler.h
+ * \brief Header file for scheduler.c
+ **/
+
+#ifndef TOR_SCHEDULER_H
+#define TOR_SCHEDULER_H
+
+#include "or.h"
+#include "channel.h"
+#include "testsupport.h"
+
+/* Global-visibility scheduler functions */
+
+/* Set up and shut down the scheduler from main.c */
+void scheduler_free_all(void);
+void scheduler_init(void);
+MOCK_DECL(void, scheduler_run, (void));
+
+/* Mark channels as having cells or wanting/not wanting writes */
+MOCK_DECL(void,scheduler_channel_doesnt_want_writes,(channel_t *chan));
+MOCK_DECL(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 */
+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);
+
+/* Notify scheduler that a channel's queue position may have changed */
+void scheduler_touch_channel(channel_t *chan);
+
+/* Adjust the watermarks from config file*/
+void scheduler_set_watermarks(uint32_t lo, uint32_t hi, uint32_t max_flush);
+
+/* Things only scheduler.c and its test suite should see */
+
+#ifdef SCHEDULER_PRIVATE_
+MOCK_DECL(STATIC int, scheduler_compare_channels,
+          (const void *c1_v, const void *c2_v));
+STATIC uint64_t scheduler_get_queue_heuristic(void);
+STATIC void scheduler_update_queue_heuristic(time_t now);
+#endif
+
+#endif /* !defined(TOR_SCHEDULER_H) */
+