Push responsibility for connection marking down as far as possible; have only a close path; add some missing end cells; change return conventions a little.

svn:r1149

8 files changed, 66 insertions, 38 deletions

diff --git a/src/or/connection.c b/src/or/connection.c
index 7387afed3a..b9b5546981 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -262,6 +262,7 @@ static int connection_handle_listener_read(connection_t *conn, int new_type) {
     }
     /* else there was a real error. */
     log_fn(LOG_WARN,"accept() failed. Closing listener.");
+    connection_mark_for_close(conn,0);
     return -1;
   }
   log(LOG_INFO,"Connection accepted on socket %d (child of fd %d).",news, conn->s);
@@ -412,6 +413,8 @@ int connection_handle_read(connection_t *conn) {
        /* XXX I suspect pollerr may make Windows not get to this point. :( */
        router_mark_as_down(conn->nickname);
     }
+    /* There's a read error; kill the connection.*/
+    connection_mark_for_close(conn, END_STREAM_REASON_MISC);
     return -1;
   }
   if(connection_process_inbuf(conn) < 0) {
@@ -516,8 +519,9 @@ int connection_handle_write(connection_t *conn) {
 
   conn->timestamp_lastwritten = time(NULL);
 
-  if(connection_speaks_cells(conn) && conn->state != OR_CONN_STATE_CONNECTING) {
-    if(conn->state == OR_CONN_STATE_HANDSHAKING) {
+  if (connection_speaks_cells(conn) &&
+      conn->state != OR_CONN_STATE_CONNECTING) {
+    if (conn->state == OR_CONN_STATE_HANDSHAKING) {
       connection_stop_writing(conn);
       return connection_tls_continue_handshake(conn);
     }
@@ -527,6 +531,7 @@ int connection_handle_write(connection_t *conn) {
       case TOR_TLS_ERROR:
       case TOR_TLS_CLOSE:
         log_fn(LOG_INFO,"tls error. breaking.");
+        connection_mark_for_close(conn, 0);
         return -1; /* XXX deal with close better */
       case TOR_TLS_WANTWRITE:
         log_fn(LOG_DEBUG,"wanted write.");
@@ -550,9 +555,10 @@ int connection_handle_write(connection_t *conn) {
        */
     }
   } else {
-    if(flush_buf(conn->s, conn->outbuf, &conn->outbuf_flushlen) < 0)
+    if (flush_buf(conn->s, conn->outbuf, &conn->outbuf_flushlen) < 0) {
+      connection_mark_for_close(conn, END_STREAM_REASON_MISC);
       return -1;
-    /* conns in CONNECTING state will fall through... */
+    }
   }
 
   if(!connection_wants_to_flush(conn)) /* it's done flushing */
diff --git a/src/or/connection_edge.c b/src/or/connection_edge.c
index 9ab1b7c11c..3a8f81b1db 100644
--- a/src/or/connection_edge.c
+++ b/src/or/connection_edge.c
@@ -305,10 +305,10 @@ int connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, connection
       conn->done_sending = 1;
       shutdown(conn->s, 1); /* XXX check return; refactor NM */
       if (conn->done_receiving) {
-        connection_mark_for_close(conn, 0);
+        connection_mark_for_close(conn, END_STREAM_REASON_DONE);
       }
 #else
-      connection_mark_for_close(conn, 0);
+      connection_mark_for_close(conn, END_STREAM_REASON_DONE);
 #endif
       return 0;
     case RELAY_COMMAND_EXTEND:
@@ -395,6 +395,7 @@ int connection_edge_finished_flushing(connection_t *conn) {
         if(!ERRNO_CONN_EINPROGRESS(errno)) {
           /* yuck. kill it. */
           log_fn(LOG_DEBUG,"in-progress exit connect failed. Removing.");
+          connection_mark_for_close(conn, END_STREAM_REASON_CONNECTFAILED);
           return -1;
         } else {
           log_fn(LOG_DEBUG,"in-progress exit connect still waiting.");
diff --git a/src/or/connection_or.c b/src/or/connection_or.c
index 07283d8e54..9b9bfe7bfb 100644
--- a/src/or/connection_or.c
+++ b/src/or/connection_or.c
@@ -7,7 +7,7 @@
 extern or_options_t options; /* command-line and config-file options */
 
 static int connection_tls_finish_handshake(connection_t *conn);
-static int connection_or_process_cell_from_inbuf(connection_t *conn);
+static int connection_or_process_cells_from_inbuf(connection_t *conn);
 
 /**************************************************************/
 
@@ -30,25 +30,28 @@ int connection_or_process_inbuf(connection_t *conn) {
   assert(conn && conn->type == CONN_TYPE_OR);
 
   if(conn->inbuf_reached_eof) {
-    log_fn(LOG_INFO,"conn reached eof. Closing.");
+    log_fn(LOG_INFO,"OR connection reached EOF. Closing.");
+    connection_mark_for_close(conn,0);
     return -1;
   }
 
   if(conn->state != OR_CONN_STATE_OPEN)
     return 0; /* don't do anything */
-  return connection_or_process_cell_from_inbuf(conn);
+  return connection_or_process_cells_from_inbuf(conn);
 }
 
 int connection_or_finished_flushing(connection_t *conn) {
   int e, len=sizeof(e);
 
   assert(conn && conn->type == CONN_TYPE_OR);
+  assert_connection_ok(conn,0);
 
   switch(conn->state) {
     case OR_CONN_STATE_CONNECTING:
       if (getsockopt(conn->s, SOL_SOCKET, SO_ERROR, (void*)&e, &len) < 0)  { /* not yet */
         if(!ERRNO_CONN_EINPROGRESS(errno)){
           log_fn(LOG_DEBUG,"in-progress connect failed. Removing.");
+          connection_mark_for_close(conn,0);
           return -1;
         } else {
           return 0; /* no change, see if next time is better */
@@ -59,14 +62,17 @@ int connection_or_finished_flushing(connection_t *conn) {
       log_fn(LOG_INFO,"OR connect() to router %s:%u finished.",
           conn->address,conn->port);
 
-      if(connection_tls_start_handshake(conn, 0) < 0)
+      if(connection_tls_start_handshake(conn, 0) < 0) {
+        /* TLS handhaking error of some kind. */
+        connection_mark_for_close(conn,0);
         return -1;
+      }
       return 0;
     case OR_CONN_STATE_OPEN:
       connection_stop_writing(conn);
       return 0;
     default:
-      log_fn(LOG_WARN,"BUG: called in unexpected state.");
+      log_fn(LOG_WARN,"BUG: called in unexpected state %d",conn->state);
       return 0;
   }
 }
@@ -254,7 +260,7 @@ void connection_or_write_cell_to_buf(const cell_t *cell, connection_t *conn) {
 }
 
 /* if there's a whole cell there, pull it off and process it. */
-static int connection_or_process_cell_from_inbuf(connection_t *conn) {
+static int connection_or_process_cells_from_inbuf(connection_t *conn) {
   char buf[CELL_NETWORK_SIZE];
   cell_t cell;
 
diff --git a/src/or/cpuworker.c b/src/or/cpuworker.c
index 0ff097f131..98dd388ba2 100644
--- a/src/or/cpuworker.c
+++ b/src/or/cpuworker.c
@@ -68,7 +68,8 @@ int connection_cpu_process_inbuf(connection_t *conn) {
     }
     num_cpuworkers--;
     spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up spinning. */
-    return -1;
+    connection_mark_for_close(conn,0);
+    return 0;
   }
 
   if(conn->state == CPUWORKER_STATE_BUSY_ONION) {
diff --git a/src/or/directory.c b/src/or/directory.c
index 2598520bd0..f7a07a81ae 100644
--- a/src/or/directory.c
+++ b/src/or/directory.c
@@ -107,9 +107,11 @@ int connection_dir_process_inbuf(connection_t *conn) {
                                    NULL, 0, &directory, MAX_DIR_SIZE)) {
           case -1: /* overflow */
             log_fn(LOG_WARN,"'fetch' response too large. Failing.");
+            connection_mark_for_close(conn,0);
             return -1;
           case 0:
             log_fn(LOG_INFO,"'fetch' response not all here, but we're at eof. Closing.");
+            connection_mark_for_close(conn,0);
             return -1;
           /* case 1, fall through */
         }
@@ -119,6 +121,7 @@ int connection_dir_process_inbuf(connection_t *conn) {
         if(directorylen == 0) {
           log_fn(LOG_INFO,"Empty directory. Ignoring.");
           free(directory);
+          connection_mark_for_close(conn,0);
           return -1;
         }
         if(router_set_routerlist_from_directory(directory, conn->identity_pkey) < 0){
@@ -130,13 +133,16 @@ int connection_dir_process_inbuf(connection_t *conn) {
           router_retry_connections();
         }
         free(directory);
-        return -1;
+        connection_mark_for_close(conn,0);
+        return 0;
       case DIR_CONN_STATE_CLIENT_READING_UPLOAD:
         /* XXX make sure there's a 200 OK on the buffer */
         log_fn(LOG_INFO,"eof while reading upload response. Finished.");
-        return -1;
+        connection_mark_for_close(conn,0);
+        return 0;
       default:
         log_fn(LOG_INFO,"conn reached eof, not reading. Closing.");
+        connection_mark_for_close(conn,0);
         return -1;
     }
   }
@@ -236,6 +242,7 @@ int connection_dir_finished_flushing(connection_t *conn) {
         if(!ERRNO_CONN_EINPROGRESS(errno)) {
           log_fn(LOG_DEBUG,"in-progress connect failed. Removing.");
           router_mark_as_down(conn->nickname); /* don't try him again */
+          connection_mark_for_close(conn,0);
           return -1;
         } else {
           return 0; /* no change, see if next time is better */
@@ -259,7 +266,8 @@ int connection_dir_finished_flushing(connection_t *conn) {
       return 0;
     case DIR_CONN_STATE_SERVER_WRITING:
       log_fn(LOG_INFO,"Finished writing server response. Closing.");
-      return -1; /* kill it */
+      connection_mark_for_close(conn,0);
+      return 0;
     default:
       log_fn(LOG_WARN,"BUG: called in unexpected state %d.", conn->state);
       return -1;
diff --git a/src/or/dns.c b/src/or/dns.c
index ce3e212c65..ad9c1fba12 100644
--- a/src/or/dns.c
+++ b/src/or/dns.c
@@ -331,7 +331,8 @@ int connection_dns_process_inbuf(connection_t *conn) {
       num_dnsworkers_busy--;
     }
     num_dnsworkers--;
-    return -1;
+    connection_mark_for_close(conn,0);
+    return 0;
   }
 
   assert(conn->state == DNSWORKER_STATE_BUSY);
diff --git a/src/or/main.c b/src/or/main.c
index c53d83d141..d7a10cf5f7 100644
--- a/src/or/main.c
+++ b/src/or/main.c
@@ -165,6 +165,8 @@ static void conn_read(int i) {
        !connection_has_pending_tls_data(conn))
       return; /* this conn should not read */
 
+  if (conn->marked_for_close)
+    return;
   log_fn(LOG_DEBUG,"socket %d wants to read.",conn->s);
 
   assert_connection_ok(conn, time(NULL));
@@ -174,18 +176,15 @@ static void conn_read(int i) {
 #ifdef MS_WINDOWS
       (poll_array[i].revents & POLLERR) ||
 #endif
-    connection_handle_read(conn) < 0)
+      (connection_handle_read(conn) < 0 && !conn->marked_for_close))
     {
       /* this connection is broken. remove it */
-      log_fn(LOG_INFO,"%s connection broken, removing.",
-             conn_type_to_string[conn->type]);
-      connection_remove(conn);
-      connection_free(conn);
-      if(i<nfds) {
-        /* we just replaced the one at i with a new one. process it too. */
-        conn_read(i);
-      }
-    } else assert_connection_ok(conn, time(NULL));
+      /* XXX This shouldn't ever happen anymore. */
+      log_fn(LOG_ERR,"Unhandled error on read for %s connection (fd %d); removing",
+             conn_type_to_string[conn->type], conn->s);
+      connection_mark_for_close(conn,0);
+    }
+    assert_connection_ok(conn, time(NULL));
 }
 
 static void conn_write(int i) {
@@ -195,18 +194,19 @@ static void conn_write(int i) {
     return; /* this conn doesn't want to write */
 
   conn = connection_array[i];
+  if (conn->marked_for_close)
+    return;
   log_fn(LOG_DEBUG,"socket %d wants to write.",conn->s);
 
   assert_connection_ok(conn, time(NULL));
 
-  if(connection_handle_write(conn) < 0) { /* this connection is broken. remove it. */
-    log_fn(LOG_INFO,"%s connection broken, removing.", conn_type_to_string[conn->type]);
-    connection_remove(conn);
-    connection_free(conn);
-    if(i<nfds) { /* we just replaced the one at i with a new one. process it too. */
-        conn_write(i);
-    }
-  } else assert_connection_ok(conn, time(NULL));
+  if(connection_handle_write(conn) < 0 && !conn->marked_for_close) {
+    /* this connection is broken. remove it. */
+    log_fn(LOG_ERR,"Unhandled error on read for %s connection (fd %d); removing",
+           conn_type_to_string[conn->type], conn->s);
+    connection_mark_for_close(conn,0);
+  }
+  assert_connection_ok(conn, time(NULL));
 }
 
 static void conn_close_if_marked(int i) {
@@ -555,7 +555,7 @@ static int do_main_loop(void) {
 
     /* do all the reads and errors first, so we can detect closed sockets */
     for(i=0;i<nfds;i++)
-      conn_read(i); /* this also blows away broken connections */
+      conn_read(i); /* this also marks broken connections */
 
     /* then do the writes */
     for(i=0;i<nfds;i++)
diff --git a/src/or/or.h b/src/or/or.h
index 6a4aa5e5bb..11834cd78c 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -314,6 +314,8 @@ struct connection_t {
   int marked_for_close; /* should we close this conn on the next
                          * iteration of the main loop?
                          */
+  char *marked_for_close_file; /* for debugging: in which file were we marked
+                                * for close? */
 
   buf_t *inbuf;
   int inbuf_reached_eof; /* did read() return 0 on this conn? */
@@ -641,8 +643,11 @@ int _connection_mark_for_close(connection_t *conn, char reason);
 #define connection_mark_for_close(c,r)                                  \
   do {                                                                  \
     if (_connection_mark_for_close(c,r)<0) {                            \
-      log(LOG_WARN,"Duplicate call to connection_mark_for_close at %s:%d", \
-          __FILE__,__LINE__);                                           \
+      log(LOG_WARN,"Duplicate call to connection_mark_for_close at %s:%d (first at %s:%d)", \
+          __FILE__,__LINE__,c->marked_for_close_file,c->marked_for_close); \
+    } else {                                                            \
+      c->marked_for_close_file = __FILE__;                              \
+      c->marked_for_close = __LINE__;                                   \
     }                                                                   \
   } while (0)