Make a function to mark a connection and set hold_open_until_flushed.

5 files changed, 33 insertions, 12 deletions

diff --git a/src/or/connection.c b/src/or/connection.c
index 6665809bc..3788348cd 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -3566,8 +3566,8 @@ assert_connection_ok(connection_t *conn, time_t now)
   if (conn->bufev) {
     tor_assert(conn->read_event == NULL);
     tor_assert(conn->write_event == NULL);
-    tor_assert(conn->inbuf == NULL);
-    tor_assert(conn->outbuf == NULL);
+    /* XXX reinstate tor_assert(conn->inbuf == NULL);
+       tor_assert(conn->outbuf == NULL);*/
   }
 #endif
 
diff --git a/src/or/connection.h b/src/or/connection.h
index 13a283122..906f9ba87 100644
--- a/src/or/connection.h
+++ b/src/or/connection.h
@@ -34,6 +34,15 @@ void _connection_mark_for_close(connection_t *conn,int line, const char *file);
 #define connection_mark_for_close(c) \
   _connection_mark_for_close((c), __LINE__, _SHORT_FILE_)
 
+#define connection_mark_and_flush(c)                                    \
+  do {                                                                  \
+    connection_t *tmp_conn_ = (c);                                      \
+    _connection_mark_for_close(tmp_conn_, __LINE__, _SHORT_FILE_);      \
+    tmp_conn_->hold_open_until_flushed = 1;                             \
+    IF_HAS_BUFFEREVENT(tmp_conn_,                                       \
+                       connection_start_writing(tmp_conn_));            \
+  } while (0)
+
 void connection_expire_held_open(void);
 
 int connection_connect(connection_t *conn, const char *address,
diff --git a/src/or/connection_edge.c b/src/or/connection_edge.c
index edc58a0cf..f90c44f58 100644
--- a/src/or/connection_edge.c
+++ b/src/or/connection_edge.c
@@ -92,6 +92,7 @@ _connection_mark_unattached_ap(edge_connection_t *conn, int endreason,
 
   _connection_mark_for_close(TO_CONN(conn), line, file);
   conn->_base.hold_open_until_flushed = 1;
+  IF_HAS_BUFFEREVENT(TO_CONN(conn), connection_start_writing(TO_CONN(conn)));
   conn->end_reason = endreason;
 }
 
@@ -191,8 +192,7 @@ connection_edge_destroy(circid_t circ_id, edge_connection_t *conn)
       conn->edge_has_sent_end = 1;
       conn->end_reason = END_STREAM_REASON_DESTROY;
       conn->end_reason |= END_STREAM_REASON_FLAG_ALREADY_SENT_CLOSED;
-      connection_mark_for_close(TO_CONN(conn));
-      conn->_base.hold_open_until_flushed = 1;
+      connection_mark_and_flush(TO_CONN(conn));
     }
   }
   conn->cpath_layer = NULL;
@@ -327,6 +327,7 @@ connection_edge_finished_flushing(edge_connection_t *conn)
     case AP_CONN_STATE_CIRCUIT_WAIT:
     case AP_CONN_STATE_CONNECT_WAIT:
     case AP_CONN_STATE_CONTROLLER_WAIT:
+    case AP_CONN_STATE_RESOLVE_WAIT:
       return 0;
     default:
       log_warn(LD_BUG, "Called in unexpected state %d.",conn->_base.state);
diff --git a/src/or/main.c b/src/or/main.c
index b8867852c..85c99fffc 100644
--- a/src/or/main.c
+++ b/src/or/main.c
@@ -172,6 +172,17 @@ connection_add(connection_t *conn)
   conn->conn_array_index = smartlist_len(connection_array);
   smartlist_add(connection_array, conn);
 
+#ifdef USE_BUFFEREVENTS
+  if (conn->type == CONN_TYPE_AP && conn->s >= 0 && !conn->linked) {
+    conn->bufev = bufferevent_socket_new(
+                         tor_libevent_get_base(),
+                         conn->s,
+                         BEV_OPT_DEFER_CALLBACKS);
+
+    connection_configure_bufferevent_callbacks(conn);
+  }
+#endif
+
   if (!HAS_BUFFEREVENT(conn) && (conn->s >= 0 || conn->linked)) {
     conn->read_event = tor_event_new(tor_libevent_get_base(),
          conn->s, EV_READ|EV_PERSIST, conn_read_callback, conn);
@@ -323,10 +334,12 @@ connection_watch_events(connection_t *conn, watchable_events_t events)
   IF_HAS_BUFFEREVENT(conn, {
       short ev = ((short)events) & (EV_READ|EV_WRITE);
       short old_ev = bufferevent_get_enabled(conn->bufev);
-      if ((ev & ~old_ev) != 0)
+      if ((ev & ~old_ev) != 0) {
         bufferevent_enable(conn->bufev, ev);
-      if ((old_ev & ~ev) != 0)
+      }
+      if ((old_ev & ~ev) != 0) {
         bufferevent_disable(conn->bufev, old_ev & ~ev);
+      }
       return;
   });
   if (events & READ_EVENT)
@@ -449,13 +462,14 @@ void
 connection_start_writing(connection_t *conn)
 {
   tor_assert(conn);
-  tor_assert(conn->write_event);
 
   IF_HAS_BUFFEREVENT(conn, {
       bufferevent_enable(conn->bufev, EV_WRITE);
       return;
   });
 
+  tor_assert(conn->write_event);
+
   if (conn->linked) {
     conn->writing_to_linked_conn = 1;
     if (conn->linked_conn &&
@@ -863,8 +877,7 @@ run_connection_housekeeping(int i, time_t now)
     log_info(LD_OR,"Expiring non-used OR connection to fd %d (%s:%d) "
              "[Hibernating or exiting].",
              conn->s,conn->address, conn->port);
-    connection_mark_for_close(conn);
-    conn->hold_open_until_flushed = 1;
+    connection_mark_and_flush(conn);
   } else if (!or_conn->n_circuits &&
              now >= or_conn->timestamp_last_added_nonpadding +
                                          IDLE_OR_CONN_TIMEOUT) {
@@ -872,7 +885,6 @@ run_connection_housekeeping(int i, time_t now)
              "[idle %d].", conn->s,conn->address, conn->port,
              (int)(now - or_conn->timestamp_last_added_nonpadding));
     connection_mark_for_close(conn);
-    conn->hold_open_until_flushed = 1;
   } else if (
       now >= or_conn->timestamp_lastempty + options->KeepalivePeriod*10 &&
       now >= conn->timestamp_lastwritten + options->KeepalivePeriod*10) {
diff --git a/src/or/relay.c b/src/or/relay.c
index 16048fa8c..f9a44cf16 100644
--- a/src/or/relay.c
+++ b/src/or/relay.c
@@ -1157,8 +1157,7 @@ connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ,
       if (!conn->_base.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(TO_CONN(conn));
-        conn->_base.hold_open_until_flushed = 1;
+        connection_mark_and_flush(TO_CONN(conn));
       }
       return 0;
     case RELAY_COMMAND_EXTEND: