Favor quiet circuits when choosing which order to relay cells in.

Each circuit is ranked in terms of how many cells from it have been
relayed recently, using a time-weighted average.

This patch has been tested this on a private Tor network on PlanetLab,
and gotten improvements of 12-35% in time it takes to fetch a small
web page while there's a simultaneous large data transfer going on
simultaneously.

[Commit msg by nickm based on mail from Ian Goldberg.]

5 files changed, 191 insertions, 6 deletions

diff --git a/src/or/circuitlist.c b/src/or/circuitlist.c
index 2c949def00..eb8c90f460 100644
--- a/src/or/circuitlist.c
+++ b/src/or/circuitlist.c
@@ -385,6 +385,10 @@ init_circuit_base(circuit_t *circ)
   circ->package_window = circuit_initial_package_window();
   circ->deliver_window = CIRCWINDOW_START;
 
+  /* Initialize the cell_ewma_t structure */
+  circ->n_cell_ewma.last_cell_time = circ->highres_created;
+  circ->n_cell_ewma.cell_count = 0.0;
+
   circuit_add(circ);
 }
 
@@ -432,6 +436,14 @@ or_circuit_new(circid_t p_circ_id, or_connection_t *p_conn)
 
   init_circuit_base(TO_CIRCUIT(circ));
 
+  /* Initialize the cell_ewma_t structure */
+
+  /* Fetch the timeval that init_circuit_base filled in. */
+  circ->p_cell_ewma.last_cell_time = TO_CIRCUIT(circ)->highres_created;
+
+  /* Initialize the cell counts to 0 */
+  circ->p_cell_ewma.cell_count = 0.0;
+
   return circ;
 }
 
diff --git a/src/or/config.c b/src/or/config.c
index deeda163b6..fe5fe9f7ee 100644
--- a/src/or/config.c
+++ b/src/or/config.c
@@ -355,6 +355,11 @@ static config_var_t _option_vars[] = {
   VAR("__HashedControlSessionPassword", LINELIST, HashedControlSessionPassword,
       NULL),
   V(MinUptimeHidServDirectoryV2, INTERVAL, "24 hours"),
+
+  /* Options for EWMA selection of circuit to write from */
+  VAR("EWMASignificance", DOUBLE, EWMASignificance, "-1.0"),
+  VAR("EWMAInterval", DOUBLE, EWMAInterval, "-1.0"),
+
   { NULL, CONFIG_TYPE_OBSOLETE, 0, NULL }
 };
 
diff --git a/src/or/connection.c b/src/or/connection.c
index 0ff1cc5876..bd12e36180 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -2275,8 +2275,8 @@ connection_read_bucket_should_increase(or_connection_t *conn)
  * Mark the connection and return -1 if you want to close it, else
  * return 0.
  */
-int
-connection_handle_read(connection_t *conn)
+static int
+connection_handle_read_impl(connection_t *conn)
 {
   int max_to_read=-1, try_to_read;
   size_t before, n_read = 0;
@@ -2371,6 +2371,17 @@ loop_again:
   return 0;
 }
 
+int
+connection_handle_read(connection_t *conn)
+{
+  int res;
+
+  tor_gettimeofday_cache_clear();
+  res = connection_handle_read_impl(conn);
+  return res;
+  
+}
+
 /** Pull in new bytes from conn-\>s or conn-\>linked_conn onto conn-\>inbuf,
  * either directly or via TLS. Reduce the token buckets by the number of bytes
  * read.
@@ -2572,8 +2583,8 @@ connection_outbuf_too_full(connection_t *conn)
  * Mark the connection and return -1 if you want to close it, else
  * return 0.
  */
-int
-connection_handle_write(connection_t *conn, int force)
+static int
+connection_handle_write_impl(connection_t *conn, int force)
 {
   int e;
   socklen_t len=(socklen_t)sizeof(e);
@@ -2740,6 +2751,15 @@ connection_handle_write(connection_t *conn, int force)
   return 0;
 }
 
+int
+connection_handle_write(connection_t *conn, int force)
+{
+    int res;
+    tor_gettimeofday_cache_clear();
+    res = connection_handle_write_impl(conn, force);
+    return res;
+}
+
 /** OpenSSL TLS record size is 16383; this is close. The goal here is to
  * push data out as soon as we know there's enough for a TLS record, so
  * during periods of high load we won't read entire megabytes from
diff --git a/src/or/or.h b/src/or/or.h
index 2e575f5ef9..003d0ebdad 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -1992,6 +1992,20 @@ typedef struct {
   time_t expiry_time;
 } cpath_build_state_t;
 
+/**
+ * The cell_ewma_t structure keeps track of how many cells a circuit has
+ * transferred recently.  It keeps an EWMA (exponentially weighted
+ * moving average) of the number of cells flushed in
+ * connection_or_flush_from_first_active_circuit().
+ */
+
+typedef struct {
+    /** The last time a cell was flushed from this circuit. */
+    struct timeval last_cell_time;
+    /** The EWMA of the cell count. */
+    double cell_count;
+} cell_ewma_t;
+
 #define ORIGIN_CIRCUIT_MAGIC 0x35315243u
 #define OR_CIRCUIT_MAGIC 0x98ABC04Fu
 
