add an ap_conn connecting state; report connecting delay

svn:r1011

5 files changed, 87 insertions, 28 deletions

diff --git a/src/or/circuit.c b/src/or/circuit.c
index 071c94726..212d0560b 100644
--- a/src/or/circuit.c
+++ b/src/or/circuit.c
@@ -736,8 +736,28 @@ void circuit_about_to_close_connection(connection_t *conn) {
   } /* end switch */
 }
 
-void circuit_dump_details(int severity, circuit_t *circ, int poll_index,
-                          char *type, int this_circid, int other_circid) {
+void circuit_log_path(int severity, circuit_t *circ) {
+  static char b[1024];
+  struct crypt_path_t *hop;
+  routerinfo_t *router;
+  assert(circ->cpath);
+  strcpy(b,"Stream is on circ: ");
+  for(hop=circ->cpath;hop->next != circ->cpath; hop=hop->next) {
+    router = router_get_by_addr_port(hop->addr,hop->port);
+    if(router) {
+      /* XXX strcat causes buffer overflow */
+      strcat(b,router->nickname);
+      strcat(b,",");
+    } else {
+      strcat(b,"UNKNOWN,");
+    }
+  }
+  log_fn(severity,"%s",b);
+}
+
+static void
+circuit_dump_details(int severity, circuit_t *circ, int poll_index,
+                     char *type, int this_circid, int other_circid) {
   struct crypt_path_t *hop;
   log(severity,"Conn %d has %s circuit: circID %d (other side %d), state %d (%s), born %d",
       poll_index, type, this_circid, other_circid, circ->state,
diff --git a/src/or/connection.c b/src/or/connection.c
index b700b784f..e2993fab3 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -45,7 +45,8 @@ char *conn_state_to_string[][_CONN_TYPE_MAX+1] = {
     "", /* 3 */
     "awaiting dest info",         /* app, 4 */
     "waiting for safe circuit",        /* 5 */
-    "open" },                          /* 6 */
+    "waiting for connected",           /* 6 */
+    "open" },                          /* 7 */
   { "ready" }, /* dir listener, 0 */
   { "",                           /* dir, 0 */
     "connecting (fetch)",              /* 1 */
diff --git a/src/or/connection_edge.c b/src/or/connection_edge.c
index 0f2736597..4755f6fd4 100644
--- a/src/or/connection_edge.c
+++ b/src/or/connection_edge.c
@@ -87,8 +87,8 @@ int connection_edge_process_inbuf(connection_t *conn) {
       log_fn(LOG_INFO,"text from server while in 'connecting' state at exit. Leaving it on buffer.");
       return 0;
   }
-
-  return 0;
+  log_fn(LOG_WARN,"Got unexpected state %d. Closing.",conn->state);
+  return -1;
 }
 
 static char *connection_edge_end_reason(char *payload, uint16_t length) {
@@ -213,6 +213,27 @@ int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, connection
       conn->marked_for_close = 1;
       conn->has_sent_end = 1;
       return 0;
+    }
+    if(conn->type == CONN_TYPE_AP && rh.command == RELAY_COMMAND_CONNECTED) {
+      if(conn->state != AP_CONN_STATE_CONNECTING) {
+        log_fn(LOG_WARN,"Got 'connected' while not in state connecting. Dropping.");
+        return 0;
+      }
+      log_fn(LOG_INFO,"Connected! Notifying application.");
+      conn->state = AP_CONN_STATE_OPEN;
+      if (rh.length >= 4) {
+        addr = ntohl(*(uint32_t*)(cell->payload + RELAY_HEADER_SIZE));
+        client_dns_set_entry(conn->socks_request->address, addr);
+      }
+      log_fn(LOG_WARN,"'connected' received after %d seconds.",
+             (int)(time(NULL) - conn->timestamp_lastread));
+      circuit_log_path(LOG_WARN,circ);
+      if(connection_ap_handshake_socks_reply(conn, NULL, 0, 1) < 0) {
+        log_fn(LOG_INFO,"Writing to socks-speaking application failed. Closing.");
+        if(connection_edge_end(conn, END_STREAM_REASON_MISC, conn->cpath_layer) < 0)
+          log_fn(LOG_WARN,"3: I called connection_edge_end redundantly.");
+      }
+      return 0;
     } else {
       log_fn(LOG_WARN,"Got an unexpected relay command %d, in state %d (%s). Closing.",
              rh.command, conn->state, conn_state_to_string[conn->type][conn->state]);
@@ -351,25 +372,8 @@ int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, connection
       circuit_truncated(circ, layer_hint);
       return 0;
     case RELAY_COMMAND_CONNECTED:
-      if(edge_type == EDGE_EXIT) {
-        log_fn(LOG_WARN,"'connected' unsupported at exit. Dropping.");
-        return 0;
-      }
-      if(!conn) {
-        log_fn(LOG_INFO,"connected cell dropped, unknown stream.");
-        return 0;
-      }
-      log_fn(LOG_INFO,"Connected! Notifying application.");
-      if (rh.length >= 4) {
-        addr = ntohl(*(uint32_t*)(cell->payload + RELAY_HEADER_SIZE));
-        client_dns_set_entry(conn->socks_request->address, addr);
-      }
-      if(connection_ap_handshake_socks_reply(conn, NULL, 0, 1) < 0) {
-        log_fn(LOG_INFO,"Writing to socks-speaking application failed. Closing.");
-        if(connection_edge_end(conn, END_STREAM_REASON_MISC, conn->cpath_layer) < 0)
-          log_fn(LOG_WARN,"3: I called connection_edge_end redundantly.");
-      }
-      return 0;
+      log_fn(LOG_WARN,"'connected' unsupported while open. Closing.");
+      return -1;
     case RELAY_COMMAND_SENDME:
       if(!conn) {
         if(edge_type == EDGE_AP) {
@@ -521,6 +525,28 @@ repeat_connection_edge_package_raw_inbuf:
   goto repeat_connection_edge_package_raw_inbuf;
 }
 
+void connection_ap_expire_beginning(void) {
+  connection_t **carray;
+  connection_t *conn;
+  int n, i;
+  time_t now = time(NULL);
+
+  get_connection_array(&carray, &n);
+
+  for (i = 0; i < n; ++i) {
+    conn = carray[i];
+    if (conn->type != CONN_TYPE_AP ||
+        conn->state != AP_CONN_STATE_CONNECTING)
+      continue;
+    if (now - conn->timestamp_lastread > 30) {
+      log_fn(LOG_WARN,"Stream is %d seconds late. Closing.",
+             (int)(now - conn->timestamp_lastread));
+      /* XXX here is where it should move back into 'pending' state */
+      conn->marked_for_close = 1;
+    }
+  }
+}
+
 /* Tell any APs that are waiting for a new circuit that one is available */
 void connection_ap_attach_pending(void)
 {
@@ -649,6 +675,9 @@ static int connection_ap_handshake_attach_circuit(connection_t *conn) {
 
   connection_start_reading(conn);
 
+  /* here, print the circ's path. so people can figure out which circs are sucking. */
+  circuit_log_path(LOG_WARN,circ);
+
   if(!circ->timestamp_dirty)
     circ->timestamp_dirty = time(NULL);
 
@@ -725,12 +754,11 @@ static void connection_ap_handshake_send_begin(connection_t *ap_conn, circuit_t
 
   ap_conn->package_window = STREAMWINDOW_START;
   ap_conn->deliver_window = STREAMWINDOW_START;
-  ap_conn->state = AP_CONN_STATE_OPEN;
+  ap_conn->state = AP_CONN_STATE_CONNECTING;
   /* XXX Right now, we rely on the socks client not to send us any data
    * XXX until we've sent back a socks reply.  (If it does, we could wind
    * XXX up packaging that data and sending it to the exit, then later having
    * XXX the exit refuse us.)
-   * XXX Perhaps we should grow an AP_CONN_STATE_CONNECTING state.
    */
   log_fn(LOG_INFO,"Address/port sent, ap socket %d, n_circ_id %d",ap_conn->s,circ->n_circ_id);
   return;
diff --git a/src/or/main.c b/src/or/main.c
index a03d85081..e8e5e7bfe 100644
--- a/src/or/main.c
+++ b/src/or/main.c
@@ -313,6 +313,13 @@ static void run_scheduled_events(time_t now) {
    */
   circuit_expire_building();
 
+  /* 2b. Also look at pending streams and prune the ones that 'began'
+   *     a long time ago but haven't gotten a 'connected' yet.
+   *     Do this before step 3, so we can put them back into pending
+   *     state to be picked up by the new circuit.
+   */
+  connection_ap_expire_beginning();
+
   /* 3. Every second, we try a new circuit if there are no valid
    *    circuits. Every NewCircuitPeriod seconds, we expire circuits
    *    that became dirty more than NewCircuitPeriod seconds ago,
diff --git a/src/or/or.h b/src/or/or.h
index 53a938e0c..1e127d517 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -164,8 +164,9 @@
 #define _AP_CONN_STATE_MIN 4
 #define AP_CONN_STATE_SOCKS_WAIT 4
 #define AP_CONN_STATE_CIRCUIT_WAIT 5
-#define AP_CONN_STATE_OPEN 6
-#define _AP_CONN_STATE_MAX 6
+#define AP_CONN_STATE_CONNECTING 6
+#define AP_CONN_STATE_OPEN 7
+#define _AP_CONN_STATE_MAX 7
 
 #define _DIR_CONN_STATE_MIN 1
 #define DIR_CONN_STATE_CONNECTING_FETCH 1
@@ -589,6 +590,7 @@ void circuit_consider_sending_sendme(circuit_t *circ, int edge_type, crypt_path_
 void circuit_close(circuit_t *circ);
 void circuit_about_to_close_connection(connection_t *conn);
 
+void circuit_log_path(int severity, circuit_t *circ);
 void circuit_dump_by_conn(connection_t *conn, int severity);
 
 void circuit_expire_unused_circuits(void);
@@ -687,6 +689,7 @@ int connection_edge_package_raw_inbuf(connection_t *conn);
 
 void connection_exit_connect(connection_t *conn);
 int connection_ap_can_use_exit(connection_t *conn, routerinfo_t *exit);
+void connection_ap_expire_beginning(void);
 void connection_ap_attach_pending(void);
 
 extern uint64_t stats_n_data_cells_packaged;