Merge branch 'tor-gitlab/mr/721'

5 files changed, 148 insertions, 32 deletions

diff --git a/src/core/or/conflux.c b/src/core/or/conflux.c
index 5f6af9268b..c979077e1f 100644
--- a/src/core/or/conflux.c
+++ b/src/core/or/conflux.c
@@ -21,6 +21,7 @@
 #include "core/or/conflux.h"
 #include "core/or/conflux_params.h"
 #include "core/or/conflux_util.h"
+#include "core/or/conflux_pool.h"
 #include "core/or/conflux_st.h"
 #include "core/or/conflux_cell.h"
 #include "lib/time/compat_time.h"
@@ -243,7 +244,9 @@ conflux_decide_circ_minrtt(const conflux_t *cfx)
   tor_assert(CONFLUX_NUM_LEGS(cfx));
 
   CONFLUX_FOR_EACH_LEG_BEGIN(cfx, leg) {
-    if (leg->circ_rtts_usec < min_rtt) {
+
+    /* Ignore circuits with no RTT measurement */
+    if (leg->circ_rtts_usec && leg->circ_rtts_usec < min_rtt) {
       circ = leg->circ;
       min_rtt = leg->circ_rtts_usec;
     }
@@ -278,7 +281,8 @@ conflux_decide_circ_lowrtt(const conflux_t *cfx)
       continue;
     }
 
-    if (leg->circ_rtts_usec < low_rtt) {
+    /* Ignore circuits with no RTT */
+    if (leg->circ_rtts_usec && leg->circ_rtts_usec < low_rtt) {
       low_rtt = leg->circ_rtts_usec;
       circ = leg->circ;
     }
@@ -398,7 +402,8 @@ conflux_decide_circ_cwndrtt(const conflux_t *cfx)
 
   /* Find the leg with the minimum RTT.*/
   CONFLUX_FOR_EACH_LEG_BEGIN(cfx, l) {
-    if (l->circ_rtts_usec < min_rtt) {
+    /* Ignore circuits with invalid RTT */
+    if (l->circ_rtts_usec && l->circ_rtts_usec < min_rtt) {
       min_rtt = l->circ_rtts_usec;
       leg = l;
     }
@@ -419,7 +424,8 @@ conflux_decide_circ_cwndrtt(const conflux_t *cfx)
     /* Pick a 'min_leg' with the lowest RTT that still has
      * room in the congestion window. Note that this works for
      * min_leg itself, up to inflight. */
-    if (cwnd_sendable(l->circ, min_rtt, l->circ_rtts_usec) > 0) {
+    if (l->circ_rtts_usec &&
+        cwnd_sendable(l->circ, min_rtt, l->circ_rtts_usec) > 0) {
       leg = l;
     }
   } CONFLUX_FOR_EACH_LEG_END(l);
@@ -548,11 +554,16 @@ conflux_pick_first_leg(conflux_t *cfx)
     }
   } CONFLUX_FOR_EACH_LEG_END(leg);
 
-  if (BUG(!min_leg)) {
+  if (!min_leg) {
     // Get the 0th leg; if it does not exist, assert
     tor_assert(smartlist_len(cfx->legs) > 0);
     min_leg = smartlist_get(cfx->legs, 0);
     tor_assert(min_leg);
+    if (BUG(min_leg->linked_sent_usec == 0)) {
+      log_warn(LD_BUG, "Conflux has no legs with non-zero RTT. "
+               "Using first leg.");
+      conflux_log_set(cfx, CIRCUIT_IS_ORIGIN(min_leg->circ));
+    }
   }
 
   // TODO-329-TUNING: We may want to initialize this to a cwnd, to
diff --git a/src/core/or/conflux_pool.c b/src/core/or/conflux_pool.c
index c84613503f..c8018f45e2 100644
--- a/src/core/or/conflux_pool.c
+++ b/src/core/or/conflux_pool.c
@@ -797,11 +797,6 @@ try_finalize_set(unlinked_circuits_t *unlinked)
   unlinked->cfx = NULL;
   unlinked_free(unlinked);
 
-  /* Now that this set is ready to use, try any pending streams again. */
-  if (is_client) {
-    connection_ap_attach_pending(1);
-  }
-
   log_info(LD_CIRC,
            "Successfully linked a conflux %s set which is now usable.",
            is_client ? "client" : "relay");
@@ -822,10 +817,29 @@ record_rtt_client(const circuit_t *circ)
   tor_assert(CIRCUIT_IS_ORIGIN(circ));
 
   leg_t *leg = unlinked_leg_find(circ, true);
-  if (leg && leg->link_sent_usec > 0) {
-    leg->rtt_usec = monotime_absolute_usec() - leg->link_sent_usec;
-    return leg->rtt_usec;
+
+  if (BUG(!leg || leg->link_sent_usec == 0)) {
+    log_warn(LD_BUG,
+             "Conflux: Trying to record client RTT without a timestamp");
+    goto err;
+  }
+
+  uint64_t now = monotime_absolute_usec();
+  tor_assert_nonfatal(now >= leg->link_sent_usec);
+  leg->rtt_usec = now - leg->link_sent_usec;
+  if (leg->rtt_usec == 0) {
+    log_warn(LD_CIRC, "Clock appears stalled for conflux.");
+    // TODO-329-TUNING: For now, let's accept this case. We need to do
+    // tuning and clean up the tests such that they use RTT in order to
+    // fail here.
+    //goto err;
   }
+  return leg->rtt_usec;
+
+ err:
+  // Avoid using this leg until a timestamp comes in
+  if (leg)
+    leg->rtt_usec = UINT64_MAX;
   return UINT64_MAX;
 }
 
@@ -842,10 +856,26 @@ record_rtt_exit(const circuit_t *circ)
   tor_assert(CIRCUIT_IS_ORCIRC(circ));
 
   conflux_leg_t *leg = conflux_get_leg(circ->conflux, circ);
-  if (leg && leg->linked_sent_usec > 0) {
-    leg->circ_rtts_usec = monotime_absolute_usec() - leg->linked_sent_usec;
-    return leg->circ_rtts_usec;
+
+  if (BUG(!leg || leg->linked_sent_usec == 0)) {
+    log_warn(LD_BUG,
+             "Conflux: Trying to record exit RTT without a timestamp");
+    goto err;
   }
+
+  uint64_t now = monotime_absolute_usec();
+  tor_assert_nonfatal(now >= leg->linked_sent_usec);
+  leg->circ_rtts_usec = now - leg->linked_sent_usec;
+
+  if (leg->circ_rtts_usec == 0) {
+    log_warn(LD_CIRC, "Clock appears stalled for conflux.");
+    goto err;
+  }
+  return leg->circ_rtts_usec;
+
+ err:
+  if (leg)
+    leg->circ_rtts_usec = UINT64_MAX;
   return UINT64_MAX;
 }
 
@@ -863,6 +893,9 @@ record_rtt(const circuit_t *circ, bool is_client)
   if (is_client) {
     rtt_usec = record_rtt_client(circ);
 
+    if (rtt_usec == UINT64_MAX)
+      return false;
+
     if (rtt_usec >= get_circuit_build_timeout_ms()*1000) {
       log_info(LD_CIRC, "Conflux leg RTT is above circuit build time out "
                         "currently at %f msec. Relaunching.",
@@ -1926,6 +1959,12 @@ conflux_process_linked(circuit_t *circ, crypt_path_t *layer_hint,
     goto end;
   }
 
+  /* If this set is ready to use with a valid conflux set, try any pending
+   * streams again. */
+  if (circ->conflux) {
+    connection_ap_attach_pending(1);
+  }
+
   goto end;
 
  close:
@@ -2020,6 +2059,51 @@ conflux_pool_init(void)
   }
 }
 
+/**
+ * Return a description of all linked and unlinked circuits associated
+ * with a conflux set.
+ *
+ * For use in rare bug cases that are hard to diagnose.
+ */
+void
+conflux_log_set(const conflux_t *cfx, bool is_client)
+{
+  tor_assert(cfx);
+
+  log_warn(LD_BUG, "Conflux %s: %d linked, %d launched",
+               fmt_nonce(cfx->nonce), smartlist_len(cfx->legs),
+               cfx->num_leg_launch);
+
+  // Log all linked legs
+  int legs = 0;
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, leg) {
+   log_warn(LD_BUG,
+            " - Linked Leg %d purpose=%d; RTT %"PRIu64", sent: %"PRIu64
+            " marked: %d",
+            legs, leg->circ->purpose, leg->circ_rtts_usec,
+            leg->linked_sent_usec, leg->circ->marked_for_close);
+   legs++;
+  } CONFLUX_FOR_EACH_LEG_END(leg);
+
+  // Look up the nonce to see if we have any unlinked circuits.
+  unlinked_circuits_t *unlinked = unlinked_pool_get(cfx->nonce, is_client);
+  if (unlinked) {
+    // Log the number of legs and the is_for_linked_set status
+    log_warn(LD_BUG, " - Unlinked set:  %d legs, for link: %d",
+             smartlist_len(unlinked->legs), unlinked->is_for_linked_set);
+    legs = 0;
+    SMARTLIST_FOREACH_BEGIN(unlinked->legs, leg_t *, leg) {
+      log_warn(LD_BUG,
+        "     Unlinked Leg: %d purpose=%d; linked: %d, RTT %"PRIu64", "
+        "sent: %"PRIu64" link ptr %p, marked: %d",
+               legs, leg->circ->purpose, leg->linked,
+               leg->rtt_usec, leg->link_sent_usec,
+               leg->link, leg->circ->marked_for_close);
+      legs++;
+    } SMARTLIST_FOREACH_END(leg);
+  }
+}
+
 /** Free and clean up the conflux pool subsystem. This is called by the subsys
  * manager AFTER all circuits have been freed which implies that all objects in
  * the pools aren't referenced anymore. */
diff --git a/src/core/or/conflux_pool.h b/src/core/or/conflux_pool.h
index 6bb858bb09..9a9701a484 100644
--- a/src/core/or/conflux_pool.h
+++ b/src/core/or/conflux_pool.h
@@ -36,6 +36,9 @@ void conflux_process_linked(circuit_t *circ, crypt_path_t *layer_hint,
                             const cell_t *cell, const uint16_t cell_len);
 void conflux_process_linked_ack(circuit_t *circ);
 
+typedef struct conflux_t conflux_t;
+void conflux_log_set(const conflux_t *cfx, bool is_client);
+
 #ifdef TOR_UNIT_TESTS
 bool launch_new_set(int num_legs);
 digest256map_t *get_linked_pool(bool is_client);
diff --git a/src/core/or/conflux_util.c b/src/core/or/conflux_util.c
index 855d477428..589db41e83 100644
--- a/src/core/or/conflux_util.c
+++ b/src/core/or/conflux_util.c
@@ -20,6 +20,7 @@
 #include "core/or/conflux.h"
 #include "core/or/conflux_params.h"
 #include "core/or/conflux_util.h"
+#include "core/or/conflux_pool.h"
 #include "core/or/conflux_st.h"
 #include "lib/time/compat_time.h"
 #include "app/config/config.h"
@@ -372,22 +373,39 @@ void
 conflux_validate_legs(const conflux_t *cfx)
 {
   tor_assert(cfx);
-  // TODO-329-UDP: Eventually we want to allow three legs for the
-  // exit case, to allow reconnection of legs to hit an RTT target.
-  // For now, this validation helps find bugs.
-  if (BUG(smartlist_len(cfx->legs) > conflux_params_get_num_legs_set())) {
-    log_warn(LD_BUG, "Number of legs is above maximum of %d allowed: %d\n",
-             conflux_params_get_num_legs_set(), smartlist_len(cfx->legs));
-  }
-
+  bool is_client = false;
+  int num_legs = 0;
   CONFLUX_FOR_EACH_LEG_BEGIN(cfx, leg) {
-    /* Ensure we have no pending nonce on the circ */
-    tor_assert_nonfatal(leg->circ->conflux_pending_nonce == NULL);
-    tor_assert_nonfatal(leg->circ->conflux != NULL);
-
     if (CIRCUIT_IS_ORIGIN(leg->circ)) {
       tor_assert_nonfatal(leg->circ->purpose ==
                           CIRCUIT_PURPOSE_CONFLUX_LINKED);
+      is_client = true;
+    }
+
+    /* Ensure we have no pending nonce on the circ */
+    if (BUG(leg->circ->conflux_pending_nonce != NULL)) {
+      conflux_log_set(cfx, is_client);
+      continue;
+    }
+
+    /* Ensure we have a conflux object */
+    if (BUG(leg->circ->conflux == NULL)) {
+      conflux_log_set(cfx, is_client);
+      continue;
+    }
+
+    /* Only count legs that have a valid RTT */
+    if (leg->circ_rtts_usec > 0) {
+      num_legs++;
     }
   } CONFLUX_FOR_EACH_LEG_END(leg);
+
+  // TODO-329-UDP: Eventually we want to allow three legs for the
+  // exit case, to allow reconnection of legs to hit an RTT target.
+  // For now, this validation helps find bugs.
+  if (BUG(num_legs > conflux_params_get_num_legs_set())) {
+    log_warn(LD_BUG, "Number of legs is above maximum of %d allowed: %d\n",
+             conflux_params_get_num_legs_set(), smartlist_len(cfx->legs));
+    conflux_log_set(cfx, is_client);
+  }
 }
diff --git a/src/core/or/relay.c b/src/core/or/relay.c
index 20336dffaf..2c722f01cc 100644
--- a/src/core/or/relay.c
+++ b/src/core/or/relay.c
@@ -2333,7 +2333,7 @@ connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial,
 
   tor_assert(conn);
 
-  if (conn->base_.marked_for_close) {
+  if (BUG(conn->base_.marked_for_close)) {
     log_warn(LD_BUG,
              "called on conn that's already marked for close at %s:%d.",
              conn->base_.marked_for_close_file, conn->base_.marked_for_close);
@@ -3081,9 +3081,9 @@ set_block_state_for_streams(circuit_t *circ, edge_connection_t *stream_list,
     if (stream_id && edge->stream_id != stream_id)
       continue;
 
-    if (!conn->read_event || edge->xoff_received) {
-      /* This connection is a placeholder for something; probably a DNS
-       * request.  It can't actually stop or start reading.*/
+    if (!conn->read_event || edge->xoff_received ||
+        conn->marked_for_close) {
+      /* This connection should not start or stop reading. */
       continue;
     }