r11552@catbus: nickm | 2007-01-27 03:55:02 -0500

 This one is a little tricky.  Our BEGIN_DIR implementation has a
 problem: the dirserv conns will decide they can flush all their data
 immediately, since the edge_conns will read greedily.
 
 For our 0.1.2 workaround, we track which or_conn a bridged dirserv
 conn is attached to, and stop writing when its outbuf is too full, and
 start writing again when the or_conn's outbuf empties out a little.
 
 This requires a bit of pointer management.  Let's hope it works.
 


svn:r9432

5 files changed, 210 insertions, 2 deletions

diff --git a/src/or/connection.c b/src/or/connection.c
index 188bfc75ee..f1962d9614 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -426,6 +426,10 @@ connection_about_to_close_connection(connection_t *conn)
       }
       if (conn->purpose == DIR_PURPOSE_FETCH_RENDDESC)
         rend_client_desc_here(dir_conn->rend_query); /* give it a try */
+      /* If this is from BEGIN_DIR, unlink it from the edge_conn and
+       * the or_conn. */
+      if (dir_conn->bridge_conn)
+        connection_dirserv_unlink_from_bridge(dir_conn);
       break;
     case CONN_TYPE_OR:
       or_conn = TO_OR_CONN(conn);
@@ -452,6 +456,13 @@ connection_about_to_close_connection(connection_t *conn)
         control_event_or_conn_status(or_conn, OR_CONN_EVENT_CLOSED,
                 control_tls_error_to_reason(or_conn->tls_error));
       }
+      /* Remove any dir_conns that are blocked on this one.  Non-blocked
+       * ones will die when the circuits do. */
+      while (or_conn->blocked_dir_connections) {
+        dir_connection_t *dir_conn = or_conn->blocked_dir_connections;
+        connection_dirserv_unlink_from_bridge(dir_conn);
+        tor_assert(or_conn->blocked_dir_connections != dir_conn);
+      }
       /* Now close all the attached circuits on it. */
       circuit_unlink_all_from_or_conn(TO_OR_CONN(conn),
                                       END_CIRC_REASON_OR_CONN_CLOSED);
@@ -485,6 +496,9 @@ connection_about_to_close_connection(connection_t *conn)
       if (conn->state == EXIT_CONN_STATE_RESOLVING) {
         connection_dns_remove(edge_conn);
       }
+      /* If we're relaying a dirserv connection, clean up any pointers */
+      if (edge_conn->bridge_for_conn)
+        connection_dirserv_unlink_from_bridge(edge_conn->bridge_for_conn);
       break;
     case CONN_TYPE_DNSWORKER:
       if (conn->state == DNSWORKER_STATE_BUSY) {
@@ -2243,6 +2257,8 @@ connection_flushed_some(connection_t *conn)
   if (conn->type == CONN_TYPE_DIR &&
       conn->state == DIR_CONN_STATE_SERVER_WRITING)
     return connection_dirserv_flushed_some(TO_DIR_CONN(conn));
+  else if (conn->type == CONN_TYPE_OR)
+    return connection_or_flushed_some(TO_OR_CONN(conn));
   else
     return 0;
 }
@@ -2425,7 +2441,35 @@ assert_connection_ok(connection_t *conn, time_t now)
       tor_assert(conn->purpose == EXIT_PURPOSE_CONNECT ||
                  conn->purpose == EXIT_PURPOSE_RESOLVE);
     }
-  } else if (conn->type != CONN_TYPE_DIR) {
+    if (edge_conn->bridge_for_conn) {
+      tor_assert(conn->type == CONN_TYPE_EXIT);
+      tor_assert(edge_conn->bridge_for_conn->bridge_conn == edge_conn);
+    }
+  } else if (conn->type == CONN_TYPE_DIR) {
+    dir_connection_t *dir_conn = TO_DIR_CONN(conn);
+
+    if (dir_conn->bridge_conn) {
+      tor_assert(DIR_CONN_IS_SERVER(conn));
+      tor_assert(dir_conn->bridge_conn->bridge_for_conn == dir_conn);
+      if (dir_conn->bridge_conn->on_circuit) {
+        dir_connection_t *d;
+        or_connection_t *or_conn;
+        tor_assert(!CIRCUIT_IS_ORIGIN(dir_conn->bridge_conn->on_circuit));
+        or_conn = TO_OR_CIRCUIT(dir_conn->bridge_conn->on_circuit)->p_conn;
+        if (dir_conn->is_blocked_on_or_conn)
+          tor_assert(or_conn);
+        for (d = or_conn->blocked_dir_connections; d;
+             d = d->next_blocked_on_same_or_conn) {
+          if (d == dir_conn) {
+            tor_assert(dir_conn->is_blocked_on_or_conn == 1);
+            break;
+          }
+        }
+        if (!d)
+          tor_assert(!dir_conn->is_blocked_on_or_conn);
+      }
+    }
+  } else {
     /* Purpose is only used for dir and exit types currently */
     tor_assert(!conn->purpose);
   }
