1 files changed, 319 insertions, 3 deletions

diff --git a/src/or/relay.c b/src/or/relay.c
index ac305ce3df..ae1b062cf6 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));
@@ -1731,6 +1752,224 @@ prev_circ_on_conn_p(circuit_t *circ, or_connection_t *conn)
   }
 }
 
+/** Helper for sorting cell_ewma_t values in their priority queue. */
+static int
+compare_cell_ewma_counts(const void *p1, const void *p2)
+{
+  const cell_ewma_t *e1=p1, *e2=p2;
+  if (e1->cell_count < e2->cell_count)
+    return -1;
+  else if (e1->cell_count > e2->cell_count)
+    return 1;
+  else
+    return 0;
+}
+
+/** Given a cell_ewma_t, return a pointer to the circuit containing it. */
+static circuit_t *
+cell_ewma_to_circuit(cell_ewma_t *ewma)
+{
+  if (ewma->is_for_p_conn) {
+    /* This is an or_circuit_t's p_cell_ewma. */
+    or_circuit_t *orcirc = SUBTYPE_P(ewma, or_circuit_t, p_cell_ewma);
+    return TO_CIRCUIT(orcirc);
+  } else {
+    /* This is some circuit's n_cell_ewma. */
+    return SUBTYPE_P(ewma, circuit_t, n_cell_ewma);
+  }
+}
+
+/* ==== Functions for scaling cell_ewma_t ====
+
+   When choosing which cells to relay first, we favor circuits that have been
+   quiet recently.  This gives better latency on connections that aren't
+   pushing lots of data, and makes the network feel more interactive.
+
+   Conceptually, we take an exponentially weighted mean average of the number
+   of cells a circuit has sent, and allow active circuits (those with cells to
+   relay) to send cells in reverse order of their exponentially-weighted mean
+   average (EWMA) cell count.  [That is, a cell sent N seconds ago 'counts'
+   F^N times as much as a cell sent now, for 0<F<1.0, and we favor the
+   circuit that has sent the fewest cells]
+
+   If 'double' had infinite precision, we could do this simply by counting a
+   cell sent at startup as having weight 1.0, and a cell sent N seconds later
+   as having weight F^-N.  This way, we would never need to re-scale
+   any already-sent cells.
+
+   To prevent double from overflowing, we could count a cell sent now as
+   having weight 1.0 and a cell sent N seconds ago as having weight F^N.
+   This, however, would mean we'd need to re-scale *ALL* old circuits every
+   time we wanted to send a cell.
+
+   So as a compromise, we divide time into 'ticks' (currently, 10-second
+   increments) and say that a cell sent at the start of a current tick is
+   worth 1.0, a cell sent N seconds before the start of the current tick is
+   worth F^N, and a cell sent N seconds after the start of the current tick is
+   worth F^-N.  This way we don't overflow, and we don't need to constantly
+   rescale.
+ */
+
+/** How long does a tick last (seconds)? */
+#define EWMA_TICK_LEN 10
+
+/** The default per-tick scale factor, if it hasn't been overridden by a
+ * consensus or a configuration setting.  zero means "disabled". */
+#define EWMA_DEFAULT_HALFLIFE 0.0
+
+/** Given a timeval <b>now</b>, compute the cell_ewma tick in which it occurs
+ * and the fraction of the tick that has elapsed between the start of the tick
+ * and <b>now</b>.  Return the former and store the latter in
+ * *<b>remainder_out</b>.
+ *
+ * These tick values are not meant to be shared between Tor instances, or used
+ * for other purposes. */
+static unsigned
+cell_ewma_tick_from_timeval(const struct timeval *now,
+                            double *remainder_out)
+{
+  unsigned res = (unsigned) (now->tv_sec / EWMA_TICK_LEN);
+  /* rem */
+  double rem = (now->tv_sec % EWMA_TICK_LEN) +
+    ((double)(now->tv_usec)) / 1.0e6;
+  *remainder_out = rem / EWMA_TICK_LEN;
+  return res;
+}
+
+/** Compute and return the current cell_ewma tick. */
+unsigned
+cell_ewma_get_tick(void)
+{
+  return ((unsigned)approx_time() / EWMA_TICK_LEN);
+}
+
+/** The per-tick scale factor to be used when computing cell-count EWMA
+ * values.  (A cell sent N ticks before the start of the current tick
+ * has value ewma_scale_factor ** N.)
+ */
+static double ewma_scale_factor = 0.1;
+static int ewma_enabled = 0;
+
+#define EPSILON 0.00001
+#define LOG_ONEHALF -0.69314718055994529
+
+/** Adjust the global cell scale factor based on <b>options</b> */
+void
+cell_ewma_set_scale_factor(or_options_t *options, networkstatus_t *consensus)
+{
+  int32_t halflife_ms;
+  double halflife;
+  const char *source;
+  if (options && options->CircuitPriorityHalflife >= -EPSILON) {
+    halflife = options->CircuitPriorityHalflife;
+    source = "CircuitPriorityHalflife in configuration";
+  } else if (consensus &&
+             (halflife_ms = networkstatus_get_param(
+                   consensus, "CircPriorityHalflifeMsec", -1) >= 0)) {
+    halflife = ((double)halflife_ms)/1000.0;
+    source = "CircPriorityHalflifeMsec in consensus";
+  } else {
+    halflife = EWMA_DEFAULT_HALFLIFE;
+    source = "Default value";
+  }
+
+  if (halflife <= EPSILON) {
+    /* The cell EWMA algorithm is disabled. */
+    ewma_scale_factor = 0.1;
+    ewma_enabled = 0;
+    log_info(LD_OR,
+             "Disabled cell_ewma algorithm because of value in %s",
+             source);
+  } else {
+    /* convert halflife into halflife-per-tick. */
+    halflife /= EWMA_TICK_LEN;
+    /* compute per-tick scale factor. */
+    ewma_scale_factor = exp( LOG_ONEHALF / halflife );
+    ewma_enabled = 1;
+    log_info(LD_OR,
+             "Enabled cell_ewma algorithm because of value in %s; "
+             "scale factor is %lf per %d seconds",
+             source, ewma_scale_factor, EWMA_TICK_LEN);
+  }
+}
+
+/** Return the multiplier necessary to convert the value of a cell sent in
+ * 'from_tick' to one sent in 'to_tick'. */
+static INLINE double
+get_scale_factor(unsigned from_tick, unsigned to_tick)
+{
+  /* This math can wrap around, but that's okay: unsigned overflow is
+     well-defined */
+  int diff = (int)(to_tick - from_tick);
+  return pow(ewma_scale_factor, diff);
+}
+
+/** Adjust the cell count of <b>ewma</b> so that it is scaled with respect to
+ * <b>cur_tick</b> */
+static void
+scale_single_cell_ewma(cell_ewma_t *ewma, unsigned cur_tick)
+{
+  double factor = get_scale_factor(ewma->last_adjusted_tick, cur_tick);
+  ewma->cell_count *= factor;
+  ewma->last_adjusted_tick = cur_tick;
+}
+
+/** Adjust the cell count of every active circuit on <b>conn</b> so
+ * that they are scaled with respect to <b>cur_tick</b> */
+static void
+scale_active_circuits(or_connection_t *conn, unsigned cur_tick)
+{
+
+  double factor = get_scale_factor(
+              conn->active_circuit_pqueue_last_recalibrated,
+              cur_tick);
+  /** Ordinarily it isn't okay to change the value of an element in a heap,
+   * but it's okay here, since we are preserving the order. */
+  SMARTLIST_FOREACH(conn->active_circuit_pqueue, cell_ewma_t *, e, {
+      tor_assert(e->last_adjusted_tick ==
+                 conn->active_circuit_pqueue_last_recalibrated);
+      e->cell_count *= factor;
+      e->last_adjusted_tick = cur_tick;
+  });
+  conn->active_circuit_pqueue_last_recalibrated = cur_tick;
+}
+
+/** Rescale <b>ewma</b> to the same scale as <b>conn</b>, and add it to
+ * <b>conn</b>'s priority queue of active circuits */
+static void
+add_cell_ewma_to_conn(or_connection_t *conn, cell_ewma_t *ewma)
+{
+  tor_assert(ewma->heap_index == -1);
+  scale_single_cell_ewma(ewma,
+                         conn->active_circuit_pqueue_last_recalibrated);
+
+  smartlist_pqueue_add(conn->active_circuit_pqueue,
+                       compare_cell_ewma_counts,
+                       STRUCT_OFFSET(cell_ewma_t, heap_index),
+                       ewma);
+}
+
+/** Remove <b>ewma</b> from <b>conn</b>'s priority queue of active circuits */
+static void
+remove_cell_ewma_from_conn(or_connection_t *conn, cell_ewma_t *ewma)
+{
+  tor_assert(ewma->heap_index != -1);
+  smartlist_pqueue_remove(conn->active_circuit_pqueue,
+                          compare_cell_ewma_counts,
+                          STRUCT_OFFSET(cell_ewma_t, heap_index),
+                          ewma);
+}
+
+/** Remove and return the first cell_ewma_t from conn's priority queue of
+ * active circuits.  Requires that the priority queue is nonempty. */
+static cell_ewma_t *
+pop_first_cell_ewma_from_conn(or_connection_t *conn)
+{
+  return smartlist_pqueue_pop(conn->active_circuit_pqueue,
+                              compare_cell_ewma_counts,
+                              STRUCT_OFFSET(cell_ewma_t, heap_index));
+}
+
 /** Add <b>circ</b> to the list of circuits with pending cells on
  * <b>conn</b>.  No effect if <b>circ</b> is already linked. */
 void
