Implement server side of Proposal 174 (optimistic data)

Ian's original message:
    The current code actually correctly handles queued data at the
    Exit; if there is queued data in a EXIT_CONN_STATE_CONNECTING
    stream, that data will be immediately sent when the connection
    succeeds. If the connection fails, the data will be correctly
    ignored and freed. The problem with the current server code is
    that the server currently drops DATA cells on streams in the
    EXIT_CONN_STATE_CONNECTING state. Also, if you try to queue data
    in the EXIT_CONN_STATE_RESOLVING state, bad things happen because
    streams in that state don't yet have conn->write_event set, and so
    some existing sanity checks (any stream with queued data is at
    least potentially writable) are no longer sound.

    The solution is to simply not drop received DATA cells while in
    the EXIT_CONN_STATE_CONNECTING state. Also do not send SENDME
    cells in this state, so that the OP cannot send more than one
    window's worth of data to be queued at the Exit. Finally, patch
    the sanity checks so that streams in the EXIT_CONN_STATE_RESOLVING
    state that have buffered data can pass.

    [...] Here is a simple patch. It seems to work with both regular
    streams and hidden services, but there may be other corner cases
    I'm not aware of. (Do streams used for directory fetches, hidden
    services, etc. take a different code path?)

src/or/connection.c

@@ -3259,7 +3259,13 @@ _connection_write_to_buf_impl(const char *string, size_t len,
     return;
   }
 
-  connection_start_writing(conn);
+  /* If we receive optimistic data in the EXIT_CONN_STATE_RESOLVING
+   * state, we don't want to try to write it right away, since
+   * conn->write_event won't be set yet.  Otherwise, write data from
+   * this conn as the socket is available. */
+  if (conn->write_event) {
+      connection_start_writing(conn);
+  }
   if (zlib) {
     conn->outbuf_flushlen += buf_datalen(conn->outbuf) - old_datalen;
   } else {
@@ -3835,7 +3841,12 @@ assert_connection_ok(connection_t *conn, time_t now)
     tor_assert(conn->s < 0);
 
   if (conn->outbuf_flushlen > 0) {
-    tor_assert(connection_is_writing(conn) || conn->write_blocked_on_bw ||
+    /* With optimistic data, we may have queued data in
+     * EXIT_CONN_STATE_RESOLVING while the conn is not yet marked to writing.
+     * */
+    tor_assert((conn->type == CONN_TYPE_EXIT &&
+		conn->state == EXIT_CONN_STATE_RESOLVING) ||
+	    connection_is_writing(conn) || conn->write_blocked_on_bw ||
             (CONN_IS_EDGE(conn) && TO_EDGE_CONN(conn)->edge_blocked_on_circ));
   }
 

src/or/relay.c

@@ -1031,6 +1031,9 @@ connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
   relay_header_t rh;
   unsigned domain = layer_hint?LD_APP:LD_EXIT;
   int reason;
+  int optimistic_data = 0;  /* Set to 1 if we receive data on a stream
+			       that's in the EXIT_CONN_STATE_RESOLVING
+			       or EXIT_CONN_STATE_CONNECTING states.*/
 
   tor_assert(cell);
   tor_assert(circ);
@@ -1050,9 +1053,20 @@ connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
   /* either conn is NULL, in which case we've got a control cell, or else
    * conn points to the recognized stream. */
 
-  if (conn && !connection_state_is_open(TO_CONN(conn)))
-    return connection_edge_process_relay_cell_not_open(
-             &rh, cell, circ, conn, layer_hint);
+  if (conn && !connection_state_is_open(TO_CONN(conn))) {
+    if (conn->_base.type == CONN_TYPE_EXIT &&
+	    (conn->_base.state == EXIT_CONN_STATE_CONNECTING ||
+		conn->_base.state == EXIT_CONN_STATE_RESOLVING) &&
+	rh.command == RELAY_COMMAND_DATA) {
+	/* We're going to allow DATA cells to be delivered to an exit
+	 * node in state EXIT_CONN_STATE_CONNECTING or
+	 * EXIT_CONN_STATE_RESOLVING.  This speeds up HTTP, for example. */
+	optimistic_data = 1;
+    } else {
+	return connection_edge_process_relay_cell_not_open(
+		 &rh, cell, circ, conn, layer_hint);
+    }
+  }
 
   switch (rh.command) {
     case RELAY_COMMAND_DROP:
@@ -1102,7 +1116,9 @@ connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
       log_debug(domain,"circ deliver_window now %d.", layer_hint ?
                 layer_hint->deliver_window : circ->deliver_window);
 
-      circuit_consider_sending_sendme(circ, layer_hint);
+      if (!optimistic_data) {
+	  circuit_consider_sending_sendme(circ, layer_hint);
+      }
 
       if (!conn) {
         log_info(domain,"data cell dropped, unknown stream (streamid %d).",
@@ -1119,7 +1135,9 @@ connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
       stats_n_data_bytes_received += rh.length;
       connection_write_to_buf((char*)(cell->payload + RELAY_HEADER_SIZE),
                               rh.length, TO_CONN(conn));
-      connection_edge_consider_sending_sendme(conn);
+      if (!optimistic_data) {
+	  connection_edge_consider_sending_sendme(conn);
+      }
       return 0;
     case RELAY_COMMAND_END:
       reason = rh.length > 0 ?