Turn cell queue watermark limits into consensus params.

This allows us to have quicker reaction to blocked orconns as
a congestion signal.

src/core/or/congestion_control_common.c

@@ -50,6 +50,9 @@
 #define OR_CONN_HIGHWATER_DFLT (32*1024)
 #define OR_CONN_LOWWATER_DFLT (16*1024)
 
+#define CELL_QUEUE_LOW_DFLT (10)
+#define CELL_QUEUE_HIGH_DFLT (256)
+
 static uint64_t congestion_control_update_circuit_rtt(congestion_control_t *,
                                                       uint64_t);
 static bool congestion_control_update_circuit_bdp(congestion_control_t *,
@@ -61,6 +64,9 @@ static uint32_t cwnd_max = CWND_MAX_DFLT;
 uint32_t or_conn_highwater = OR_CONN_HIGHWATER_DFLT;
 uint32_t or_conn_lowwater = OR_CONN_LOWWATER_DFLT;
 
+int32_t cell_queue_high = CELL_QUEUE_HIGH_DFLT;
+int32_t cell_queue_low = CELL_QUEUE_LOW_DFLT;
+
 /**
  * Update global congestion control related consensus parameter values,
  * every consensus update.
@@ -68,6 +74,20 @@ uint32_t or_conn_lowwater = OR_CONN_LOWWATER_DFLT;
 void
 congestion_control_new_consensus_params(const networkstatus_t *ns)
 {
+#define CELL_QUEUE_HIGH_MIN (1)
+#define CELL_QUEUE_HIGH_MAX (1000)
+  cell_queue_high = networkstatus_get_param(ns, "cellq_high",
+      CELL_QUEUE_HIGH_DFLT,
+      CELL_QUEUE_HIGH_MIN,
+      CELL_QUEUE_HIGH_MAX);
+
+#define CELL_QUEUE_LOW_MIN (1)
+#define CELL_QUEUE_LOW_MAX (1000)
+  cell_queue_low = networkstatus_get_param(ns, "cellq_low",
+      CELL_QUEUE_LOW_DFLT,
+      CELL_QUEUE_LOW_MIN,
+      CELL_QUEUE_LOW_MAX);
+
 #define OR_CONN_HIGHWATER_MIN (CELL_PAYLOAD_SIZE)
 #define OR_CONN_HIGHWATER_MAX (INT32_MAX)
   or_conn_highwater =

src/core/or/congestion_control_common.h

@@ -43,10 +43,12 @@ bool is_monotime_clock_reliable(void);
 
 void congestion_control_new_consensus_params(const networkstatus_t *ns);
 
-/* Ugh, C.. these two are private. Use the getter instead, when
+/* Ugh, C.. these four are private. Use the getter instead, when
  * external to the congestion control code. */
 extern uint32_t or_conn_highwater;
 extern uint32_t or_conn_lowwater;
+extern int32_t cell_queue_high;
+extern int32_t cell_queue_low;
 
 /** Stop writing on an orconn when its outbuf is this large */
 static inline uint32_t
@@ -62,6 +64,22 @@ or_conn_lowwatermark(void)
   return or_conn_lowwater;
 }
 
+/** Stop reading on edge connections when we have this many cells
+ * waiting on the appropriate queue. */
+static inline int32_t
+cell_queue_highwatermark(void)
+{
+  return cell_queue_high;
+}
+
+/** Start reading from edge connections again when we get down to this many
+ * cells. */
+static inline int32_t
+cell_queue_lowwatermark(void)
+{
+  return cell_queue_low;
+}
+
 /**
  * Compute an N-count EWMA, aka N-EWMA. N-EWMA is defined as:
  *  EWMA = alpha*value + (1-alpha)*EWMA_prev

src/core/or/relay.c

@@ -117,13 +117,6 @@ static void adjust_exit_policy_from_exitpolicy_failure(origin_circuit_t *circ,
                                                   node_t *node,
                                                   const tor_addr_t *addr);
 
-/** Stop reading on edge connections when we have this many cells
- * waiting on the appropriate queue. */
-#define CELL_QUEUE_HIGHWATER_SIZE 256
-/** Start reading from edge connections again when we get down to this many
- * cells. */
-#define CELL_QUEUE_LOWWATER_SIZE 64
-
 /** Stats: how many relay cells have originated at this hop, or have
  * been relayed onward (not recognized at this hop)?
  */
@@ -2400,8 +2393,8 @@ circuit_resume_edge_reading_helper(edge_connection_t *first_conn,
     or_circuit_t *or_circ = TO_OR_CIRCUIT(circ);
     cells_on_queue = or_circ->p_chan_cells.n;
   }
-  if (CELL_QUEUE_HIGHWATER_SIZE - cells_on_queue < max_to_package)
+  if (cell_queue_highwatermark() - cells_on_queue < max_to_package)
-    max_to_package = CELL_QUEUE_HIGHWATER_SIZE - cells_on_queue;
+    max_to_package = cell_queue_highwatermark() - cells_on_queue;
 
   /* Once we used to start listening on the streams in the order they
    * appeared in the linked list.  That leads to starvation on the
@@ -3121,7 +3114,7 @@ channel_flush_from_first_active_circuit, (channel_t *chan, int max))
 
     /* Is the cell queue low enough to unblock all the streams that are waiting
      * to write to this circuit? */
-    if (streams_blocked && queue->n <= CELL_QUEUE_LOWWATER_SIZE)
+    if (streams_blocked && queue->n <= cell_queue_lowwatermark())
       set_streams_blocked_on_circ(circ, chan, 0, 0); /* unblock streams */
 
     /* If n_flushed < max still, loop around and pick another circuit */
@@ -3239,7 +3232,7 @@ append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan,
 
   /* If we have too many cells on the circuit, we should stop reading from
    * the edge streams for a while. */
-  if (!streams_blocked && queue->n >= CELL_QUEUE_HIGHWATER_SIZE)
+  if (!streams_blocked && queue->n >= cell_queue_highwatermark())
     set_streams_blocked_on_circ(circ, chan, 1, 0); /* block streams */
 
   if (streams_blocked && fromstream) {