@@ -1744,6 +1983,8 @@ make_circuit_active_on_conn(circuit_t *circ, or_connection_t *conn)
     return;
   }
 
+  assert_active_circuits_ok_paranoid(conn);
+
   if (! conn->active_circuits) {
     conn->active_circuits = circ;
     *prevp = *nextp = circ;
@@ -1755,6 +1996,15 @@ make_circuit_active_on_conn(circuit_t *circ, or_connection_t *conn)
     *prev_circ_on_conn_p(head, conn) = circ;
     *prevp = old_tail;
   }
+
+  if (circ->n_conn == conn) {
+    add_cell_ewma_to_conn(conn, &circ->n_cell_ewma);
+  } else {
+    or_circuit_t *orcirc = TO_OR_CIRCUIT(circ);
+    tor_assert(conn == orcirc->p_conn);
+    add_cell_ewma_to_conn(conn, &orcirc->p_cell_ewma);
+  }
+
   assert_active_circuits_ok_paranoid(conn);
 }
 
@@ -1772,6 +2022,8 @@ make_circuit_inactive_on_conn(circuit_t *circ, or_connection_t *conn)
     return;
   }
 
+  assert_active_circuits_ok_paranoid(conn);
+
   tor_assert(next && prev);
   tor_assert(*prev_circ_on_conn_p(next, conn) == circ);
   tor_assert(*next_circ_on_conn_p(prev, conn) == circ);
@@ -1785,6 +2037,15 @@ make_circuit_inactive_on_conn(circuit_t *circ, or_connection_t *conn)
       conn->active_circuits = next;
   }
   *prevp = *nextp = NULL;
+
+  if (circ->n_conn == conn) {
+    remove_cell_ewma_from_conn(conn, &circ->n_cell_ewma);
+  } else {
+    or_circuit_t *orcirc = TO_OR_CIRCUIT(circ);
+    tor_assert(conn == orcirc->p_conn);
+    remove_cell_ewma_from_conn(conn, &orcirc->p_cell_ewma);
+  }
+
   assert_active_circuits_ok_paranoid(conn);
 }
 
@@ -1804,6 +2065,10 @@ connection_or_unlink_all_active_circs(or_connection_t *orconn)
     cur = next;
   } while (cur != head);
   orconn->active_circuits = NULL;
+
+  SMARTLIST_FOREACH(orconn->active_circuit_pqueue, cell_ewma_t *, e,
+                    e->heap_index = -1);
+  smartlist_clear(orconn->active_circuit_pqueue);
 }
 
 /** Block (if <b>block</b> is true) or unblock (if <b>block</b> is false)
@@ -1857,9 +2122,35 @@ 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;
+  double ewma_increment = -1;
+
   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 (ewma_enabled) {
+    unsigned tick;
+    double fractional_tick;
+    tor_gettimeofday_cached(&now_hires);
+    tick = cell_ewma_tick_from_timeval(&now_hires, &fractional_tick);
+
+    if (tick != conn->active_circuit_pqueue_last_recalibrated) {
+      scale_active_circuits(conn, tick);
+    }
+
+    ewma_increment = pow(ewma_scale_factor, -fractional_tick);
+
+    cell_ewma = smartlist_get(conn->active_circuit_pqueue, 0);
+    circ = cell_ewma_to_circuit(cell_ewma);
+  }
+
   if (circ->n_conn == conn) {
     queue = &circ->n_conn_cells;
     streams_blocked = circ->streams_blocked_on_n_conn;
@@ -1879,7 +2170,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 +2206,16 @@ 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_t *tmp;
+      cell_ewma->cell_count += ewma_increment;
+      /* We pop and re-add the cell_ewma_t here, not above, since we need to
+       * re-add it immediately to keep the priority queue consistent with
+       * the linked-list implementation */
+      tmp = pop_first_cell_ewma_from_conn(conn);
+      tor_assert(tmp == cell_ewma);
+      add_cell_ewma_to_conn(conn, cell_ewma);
+    }
     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:
