Move destroy cells into a separate queue type of their own, to save RAM

We've been seeing problems with destroy cells queues taking up a
huge amount of RAM.  We can mitigate this, since while a full packed
destroy cell takes 514 bytes, we only need 5 bytes to remember a
circuit ID and a reason.

Fixes bug 24666. Bugfix on 0.2.5.1-alpha, when destroy cell queues
were introduced.

changes/bug24666

@@ -0,0 +1,7 @@
+  o Minor bugfixes (memory usage):
+
+    - When queuing DESTROY cells on a channel, only queue the
+      circuit-id and reason fields: not the entire 514-byte
+      cell. This fix should help mitigate any bugs or attacks that
+      fill up these queues, and free more RAM for other uses. Fixes
+      bug 24666; bugfix on 0.2.5.1-alpha.

src/or/circuitmux.c

@@ -117,7 +117,7 @@ struct circuitmux_s {
   struct circuit_t *active_circuits_head, *active_circuits_tail;
 
   /** List of queued destroy cells */
-  cell_queue_t destroy_cell_queue;
+  destroy_cell_queue_t destroy_cell_queue;
   /** Boolean: True iff the last cell to circuitmux_get_first_active_circuit
    * returned the destroy queue. Used to force alternation between
    * destroy/non-destroy cells.
@@ -383,7 +383,7 @@ circuitmux_alloc(void)
   rv = tor_malloc_zero(sizeof(*rv));
   rv->chanid_circid_map = tor_malloc_zero(sizeof(*( rv->chanid_circid_map)));
   HT_INIT(chanid_circid_muxinfo_map, rv->chanid_circid_map);
-  cell_queue_init(&rv->destroy_cell_queue);
+  destroy_cell_queue_init(&rv->destroy_cell_queue);
 
   return rv;
 }
@@ -522,19 +522,10 @@ circuitmux_detach_all_circuits(circuitmux_t *cmux, smartlist_t *detached_out)
 void
 circuitmux_mark_destroyed_circids_usable(circuitmux_t *cmux, channel_t *chan)
 {
-  packed_cell_t *cell;
-  int n_bad = 0;
+  destroy_cell_t *cell;
   TOR_SIMPLEQ_FOREACH(cell, &cmux->destroy_cell_queue.head, next) {
-    circid_t circid = 0;
-    if (packed_cell_is_destroy(chan, cell, &circid)) {
-      channel_mark_circid_usable(chan, circid);
-    } else {
-      ++n_bad;
-    }
+    channel_mark_circid_usable(chan, cell->circid);
   }
-  if (n_bad)
-    log_warn(LD_BUG, "%d cell(s) on destroy queue did not look like a "
-             "DESTROY cell.", n_bad);
 }
 
 /**
@@ -591,7 +582,7 @@ circuitmux_free(circuitmux_t *cmux)
               I64_PRINTF_ARG(global_destroy_ctr));
   }
 
-  cell_queue_clear(&cmux->destroy_cell_queue);
+  destroy_cell_queue_clear(&cmux->destroy_cell_queue);
 
   tor_free(cmux);
 }
@@ -1469,7 +1460,7 @@ circuitmux_set_num_cells(circuitmux_t *cmux, circuit_t *circ,
 
 circuit_t *
 circuitmux_get_first_active_circuit(circuitmux_t *cmux,
-                                    cell_queue_t **destroy_queue_out)
+                                    destroy_cell_queue_t **destroy_queue_out)
 {
   circuit_t *circ = NULL;
 
@@ -1885,14 +1876,7 @@ circuitmux_append_destroy_cell(channel_t *chan,
                                circid_t circ_id,
                                uint8_t reason)
 {
-  cell_t cell;
-  memset(&cell, 0, sizeof(cell_t));
-  cell.circ_id = circ_id;
-  cell.command = CELL_DESTROY;
-  cell.payload[0] = (uint8_t) reason;
-
-  cell_queue_append_packed_copy(NULL, &cmux->destroy_cell_queue, 0, &cell,
-                                chan->wide_circ_ids, 0);
+  destroy_cell_queue_append(&cmux->destroy_cell_queue, circ_id, reason);
 
   /* Destroy entering the queue, update counters */
   ++(cmux->destroy_ctr);
@@ -1925,13 +1909,13 @@ circuitmux_count_queued_destroy_cells(const channel_t *chan,
 
   int64_t manual_total = 0;
   int64_t manual_total_in_map = 0;
-  packed_cell_t *cell;
+  destroy_cell_t *cell;
 
   TOR_SIMPLEQ_FOREACH(cell, &cmux->destroy_cell_queue.head, next) {
     circid_t id;
     ++manual_total;
 
-    id = packed_cell_get_circid(cell, chan->wide_circ_ids);
+    id = cell->circid;
     if (circuit_id_in_use_on_channel(id, (channel_t*)chan))
       ++manual_total_in_map;
   }

src/or/circuitmux.h

@@ -127,7 +127,7 @@ int64_t circuitmux_count_queued_destroy_cells(const channel_t *chan,
 
 /* Channel interface */
 circuit_t * circuitmux_get_first_active_circuit(circuitmux_t *cmux,
-                                           cell_queue_t **destroy_queue_out);
+                                    destroy_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);

src/or/or.h

@@ -1121,6 +1121,21 @@ typedef struct cell_queue_t {
   int n; /**< The number of cells in the queue. */
 } cell_queue_t;
 
+/** A single queued destroy cell. */
+typedef struct destroy_cell_t {
+  TOR_SIMPLEQ_ENTRY(destroy_cell_t) next;
+  circid_t circid;
+  uint32_t inserted_time; /** Timestamp when this was queued. */
+  uint8_t reason;
+} destroy_cell_t;
+
+/** A queue of destroy cells on a channel. */
+typedef struct destroy_cell_queue_t {
+  /** Linked list of packed_cell_t */
+  TOR_SIMPLEQ_HEAD(dcell_simpleq, destroy_cell_t) head;
+  int n; /**< The number of cells in the queue. */
+} destroy_cell_queue_t;
+
 /** Beginning of a RELAY cell payload. */
 typedef struct {
   uint8_t command; /**< The end-to-end relay command. */

src/or/relay.c

@@ -2418,6 +2418,75 @@ cell_queue_pop(cell_queue_t *queue)
   return cell;
 }
 
+/** Initialize <b>queue</b> as an empty cell queue. */
+void
+destroy_cell_queue_init(destroy_cell_queue_t *queue)
+{
+  memset(queue, 0, sizeof(destroy_cell_queue_t));
+  TOR_SIMPLEQ_INIT(&queue->head);
+}
+
+/** Remove and free every cell in <b>queue</b>. */
+void
+destroy_cell_queue_clear(destroy_cell_queue_t *queue)
+{
+  destroy_cell_t *cell;
+  while ((cell = TOR_SIMPLEQ_FIRST(&queue->head))) {
+    TOR_SIMPLEQ_REMOVE_HEAD(&queue->head, next);
+    tor_free(cell);
+  }
+  TOR_SIMPLEQ_INIT(&queue->head);
+  queue->n = 0;
+}
+
+/** Extract and return the cell at the head of <b>queue</b>; return NULL if
+ * <b>queue</b> is empty. */
+STATIC destroy_cell_t *
+destroy_cell_queue_pop(destroy_cell_queue_t *queue)
+{
+  destroy_cell_t *cell = TOR_SIMPLEQ_FIRST(&queue->head);
+  if (!cell)
+    return NULL;
+  TOR_SIMPLEQ_REMOVE_HEAD(&queue->head, next);
+  --queue->n;
+  return cell;
+}
+
+/** Append a destroy cell for <b>circid</b> to <b>queue</b>. */
+void
+destroy_cell_queue_append(destroy_cell_queue_t *queue,
+                          circid_t circid,
+                          uint8_t reason)
+{
+  struct timeval now;
+
+  destroy_cell_t *cell = tor_malloc_zero(sizeof(destroy_cell_t));
+  cell->circid = circid;
+  cell->reason = reason;
+  tor_gettimeofday_cached_monotonic(&now);
+  /* Not yet used, but will be required for OOM handling. */
+  cell->inserted_time = (uint32_t)tv_to_msec(&now);
+
+  TOR_SIMPLEQ_INSERT_TAIL(&queue->head, cell, next);
+  ++queue->n;
+}
+
+/** Convert a destroy_cell_t to a newly allocated cell_t. Frees its input. */
+static packed_cell_t *
+destroy_cell_to_packed_cell(destroy_cell_t *inp, int wide_circ_ids)
+{
+  packed_cell_t *packed = packed_cell_new();
+  cell_t cell;
+  memset(&cell, 0, sizeof(cell));
+  cell.circ_id = inp->circid;
+  cell.command = CELL_DESTROY;
+  cell.payload[0] = inp->reason;
+  cell_pack(packed, &cell, wide_circ_ids);
+
+  tor_free(inp);
+  return packed;
+}
+
 /** Return the total number of bytes used for each packed_cell in a queue.
  * Approximate. */
 size_t
@@ -2596,7 +2665,8 @@ channel_flush_from_first_active_circuit(channel_t *chan, int max)
 {
   circuitmux_t *cmux = NULL;
   int n_flushed = 0;
-  cell_queue_t *queue, *destroy_queue=NULL;
+  cell_queue_t *queue;
+  destroy_cell_queue_t *destroy_queue=NULL;
   circuit_t *circ;
   or_circuit_t *or_circ;
   int streams_blocked;
@@ -2611,9 +2681,17 @@ channel_flush_from_first_active_circuit(channel_t *chan, int max)
   while (n_flushed < max) {
     circ = circuitmux_get_first_active_circuit(cmux, &destroy_queue);
     if (destroy_queue) {
+      destroy_cell_t *dcell;
       /* this code is duplicated from some of the logic below. Ugly! XXXX */
+      /* If we are given a destroy_queue here, then it is required to be
+       * nonempty... */
       tor_assert(destroy_queue->n > 0);
-      cell = cell_queue_pop(destroy_queue);
+      dcell = destroy_cell_queue_pop(destroy_queue);
+      /* ...and pop() will always yield a cell from a nonempty queue. */
+      tor_assert(dcell);
+      /* frees dcell */
+      cell = destroy_cell_to_packed_cell(dcell, chan->wide_circ_ids);
+      /* frees cell */
       channel_write_packed_cell(chan, cell);
       /* Update the cmux destroy counter */
       circuitmux_notify_xmit_destroy(cmux);

src/or/relay.h

@@ -63,6 +63,13 @@ void cell_queue_append_packed_copy(circuit_t *circ, cell_queue_t *queue,
 void append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan,
                                   cell_t *cell, cell_direction_t direction,
                                   streamid_t fromstream);
+
+void destroy_cell_queue_init(destroy_cell_queue_t *queue);
+void destroy_cell_queue_clear(destroy_cell_queue_t *queue);
+void destroy_cell_queue_append(destroy_cell_queue_t *queue,
+                               circid_t circid,
+                               uint8_t reason);
+
 void channel_unlink_all_circuits(channel_t *chan, smartlist_t *detached_out);
 int channel_flush_from_first_active_circuit(channel_t *chan, int max);
 void assert_circuit_mux_okay(channel_t *chan);
@@ -101,6 +108,7 @@ STATIC int connection_edge_process_resolved_cell(edge_connection_t *conn,
                                                  const relay_header_t *rh);
 STATIC packed_cell_t *packed_cell_new(void);
 STATIC packed_cell_t *cell_queue_pop(cell_queue_t *queue);
+STATIC destroy_cell_t *destroy_cell_queue_pop(destroy_cell_queue_t *queue);
 STATIC size_t cell_queues_get_total_allocation(void);
 STATIC int cell_queues_check_size(void);
 #endif

src/test/test_circuitmux.c

@@ -33,8 +33,9 @@ test_cmux_destroy_cell_queue(void *arg)
   circuitmux_t *cmux = NULL;
   channel_t *ch = NULL;
   circuit_t *circ = NULL;
-  cell_queue_t *cq = NULL;
+  destroy_cell_queue_t *cq = NULL;
   packed_cell_t *pc = NULL;
+  destroy_cell_t *dc = NULL;
 
 #ifdef ENABLE_MEMPOOLS
   init_cell_pool();
@@ -63,11 +64,10 @@ test_cmux_destroy_cell_queue(void *arg)
 
   tt_int_op(cq->n, ==, 3);
 
-  pc = cell_queue_pop(cq);
-  tt_assert(pc);
-  test_mem_op(pc->body, ==, "\x00\x00\x00\x64\x04\x0a\x00\x00\x00", 9);
-  packed_cell_free(pc);
-  pc = NULL;
+  dc = destroy_cell_queue_pop(cq);
+  tt_assert(dc);
+  tt_uint_op(dc->circid, ==, 100);
+  tor_free(dc);
 
   tt_int_op(circuitmux_num_cells(cmux), ==, 2);