break reached_eof() out of process_inbuf()

svn:r2930

8 files changed, 125 insertions, 94 deletions

diff --git a/src/or/connection.c b/src/or/connection.c
index e9d39bc509..daf51a0c5f 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -87,6 +87,7 @@ static int connection_handle_listener_read(connection_t *conn, int new_type);
 static int connection_receiver_bucket_should_increase(connection_t *conn);
 static int connection_finished_flushing(connection_t *conn);
 static int connection_finished_connecting(connection_t *conn);
+static int connection_reached_eof(connection_t *conn);
 static int connection_read_to_buf(connection_t *conn, int *max_to_read);
 static int connection_process_inbuf(connection_t *conn, int package_partial);
 static int connection_bucket_read_limit(connection_t *conn);
@@ -856,6 +857,9 @@ loop_again:
   if (connection_process_inbuf(conn, 1) < 0) {
     return -1;
   }
+  if (conn->inbuf_reached_eof && connection_reached_eof(conn) < 0) {
+    return -1;
+  }
   return 0;
 }
 
@@ -1294,7 +1298,7 @@ static int connection_process_inbuf(connection_t *conn, int package_partial) {
     case CONN_TYPE_CONTROL:
       return connection_control_process_inbuf(conn);
     default:
-      log_fn(LOG_WARN,"got unexpected conn->type %d.", conn->type);
+      log_fn(LOG_WARN,"got unexpected conn type %d.", conn->type);
       return -1;
   }
 }
@@ -1326,7 +1330,7 @@ static int connection_finished_flushing(connection_t *conn) {
     case CONN_TYPE_CONTROL:
       return connection_control_finished_flushing(conn);
     default:
-      log_fn(LOG_WARN,"got unexpected conn->type %d.", conn->type);
+      log_fn(LOG_WARN,"got unexpected conn type %d.", conn->type);
       return -1;
   }
 }
@@ -1349,7 +1353,29 @@ static int connection_finished_connecting(connection_t *conn)
     case CONN_TYPE_DIR:
       return connection_dir_finished_connecting(conn);
     default:
-      tor_assert(0);
+      log_fn(LOG_WARN,"got unexpected conn type %d.", conn->type);
+      return -1;
+  }
+}
+
+static int connection_reached_eof(connection_t *conn)
+{
+  switch (conn->type) {
+    case CONN_TYPE_OR:
+      return connection_or_reached_eof(conn);
+    case CONN_TYPE_AP:
+    case CONN_TYPE_EXIT:
+      return connection_edge_reached_eof(conn);
+    case CONN_TYPE_DIR:
+      return connection_dir_reached_eof(conn);
+    case CONN_TYPE_DNSWORKER:
+      return connection_dns_reached_eof(conn);
+    case CONN_TYPE_CPUWORKER:
+      return connection_cpu_reached_eof(conn);
+    case CONN_TYPE_CONTROL:
+      return connection_control_reached_eof(conn);
+    default:
+      log_fn(LOG_WARN,"got unexpected conn type %d.", conn->type);
       return -1;
   }
 }
diff --git a/src/or/connection_edge.c b/src/or/connection_edge.c
index ecd2b0d0ee..0bd8247d7a 100644
--- a/src/or/connection_edge.c
+++ b/src/or/connection_edge.c
@@ -18,12 +18,37 @@ static smartlist_t *redirect_exit_list = NULL;
 
 static int connection_ap_handshake_process_socks(connection_t *conn);
 
