Prop#329 Algs: Conflux multiplexed cell sending decision algs

src/core/or/circuit_st.h

@@ -248,6 +248,19 @@ struct circuit_t {
 
   /** Congestion control fields */
   struct congestion_control_t *ccontrol;
+
+  /** Conflux linked circuit information.
+   *
+   * If this is non-NULL, the circuit is linked and part of a usable set,
+   * and for origin_circuit_t subtypes, the circuit purpose is
+   * CIRCUIT_PURPOSE_CONFLUX_LINKED.
+   *
+   * If this is NULL, the circuit could still be part of a pending conflux
+   * object, in which case the conflux_pending_nonce field is set, and for
+   * origin_circuit_t subtypes, the purpose is
+   * CIRCUIT_PURPOSE_CONFLUX_UNLINKED.
+  */
+  struct conflux_t *conflux;
 };
 
 #endif /* !defined(CIRCUIT_ST_H) */

src/core/or/conflux.c

@@ -12,6 +12,7 @@
 
 #include "core/or/circuit_st.h"
 #include "core/or/sendme.h"
+#include "core/or/relay.h"
 #include "core/or/congestion_control_common.h"
 #include "core/or/congestion_control_st.h"
 #include "core/or/origin_circuit_st.h"
@@ -21,14 +22,16 @@
 #include "core/or/conflux_params.h"
 #include "core/or/conflux_util.h"
 #include "core/or/conflux_st.h"
+#include "core/or/conflux_cell.h"
 #include "lib/time/compat_time.h"
 #include "app/config/config.h"
 
-#include "trunnel/extension.h"
-
 /** One million microseconds in a second */
 #define USEC_PER_SEC 1000000
 
+static inline uint64_t cwnd_sendable(const circuit_t *on_circ,
+                                     uint64_t in_usec, uint64_t our_usec);
+
 /**
  * Determine if we should multiplex a specific relay command or not.
  *
@@ -119,6 +122,612 @@ conflux_get_leg(conflux_t *cfx, const circuit_t *circ)
   return leg_found;
 }
 
+/**
+ * Gets the maximum last_seq_sent from all legs.
+ */
+uint64_t
+conflux_get_max_seq_sent(const conflux_t *cfx)
+{
+  uint64_t max_seq_sent = 0;
+
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, leg) {
+    if (leg->last_seq_sent > max_seq_sent) {
+      max_seq_sent = leg->last_seq_sent;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(leg);
+
+  return max_seq_sent;
+}
+
+/**
+ * Gets the maximum last_seq_recv from all legs.
+ */
+uint64_t
+conflux_get_max_seq_recv(const conflux_t *cfx)
+{
+  uint64_t max_seq_recv = 0;
+
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, leg) {
+    if (leg->last_seq_recv > max_seq_recv) {
+      max_seq_recv = leg->last_seq_recv;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(leg);
+
+  return max_seq_recv;
+}
+
+/**
+ * Returns true if a circuit has package window space to send, and is
+ * not blocked locally.
+ */
+static inline bool
+circuit_ready_to_send(const circuit_t *circ)
+{
+  const congestion_control_t *cc = circuit_ccontrol(circ);
+  bool cc_sendable = true;
+
+  /* We consider ourselves blocked if we're within 1 sendme of the
+   * cwnd, because inflight is decremented before this check */
+  // TODO-329-TUNING: This subtraction not be right.. It depends
+  // on call order wrt decisions and sendme arrival
+  if (cc->inflight + cc->sendme_inc >= cc->cwnd) {
+    cc_sendable = false;
+  }
+
+  /* Origin circuits use the package window of the last hop, and
+   * have an outbound cell direction (towards exit). Otherwise,
+   * there is no cpath and direction is inbound. */
+  if (CIRCUIT_IS_ORIGIN(circ)) {
+    return cc_sendable && !circ->circuit_blocked_on_n_chan;
+  } else {
+    return cc_sendable && !circ->circuit_blocked_on_p_chan;
+  }
+}
+
+/**
+ * Return the circuit with the minimum RTT. Do not use any
+ * other circuit.
+ *
+ * This algorithm will minimize RTT always, and will not provide
+ * any throughput benefit. We expect it to be useful for VoIP/UDP
+ * use cases. Because it only uses one circuit on a leg at a time,
+ * it can have more than one circuit per guard (ie: to find
+ * lower-latency middles for the path).
+ */
+static const circuit_t *
+conflux_decide_circ_minrtt(const conflux_t *cfx)
+{
+  uint64_t min_rtt = UINT64_MAX;
+  const circuit_t *circ = NULL;
+
+  /* Can't get here without any legs. */
+  tor_assert(CONFLUX_NUM_LEGS(cfx));
+
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, leg) {
+    if (leg->circ_rtts_usec < min_rtt) {
+      circ = leg->circ;
+      min_rtt = leg->circ_rtts_usec;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(leg);
+
+  /* If the minRTT circuit can't send, dont send on any circuit. */
+  if (!circ || !circuit_ready_to_send(circ)) {
+    return NULL;
+  }
+  return circ;
+}
+
+/**
+ * Favor the circuit with the lowest RTT that still has space in the
+ * congestion window.
+ *
+ * This algorithm will maximize total throughput at the expense of
+ * bloating out-of-order queues.
+ */
+static const circuit_t *
+conflux_decide_circ_lowrtt(const conflux_t *cfx)
+{
+  uint64_t low_rtt = UINT64_MAX;
+  const circuit_t *circ = NULL;
+
+  /* Can't get here without any legs. */
+  tor_assert(CONFLUX_NUM_LEGS(cfx));
+
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, leg) {
+    /* If the package window is full, skip it */
+    if (!circuit_ready_to_send(leg->circ)) {
+      continue;
+    }
+
+    if (leg->circ_rtts_usec < low_rtt) {
+      low_rtt = leg->circ_rtts_usec;
+      circ = leg->circ;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(leg);
+
+  /* At this point, if we found a circuit, we've already validated that its
+   * congestion window has room. */
+  return circ;
+}
+
+/**
+ * Return the amount of congestion window we can send on
+ * on_circ during in_usec. However, if we're still in
+ * slow-start, send the whole window to establish the true
+ * cwnd.
+ */
+static inline uint64_t
+cwnd_sendable(const circuit_t *on_circ, uint64_t in_usec,
+              uint64_t our_usec)
+{
+  const congestion_control_t *cc = circuit_ccontrol(on_circ);
+
+  tor_assert(cc);
+
+  // TODO-329-TUNING: This function may want to consider inflight?
+
+  if (our_usec == 0 || in_usec == 0) {
+    log_fn(LOG_PROTOCOL_WARN, LD_CIRC,
+       "cwnd_sendable: Missing RTT data. in_usec: %" PRIu64
+       " our_usec: %" PRIu64, in_usec, our_usec);
+    return cc->cwnd;
+  }
+
+  if (cc->in_slow_start) {
+    return cc->cwnd;
+  } else {
+    uint64_t sendable =
+      conflux_params_get_send_pct()*cc->cwnd*in_usec/(100*our_usec);
+    return MIN(cc->cwnd, sendable);
+  }
+}
+
+/**
+ * Returns the amount of room in a cwnd on a circuit.
+ */
+static inline uint64_t
+cwnd_available(const circuit_t *on_circ)
+{
+  const congestion_control_t *cc = circuit_ccontrol(on_circ);
+  tor_assert(cc);
+
+  if (cc->cwnd < cc->inflight)
+    return 0;
+
+  return cc->cwnd - cc->inflight;
+}
+
+/**
+ * Returns true if we can switch to a new circuit, false otherwise.
+ *
+ * This function assumes we're primarily switching between two circuits,
+ * the current and the prev. If we're using more than two circuits, we
+ * need to set cfx_drain_pct to 100.
+ */
+static inline bool
+conflux_can_switch(const conflux_t *cfx)
+{
+  /* If we still expected to send more cells on this circuit,
+   * we're only allowed to switch if the previous circuit emptied. */
+  if (cfx->cells_until_switch > 0) {
+    /* If there is no prev leg, skip the inflight check. */
+    if (!cfx->prev_leg) {
+      return false;
+    }
+    const congestion_control_t *ccontrol =
+      circuit_ccontrol(cfx->prev_leg->circ);
+
+    /* If the inflight count has drained to below cfx_drain_pct
+     * of the congestion window, then we can switch.
+     * We check the sendme_inc because there may be un-ackable
+     * data in inflight as well, and we can still switch then. */
+    if (ccontrol->inflight < ccontrol->sendme_inc ||
+        100*ccontrol->inflight <=
+        conflux_params_get_drain_pct()*ccontrol->cwnd) {
+      return true;
+    }
+
+    // TODO-329-TUNING: Should we try to switch if the prev_leg is
+    // ready to send?
+
+    return false;
+  }
+
+  return true;
+}
+
+/**
+ * Favor the circuit with the lowest RTT that still has space in the
+ * congestion window up to the ratio of RTTs.
+ *
+ * This algorithm should only use auxillary legs up to the point
+ * where their data arrives roughly the same time as the lowest
+ * RTT leg. It will not utilize the full cwnd of auxillary legs,
+ * except in slow start. Therefore, out-of-order queue bloat should
+ * be minimized to just the slow-start phase.
+ */
+static const circuit_t *
+conflux_decide_circ_cwndrtt(const conflux_t *cfx)
+{
+  uint64_t min_rtt = UINT64_MAX;
+  const conflux_leg_t *leg = NULL;
+
+  /* Can't get here without any legs. */
+  tor_assert(!CONFLUX_NUM_LEGS(cfx));
+
+  /* Find the leg with the minimum RTT.*/
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, l) {
+    if (l->circ_rtts_usec < min_rtt) {
+      min_rtt = l->circ_rtts_usec;
+      leg = l;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(l);
+
+  /* If the package window is has room, use it */
+  if (leg && circuit_ready_to_send(leg->circ)) {
+    return leg->circ;
+  }
+
+  /* For any given leg, it has min_rtt/2 time before the 'primary'
+   * leg's acks start arriving. So, the amount of data this
+   * 'secondary' leg can send while the min_rtt leg transmits these
+   * acks is:
+   *   (cwnd_leg/(leg_rtt/2))*min_rtt/2 = cwnd_leg*min_rtt/leg_rtt.
+   * So any leg with available room below that is no good.
+   */
+
+  leg = NULL;
+
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, l) {
+    if (!circuit_ready_to_send(l->circ)) {
+      continue;
+    }
+
+    /* 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) <=
+        cwnd_available(l->circ)) {
+      leg = l;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(l);
+
+  /* If the circuit can't send, don't send on any circuit. */
+  if (!leg || !circuit_ready_to_send(leg->circ)) {
+    return NULL;
+  }
+  return leg->circ;
+}
+
+/**
+ * Favor the circuit with the highest send rate.
+ *
+ * Only spill over to other circuits if they are still in slow start.
+ * In steady-state, we only use the max throughput circuit.
+ */
+static const circuit_t *
+conflux_decide_circ_maxrate(const conflux_t *cfx)
+{
+  uint64_t max_rate = 0;
+  const conflux_leg_t *leg = NULL;
+
+  /* Find the highest bandwidth leg */
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, l) {
+    uint64_t rate;
+    const congestion_control_t *cc = circuit_ccontrol(l->circ);
+
+    rate = CELL_MAX_NETWORK_SIZE*USEC_PER_SEC *
+           cc->cwnd / l->circ_rtts_usec;
+    if (rate > max_rate) {
+      max_rate = rate;
+      leg = l;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(l);
+
+  /* If the package window is has room, use it */
+  if (leg && circuit_ready_to_send(leg->circ)) {
+    return leg->circ;
+  }
+
+  leg = NULL;
+  max_rate = 0;
+
+  /* Find the circuit with the max rate where in_slow_start == 1: */
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, l) {
+    uint64_t rate;
+    /* Ignore circuits with no room in the package window */
+    if (!circuit_ready_to_send(l->circ)) {
+      continue;
+    }
+
+    const congestion_control_t *cc = circuit_ccontrol(l->circ);
+
+    rate = CELL_MAX_NETWORK_SIZE*USEC_PER_SEC *
+                 cc->cwnd / l->circ_rtts_usec;
+
+    if (rate > max_rate && cc->in_slow_start) {
+      max_rate = rate;
+      leg = l;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(l);
+
+  /* If no sendable leg was found, don't send on any circuit. */
+  if (!leg) {
+    return NULL;
+  }
+  return leg->circ;
+}
+
+/**
+ * Favor the circuit with the highest send rate that still has space
+ * in the congestion window, but when it is full, pick the next
+ * highest.
+ */
+static const circuit_t *
+conflux_decide_circ_highrate(const conflux_t *cfx)
+{
+  uint64_t max_rate = 0;
+  uint64_t primary_leg_rtt = 0;
+  const conflux_leg_t *leg = NULL;
+
+  /* Find the highest bandwidth leg */
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, l) {
+    uint64_t rate;
+    const congestion_control_t *cc = circuit_ccontrol(l->circ);
+
+    rate = CELL_MAX_NETWORK_SIZE*USEC_PER_SEC *
+              cc->cwnd / l->circ_rtts_usec;
+
+    if (rate > max_rate) {
+      max_rate = rate;
+      primary_leg_rtt = l->circ_rtts_usec;
+      leg = l;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(l);
+
+  /* If the package window is has room, use it */
+  if (leg && circuit_ready_to_send(leg->circ)) {
+    return leg->circ;
+  }
+
+  /* Reset the max rate to find a new max */
+  max_rate = 0;
+  leg = NULL;
+
+  /* For any given leg, it has primary_leg_rtt/2 time before the 'primary'
+   * leg's acks start arriving. So, the amount of data a 'secondary'
+   * leg can send while the primary leg transmits these acks is:
+   *   (cwnd_leg/(secondary_rtt/2))*primary_rtt/2
+   *     = cwnd_leg*primary_rtt/secondary_rtt.
+   * So any leg with available room below that that is no good.
+   */
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, l) {
+    if (!circuit_ready_to_send(l->circ)) {
+      continue;
+    }
+    const congestion_control_t *cc = circuit_ccontrol(l->circ);
+
+    uint64_t rate = CELL_MAX_NETWORK_SIZE*USEC_PER_SEC *
+                    cc->cwnd / l->circ_rtts_usec;
+
+    /* Pick the leg with the highest rate that still has room */
+    if (rate > max_rate &&
+        cwnd_sendable(l->circ, primary_leg_rtt, l->circ_rtts_usec) <=
+        cwnd_available(l->circ)) {
+      leg = l;
+      max_rate = rate;
+    }
+  } CONFLUX_FOR_EACH_LEG_END(l);
+
+  /* If no sendable leg was found, don't send on any circuit. */
+  if (!leg) {
+    return NULL;
+  }
+  return leg->circ;
+}
+
+/**
+ * This function is called when we want to send a relay cell on a
+ * conflux, as well as when we want to compute available space in
+ * to package from streams.
+ *
+ * It determines the circuit that relay command should be sent on,
+ * and sends a SWITCH cell if necessary.
+ *
+ * It returns the circuit we should send on. If no circuits are ready
+ * to send, it returns NULL.
+ */
+circuit_t *
+conflux_decide_circ_for_send(conflux_t *cfx,
+                             circuit_t *orig_circ,
+                             uint8_t relay_command)
+{
+  /* If this command should not be multiplexed, send it on the original
+   * circuit */
+  if (!conflux_should_multiplex(relay_command)) {
+    return orig_circ;
+  }
+
+  circuit_t *new_circ = conflux_decide_next_circ(cfx);
+
+  /* Because our congestion window only cover relay data command, we can end up
+   * in a situation where we need to send non data command when all circuits
+   * are at capacity. For those cases, keep using the *current* leg,
+   * so these commands arrive in-order. */
+  if (!new_circ && relay_command != RELAY_COMMAND_DATA) {
+    /* Curr leg should be set, because conflux_decide_next_circ() should
+     * have set it earlier. */
+    tor_assert(cfx->curr_leg);
+    return cfx->curr_leg->circ;
+  }
+
+  /*
+   * If we are switching to a new circuit, we need to send a SWITCH command.
+   * We also need to compute an estimate of how much data we can send on
+   * the new circuit before we are allowed to switch again, to rate
+   * limit the frequency of switching.
+   */
+  if (new_circ) {
+    conflux_leg_t *new_leg = conflux_get_leg(cfx, new_circ);
+    tor_assert(cfx->curr_leg);
+
+    if (new_circ != cfx->curr_leg->circ) {
+      cfx->cells_until_switch =
+        cwnd_sendable(new_circ,cfx->curr_leg->circ_rtts_usec,
+                                 new_leg->circ_rtts_usec);
+
+      conflux_validate_stream_lists(cfx);
+
+      cfx->prev_leg = cfx->curr_leg;
+      cfx->curr_leg = new_leg;
+
+      tor_assert(cfx->prev_leg);
+      tor_assert(cfx->curr_leg);
+
+      uint64_t relative_seq = cfx->prev_leg->last_seq_sent -
+                              cfx->curr_leg->last_seq_sent;
+
+      tor_assert(cfx->prev_leg->last_seq_sent >=
+                 cfx->curr_leg->last_seq_sent);
+      conflux_send_switch_command(cfx->curr_leg->circ, relative_seq);
+      cfx->curr_leg->last_seq_sent = cfx->prev_leg->last_seq_sent;
+    }
+  }
+
+  return new_circ;
+}
+
+/** Called after conflux actually sent a cell on a circuit.
+ * This function updates sequence number counters, and
+ * switch counters.
+ */
+void
+conflux_note_cell_sent(conflux_t *cfx, circuit_t *circ, uint8_t relay_command)
+{
+  conflux_leg_t *leg = NULL;
+
+  if (!conflux_should_multiplex(relay_command)) {
+    return;
+  }
+
+  leg = conflux_get_leg(cfx, circ);
+  tor_assert(leg);
+
+  leg->last_seq_sent++;
+
+  if (cfx->cells_until_switch > 0) {
+    cfx->cells_until_switch--;
+  }
+}
+
+/** Find the leg with lowest non-zero curr_rtt_usec, and
+ * pick it for our current leg. */
+static inline void
+conflux_pick_first_leg(conflux_t *cfx)
+{
+  conflux_leg_t *min_leg = NULL;
+
+  CONFLUX_FOR_EACH_LEG_BEGIN(cfx, leg) {
+    /* We need to skip 0-RTT legs, since this can happen at the exit
+     * when there is a race between BEGIN and LINKED_ACK, and BEGIN
+     * wins the race. The good news is that because BEGIN won,
+     * we don't need to consider those other legs, since they are
+     * slower. */
+    if (leg->circ_rtts_usec > 0) {
+      if (!min_leg || leg->circ_rtts_usec < min_leg->circ_rtts_usec) {
+        min_leg = leg;
+      }
+    }
+  } CONFLUX_FOR_EACH_LEG_END(leg);
+
+  if (BUG(!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);
+  }
+
+  // TODO-329-TUNING: Does this create an edge condition by getting blocked,
+  // is it possible that we get full before this point and block?
+  // Esp if we switch to a new circuit that is not ready to
+  // send because it has unacked inflight data.... This might cause
+  // stalls?
+  // That is the thinking with this -1 here, but maybe it is not needed.
+  cfx->cells_until_switch = circuit_ccontrol(min_leg->circ)->cwnd - 1;
+
+  cfx->curr_leg = min_leg;
+}
+
+/**
+ * Returns the circuit that conflux would send on next, if
+ * conflux_decide_circ_for_send were called. This is used to compute
+ * available space in the package window.
+ */
+circuit_t *
+conflux_decide_next_circ(conflux_t *cfx)
+{
+  // TODO-329-TUNING: Temporarily validate legs here. We can remove
+  // this once tuning is complete.
+  conflux_validate_legs(cfx);
+
+  /* If we don't have a current leg yet, pick one.
+   * (This is the only non-const operation in this function). */
+  if (!cfx->curr_leg) {
+    conflux_pick_first_leg(cfx);
+  }
+
+  /* First, check if we can switch. */
+  if (!conflux_can_switch(cfx)) {
+    tor_assert(cfx->curr_leg);
+    circuit_t *curr_circ = cfx->curr_leg->circ;
+
+    /* If we can't switch, and the current circuit can't send,
+     * then return null. */
+    if (circuit_ready_to_send(curr_circ)) {
+      return curr_circ;
+    }
+    log_info(LD_CIRC, "Conflux can't switch; no circuit to send on.");
+    return NULL;
+  }
+
+  switch (cfx->params.alg) {
+    case CONFLUX_ALG_MINRTT: // latency (no ooq)
+      return (circuit_t*)conflux_decide_circ_minrtt(cfx);
+    case CONFLUX_ALG_LOWRTT: // high throughput (high oooq)
+      return (circuit_t*)conflux_decide_circ_lowrtt(cfx);
+    case CONFLUX_ALG_CWNDRTT: // throughput (low oooq)
+      return (circuit_t*)conflux_decide_circ_cwndrtt(cfx);
+    case CONFLUX_ALG_MAXRATE: // perf test (likely high ooq)
+      return (circuit_t*)conflux_decide_circ_maxrate(cfx);
+    case CONFLUX_ALG_HIGHRATE: // perf test (likely high ooq)
+      return (circuit_t*)conflux_decide_circ_highrate(cfx);
+    default:
+      return NULL;
+  }
+}
+
+/**
+ * Called when we have a new RTT estimate for a circuit.
+ */
+void
+conflux_update_rtt(conflux_t *cfx, circuit_t *circ, uint64_t rtt_usec)
+{
+  conflux_leg_t *leg = conflux_get_leg(cfx, circ);
+
+  if (!leg) {
+    log_warn(LD_BUG, "Got RTT update for circuit not in conflux");
+    return;
+  }
+
+  // Update RTT
+  leg->circ_rtts_usec = rtt_usec;
+
+  // TODO-329-ARTI: For UDP latency targeting, arti could decide to launch
+  // new a test leg to potentially replace this one, if a latency target
+  // was requested and we now exceed it. Since C-Tor client likely
+  // will not have UDP support, we aren't doing this here.
+}
+
 /**
  * Comparison function for ooo_q pqueue.
  *
@@ -171,6 +780,106 @@ circuit_ccontrol(const circuit_t *circ)
   return ccontrol;
 }
 
+// TODO-329-TUNING: For LowRTT, we can at most switch every SENDME,
+// but for BLEST, we should switch at most every cwnd.. But
+// we do not know the other side's CWND here.. We can at best
+// asssume it is above the cwnd_min
+#define CONFLUX_MIN_LINK_INCREMENT 31
+/**
+ * Validate and handle RELAY_COMMAND_CONFLUX_SWITCH.
+ */
+int
+conflux_process_switch_command(circuit_t *in_circ,
+                               crypt_path_t *layer_hint, cell_t *cell,
+                               relay_header_t *rh)
+{
+  tor_assert(in_circ);
+  tor_assert(cell);
+  tor_assert(rh);
+
+  conflux_t *cfx = in_circ->conflux;
+  uint32_t relative_seq;
+  conflux_leg_t *leg;
+
+  if (!conflux_is_enabled(in_circ)) {
+    circuit_mark_for_close(in_circ, END_CIRC_REASON_TORPROTOCOL);
+    return -1;
+  }
+
+  /* If there is no conflux object negotiated, this is invalid.
+   * log and close circ */
+  if (!cfx) {
+    log_warn(LD_BUG, "Got a conflux switch command on a circuit without "
+             "conflux negotiated. Closing circuit.");
+
+    circuit_mark_for_close(in_circ, END_CIRC_REASON_TORPROTOCOL);
+    return -1;
+  }
+
+  // TODO-329-TUNING: Temporarily validate that we have all legs.
+  // After tuning is complete, we can remove this.
+  conflux_validate_legs(cfx);
+
+  leg = conflux_get_leg(cfx, in_circ);
+
+  /* If we can't find the conflux leg, we got big problems..
+   * Close the circuit. */
+  if (!leg) {
+    log_warn(LD_BUG, "Got a conflux switch command on a circuit without "
+             "conflux leg. Closing circuit.");
+    circuit_mark_for_close(in_circ, END_CIRC_REASON_INTERNAL);
+    return -1;
+  }
+
+  // Check source hop via layer_hint
+  if (!conflux_validate_source_hop(in_circ, layer_hint)) {
+    log_warn(LD_BUG, "Got a conflux switch command on a circuit with "
+             "invalid source hop. Closing circuit.");
+    circuit_mark_for_close(in_circ, END_CIRC_REASON_TORPROTOCOL);
+    return -1;
+  }
+
+  relative_seq = conflux_cell_parse_switch(cell, rh->length);
+
+  /*
+   * We have to make sure that the switch command is truely
+   * incrementing the sequence number, or else it becomes
+   * a side channel that can be spammed for traffic analysis.
+   */
+  // TODO-329-TUNING: This can happen. Disabling for now..
+  //if (relative_seq < CONFLUX_MIN_LINK_INCREMENT) {
+  // log_warn(LD_CIRC, "Got a conflux switch command with a relative "
+  //          "sequence number less than the minimum increment. Closing "
+  //          "circuit.");
+  // circuit_mark_for_close(in_circ, END_CIRC_REASON_TORPROTOCOL);
+  // return -1;
+  //}
+
+  // TODO-329-UDP: When Prop#340 exits and was negotiated, ensure we're
+  // in a packed cell, with another cell following, otherwise
+  // this is a spammed side-channel.
+  //   - We definitely should never get switches back-to-back.
+  //   - We should not get switches across all legs with no data
+  // But before Prop#340, it doesn't make much sense to do this.
+  // C-Tor is riddled with side-channels like this anyway, unless
+  // vanguards is in use. And this feature is not supported by
+  // onion servicees in C-Tor, so we're good there.
+
+  /* Update the absolute sequence number on this leg by the delta.
+   * Since this cell is not multiplexed, we do not count it towards
+   * absolute sequence numbers. We only increment the sequence
+   * numbers for multiplexed cells. Hence there is no +1 here. */
+  leg->last_seq_recv += relative_seq;
+
+  /* Mark this data as validated for controlport and vanguards
+   * dropped cell handling */
+  if (CIRCUIT_IS_ORIGIN(in_circ)) {
+    circuit_read_valid_data(TO_ORIGIN_CIRCUIT(in_circ), rh->length);
+  }
+
+  return 0;
+}
+
 /**
  * Process an incoming relay cell for conflux. Called from
  * connection_edge_process_relay_cell().

src/core/or/congestion_control_common.c

@@ -23,6 +23,7 @@
 #include "core/or/congestion_control_nola.h"
 #include "core/or/congestion_control_westwood.h"
 #include "core/or/congestion_control_st.h"
+#include "core/or/conflux.h"
 #include "core/or/trace_probes_cc.h"
 #include "lib/time/compat_time.h"
 #include "feature/nodelist/networkstatus.h"
@@ -1188,7 +1189,7 @@ congestion_control_update_circuit_bdp(congestion_control_t *cc,
  */
 int
 congestion_control_dispatch_cc_alg(congestion_control_t *cc,
-                                   const circuit_t *circ,
+                                   circuit_t *circ,
                                    const crypt_path_t *layer_hint)
 {
   int ret = -END_CIRC_REASON_INTERNAL;
@@ -1218,6 +1219,10 @@ congestion_control_dispatch_cc_alg(congestion_control_t *cc,
     cc->cwnd = cwnd_max;
   }
 
+  /* If we have a non-zero RTT measurement, update conflux. */
+  if (circ->conflux && cc->ewma_rtt_usec)
+    conflux_update_rtt(circ->conflux, circ, cc->ewma_rtt_usec);
+
   return ret;
 }
 

src/core/or/congestion_control_common.h

@@ -46,7 +46,7 @@ congestion_control_t *congestion_control_new(
                                     cc_path_t path);
 
 int congestion_control_dispatch_cc_alg(congestion_control_t *cc,
-                                       const circuit_t *circ,
+                                       circuit_t *circ,
                                        const crypt_path_t *layer_hint);
 
 void congestion_control_note_cell_sent(congestion_control_t *cc,

src/core/or/relay.c

@@ -624,6 +624,23 @@ relay_send_command_from_edge_,(streamid_t stream_id, circuit_t *orig_circ,
   cell_t cell;
   relay_header_t rh;
   cell_direction_t cell_direction;
+  circuit_t *circ = orig_circ;
+
+  /* If conflux is enabled, decide which leg to send on, and use that */
+  if (orig_circ->conflux && conflux_should_multiplex(relay_command)) {
+    circ = conflux_decide_circ_for_send(orig_circ->conflux, orig_circ,
+                                        relay_command);
+    if (BUG(!circ)) {
+      log_warn(LD_BUG, "No circuit to send on for conflux");
+      circ = orig_circ;
+    } else {
+      /* Conflux circuits always send multiplexed relay commands to
+       * to the last hop. (Non-multiplexed commands go on their
+       * original circuit and hop). */
+      cpath_layer = conflux_get_destination_hop(circ);
+    }
+  }
+
   /* XXXX NM Split this function into a separate versions per circuit type? */
 
   tor_assert(circ);
@@ -721,6 +738,10 @@ relay_send_command_from_edge_,(streamid_t stream_id, circuit_t *orig_circ,
     return -1;
   }
 
+  if (circ->conflux) {
+    conflux_note_cell_sent(circ->conflux, circ, relay_command);
+  }
+
   /* If applicable, note the cell digest for the SENDME version 1 purpose if
    * we need to. This call needs to be after the circuit_package_relay_cell()
    * because the cell digest is set within that function. */
@@ -1639,6 +1660,8 @@ handle_relay_cell_command(cell_t *cell, circuit_t *circ,
 
   /* Now handle all the other commands */
   switch (rh->command) {
+    case RELAY_COMMAND_CONFLUX_SWITCH:
+      return conflux_process_switch_command(circ, layer_hint, cell, rh);
     case RELAY_COMMAND_BEGIN:
     case RELAY_COMMAND_BEGIN_DIR:
       if (layer_hint &&