Add destroy balance tracking and logging to circuitmux

src/or/channel.c

@@ -2652,7 +2652,6 @@ is_destroy_cell(channel_t *chan,
   return 0;
 }
 
-
 /**
  * Send destroy cell on a channel
  *

src/or/circuitlist.c

@@ -1717,8 +1717,9 @@ assert_circuit_ok(const circuit_t *c)
   if (or_circ && or_circ->p_chan) {
     if (or_circ->p_circ_id) {
       /* ibid */
-      circuit_t *c2 = circuit_get_by_circid_channel_impl(or_circ->p_circ_id,
-                                                         or_circ->p_chan, NULL);
+      circuit_t *c2 =
+        circuit_get_by_circid_channel_impl(or_circ->p_circ_id,
+                                           or_circ->p_chan, NULL);
       tor_assert(c == c2);
     }
   }

src/or/circuitmux.c

@@ -127,6 +127,10 @@ struct circuitmux_s {
    * cells completely.
    */
   unsigned int last_cell_was_destroy : 1;
+  /** Destroy counter: increment this when a destroy gets queued, decrement
+   * when we unqueue it, so we can test to make sure they don't starve.
+   */
+  int64_t destroy_ctr;
 
   /*
    * Circuitmux policy; if this is non-NULL, it can override the built-
@@ -206,6 +210,11 @@ static void circuitmux_assert_okay_pass_one(circuitmux_t *cmux);
 static void circuitmux_assert_okay_pass_two(circuitmux_t *cmux);
 static void circuitmux_assert_okay_pass_three(circuitmux_t *cmux);
 
+/* Static global variables */
+
+/** Count the destroy balance to debug destroy queue logic */
+static int64_t global_destroy_ctr = 0;
+
 /* Function definitions */
 
 /**
@@ -521,6 +530,25 @@ circuitmux_free(circuitmux_t *cmux)
     tor_free(cmux->chanid_circid_map);
   }
 
+  /*
+   * We're throwing away some destroys; log the counter and
+   * adjust the global counter by the queue size.
+   */
+  if (cmux->destroy_cell_queue.n > 0) {
+    cmux->destroy_ctr -= cmux->destroy_cell_queue.n;
+    global_destroy_ctr -= cmux->destroy_cell_queue.n;
+    log_debug(LD_CIRC,
+              "Freeing cmux at %p with %u queued destroys; the last cmux "
+              "destroy balance was %ld, global is %ld\n",
+              cmux, cmux->destroy_cell_queue.n,
+              cmux->destroy_ctr, global_destroy_ctr);
+  } else {
+    log_debug(LD_CIRC,
+              "Freeing cmux at %p with no queued destroys, the cmux destroy "
+              "balance was %ld, global is %ld\n",
+              cmux, cmux->destroy_ctr, global_destroy_ctr);
+  }
+
   cell_queue_clear(&cmux->destroy_cell_queue);
 
   tor_free(cmux);
@@ -1502,6 +1530,24 @@ circuitmux_notify_xmit_cells(circuitmux_t *cmux, circuit_t *circ,
   circuitmux_assert_okay_paranoid(cmux);
 }
 
+/**
+ * Notify the circuitmux that a destroy was sent, so we can update
+ * the counter.
+ */
+
+void
+circuitmux_notify_xmit_destroy(circuitmux_t *cmux)
+{
+  tor_assert(cmux);
+
+  --(cmux->destroy_ctr);
+  --(global_destroy_ctr);
+  log_debug(LD_CIRC,
+            "Cmux at %p sent a destroy, cmux counter is now %ld, "
+            "global counter is now %ld\n",
+            cmux, cmux->destroy_ctr, global_destroy_ctr);
+}
+
 /*
  * Circuitmux consistency checking assertions
  */
@@ -1798,6 +1844,14 @@ circuitmux_append_destroy_cell(channel_t *chan,
   cell_queue_append_packed_copy(&cmux->destroy_cell_queue, &cell,
                                 chan->wide_circ_ids, 0);
 
+  /* Destroy entering the queue, update counters */
+  ++(cmux->destroy_ctr);
+  ++global_destroy_ctr;
+  log_debug(LD_CIRC,
+            "Cmux at %p queued a destroy for circ %u, "
+            "cmux counter is now %ld, global counter is now %ld\n",
+            cmux, circ_id, cmux->destroy_ctr, global_destroy_ctr);
+
   /* XXXX Duplicate code from append_cell_to_circuit_queue */
   if (!channel_has_queued_writes(chan)) {
     /* There is no data at all waiting to be sent on the outbuf.  Add a
@@ -1808,3 +1862,4 @@ circuitmux_append_destroy_cell(channel_t *chan,
     channel_flush_from_first_active_circuit(chan, 1);
   }
 }
+

src/or/circuitmux.h

@@ -124,6 +124,7 @@ circuit_t * circuitmux_get_first_active_circuit(circuitmux_t *cmux,
                                            cell_queue_t **destroy_queue_out);
 void circuitmux_notify_xmit_cells(circuitmux_t *cmux, circuit_t *circ,
                                   unsigned int n_cells);
+void circuitmux_notify_xmit_destroy(circuitmux_t *cmux);
 
 /* Circuit interface */
 void circuitmux_attach_circuit(circuitmux_t *cmux, circuit_t *circ,

src/or/relay.c

@@ -2358,6 +2358,8 @@ channel_flush_from_first_active_circuit(channel_t *chan, int max)
       tor_assert(destroy_queue->n > 0);
       cell = cell_queue_pop(destroy_queue);
       channel_write_packed_cell(chan, cell);
+      /* Update the cmux destroy counter */
+      circuitmux_notify_xmit_destroy(cmux);
       cell = NULL;
       ++n_flushed;
       continue;