diff --git a/src/or/connection_edge.c b/src/or/connection_edge.c
index 22ef529c36..7b2f95eea3 100644
--- a/src/or/connection_edge.c
+++ b/src/or/connection_edge.c
@@ -2461,6 +2461,9 @@ connection_exit_connect_dir(edge_connection_t *exit_conn)
     return 0;
   }
 
+  dir_conn->bridge_conn = exit_conn;
+  exit_conn->bridge_for_conn = dir_conn;
+
   connection_start_reading(TO_CONN(dir_conn));
   connection_start_reading(TO_CONN(exit_conn));
 
diff --git a/src/or/connection_or.c b/src/or/connection_or.c
index 295a060fed..036a1b1ddc 100644
--- a/src/or/connection_or.c
+++ b/src/or/connection_or.c
@@ -21,6 +21,7 @@ const char connection_or_c_id[] =
 
 static int connection_tls_finish_handshake(or_connection_t *conn);
 static int connection_or_process_cells_from_inbuf(or_connection_t *conn);
+static int connection_or_empty_enough_for_dirserv_data(or_connection_t *conn);
 
 /**************************************************************/
 
@@ -224,6 +225,17 @@ connection_or_process_inbuf(or_connection_t *conn)
   }
 }
 
+/** Called whenever we have flushed some data on an or_conn. */
+int
+connection_or_flushed_some(or_connection_t *conn)
+{
+  if (conn->blocked_dir_connections &&
+      connection_or_empty_enough_for_dirserv_data(conn)) {
+    connection_dirserv_stop_blocking_all_on_or_conn(conn);
+  }
+  return 0;
+}
+
 /** Connection <b>conn</b> has finished writing and has no bytes left on
  * its outbuf.
  *
@@ -798,3 +810,25 @@ connection_or_count_pending_circs(or_connection_t *or_conn)
   return cnt;
 }
 
+#define BUF_FULLNESS_THRESHOLD (128*1024)
+#define BUF_EMPTINESS_THRESHOLD (96*1024)
+
+/** Return true iff there is so much data waiting to be flushed on <b>conn</b>
+ * that we should stop writing directory data to it. */
+int
+connection_or_too_full_for_dirserv_data(or_connection_t *conn)
+{
+  return buf_datalen(conn->_base.outbuf) > BUF_FULLNESS_THRESHOLD;
+}
+
+/** Return true iff there is no longer so much data waiting to be flushed on
+ * <b>conn</b> that we should not write directory data to it. */
+static int
+connection_or_empty_enough_for_dirserv_data(or_connection_t *conn)
+{
+  /* Note that the threshold to stop writing is a bit higher than the
+   * threshold to start again: this should (with any luck) keep us from
+   * flapping about indefinitely. */
+  return buf_datalen(conn->_base.outbuf) < BUF_EMPTINESS_THRESHOLD;
+}
+
diff --git a/src/or/dirserv.c b/src/or/dirserv.c
index bf0084be12..dcaccd9d2b 100644
--- a/src/or/dirserv.c
+++ b/src/or/dirserv.c
@@ -1951,6 +1951,99 @@ dirserv_test_reachability(int try_all)
     ctr = (ctr + 1) % 128;
 }
 
+/** If <b>conn</b> is a dirserv connection tunneled over an or_connection,
+ * return that connection.  Otherwise, return NULL. */
+static INLINE or_connection_t *
+connection_dirserv_get_target_or_conn(dir_connection_t *conn)
+{
+  if (conn->bridge_conn &&
+      conn->bridge_conn->on_circuit &&
+      !CIRCUIT_IS_ORIGIN(conn->bridge_conn->on_circuit)) {
+    or_circuit_t *circ = TO_OR_CIRCUIT(conn->bridge_conn->on_circuit);
+    return circ->p_conn;
+  } else {
+    return NULL;
+  }
+}
+
+/** Remove <b>dir_conn</b> from the list of bridged dirserv connections
+ * blocking on <b>or_conn</b>, and set its status to nonblocked. */
+static INLINE void
+connection_dirserv_remove_from_blocked_list(or_connection_t *or_conn,
+                                            dir_connection_t *dir_conn)
+{
+  dir_connection_t **c;
+  for (c = &or_conn->blocked_dir_connections; *c;
+       c = &(*c)->next_blocked_on_same_or_conn) {
+    if (*c == dir_conn) {
+      tor_assert(dir_conn->is_blocked_on_or_conn == 1);
+      *c = dir_conn->next_blocked_on_same_or_conn;
+      dir_conn->next_blocked_on_same_or_conn = NULL;
+      dir_conn->is_blocked_on_or_conn = 0;
+      return;
+    }
+  }
+  tor_assert(!dir_conn->is_blocked_on_or_conn);
+}
+
+/* If <b>dir_conn</b> is a dirserv connection that's bridged over an edge_conn
+ * onto an or_conn, remove it from the blocked list (if it's blocked) and
+ * unlink it and the edge_conn from one another. */
+void
+connection_dirserv_unlink_from_bridge(dir_connection_t *dir_conn)
+{
+  edge_connection_t *edge_conn;
+  or_connection_t *or_conn;
+  tor_assert(dir_conn);
+  edge_conn = dir_conn->bridge_conn;
+  or_conn = connection_dirserv_get_target_or_conn(dir_conn);
+  if (or_conn) {
+    /* XXXX Really, this is only necessary if dir_conn->is_blocked_on_or_conn.
+     * But for now, let's leave it in, so the assert can catch  */
+    connection_dirserv_remove_from_blocked_list(or_conn, dir_conn);
+  }
+  dir_conn->is_blocked_on_or_conn = 0; /* Probably redundant. */
+  edge_conn->bridge_for_conn = NULL;
+  dir_conn->bridge_conn = NULL;
+}
+
+/** Stop writing on a bridged dir_conn, and remember that it's blocked because
+ * its or_conn was too full. */
+static void
+connection_dirserv_mark_as_blocked(dir_connection_t *dir_conn)
+{
+  or_connection_t *or_conn;
+  if (dir_conn->is_blocked_on_or_conn)
+    return;
+  tor_assert(! dir_conn->next_blocked_on_same_or_conn);
+  or_conn = connection_dirserv_get_target_or_conn(dir_conn);
+  if (!or_conn)
+    return;
+  dir_conn->next_blocked_on_same_or_conn = or_conn->blocked_dir_connections;
+  or_conn->blocked_dir_connections = dir_conn;
+  dir_conn->is_blocked_on_or_conn = 1;
+  connection_stop_writing(TO_CONN(dir_conn));
+}
+
+/** Tell all bridged dir_conns that were blocked because or_conn's outbuf was
+ * too full that they can write again. */
+void
+connection_dirserv_stop_blocking_all_on_or_conn(or_connection_t *or_conn)
+{
+  dir_connection_t *dir_conn, *next;
+
+  while (or_conn->blocked_dir_connections) {
+    dir_conn = or_conn->blocked_dir_connections;
+    next = dir_conn->next_blocked_on_same_or_conn;
+
+    dir_conn->is_blocked_on_or_conn = 0;
+    dir_conn->next_blocked_on_same_or_conn = NULL;
+    connection_start_writing(TO_CONN(dir_conn));
+    dir_conn = next;
+  }
+  or_conn->blocked_dir_connections = NULL;
+}
+
 /** Return an approximate estimate of the number of bytes that will
  * be needed to transmit the server descriptors (if is_serverdescs --
  * they can be either d/ or fp/ queries) or networkstatus objects (if
@@ -2155,11 +2248,18 @@ connection_dirserv_add_networkstatus_bytes_to_outbuf(dir_connection_t *conn)
 int
 connection_dirserv_flushed_some(dir_connection_t *conn)
 {
+  or_connection_t *or_conn;
   tor_assert(conn->_base.state == DIR_CONN_STATE_SERVER_WRITING);
 
   if (buf_datalen(conn->_base.outbuf) >= DIRSERV_BUFFER_MIN)
     return 0;
 
+  if ((or_conn = connection_dirserv_get_target_or_conn(conn)) &&
+      connection_or_too_full_for_dirserv_data(or_conn)) {
+    connection_dirserv_mark_as_blocked(conn);
+    return 0;
+  }
+
   switch (conn->dir_spool_src) {
     case DIR_SPOOL_SERVER_BY_DIGEST:
     case DIR_SPOOL_SERVER_BY_FP:
@@ -2174,6 +2274,7 @@ connection_dirserv_flushed_some(dir_connection_t *conn)
   }
 }
 
+
 /** Release all storage used by the directory server. */
 void
 dirserv_free_all(void)
diff --git a/src/or/or.h b/src/or/or.h
index 4e4ba5557b..04f24a4c98 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -752,6 +752,9 @@ typedef struct or_connection_t {
                    * n_conn ? */
   struct or_connection_t *next_with_same_id; /**< Next connection with same
                                            * identity digest as this one. */
+  /** Linked list of bridged dirserver connections that can't write until
+   * this connection's outbuf is less full. */
+  struct dir_connection_t *blocked_dir_connections;
   uint16_t next_circ_id; /**< Which circ_id do we try to use next on
                           * this connection?  This is always in the
                           * range 0..1<<15-1. */
@@ -791,6 +794,10 @@ typedef struct edge_connection_t {
   /* XXXX NM This can get re-used after 2**32 streams */
   uint32_t global_identifier;
 
+  /** Exit only: a dirserv connection that is tunneled over this connection
+   * using a socketpair. */
+  struct dir_connection_t *bridge_for_conn;
+
   char rend_query[REND_SERVICE_ID_LEN+1]; /**< What rendezvous service are we
                                            * querying for? (AP only) */
 
@@ -809,6 +816,10 @@ typedef struct dir_connection_t {
   char *requested_resource; /**< Which 'resource' did we ask the directory
                              * for? */
   unsigned int dirconn_direct:1; /**< Is this dirconn direct, or via Tor? */
+  /** True iff this is a dirserv conn, and it's tunneled over an or_conn,
+   * and we've stopped writing because the or_conn had too much pending
+   * data to write */
+  unsigned int is_blocked_on_or_conn : 1;
 
   /* Used only for server sides of some dir connections, to implement
    * "spooling" of directory material to the outbuf.  Otherwise, we'd have
@@ -817,7 +828,7 @@ typedef struct dir_connection_t {
   enum {
     DIR_SPOOL_NONE=0, DIR_SPOOL_SERVER_BY_DIGEST, DIR_SPOOL_SERVER_BY_FP,
     DIR_SPOOL_CACHED_DIR, DIR_SPOOL_NETWORKSTATUS
-  } dir_spool_src;
+  } dir_spool_src : 3;
   /** List of fingerprints for networkstatuses or desriptors to be spooled. */
   smartlist_t *fingerprint_stack;
   /** A cached_dir_t object that we're currently spooling out */
@@ -832,6 +843,16 @@ typedef struct dir_connection_t {
 
   char identity_digest[DIGEST_LEN]; /**< Hash of the public RSA key for
                                      * the directory server's signing key. */
+
+  /** If this is a dirserv conn created with a BEGIN_DIR (a "bridged" dirserv
+   * connection), a pointer to the edge_conn at the other end of its
+   * socketpair. */
+  edge_connection_t *bridge_conn;
+  /** Next connection in linked list of dirserv connections blocked until
+   * the or_conns over which they're bridged have enough space in their
+   * outbufs. */
+  struct dir_connection_t *next_blocked_on_same_or_conn;
+
 } dir_connection_t;
 
 /** Subtype of connection_t for an connection to a controller. */
@@ -2093,6 +2114,7 @@ char *alloc_http_authenticator(const char *authenticator);
 
 void assert_connection_ok(connection_t *conn, time_t now);
 int connection_or_nonopen_was_started_here(or_connection_t *conn);
+int connection_or_too_full_for_dirserv_data(or_connection_t *conn);
 
 /********************************* connection_edge.c *************************/
 
@@ -2179,6 +2201,7 @@ or_connection_t *connection_or_get_by_identity_digest(const char *digest);
 
 int connection_or_reached_eof(or_connection_t *conn);
 int connection_or_process_inbuf(or_connection_t *conn);
+int connection_or_flushed_some(or_connection_t *conn);
 int connection_or_finished_flushing(or_connection_t *conn);
 int connection_or_finished_connecting(or_connection_t *conn);
 
@@ -2333,6 +2356,9 @@ char *directory_dump_request_log(void);
 #define UNNAMED_ROUTER_NICKNAME "Unnamed"
 
 int connection_dirserv_flushed_some(dir_connection_t *conn);
+void connection_dirserv_unlink_from_bridge(dir_connection_t *dir_conn);
+void connection_dirserv_stop_blocking_all_on_or_conn(or_connection_t *or_conn);
+
 int dirserv_add_own_fingerprint(const char *nickname, crypto_pk_env_t *pk);
 int dirserv_load_fingerprint_file(void);
 void dirserv_free_fingerprint_list(void);