@@ -2081,6 +2095,10 @@ typedef struct circuit_t {
 
   /** Unique ID for measuring tunneled network status requests. */
   uint64_t dirreq_id;
+
+  /** The EWMA count for the number of cells flushed from the
+   * n_conn_cells queue. */
+  cell_ewma_t n_cell_ewma;
 } circuit_t;
 
 /** Largest number of relay_early cells that we can send on a given
@@ -2212,6 +2230,10 @@ typedef struct or_circuit_t {
    * exit-ward queues of this circuit; reset every time when writing
    * buffer stats to disk. */
   uint64_t total_cell_waiting_time;
+
+  /** The EWMA count for the number of cells flushed from the
+   * p_conn_cells queue. */
+  cell_ewma_t p_cell_ewma;
 } or_circuit_t;
 
 /** Convert a circuit subtype to a circuit_t.*/
@@ -2740,6 +2762,14 @@ typedef struct {
    * to make this false. */
   int ReloadTorrcOnSIGHUP;
 
+  /* The EWMA parameters for circuit selection within a connection.
+   * The most recent EWMAInterval seconds will account for an
+   * EWMASignificance (between 0 and 1) portion of the weight.
+   * If these values are negative, use the global defaults (soon to be
+   * set in the consensus). */
+  double EWMASignificance;
+  double EWMAInterval;
+
 } or_options_t;
 
 /** Persistent state for an onion router, as saved to disk. */
@@ -5122,5 +5152,7 @@ int rend_parse_introduction_points(rend_service_descriptor_t *parsed,
                                    size_t intro_points_encoded_size);
 int rend_parse_client_keys(strmap_t *parsed_clients, const char *str);
 
+void tor_gettimeofday_cache_clear(void);
+
 #endif
 
diff --git a/src/or/relay.c b/src/or/relay.c
index ac305ce3df..147412f596 100644
--- a/src/or/relay.c
+++ b/src/or/relay.c
@@ -10,6 +10,7 @@
  *    receiving from circuits, plus queuing on circuits.
  **/
 
+#include <math.h>
 #include "or.h"
 #include "mempool.h"
 