-/** Handle new bytes on conn->inbuf, or notification of eof.
- *
- * If there was an EOF, then send an end and mark the connection
- * for close.
- *
- * Otherwise handle it based on state:
+/** There was an EOF. Send an end and mark the connection for close.
+ */
+int connection_edge_reached_eof(connection_t *conn) {
+#ifdef HALF_OPEN
+  /* 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) {
+    connection_edge_end(conn, END_STREAM_REASON_DONE, conn->cpath_layer);
+    connection_mark_for_close(conn);
+  } 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 (stream size %d). Closing.", conn->s, (int)conn->stream_size);
+  connection_edge_end(conn, END_STREAM_REASON_DONE, conn->cpath_layer);
+  if(!conn->marked_for_close) {
+    /* only mark it if not already marked. it's possible to
+     * get the 'end' right around when the client hangs up on us. */
+    connection_mark_for_close(conn);
+  }
+  conn->hold_open_until_flushed = 1; /* just because we shouldn't read
+                                        doesn't mean we shouldn't write */
+  return 0;
+#endif
+}
+
+/** Handle new bytes on conn->inbuf based on state:
  *   - If it's waiting for socks info, try to read another step of the
  *     socks handshake out of conn->inbuf.
  *   - If it's open, then package more relay cells from the stream.
@@ -37,34 +62,6 @@ int connection_edge_process_inbuf(connection_t *conn, int package_partial) {
   tor_assert(conn);
   tor_assert(conn->type == CONN_TYPE_AP || conn->type == CONN_TYPE_EXIT);
 
-  if(conn->inbuf_reached_eof) {
-#ifdef HALF_OPEN
-    /* 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) {
-      connection_edge_end(conn, END_STREAM_REASON_DONE, conn->cpath_layer);
-      connection_mark_for_close(conn);
-    } 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 (stream size %d). Closing.", conn->s, (int)conn->stream_size);
-    connection_edge_end(conn, END_STREAM_REASON_DONE, conn->cpath_layer);
-    if(!conn->marked_for_close) {
-      /* only mark it if not already marked. it's possible to
-       * get the 'end' right around when the client hangs up on us. */
-      connection_mark_for_close(conn);
-    }
-    conn->hold_open_until_flushed = 1; /* just because we shouldn't read
-                                          doesn't mean we shouldn't write */
-    return 0;
-#endif
-  }
-
   switch(conn->state) {
     case AP_CONN_STATE_SOCKS_WAIT:
       if(connection_ap_handshake_process_socks(conn) < 0) {
diff --git a/src/or/connection_or.c b/src/or/connection_or.c
index 273867e054..73b635d593 100644
--- a/src/or/connection_or.c
+++ b/src/or/connection_or.c
@@ -40,6 +40,12 @@ static void cell_unpack(cell_t *dest, const char *src) {
   memcpy(dest->payload, src+3, CELL_PAYLOAD_SIZE);
 }
 
+int connection_or_reached_eof(connection_t *conn) {
+  log_fn(LOG_INFO,"OR connection reached EOF. Closing.");
+  connection_mark_for_close(conn);
+  return 0;
+}
+
 /** Handle any new bytes that have come in on connection <b>conn</b>.
  * If conn is in 'open' state, hand it to
  * connection_or_process_cells_from_inbuf()
@@ -50,12 +56,6 @@ int connection_or_process_inbuf(connection_t *conn) {
   tor_assert(conn);
   tor_assert(conn->type == CONN_TYPE_OR);
 
-  if(conn->inbuf_reached_eof) {
-    log_fn(LOG_INFO,"OR connection reached EOF. Closing.");
-    connection_mark_for_close(conn);
-    return 0;
-  }
-
   if(conn->state != OR_CONN_STATE_OPEN)
     return 0; /* don't do anything */
   return connection_or_process_cells_from_inbuf(conn);
diff --git a/src/or/control.c b/src/or/control.c
index a25e0806d0..92297f3501 100644
--- a/src/or/control.c
+++ b/src/or/control.c
@@ -391,8 +391,15 @@ connection_control_finished_flushing(connection_t *conn) {
   return 0;
 }
 
-/** Called when <b>conn</b> has received more bytes on its inbuf, or has
- * gotten its socket closed. */
+/** Called when <b>conn</b> has gotten its socket closed. */
+int connection_control_reached_eof(connection_t *conn) {
+  log_fn(LOG_INFO,"Control connection reached EOF. Closing.");
+  connection_mark_for_close(conn);
+  return 0;
+}
+
+/** Called when <b>conn</b> has received more bytes on its inbuf.
+ */
 int
 connection_control_process_inbuf(connection_t *conn) {
   uint16_t body_len, command_type;
@@ -401,12 +408,6 @@ connection_control_process_inbuf(connection_t *conn) {
   tor_assert(conn);
   tor_assert(conn->type == CONN_TYPE_CONTROL);
 
-  if(conn->inbuf_reached_eof) {
-    log_fn(LOG_INFO,"Control connection reached EOF. Closing.");
-    connection_mark_for_close(conn);
-    return 0;
-  }
-
  again:
   /* Try to suck a control message from the buffer. */
   switch(fetch_from_buf_control(conn->inbuf, &body_len, &command_type, &body))
diff --git a/src/or/cpuworker.c b/src/or/cpuworker.c
index 998b4b3791..b72343a811 100644
--- a/src/or/cpuworker.c
+++ b/src/or/cpuworker.c
@@ -91,9 +91,25 @@ void cpuworkers_rotate(void)
   spawn_enough_cpuworkers();
 }
 
+/** If the cpuworker closes the connection,
+ * mark it as closed and spawn a new one as needed. */
+int connection_cpu_reached_eof(connection_t *conn) {
+  log_fn(LOG_WARN,"Read eof. Worker died unexpectedly.");
+  if(conn->state != CPUWORKER_STATE_IDLE) {
+    /* the circ associated with this cpuworker will have to wait until
+     * it gets culled in run_connection_housekeeping(), since we have
+     * no way to find out which circ it was. */
+    log_fn(LOG_WARN,"...and it left a circuit queued; abandoning circ.");
+    num_cpuworkers_busy--;
+  }
+  num_cpuworkers--;
+  spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up spinning. */
+  connection_mark_for_close(conn);
+  return 0;
+}
+
 /** Called when we get data from a cpuworker.  If the answer is not complete,
- * wait for a complete answer.  If the cpuworker closes the connection,
- * mark it as closed and spawn a new one as needed.  If the answer is complete,
+ * wait for a complete answer. If the answer is complete,
  * process it as appropriate.
  */
 int connection_cpu_process_inbuf(connection_t *conn) {
@@ -108,21 +124,6 @@ int connection_cpu_process_inbuf(connection_t *conn) {
   tor_assert(conn);
   tor_assert(conn->type == CONN_TYPE_CPUWORKER);
 
-  if(conn->inbuf_reached_eof) {
-    log_fn(LOG_WARN,"Read eof. Worker died unexpectedly.");
-    if(conn->state != CPUWORKER_STATE_IDLE) {
-      /* the circ associated with this cpuworker will have to wait until
-       * it gets culled in run_connection_housekeeping(), since we have
-       * no way to find out which circ it was. */
-      log_fn(LOG_WARN,"...and it left a circuit queued; abandoning circ.");
-      num_cpuworkers_busy--;
-    }
-    num_cpuworkers--;
-    spawn_enough_cpuworkers(); /* try to regrow. hope we don't end up spinning. */
-    connection_mark_for_close(conn);
-    return 0;
-  }
-
   if(conn->state == CPUWORKER_STATE_BUSY_ONION) {
     if(buf_datalen(conn->inbuf) < LEN_ONION_RESPONSE) /* entire answer available? */
       return 0; /* not yet */
diff --git a/src/or/directory.c b/src/or/directory.c
index ad31676e7d..6273b904c1 100644
--- a/src/or/directory.c
+++ b/src/or/directory.c
@@ -683,11 +683,24 @@ connection_dir_client_reached_eof(connection_t *conn)
   return 0;
 }
 
+int connection_dir_reached_eof(connection_t *conn) {
+  int retval;
+  if(conn->state != DIR_CONN_STATE_CLIENT_READING) {
+    log_fn(LOG_INFO,"conn reached eof, not reading. Closing.");
+    connection_close_immediate(conn); /* it was an error; give up on flushing */
+    connection_mark_for_close(conn);
+    return -1;
+  }
+
+  retval = connection_dir_client_reached_eof(conn);
+  connection_mark_for_close(conn);
+  return retval;
+}
+
 /** Read handler for directory connections.  (That's connections <em>to</em>
  * directory servers and connections <em>at</em> directory servers.)
  */
 int connection_dir_process_inbuf(connection_t *conn) {
-  int retval;
 
   tor_assert(conn);
   tor_assert(conn->type == CONN_TYPE_DIR);
@@ -697,18 +710,6 @@ int connection_dir_process_inbuf(connection_t *conn) {
    * write their response (when it's finished flushing, they mark for
    * close).
    */
-  if(conn->inbuf_reached_eof) {
-    if(conn->state != DIR_CONN_STATE_CLIENT_READING) {
-      log_fn(LOG_INFO,"conn reached eof, not reading. Closing.");
-      connection_close_immediate(conn); /* it was an error; give up on flushing */
-      connection_mark_for_close(conn);
-      return -1;
-    }
-
-    retval = connection_dir_client_reached_eof(conn);
-    connection_mark_for_close(conn);
-    return retval;
-  } /* endif 'reached eof' */
 
   /* If we're on the dirserver side, look for a command. */
   if(conn->state == DIR_CONN_STATE_SERVER_COMMAND_WAIT) {
diff --git a/src/or/dns.c b/src/or/dns.c
index 8e6dfca5e1..7593d2cf00 100644
--- a/src/or/dns.c
+++ b/src/or/dns.c
@@ -559,8 +559,18 @@ int connection_dns_finished_flushing(connection_t *conn) {
   return 0;
 }
 
-/** Read handler: called when we get data from a dnsworker.  If the
- * connection is closed, mark the dnsworker as dead.  Otherwise, see
+int connection_dns_reached_eof(connection_t *conn) {
+  log_fn(LOG_WARN,"Read eof. Worker died unexpectedly.");
+  if(conn->state == DNSWORKER_STATE_BUSY) {
+    dns_cancel_pending_resolve(conn->address);
+    num_dnsworkers_busy--;
+  }
+  num_dnsworkers--;
+  connection_mark_for_close(conn);
+  return 0;
+}
+
+/** Read handler: called when we get data from a dnsworker. See
  * if we have a complete answer.  If so, call dns_found_answer on the
  * result.  If not, wait.  Returns 0. */
 int connection_dns_process_inbuf(connection_t *conn) {
@@ -570,17 +580,6 @@ int connection_dns_process_inbuf(connection_t *conn) {
   tor_assert(conn);
   tor_assert(conn->type == CONN_TYPE_DNSWORKER);
 
-  if(conn->inbuf_reached_eof) {
-    log_fn(LOG_WARN,"Read eof. Worker died unexpectedly.");
-    if(conn->state == DNSWORKER_STATE_BUSY) {
-      dns_cancel_pending_resolve(conn->address);
-      num_dnsworkers_busy--;
-    }
-    num_dnsworkers--;
-    connection_mark_for_close(conn);
-    return 0;
-  }
-
   if(conn->state != DNSWORKER_STATE_BUSY) {
     log_fn(LOG_WARN,"Bug: poll() indicated than an idle dns worker was readable. Please report.");
     return 0;
diff --git a/src/or/or.h b/src/or/or.h
index fdcd3200a2..b230cb1bf0 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -1196,6 +1196,7 @@ int connection_or_nonopen_was_started_here(connection_t *conn);
 
 /********************************* connection_edge.c ***************************/
 
+int connection_edge_reached_eof(connection_t *conn);
 int connection_edge_process_inbuf(connection_t *conn, int package_partial);
 int connection_edge_destroy(uint16_t circ_id, connection_t *conn);
 int connection_edge_end(connection_t *conn, char reason, crypt_path_t *cpath_layer);
@@ -1233,6 +1234,7 @@ void parse_socks_policy(void);
 
 /********************************* connection_or.c ***************************/
 
+int connection_or_reached_eof(connection_t *conn);
 int connection_or_process_inbuf(connection_t *conn);
 int connection_or_finished_flushing(connection_t *conn);
 int connection_or_finished_connecting(connection_t *conn);
@@ -1272,6 +1274,7 @@ typedef enum or_conn_status_event_t {
 } or_conn_status_event_t;
 
 int connection_control_finished_flushing(connection_t *conn);
+int connection_control_reached_eof(connection_t *conn);
 int connection_control_process_inbuf(connection_t *conn);
 
 int control_event_circuit_status(circuit_t *circ, circuit_status_event_t e);
@@ -1287,6 +1290,7 @@ int init_cookie_authentication(int enabled);
 void cpu_init(void);
 void cpuworkers_rotate(void);
 int connection_cpu_finished_flushing(connection_t *conn);
+int connection_cpu_reached_eof(connection_t *conn);
 int connection_cpu_process_inbuf(connection_t *conn);
 int assign_to_cpuworker(connection_t *cpuworker, unsigned char question_type,
                         void *task);
@@ -1297,6 +1301,7 @@ int dir_policy_permits_address(uint32_t addr);
 void directory_post_to_dirservers(uint8_t purpose, const char *payload,
                                   size_t payload_len);
 void directory_get_from_dirserver(uint8_t purpose, const char *resource);
+int connection_dir_reached_eof(connection_t *conn);
 int connection_dir_process_inbuf(connection_t *conn);
 int connection_dir_finished_flushing(connection_t *conn);
 int connection_dir_finished_connecting(connection_t *conn);
@@ -1325,6 +1330,7 @@ void dirserv_set_cached_directory(const char *directory, time_t when,
 
 void dns_init(void);
 int connection_dns_finished_flushing(connection_t *conn);
+int connection_dns_reached_eof(connection_t *conn);
 int connection_dns_process_inbuf(connection_t *conn);
 void dnsworkers_rotate(void);
 void connection_dns_remove(connection_t *conn);