@@ -1934,7 +2235,7 @@ connection_or_flush_from_first_active_circuit(or_connection_t *conn, int max,
   if (streams_blocked && queue->n <= CELL_QUEUE_LOWWATER_SIZE)
     set_streams_blocked_on_circ(circ, conn, 0); /* unblock streams */
 
-  /* Did we just ran out of cells on this queue? */
+  /* Did we just run out of cells on this circuit's queue? */
   if (queue->n == 0) {
     log_debug(LD_GENERAL, "Made a circuit inactive.");
     make_circuit_inactive_on_conn(circ, conn);
@@ -2057,16 +2358,31 @@ assert_active_circuits_ok(or_connection_t *orconn)
 {
   circuit_t *head = orconn->active_circuits;
   circuit_t *cur = head;
+  int n = 0;
   if (! head)
     return;
   do {
     circuit_t *next = *next_circ_on_conn_p(cur, orconn);
     circuit_t *prev = *prev_circ_on_conn_p(cur, orconn);
+    cell_ewma_t *ewma;
     tor_assert(next);
     tor_assert(prev);
     tor_assert(*next_circ_on_conn_p(prev, orconn) == cur);
     tor_assert(*prev_circ_on_conn_p(next, orconn) == cur);
+    if (orconn == cur->n_conn) {
+      ewma = &cur->n_cell_ewma;
+      tor_assert(!ewma->is_for_p_conn);
+    } else {
+      ewma = &TO_OR_CIRCUIT(cur)->p_cell_ewma;
+      tor_assert(ewma->is_for_p_conn);
+    }
+    tor_assert(ewma->heap_index != -1);
+    tor_assert(ewma == smartlist_get(orconn->active_circuit_pqueue,
+                                     ewma->heap_index));
+    n++;
     cur = next;
   } while (cur != head);
+
+  tor_assert(n == smartlist_len(orconn->active_circuit_pqueue));
 }