@@ -35,6 +36,26 @@ circuit_resume_edge_reading_helper(edge_connection_t *conn,
 static int
 circuit_consider_stop_edge_reading(circuit_t *circ, crypt_path_t *layer_hint);
 
+/** Cache the current hi-res time; the cache gets reset when libevent
+ * calls us. */
+
+static struct timeval cached_time_hires = {0, 0};
+
+static void
+tor_gettimeofday_cached(struct timeval *tv)
+{
+  if (cached_time_hires.tv_sec == 0) {
+    tor_gettimeofday(&cached_time_hires);
+  }
+  *tv = cached_time_hires;
+}
+
+void
+tor_gettimeofday_cache_clear(void)
+{
+    cached_time_hires.tv_sec = 0;
+}
+
 /** Stats: how many relay cells have originated at this hop, or have
  * been relayed onward (not recognized at this hop)?
  */
@@ -1633,7 +1654,7 @@ cell_queue_append_packed_copy(cell_queue_t *queue, const cell_t *cell)
     insertion_time_queue_t *it_queue = queue->insertion_times;
     if (!it_pool)
       it_pool = mp_pool_new(sizeof(insertion_time_elem_t), 1024);
-    tor_gettimeofday(&now);
+    tor_gettimeofday_cached(&now);
 #define SECONDS_IN_A_DAY 86400L
     added = (uint32_t)(((now.tv_sec % SECONDS_IN_A_DAY) * 100L)
             + ((uint32_t)now.tv_usec / (uint32_t)10000L));
@@ -1857,9 +1878,101 @@ connection_or_flush_from_first_active_circuit(or_connection_t *conn, int max,
   cell_queue_t *queue;
   circuit_t *circ;
   int streams_blocked;
+
+  /* The current (hi-res) time */
+  struct timeval now_hires;
+
+  /* The EWMA cell counter for the circuit we're flushing. */
+  cell_ewma_t *cell_ewma = NULL;
+
+  /* The global cell EWMA parameters.  The algorithm is parameterized by
+   * two values (type double):
+   *
+   * "significance" (between 0 and 1) and "interval"
+   *
+   * The cell count is weighted so that the most recent "interval"
+   * seconds will account for "significance" of the weight.
+   *
+   * If "interval" is set to 0, it disables the algorithm, and the old
+   * algorithm (round-robin) is used.
+   *
+   * These parameters should really be set by the consensus, but can be
+   * overridden by the torrc (in which case the options values will be
+   * >= 0.0).
+   */
+  static double cell_ewma_significance = 0.9;
+  static double cell_ewma_interval = 10.0;
+
+  double significance_override = get_options()->EWMASignificance;
+  double interval_override = get_options()->EWMAInterval;
+  if (significance_override >= 0.0) {
+    cell_ewma_significance = significance_override;
+  }
+  if (interval_override >= 0.0) {
+    cell_ewma_interval = interval_override;
+  }
+
   circ = conn->active_circuits;
   if (!circ) return 0;
   assert_active_circuits_ok_paranoid(conn);
+
+  /* See if we're doing the ewma circuit selection algorithm. */
+  if (cell_ewma_interval > 0.0) {
+    /* Is there another circuit we might like better? */
+    circuit_t *circ_iter, *circ_start;
+    circuit_t *circ_min_cell_count = NULL;
+    double min_cell_count = 0.0;
+    tor_gettimeofday_cached(&now_hires);
+
+    /* Start with circ, and go around the circular linked list */
+    circ_start = circ_iter = circ;
+    do {
+      double delta_t;
+
+      /* Find the appropriate EWMA cell counter to use. */
+      if (circ_iter->n_conn == conn) {
+        cell_ewma = &(circ_iter->n_cell_ewma);
+      } else {
+        cell_ewma = &(TO_OR_CIRCUIT(circ_iter)->p_cell_ewma);
+      }
+
+      /* Update the EWMA cell counter to account for the passage of time. */
+      delta_t = (double)(now_hires.tv_sec -
+			    cell_ewma->last_cell_time.tv_sec);
+      delta_t += ((double)(now_hires.tv_usec -
+			    cell_ewma->last_cell_time.tv_usec)) / 1000000.0;
+
+      if (delta_t > 0.0) {
+        cell_ewma->cell_count *=
+	    pow(cell_ewma_significance, delta_t / cell_ewma_interval);
+		//printf("cc: %f ", cell_ewma->cell_count);
+      }
+      cell_ewma->last_cell_time = now_hires;
+
+      /* Now keep track of the lowest cell count we've seen. */
+      if (circ_min_cell_count == NULL ||
+	    cell_ewma->cell_count < min_cell_count) {
+		min_cell_count = cell_ewma->cell_count;
+		circ_min_cell_count = circ_iter;
+      }
+      
+      circ_iter = *next_circ_on_conn_p(circ_iter, conn);
+    } while (circ_iter != circ_start);
+
+    /* OK, we've gone all the way around.  Let's use the circ with the
+     * lowest (recent) cell count. */
+    circ = circ_min_cell_count;
+
+    /* Now set the appropriate EWMA cell counter to use below to add the
+     * cells we actually send. */
+    if (circ_min_cell_count->n_conn == conn) {
+      cell_ewma = &(circ_min_cell_count->n_cell_ewma);
+    } else {
+      cell_ewma = &(TO_OR_CIRCUIT(circ_min_cell_count)->p_cell_ewma);
+    }
+  }
+  
+
   if (circ->n_conn == conn) {
     queue = &circ->n_conn_cells;
     streams_blocked = circ->streams_blocked_on_n_conn;
@@ -1879,7 +1992,7 @@ connection_or_flush_from_first_active_circuit(or_connection_t *conn, int max,
       uint32_t flushed;
       uint32_t cell_waiting_time;
       insertion_time_queue_t *it_queue = queue->insertion_times;
-      tor_gettimeofday(&now);
+      tor_gettimeofday_cached(&now);
       flushed = (uint32_t)((now.tv_sec % SECONDS_IN_A_DAY) * 100L +
                  (uint32_t)now.tv_usec / (uint32_t)10000L);
       if (!it_queue || !it_queue->first) {
@@ -1915,6 +2028,9 @@ connection_or_flush_from_first_active_circuit(or_connection_t *conn, int max,
 
     packed_cell_free_unchecked(cell);
     ++n_flushed;
+    if (cell_ewma) {
+      cell_ewma->cell_count += 1.0;
+    }
     if (circ != conn->active_circuits) {
       /* If this happens, the current circuit just got made inactive by
        * a call in connection_write_to_buf().  That's nothing to worry about: