send the end cell when we realize we're going to end,

not when we're closing the stream.

this lets us put a payload in the end cell if we want to,
to describe why we're closing the stream.

there are still some places where we don't send the end cell
immediately. i need to track them down. but it's a low priority,
since i've made it send the end cell when we close the stream if
we haven't already sent it.


svn:r640

6 files changed, 125 insertions, 68 deletions

diff --git a/src/or/circuit.c b/src/or/circuit.c
index be05caafd..d3053eb27 100644
--- a/src/or/circuit.c
+++ b/src/or/circuit.c
@@ -512,52 +512,59 @@ void circuit_about_to_close_connection(connection_t *conn) {
   circuit_t *circ;
   connection_t *prevconn;
 
-  if(!connection_speaks_cells(conn)) {
-    /* it's an edge conn. need to remove it from the linked list of
-     * conn's for this circuit. Send an 'end' relay command.
-     * But don't kill the circuit.
-     */
-
-    circ = circuit_get_by_conn(conn);
-    if(!circ)
+  switch(conn->type) {
+    case CONN_TYPE_OR:
+      /* We must close all the circuits on it. */
+      while((circ = circuit_get_by_conn(conn))) {
+        if(circ->n_conn == conn) /* it's closing in front of us */
+          circ->n_conn = NULL;
+        if(circ->p_conn == conn) /* it's closing behind us */
+          circ->p_conn = NULL;
+        circuit_close(circ);
+      }  
       return;
+    case CONN_TYPE_AP:
+    case CONN_TYPE_EXIT:
 
-    if(conn == circ->p_streams) {
-      circ->p_streams = conn->next_stream;
-      goto send_end;
-    }
-    if(conn == circ->n_streams) {
-      circ->n_streams = conn->next_stream;
-      goto send_end;
-    }
-    for(prevconn = circ->p_streams; prevconn && prevconn->next_stream && prevconn->next_stream != conn; prevconn = prevconn->next_stream) ;
-    if(prevconn && prevconn->next_stream) {
-      prevconn->next_stream = conn->next_stream;
-      goto send_end;
-    }
-    for(prevconn = circ->n_streams; prevconn && prevconn->next_stream && prevconn->next_stream != conn; prevconn = prevconn->next_stream) ;
-    if(prevconn && prevconn->next_stream) {
-      prevconn->next_stream = conn->next_stream;
-      goto send_end;
-    }
-    log_fn(LOG_ERR,"edge conn not in circuit's list?");
-    assert(0); /* should never get here */
-send_end:
-    connection_edge_send_command(conn, circ, RELAY_COMMAND_END,
-                                 NULL, 0, conn->cpath_layer);
-    return;
-  }
+      /* It's an edge conn. Need to remove it from the linked list of
+       * conn's for this circuit. Confirm that 'end' relay command has
+       * been sent. But don't kill the circuit.
+       */
 
-  /* this connection speaks cells. We must close all the circuits on it. */
-  while((circ = circuit_get_by_conn(conn))) {
-    if(circ->n_conn == conn) /* it's closing in front of us */
-      circ->n_conn = NULL;
-    if(circ->p_conn == conn) /* it's closing behind us */
-      circ->p_conn = NULL;
-    circuit_close(circ);
-  }  
+      circ = circuit_get_by_conn(conn);
+      if(!circ)
+        return;
+
+      if(!conn->has_sent_end) {
+        log_fn(LOG_INFO,"Edge connection hasn't sent end yet? Bug.");
+        connection_edge_send_command(conn, circ, RELAY_COMMAND_END,
+                                     NULL, 0, conn->cpath_layer);
+      }
+
+      if(conn == circ->p_streams) {
+        circ->p_streams = conn->next_stream;
+        return;
+      }
+      if(conn == circ->n_streams) {
+        circ->n_streams = conn->next_stream;
+        return;
+      }
+      for(prevconn = circ->p_streams; prevconn && prevconn->next_stream && prevconn->next_stream != conn; prevconn = prevconn->next_stream) ;
+      if(prevconn && prevconn->next_stream) {
+        prevconn->next_stream = conn->next_stream;
+        return;
+      }
+      for(prevconn = circ->n_streams; prevconn && prevconn->next_stream && prevconn->next_stream != conn; prevconn = prevconn->next_stream) ;
+      if(prevconn && prevconn->next_stream) {
+        prevconn->next_stream = conn->next_stream;
+        return;
+      }
+      log_fn(LOG_ERR,"edge conn not in circuit's list?");
+      assert(0); /* should never get here */
+  } /* end switch */
 }
 
+
 /* FIXME this now leaves some out */
 void circuit_dump_by_conn(connection_t *conn, int severity) {
   circuit_t *circ;
@@ -903,7 +910,11 @@ int circuit_truncated(circuit_t *circ, crypt_path_t *layer) {
     for(stream = circ->p_streams; stream; stream=stream->next_stream) {
       if(stream->cpath_layer == victim) {
         log_fn(LOG_INFO, "Marking stream %d for close.", *(int*)stream->stream_id);
-/*ENDCLOSE*/    stream->marked_for_close = 1;
+        /* no need to send 'end' relay cells,
+         * because the other side's already dead
+         */
+        stream->marked_for_close = 1;
+        stream->has_sent_end = 1;
       }
     }
 
diff --git a/src/or/connection.c b/src/or/connection.c
index 2c8a0032a..a7bd89a71 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -353,9 +353,9 @@ int connection_read_to_buf(connection_t *conn) {
   } else {
     /* do a rudimentary round-robin so one connection can't hog a thickpipe */
     if(connection_speaks_cells(conn)) {
-      at_most = 10*(CELL_NETWORK_SIZE);
+      at_most = 30*(CELL_NETWORK_SIZE);
     } else {
-      at_most = 10*(CELL_PAYLOAD_SIZE - RELAY_HEADER_SIZE);
+      at_most = 30*(CELL_PAYLOAD_SIZE - RELAY_HEADER_SIZE);
     }
 
     if(at_most > global_read_bucket)
@@ -644,7 +644,8 @@ int connection_send_destroy(aci_t aci, connection_t *conn) {
 
   if(!connection_speaks_cells(conn)) {
      log_fn(LOG_INFO,"Aci %d: At an edge. Marking connection for close.", aci);
-/*ENDCLOSE*/ conn->marked_for_close = 1;
+     connection_edge_end(conn, NULL, 0, conn->cpath_layer);
+     /* if they already sent a destroy, they know. XXX can just close? */
      return 0;
   }
 
diff --git a/src/or/connection_edge.c b/src/or/connection_edge.c
index b79da2d08..394139932 100644
--- a/src/or/connection_edge.c
+++ b/src/or/connection_edge.c
@@ -26,27 +26,34 @@ int connection_edge_process_inbuf(connection_t *conn) {
     /* eof reached; we're done reading, but we might want to write more. */ 
     conn->done_receiving = 1;
     shutdown(conn->s, 0); /* XXX check return, refactor NM */
-    if (conn->done_sending)
-/*ENDCLOSE*/  conn->marked_for_close = 1;
-
-    /* XXX Factor out common logic here and in circuit_about_to_close NM */
-    connection_edge_send_command(conn, circuit_get_by_conn(conn), RELAY_COMMAND_END,
-                                 NULL, 0, conn->cpath_layer);
+    if (conn->done_sending) {
+      connection_edge_end(conn, NULL, 0, conn->cpath_layer);
+    } else {
+      connection_edge_send_command(conn, circuit_get_by_conn(conn), RELAY_COMMAND_END,
+                                   NULL, 0, conn->cpath_layer);
+    }
     return 0;
 #else 
     /* eof reached, kill it. */
     log_fn(LOG_INFO,"conn (fd %d) reached eof. Closing.", conn->s);
-/*ENDCLOSE*/ return -1;
+    connection_edge_end(conn, NULL, 0, conn->cpath_layer);
+    return -1;
 #endif
   }
 
   switch(conn->state) {
     case AP_CONN_STATE_SOCKS_WAIT:
-/*ENDCLOSE*/  return connection_ap_handshake_process_socks(conn);
+      if(connection_ap_handshake_process_socks(conn) < 0) {
+        connection_edge_end(conn, NULL, 0, conn->cpath_layer);
+        return -1;
+      }
+      return 0;
     case AP_CONN_STATE_OPEN:
     case EXIT_CONN_STATE_OPEN:
-      if(connection_edge_package_raw_inbuf(conn) < 0)
-/*ENDCLOSE*/  return -1;
+      if(connection_edge_package_raw_inbuf(conn) < 0) {
+        connection_edge_end(conn, NULL, 0, conn->cpath_layer);
+        return -1;
+      }
       return 0;
     case EXIT_CONN_STATE_CONNECTING:
       log_fn(LOG_INFO,"text from server while in 'connecting' state at exit. Leaving it on buffer.");
@@ -56,6 +63,25 @@ int connection_edge_process_inbuf(connection_t *conn) {
   return 0;
 }
 
+void connection_edge_end(connection_t *conn, void *payload, int payload_len,
+                         crypt_path_t *cpath_layer) {
+  circuit_t *circ = circuit_get_by_conn(conn);
+
+  if(conn->has_sent_end) {
+    log_fn(LOG_WARN,"It appears I've already sent the end. Are you calling me twice?");
+    return;
+  }
+
+  if(circ) {
+    log_fn(LOG_DEBUG,"Marking conn (fd %d) and sending end.",conn->s);
+    connection_edge_send_command(conn, circ, RELAY_COMMAND_END,
+                                 payload, payload_len, cpath_layer);
+  }
+
+  conn->marked_for_close = 1;
+  conn->has_sent_end = 1;
+}
+
 void connection_edge_send_command(connection_t *fromconn, circuit_t *circ, int relay_command,
                                   void *payload, int payload_len, crypt_path_t *cpath_layer) {
   cell_t cell;
@@ -128,6 +154,7 @@ int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, connection
       return 0;
     } else {
       log_fn(LOG_WARN,"Got an unexpected relay cell, not in 'open' state. Closing.");
+      connection_edge_end(conn, NULL, 0, conn->cpath_layer);
       return -1;
     }
   }
@@ -148,12 +175,15 @@ int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, connection
       if((edge_type == EDGE_AP && --layer_hint->deliver_window < 0) ||
          (edge_type == EDGE_EXIT && --circ->deliver_window < 0)) {
         log_fn(LOG_WARN,"(relay data) circ deliver_window below 0. Killing.");
+        connection_edge_end(conn, NULL, 0, conn->cpath_layer);
         return -1;
       }
       log_fn(LOG_DEBUG,"circ deliver_window now %d.", edge_type == EDGE_AP ? layer_hint->deliver_window : circ->deliver_window);
 
-      if(circuit_consider_sending_sendme(circ, edge_type, layer_hint) < 0)
+      if(circuit_consider_sending_sendme(circ, edge_type, layer_hint) < 0) {
+        conn->has_sent_end = 1; /* we failed because conn is broken. can't send end. */
         return -1;
+      }
 
       if(!conn) {
         log_fn(LOG_INFO,"relay cell dropped, unknown stream %d.",*(int*)conn->stream_id);
@@ -181,10 +211,14 @@ int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, connection
 #ifdef HALF_OPEN
       conn->done_sending = 1;
       shutdown(conn->s, 1); /* XXX check return; refactor NM */
-      if (conn->done_receiving)
-/*ENDCLOSE*/  conn->marked_for_close = 1;
+      if (conn->done_receiving) {
+        conn->marked_for_close = 1;
+        conn->has_sent_end = 1; /* no need to send end, we just got one! */
+      }
+#else
+      conn->marked_for_close = 1;
+      conn->has_sent_end = 1; /* no need to send end, we just got one! */
 #endif
-/*ENDCLOSE*/  conn->marked_for_close = 1;
       break;
     case RELAY_COMMAND_EXTEND:
       if(conn) {
@@ -233,7 +267,8 @@ int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, connection
       }
       log_fn(LOG_INFO,"Connected! Notifying application.");
       if(connection_ap_handshake_socks_reply(conn, NULL, 0, 1) < 0) {
-/*ENDCLOSE*/    conn->marked_for_close = 1;
+        log_fn(LOG_INFO,"Writing to socks-speaking application failed. Closing.");
+        connection_edge_end(conn, NULL, 0, conn->cpath_layer);
       }
       break;
     case RELAY_COMMAND_SENDME:
@@ -529,10 +564,14 @@ static int connection_ap_handshake_socks_reply(connection_t *conn, char *reply,
   return 0; /* if socks_version isn't 4 or 5, don't send anything */
 }
 
-/*ENDCLOSE*/ static int connection_exit_begin_conn(cell_t *cell, circuit_t *circ) {
+static int connection_exit_begin_conn(cell_t *cell, circuit_t *circ) {
   connection_t *n_stream;
   char *colon;
 
+  /* XXX currently we don't send an end cell back if we drop the
+   * begin because it's malformed.
+   */
+
   if(!memchr(cell->payload+RELAY_HEADER_SIZE+STREAM_ID_SIZE,0,
              cell->length-RELAY_HEADER_SIZE-STREAM_ID_SIZE)) {
     log_fn(LOG_WARN,"relay begin cell has no \\0. Dropping.");
@@ -578,6 +617,7 @@ static int connection_ap_handshake_socks_reply(connection_t *conn, char *reply,
       /* else fall through */
     case -1: /* resolve failed */
       log_fn(LOG_WARN,"Resolve or connect failed (%s).", n_stream->address);
+      connection_edge_end(n_stream, NULL, 0, NULL);
       connection_remove(n_stream);
       connection_free(n_stream);
     case 0: /* resolve added to pending list */
diff --git a/src/or/dns.c b/src/or/dns.c
index 7c0566c80..ac25a9dd5 100644
--- a/src/or/dns.c
+++ b/src/or/dns.c
@@ -220,7 +220,7 @@ void dns_cancel_pending_resolve(char *question, connection_t *onlyconn) {
     /* mark all pending connections to fail */
     while(resolve->pending_connections) {
       pend = resolve->pending_connections;
-/*ENDCLOSE*/  pend->conn->marked_for_close = 1;
+      connection_edge_end(pend->conn, NULL, 0, NULL);
       resolve->pending_connections = pend->next;
       free(pend);
     }
@@ -273,7 +273,7 @@ static void dns_found_answer(char *question, uint32_t answer) {
     pend = resolve->pending_connections;
     pend->conn->addr = resolve->answer;
     if(resolve->state == CACHE_STATE_FAILED || connection_exit_connect(pend->conn) < 0) {
-/*ENDCLOSE*/  pend->conn->marked_for_close = 1;
+      connection_edge_end(pend->conn, NULL, 0, NULL);
     }
     resolve->pending_connections = pend->next;
     free(pend);
diff --git a/src/or/or.h b/src/or/or.h
index 639a68ff3..45631e272 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -310,6 +310,8 @@ struct connection_t {
 
   int done_sending; /* for half-open connections; not used currently */
   int done_receiving;
+  char has_sent_end; /* for debugging: set once we've set the stream end,
+                        and check in circuit_about_to_close_connection() */
 };
 
 typedef struct connection_t connection_t;
@@ -581,6 +583,9 @@ void assert_connection_ok(connection_t *conn, time_t now);
 /********************************* connection_edge.c ***************************/
 
 int connection_edge_process_inbuf(connection_t *conn);
+void connection_edge_end(connection_t *conn, void *payload, int payload_len,
+                         crypt_path_t *cpath_layer);
+
 void connection_edge_send_command(connection_t *fromconn, circuit_t *circ, int relay_command,
                                   void *payload, int payload_len, crypt_path_t *cpath_layer);
 
diff --git a/src/or/test.c b/src/or/test.c
index 269022b5b..7d692c1f3 100644
--- a/src/or/test.c
+++ b/src/or/test.c
@@ -46,7 +46,7 @@ setup_directory() {
 
 void
 test_buffers() {
-#define MAX_BUF_SIZE 640*1024
+#define MAX_BUF_SIZE 1024*1024
   char str[256];
   char str2[256];
 
@@ -61,7 +61,7 @@ test_buffers() {
   if (!(buf = buf_new()))
     test_fail();
 
-  test_eq(buf_capacity(buf), 2*1024);
+  test_eq(buf_capacity(buf), 512*1024);
   test_eq(buf_datalen(buf), 0);
 
   /****
@@ -77,7 +77,7 @@ test_buffers() {
   s = open("/tmp/tor_test/data", O_RDONLY, 0);
   eof = 0;
   i = read_to_buf(s, 10, buf, &eof);
-  test_eq(buf_capacity(buf), 2*1024);
+  test_eq(buf_capacity(buf), 512*1024);
   test_eq(buf_datalen(buf), 10);
   test_eq(eof, 0);
   test_eq(i, 10);
@@ -85,7 +85,7 @@ test_buffers() {
 
   /* Test reading 0 bytes. */
   i = read_to_buf(s, 0, buf, &eof);
-  test_eq(buf_capacity(buf), MAX_BUF_SIZE);
+  test_eq(buf_capacity(buf), 512*1024);
   test_eq(buf_datalen(buf), 10);
   test_eq(eof, 0);
   test_eq(i, 0);
@@ -103,7 +103,7 @@ test_buffers() {
   /* Now test when buffer is filled with more data to read. */
   buf2 = buf_new_with_capacity(32);
   i = read_to_buf(s, 128, buf2, &eof);
-  test_eq(buf_capacity(buf2), 32);
+  test_eq(buf_capacity(buf2), 128);
   test_eq(buf_datalen(buf2), 32);
   test_eq(eof, 0);
   test_eq(i, 32);