1 files changed, 174 insertions, 51 deletions

diff --git a/src/or/relay.c b/src/or/relay.c
index 6d51f18a3a..a4f7718d90 100644
--- a/src/or/relay.c
+++ b/src/or/relay.c
@@ -38,26 +38,32 @@ static edge_connection_t *relay_lookup_conn(circuit_t *circ, cell_t *cell,
                                             cell_direction_t cell_direction,
                                             crypt_path_t *layer_hint);
 
-static int
-connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
-                                   edge_connection_t *conn,
-                                   crypt_path_t *layer_hint);
-static void
-circuit_consider_sending_sendme(circuit_t *circ, crypt_path_t *layer_hint);
-static void
-circuit_resume_edge_reading(circuit_t *circ, crypt_path_t *layer_hint);
-static int
-circuit_resume_edge_reading_helper(edge_connection_t *conn,
-                                   circuit_t *circ,
-                                   crypt_path_t *layer_hint);
-static int
-circuit_consider_stop_edge_reading(circuit_t *circ, crypt_path_t *layer_hint);
+static int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
+                                              edge_connection_t *conn,
+                                              crypt_path_t *layer_hint);
+static void circuit_consider_sending_sendme(circuit_t *circ,
+                                            crypt_path_t *layer_hint);
+static void circuit_resume_edge_reading(circuit_t *circ,
+                                        crypt_path_t *layer_hint);
+static int circuit_resume_edge_reading_helper(edge_connection_t *conn,
+                                              circuit_t *circ,
+                                              crypt_path_t *layer_hint);
+static int circuit_consider_stop_edge_reading(circuit_t *circ,
+                                              crypt_path_t *layer_hint);
+static int circuit_queue_streams_are_blocked(circuit_t *circ);
 
 /** Cache the current hi-res time; the cache gets reset when libevent
  * calls us. */
 
 static struct timeval cached_time_hires = {0, 0};
 
+/** 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
+
 static void
 tor_gettimeofday_cached(struct timeval *tv)
 {
@@ -268,7 +274,7 @@ circuit_receive_relay_cell(cell_t *cell, circuit_t *circ,
                                   * we might kill the circ before we relay
                                   * the cells. */
 
-  append_cell_to_circuit_queue(circ, or_conn, cell, cell_direction);
+  append_cell_to_circuit_queue(circ, or_conn, cell, cell_direction, 0);
   return 0;
 }
 
@@ -365,7 +371,7 @@ relay_crypt(circuit_t *circ, cell_t *cell, cell_direction_t cell_direction,
 static int
 circuit_package_relay_cell(cell_t *cell, circuit_t *circ,
                            cell_direction_t cell_direction,
-                           crypt_path_t *layer_hint)
+                           crypt_path_t *layer_hint, streamid_t on_stream)
 {
   or_connection_t *conn; /* where to send the cell */
 
@@ -409,7 +415,7 @@ circuit_package_relay_cell(cell_t *cell, circuit_t *circ,
   }
   ++stats_n_relay_cells_relayed;
 
-  append_cell_to_circuit_queue(circ, conn, cell, cell_direction);
+  append_cell_to_circuit_queue(circ, conn, cell, cell_direction, on_stream);
   return 0;
 }
 
@@ -536,7 +542,7 @@ relay_command_to_string(uint8_t command)
  * return 0.
  */
 int
-relay_send_command_from_edge(uint16_t stream_id, circuit_t *circ,
+relay_send_command_from_edge(streamid_t stream_id, circuit_t *circ,
                              uint8_t relay_command, const char *payload,
                              size_t payload_len, crypt_path_t *cpath_layer)
 {
@@ -624,8 +630,8 @@ relay_send_command_from_edge(uint16_t stream_id, circuit_t *circ,
     }
   }
 
-  if (circuit_package_relay_cell(&cell, circ, cell_direction, cpath_layer)
-      < 0) {
+  if (circuit_package_relay_cell(&cell, circ, cell_direction, cpath_layer,
+                                 stream_id) < 0) {
     log_warn(LD_BUG,"circuit_package_relay_cell failed. Closing.");
     circuit_mark_for_close(circ, END_CIRC_REASON_INTERNAL);
     return -1;
@@ -692,7 +698,8 @@ edge_reason_is_retriable(int reason)
          reason == END_STREAM_REASON_RESOURCELIMIT ||
          reason == END_STREAM_REASON_EXITPOLICY ||
          reason == END_STREAM_REASON_RESOLVEFAILED ||
-         reason == END_STREAM_REASON_MISC;
+         reason == END_STREAM_REASON_MISC ||
+         reason == END_STREAM_REASON_NOROUTE;
 }
 
 /** Called when we receive an END cell on a stream that isn't open yet,
@@ -787,6 +794,7 @@ connection_ap_process_end_not_open(
       case END_STREAM_REASON_RESOLVEFAILED:
       case END_STREAM_REASON_TIMEOUT:
       case END_STREAM_REASON_MISC:
+      case END_STREAM_REASON_NOROUTE:
         if (client_dns_incr_failures(conn->socks_request->address)
             < MAX_RESOLVE_FAILURES) {
           /* We haven't retried too many times; reattach the connection. */
@@ -945,7 +953,7 @@ connection_edge_process_relay_cell_not_open(
     }
 
     /* handle anything that might have queued */
-    if (connection_edge_package_raw_inbuf(conn, 1) < 0) {
+    if (connection_edge_package_raw_inbuf(conn, 1, NULL) < 0) {
       /* (We already sent an end cell if possible) */
       connection_mark_for_close(TO_CONN(conn));
       return 0;
@@ -1235,9 +1243,13 @@ connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
       conn->package_window += STREAMWINDOW_INCREMENT;
       log_debug(domain,"stream-level sendme, packagewindow now %d.",
                 conn->package_window);
+      if (circuit_queue_streams_are_blocked(circ)) {
+        /* Still waiting for queue to flush; don't touch conn */
+        return 0;
+      }
       connection_start_reading(TO_CONN(conn));
       /* handle whatever might still be on the inbuf */
-      if (connection_edge_package_raw_inbuf(conn, 1) < 0) {
+      if (connection_edge_package_raw_inbuf(conn, 1, NULL) < 0) {
         /* (We already sent an end cell if possible) */
         connection_mark_for_close(TO_CONN(conn));
         return 0;
@@ -1303,15 +1315,19 @@ uint64_t stats_n_data_cells_received = 0;
  * ever received were completely full of data. */
 uint64_t stats_n_data_bytes_received = 0;
 
-/** While conn->inbuf has an entire relay payload of bytes on it,
- * and the appropriate package windows aren't empty, grab a cell
- * and send it down the circuit.
+/** If <b>conn</b> has an entire relay payload of bytes on its inbuf (or
+ * <b>package_partial</b> is true), and the appropriate package windows aren't
+ * empty, grab a cell and send it down the circuit.
+ *
+ * If *<b>max_cells</b> is given, package no more than max_cells.  Decrement
+ * *<b>max_cells</b> by the number of cells packaged.
  *
  * Return -1 (and send a RELAY_COMMAND_END cell if necessary) if conn should
  * be marked for close, else return 0.
  */
 int
-connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial)
+connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial,
+                                  int *max_cells)
 {
   size_t amount_to_process, length;
   char payload[CELL_PAYLOAD_SIZE];
@@ -1327,7 +1343,10 @@ connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial)
     return 0;
   }
 
-repeat_connection_edge_package_raw_inbuf:
+  if (max_cells && *max_cells <= 0)
+    return 0;
+
+ repeat_connection_edge_package_raw_inbuf:
 
   circ = circuit_get_by_edge_conn(conn);
   if (!circ) {
@@ -1388,6 +1407,12 @@ repeat_connection_edge_package_raw_inbuf:
   }
   log_debug(domain,"conn->package_window is now %d",conn->package_window);
 
+  if (max_cells) {
+    *max_cells -= 1;
+    if (*max_cells <= 0)
+      return 0;
+  }
+
   /* handle more if there's more, or return 0 if there isn't */
   goto repeat_connection_edge_package_raw_inbuf;
 }
@@ -1435,7 +1460,10 @@ connection_edge_consider_sending_sendme(edge_connection_t *conn)
 static void
 circuit_resume_edge_reading(circuit_t *circ, crypt_path_t *layer_hint)
 {
-
+  if (circuit_queue_streams_are_blocked(circ)) {
+    log_debug(layer_hint?LD_APP:LD_EXIT,"Too big queue, no resuming");
+    return;
+  }
   log_debug(layer_hint?LD_APP:LD_EXIT,"resuming");
 
   if (CIRCUIT_IS_ORIGIN(circ))
@@ -1451,31 +1479,100 @@ circuit_resume_edge_reading(circuit_t *circ, crypt_path_t *layer_hint)
  * of a linked list of edge streams that should each be considered.
  */
 static int
-circuit_resume_edge_reading_helper(edge_connection_t *conn,
+circuit_resume_edge_reading_helper(edge_connection_t *first_conn,
                                    circuit_t *circ,
                                    crypt_path_t *layer_hint)
 {
-  for ( ; conn; conn=conn->next_stream) {
-    if (conn->_base.marked_for_close)
+  edge_connection_t *conn;
+  int n_streams, n_streams_left;
+  int packaged_this_round;
+  int cells_on_queue;
+  int cells_per_conn;
+
+  /* How many cells do we have space for?  It will be the minimum of
+   * the number needed to exhaust the package window, and the minimum
+   * needed to fill the cell queue. */
+  int max_to_package = circ->package_window;
+  if (CIRCUIT_IS_ORIGIN(circ)) {
+    cells_on_queue = circ->n_conn_cells.n;
+  } else {
+    or_circuit_t *or_circ = TO_OR_CIRCUIT(circ);
+    cells_on_queue = or_circ->p_conn_cells.n;
+  }
+  if (CELL_QUEUE_HIGHWATER_SIZE - cells_on_queue < max_to_package)
+    max_to_package = CELL_QUEUE_HIGHWATER_SIZE - cells_on_queue;
+
+  /* Count how many non-marked streams there are that have anything on
+   * their inbuf, and enable reading on all of the connections. */
+  n_streams = 0;
+  for (conn=first_conn; conn; conn=conn->next_stream) {
+    if (conn->_base.marked_for_close || conn->package_window <= 0)
       continue;
-    if ((!layer_hint && conn->package_window > 0) ||
-        (layer_hint && conn->package_window > 0 &&
-         conn->cpath_layer == layer_hint)) {
+    if (!layer_hint || conn->cpath_layer == layer_hint) {
       connection_start_reading(TO_CONN(conn));
+
+      if (buf_datalen(conn->_base.inbuf) > 0)
+        ++n_streams;
+    }
+  }
+
+  if (n_streams == 0) /* avoid divide-by-zero */
+    return 0;
+
+ again:
+
+  cells_per_conn = CEIL_DIV(max_to_package, n_streams);
+
+  packaged_this_round = 0;
+  n_streams_left = 0;
+
+  /* Iterate over all connections.  Package up to cells_per_conn cells on
+   * each.  Update packaged_this_round with the total number of cells
+   * packaged, and n_streams_left with the number that still have data to
+   * package.
+   */
+  for (conn=first_conn; conn; conn=conn->next_stream) {
+    if (conn->_base.marked_for_close || conn->package_window <= 0)
+      continue;
+    if (!layer_hint || conn->cpath_layer == layer_hint) {
+      int n = cells_per_conn, r;
       /* handle whatever might still be on the inbuf */
-      if (connection_edge_package_raw_inbuf(conn, 1)<0) {
-        /* (We already sent an end cell if possible) */
+      r = connection_edge_package_raw_inbuf(conn, 1, &n);
+
+      /* Note how many we packaged */
+      packaged_this_round += (cells_per_conn-n);
+
+      if (r<0) {
+        /* Problem while packaging. (We already sent an end cell if
+         * possible) */
         connection_mark_for_close(TO_CONN(conn));
         continue;
       }
 
+      /* If there's still data to read, we'll be coming back to this stream. */
+      if (buf_datalen(conn->_base.inbuf))
+          ++n_streams_left;
+
       /* If the circuit won't accept any more data, return without looking
        * at any more of the streams. Any connections that should be stopped
        * have already been stopped by connection_edge_package_raw_inbuf. */
       if (circuit_consider_stop_edge_reading(circ, layer_hint))
         return -1;
+      /* XXXX should we also stop immediately if we fill up the cell queue?
+       * Probably. */
     }
   }
+
+  /* If we made progress, and we are willing to package more, and there are
+   * any streams left that want to package stuff... try again!
+   */
+  if (packaged_this_round && packaged_this_round < max_to_package &&
+      n_streams_left) {
+    max_to_package -= packaged_this_round;
+    n_streams = n_streams_left;
+    goto again;
+  }
+
   return 0;
 }
 
@@ -1544,13 +1641,6 @@ circuit_consider_sending_sendme(circuit_t *circ, crypt_path_t *layer_hint)
   }
 }
 
-/** 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
-
 #ifdef ACTIVE_CIRCUITS_PARANOIA
 #define assert_active_circuits_ok_paranoid(conn) \
      assert_active_circuits_ok(conn)
@@ -2091,12 +2181,19 @@ connection_or_unlink_all_active_circs(or_connection_t *orconn)
 
 /** Block (if <b>block</b> is true) or unblock (if <b>block</b> is false)
  * every edge connection that is using <b>circ</b> to write to <b>orconn</b>,
- * and start or stop reading as appropriate. */
-static void
+ * and start or stop reading as appropriate.
+ *
+ * If <b>stream_id</b> is nonzero, block only the edge connection whose
+ * stream_id matches it.
+ *
+ * Returns the number of streams whose status we changed.
+ */
+static int
 set_streams_blocked_on_circ(circuit_t *circ, or_connection_t *orconn,
-                            int block)
+                            int block, streamid_t stream_id)
 {
   edge_connection_t *edge = NULL;
+  int n = 0;
   if (circ->n_conn == orconn) {
     circ->streams_blocked_on_n_conn = block;
     if (CIRCUIT_IS_ORIGIN(circ))
@@ -2109,7 +2206,13 @@ set_streams_blocked_on_circ(circuit_t *circ, or_connection_t *orconn,
 
   for (; edge; edge = edge->next_stream) {
     connection_t *conn = TO_CONN(edge);
-    edge->edge_blocked_on_circ = block;
+    if (stream_id && edge->stream_id != stream_id)
+      continue;
+
+    if (edge->edge_blocked_on_circ != block) {
+      ++n;
+      edge->edge_blocked_on_circ = block;
+    }
 
     if (!conn->read_event) {
       /* This connection is a placeholder for something; probably a DNS
@@ -2126,6 +2229,8 @@ set_streams_blocked_on_circ(circuit_t *circ, or_connection_t *orconn,
         connection_start_reading(conn);
     }
   }
+
+  return n;
 }
 
 /** Pull as many cells as possible (but no more than <b>max</b>) from the
@@ -2251,7 +2356,7 @@ connection_or_flush_from_first_active_circuit(or_connection_t *conn, 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)
-    set_streams_blocked_on_circ(circ, conn, 0); /* unblock streams */
+    set_streams_blocked_on_circ(circ, conn, 0, 0); /* unblock streams */
 
   /* Did we just run out of cells on this circuit's queue? */
   if (queue->n == 0) {
@@ -2268,7 +2373,8 @@ connection_or_flush_from_first_active_circuit(or_connection_t *conn, int max,
  * transmitting in <b>direction</b>. */
 void
 append_cell_to_circuit_queue(circuit_t *circ, or_connection_t *orconn,
-                             cell_t *cell, cell_direction_t direction)
+                             cell_t *cell, cell_direction_t direction,
+                             streamid_t fromstream)
 {
   cell_queue_t *queue;
   int streams_blocked;
@@ -2293,7 +2399,12 @@ append_cell_to_circuit_queue(circuit_t *circ, or_connection_t *orconn,
   /* 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)
-    set_streams_blocked_on_circ(circ, orconn, 1); /* block streams */
+    set_streams_blocked_on_circ(circ, orconn, 1, 0); /* block streams */
+
+  if (streams_blocked && fromstream) {
+    /* This edge connection is apparently not blocked; block it. */
+    set_streams_blocked_on_circ(circ, orconn, 1, fromstream);
+  }
 
   if (queue->n == 1) {
     /* This was the first cell added to the queue.  We need to make this
@@ -2426,3 +2537,15 @@ assert_active_circuits_ok(or_connection_t *orconn)
   tor_assert(n == smartlist_len(orconn->active_circuit_pqueue));
 }
 
+/** Return 1 if we shouldn't restart reading on this circuit, even if
+ * we get a SENDME.  Else return 0.
+*/
+static int
+circuit_queue_streams_are_blocked(circuit_t *circ)
+{
+  if (CIRCUIT_IS_ORIGIN(circ)) {
+    return circ->streams_blocked_on_n_conn;
+  } else {
+    return circ->streams_blocked_on_p_conn;
+  }
+}