October 2021 - tor-commits - lists.torproject.org

[tor/main] Support time-based half-closed connection handling.
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit 5e17f8acab8bf7380a626e5ef1a162b7e4351dcc Author: Mike Perry <mikeperry-git(a)torproject.org> Date: Thu Sep 23 19:36:11 2021 +0000 Support time-based half-closed connection handling. Since we no longer use stream SENDMEs for congestion control, we must now use time to decide when data should stop arriving on a half-closed stream. --- src/core/or/connection_edge.c | 51 ++++++++++++++++++++++++++++++++++--------- src/core/or/half_edge_st.h | 12 ++++++++++ 2 files changed, 53 insertions(+), 10 deletions(-) diff --git a/src/core/or/connection_edge.c b/src/core/or/connection_edge.c index 6f6f22a0d4..d4d9d2f759 100644 --- a/src/core/or/connection_edge.c +++ b/src/core/or/connection_edge.c @@ -69,6 +69,8 @@ #include "core/or/circuituse.h" #include "core/or/circuitpadding.h" #include "core/or/connection_edge.h" +#include "core/or/congestion_control_flow.h" +#include "core/or/circuitstats.h" #include "core/or/connection_or.h" #include "core/or/extendinfo.h" #include "core/or/policies.h" @@ -614,20 +616,39 @@ connection_half_edge_add(const edge_connection_t *conn, half_conn->stream_id = conn->stream_id; - // How many sendme's should I expect? - half_conn->sendmes_pending = - (STREAMWINDOW_START-conn->package_window)/STREAMWINDOW_INCREMENT; - // Is there a connected cell pending? half_conn->connected_pending = conn->base_.state == AP_CONN_STATE_CONNECT_WAIT; - /* Data should only arrive if we're not waiting on a resolved cell. - * It can arrive after waiting on connected, because of optimistic - * data. */ - if (conn->base_.state != AP_CONN_STATE_RESOLVE_WAIT) { - // How many more data cells can arrive on this id? - half_conn->data_pending = conn->deliver_window; + if (edge_uses_flow_control(conn)) { + /* If the edge uses the new congestion control flow control, we must use + * time-based limits on half-edge activity. */ + uint64_t timeout_usec = (uint64_t)(get_circuit_build_timeout_ms()*1000); + half_conn->used_ccontrol = 1; + + /* If this is an onion service circuit, double the CBT as an approximate + * value for the other half of the circuit */ + if (conn->hs_ident) { + timeout_usec *= 2; + } + + /* The stream should stop seeing any use after the larger of the circuit + * RTT and the overall circuit build timeout */ + half_conn->end_ack_expected_usec = MAX(timeout_usec, + edge_get_max_rtt(conn)) + + monotime_absolute_usec(); + } else { + // How many sendme's should I expect? + half_conn->sendmes_pending = + (STREAMWINDOW_START-conn->package_window)/STREAMWINDOW_INCREMENT; + + /* Data should only arrive if we're not waiting on a resolved cell. + * It can arrive after waiting on connected, because of optimistic + * data. */ + if (conn->base_.state != AP_CONN_STATE_RESOLVE_WAIT) { + // How many more data cells can arrive on this id? + half_conn->data_pending = conn->deliver_window; + } } insert_at = smartlist_bsearch_idx(circ->half_streams, &half_conn->stream_id, @@ -688,6 +709,12 @@ connection_half_edge_is_valid_data(const smartlist_t *half_conns, if (!half) return 0; + if (half->used_ccontrol) { + if (monotime_absolute_usec() > half->end_ack_expected_usec) + return 0; + return 1; + } + if (half->data_pending > 0) { half->data_pending--; return 1; @@ -740,6 +767,10 @@ connection_half_edge_is_valid_sendme(const smartlist_t *half_conns, if (!half) return 0; + /* congestion control edges don't use sendmes */ + if (half->used_ccontrol) + return 0; + if (half->sendmes_pending > 0) { half->sendmes_pending--; return 1; diff --git a/src/core/or/half_edge_st.h b/src/core/or/half_edge_st.h index c956c7434a..ac97eb19f1 100644 --- a/src/core/or/half_edge_st.h +++ b/src/core/or/half_edge_st.h @@ -31,6 +31,18 @@ typedef struct half_edge_t { * our deliver window */ int data_pending; + /** + * Monotime timestamp of when the other end should have successfuly + * shut down the stream and stop sending data, based on the larger + * of circuit RTT and CBT. Used if 'used_ccontrol' is true, to expire + * the half_edge at this monotime timestamp. */ + uint64_t end_ack_expected_usec; + + /** + * Did this edge use congestion control? If so, use + * timer instead of pending data approach */ + int used_ccontrol : 1; + /** Is there a connected cell pending? */ int connected_pending : 1; } half_edge_t;

1 0

[tor/main] Turn CircEWMA tick len into consensus parameter.
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit 98be8634faf4405edc5388c64d9f834b6ae5b4b1 Author: Mike Perry <mikeperry-git(a)torproject.org> Date: Tue Sep 28 17:44:33 2021 +0000 Turn CircEWMA tick len into consensus parameter. This will assist tuning of CircEWMA in Shadow and Live. --- src/core/or/circuitmux_ewma.c | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/src/core/or/circuitmux_ewma.c b/src/core/or/circuitmux_ewma.c index 0382e62f75..adf256ab05 100644 --- a/src/core/or/circuitmux_ewma.c +++ b/src/core/or/circuitmux_ewma.c @@ -45,7 +45,10 @@ /*** EWMA parameter #defines ***/ /** How long does a tick last (seconds)? */ -#define EWMA_TICK_LEN 10 +#define EWMA_TICK_LEN_DEFAULT 10 +#define EWMA_TICK_LEN_MIN 1 +#define EWMA_TICK_LEN_MAX 600 +static int ewma_tick_len = EWMA_TICK_LEN_DEFAULT; /** The default per-tick scale factor, if it hasn't been overridden by a * consensus or a configuration setting. zero means "disabled". */ @@ -148,7 +151,7 @@ cell_ewma_get_tick(void) monotime_coarse_get(&now); int32_t msec_diff = monotime_coarse_diff_msec32(&start_of_current_tick, &now); - return current_tick_num + msec_diff / (1000*EWMA_TICK_LEN); + return current_tick_num + msec_diff / (1000*ewma_tick_len); } /** @@ -527,15 +530,15 @@ cell_ewma_get_current_tick_and_fraction(double *remainder_out) monotime_coarse_get(&now); int32_t msec_diff = monotime_coarse_diff_msec32(&start_of_current_tick, &now); - if (msec_diff > (1000*EWMA_TICK_LEN)) { - unsigned ticks_difference = msec_diff / (1000*EWMA_TICK_LEN); + if (msec_diff > (1000*ewma_tick_len)) { + unsigned ticks_difference = msec_diff / (1000*ewma_tick_len); monotime_coarse_add_msec(&start_of_current_tick, &start_of_current_tick, - ticks_difference * 1000 * EWMA_TICK_LEN); + ticks_difference * 1000 * ewma_tick_len); current_tick_num += ticks_difference; - msec_diff %= 1000*EWMA_TICK_LEN; + msec_diff %= 1000*ewma_tick_len; } - *remainder_out = ((double)msec_diff) / (1.0e3 * EWMA_TICK_LEN); + *remainder_out = ((double)msec_diff) / (1.0e3 * ewma_tick_len); return current_tick_num; } @@ -605,15 +608,20 @@ cmux_ewma_set_options(const or_options_t *options, /* Both options and consensus can be NULL. This assures us to either get a * valid configured value or the default one. */ halflife = get_circuit_priority_halflife(options, consensus, &source); + ewma_tick_len = networkstatus_get_param(consensus, + "CircuitPriorityTickSecs", + EWMA_TICK_LEN_DEFAULT, + EWMA_TICK_LEN_MIN, + EWMA_TICK_LEN_MAX); /* convert halflife into halflife-per-tick. */ - halflife /= EWMA_TICK_LEN; + halflife /= ewma_tick_len; /* compute per-tick scale factor. */ ewma_scale_factor = exp(LOG_ONEHALF / halflife); log_info(LD_OR, "Enabled cell_ewma algorithm because of value in %s; " "scale factor is %f per %d seconds", - source, ewma_scale_factor, EWMA_TICK_LEN); + source, ewma_scale_factor, ewma_tick_len); } /** Return the multiplier necessary to convert the value of a cell sent in

1 0

[tor/main] oom: Consider edge connections as well
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit bd0aabe20f7b0b89a1f91ac11f0bf4f72034b27d Author: David Goulet <dgoulet(a)torproject.org> Date: Thu Sep 9 10:32:55 2021 -0400 oom: Consider edge connections as well Signed-off-by: David Goulet <dgoulet(a)torproject.org> --- src/core/or/circuitlist.c | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/core/or/circuitlist.c b/src/core/or/circuitlist.c index 35d810c660..88c4159294 100644 --- a/src/core/or/circuitlist.c +++ b/src/core/or/circuitlist.c @@ -2602,6 +2602,7 @@ circuits_handle_oom(size_t current_allocation) size_t mem_recovered=0; int n_circuits_killed=0; int n_dirconns_killed=0; + int n_edgeconns_killed = 0; uint32_t now_ts; log_notice(LD_GENERAL, "We're low on memory (cell queues total alloc:" " %"TOR_PRIuSZ" buffer total alloc: %" TOR_PRIuSZ "," @@ -2668,12 +2669,19 @@ circuits_handle_oom(size_t current_allocation) if (conn_age < circ->age_tmp) { break; } - if (conn->type == CONN_TYPE_DIR && conn->linked_conn == NULL) { + /* Also consider edge connections so we don't accumulate bytes on the + * outbuf due to a malicious destination holding off the read on us. */ + if ((conn->type == CONN_TYPE_DIR && conn->linked_conn == NULL) || + CONN_IS_EDGE(conn)) { if (!conn->marked_for_close) connection_mark_for_close(conn); mem_recovered += single_conn_free_bytes(conn); - ++n_dirconns_killed; + if (conn->type == CONN_TYPE_DIR) { + ++n_dirconns_killed; + } else { + ++n_edgeconns_killed; + } if (mem_recovered >= mem_to_recover) goto done_recovering_mem; @@ -2703,11 +2711,12 @@ circuits_handle_oom(size_t current_allocation) done_recovering_mem: log_notice(LD_GENERAL, "Removed %"TOR_PRIuSZ" bytes by killing %d circuits; " "%d circuits remain alive. Also killed %d non-linked directory " - "connections.", + "connections. Killed %d edge connections", mem_recovered, n_circuits_killed, smartlist_len(circlist) - n_circuits_killed, - n_dirconns_killed); + n_dirconns_killed, + n_edgeconns_killed); return mem_recovered; }

1 0

[tor/main] Turn orconn watermarks into consensus parameters.
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit 6dae9903b1d5fe3c5df712eb99076a9adf9d5c8b Author: Mike Perry <mikeperry-git(a)torproject.org> Date: Tue Sep 28 21:19:27 2021 +0000 Turn orconn watermarks into consensus parameters. Tuning these may reduce memory usage and latency. --- src/app/main/main.c | 2 ++ src/core/or/channeltls.c | 3 ++- src/core/or/congestion_control_common.c | 31 +++++++++++++++++++++++++++++++ src/core/or/congestion_control_common.h | 21 +++++++++++++++++++++ src/core/or/connection_or.c | 7 ++++--- src/core/or/or.h | 12 ------------ src/feature/nodelist/networkstatus.c | 2 ++ src/test/test_channeltls.c | 3 ++- 8 files changed, 64 insertions(+), 17 deletions(-) diff --git a/src/app/main/main.c b/src/app/main/main.c index 0742abe70a..6043b9589f 100644 --- a/src/app/main/main.c +++ b/src/app/main/main.c @@ -27,6 +27,7 @@ #include "core/or/channel.h" #include "core/or/channelpadding.h" #include "core/or/circuitpadding.h" +#include "core/or/congestion_control_common.h" #include "core/or/congestion_control_flow.h" #include "core/or/circuitlist.h" #include "core/or/command.h" @@ -631,6 +632,7 @@ tor_init(int argc, char *argv[]) * until we get a consensus */ channelpadding_new_consensus_params(NULL); circpad_new_consensus_params(NULL); + congestion_control_new_consensus_params(NULL); flow_control_new_consensus_params(NULL); /* Initialize circuit padding to defaults+torrc until we get a consensus */ diff --git a/src/core/or/channeltls.c b/src/core/or/channeltls.c index 481dafef91..9db8e2392d 100644 --- a/src/core/or/channeltls.c +++ b/src/core/or/channeltls.c @@ -64,6 +64,7 @@ #include "trunnel/netinfo.h" #include "core/or/channelpadding.h" #include "core/or/extendinfo.h" +#include "core/or/congestion_control_common.h" #include "core/or/cell_st.h" #include "core/or/cell_queue_st.h" @@ -793,7 +794,7 @@ channel_tls_num_cells_writeable_method(channel_t *chan) cell_network_size = get_cell_network_size(tlschan->conn->wide_circ_ids); outbuf_len = connection_get_outbuf_len(TO_CONN(tlschan->conn)); /* Get the number of cells */ - n = CEIL_DIV(OR_CONN_HIGHWATER - outbuf_len, cell_network_size); + n = CEIL_DIV(or_conn_highwatermark() - outbuf_len, cell_network_size); if (n < 0) n = 0; #if SIZEOF_SIZE_T > SIZEOF_INT if (n > INT_MAX) n = INT_MAX; diff --git a/src/core/or/congestion_control_common.c b/src/core/or/congestion_control_common.c index 51691c106d..393b79459d 100644 --- a/src/core/or/congestion_control_common.c +++ b/src/core/or/congestion_control_common.c @@ -46,6 +46,9 @@ #define BWE_SENDME_MIN_DFLT 5 +#define OR_CONN_HIGHWATER_DFLT (32*1024) +#define OR_CONN_LOWWATER_DFLT (16*1024) + static uint64_t congestion_control_update_circuit_rtt(congestion_control_t *, uint64_t); static bool congestion_control_update_circuit_bdp(congestion_control_t *, @@ -53,6 +56,34 @@ static bool congestion_control_update_circuit_bdp(congestion_control_t *, const crypt_path_t *, uint64_t, uint64_t); +static uint32_t cwnd_max = CWND_MAX_DFLT; +uint32_t or_conn_highwater = OR_CONN_HIGHWATER_DFLT; +uint32_t or_conn_lowwater = OR_CONN_LOWWATER_DFLT; + +/** + * Update global congestion control related consensus parameter values, + * every consensus update. + */ +void +congestion_control_new_consensus_params(const networkstatus_t *ns) +{ +#define OR_CONN_HIGHWATER_MIN (CELL_PAYLOAD_SIZE) +#define OR_CONN_HIGHWATER_MAX (INT32_MAX) + or_conn_highwater = + networkstatus_get_param(ns, "orconn_high", + OR_CONN_HIGHWATER_DFLT, + OR_CONN_HIGHWATER_MIN, + OR_CONN_HIGHWATER_MAX); + +#define OR_CONN_LOWWATER_MIN (CELL_PAYLOAD_SIZE) +#define OR_CONN_LOWWATER_MAX (INT32_MAX) + or_conn_lowwater = + networkstatus_get_param(ns, "orconn_low", + OR_CONN_LOWWATER_DFLT, + OR_CONN_LOWWATER_MIN, + OR_CONN_LOWWATER_MAX); +} + /** * Set congestion control parameters on a circuit's congestion * control object based on values from the consensus. diff --git a/src/core/or/congestion_control_common.h b/src/core/or/congestion_control_common.h index e8b9681ac6..02fee78013 100644 --- a/src/core/or/congestion_control_common.h +++ b/src/core/or/congestion_control_common.h @@ -41,6 +41,27 @@ int sendme_get_inc_count(const circuit_t *, const crypt_path_t *); bool circuit_sent_cell_for_sendme(const circuit_t *, const crypt_path_t *); bool is_monotime_clock_reliable(void); +void congestion_control_new_consensus_params(const networkstatus_t *ns); + +/* Ugh, C.. these two are private. Use the getter instead, when + * external to the congestion control code. */ +extern uint32_t or_conn_highwater; +extern uint32_t or_conn_lowwater; + +/** Stop writing on an orconn when its outbuf is this large */ +static inline uint32_t +or_conn_highwatermark(void) +{ + return or_conn_highwater; +} + +/** Resume writing on an orconn when its outbuf is less than this */ +static inline uint32_t +or_conn_lowwatermark(void) +{ + return or_conn_lowwater; +} + /** * Compute an N-count EWMA, aka N-EWMA. N-EWMA is defined as: * EWMA = alpha*value + (1-alpha)*EWMA_prev diff --git a/src/core/or/connection_or.c b/src/core/or/connection_or.c index dd31638eb3..db9f93e6f6 100644 --- a/src/core/or/connection_or.c +++ b/src/core/or/connection_or.c @@ -65,6 +65,7 @@ #include "core/or/scheduler.h" #include "feature/nodelist/torcert.h" #include "core/or/channelpadding.h" +#include "core/or/congestion_control_common.h" #include "feature/dirauth/authmode.h" #include "feature/hs/hs_service.h" @@ -636,7 +637,7 @@ connection_or_flushed_some(or_connection_t *conn) /* If we're under the low water mark, add cells until we're just over the * high water mark. */ datalen = connection_get_outbuf_len(TO_CONN(conn)); - if (datalen < OR_CONN_LOWWATER) { + if (datalen < or_conn_lowwatermark()) { /* Let the scheduler know */ scheduler_channel_wants_writes(TLS_CHAN_TO_BASE(conn->chan)); } @@ -660,9 +661,9 @@ connection_or_num_cells_writeable(or_connection_t *conn) * used to trigger when to start writing after we've stopped. */ datalen = connection_get_outbuf_len(TO_CONN(conn)); - if (datalen < OR_CONN_HIGHWATER) { + if (datalen < or_conn_highwatermark()) { cell_network_size = get_cell_network_size(conn->wide_circ_ids); - n = CEIL_DIV(OR_CONN_HIGHWATER - datalen, cell_network_size); + n = CEIL_DIV(or_conn_highwatermark() - datalen, cell_network_size); } return n; diff --git a/src/core/or/or.h b/src/core/or/or.h index ad82130301..392a848ee7 100644 --- a/src/core/or/or.h +++ b/src/core/or/or.h @@ -594,18 +594,6 @@ typedef struct or_handshake_state_t or_handshake_state_t; /** Length of Extended ORPort connection identifier. */ #define EXT_OR_CONN_ID_LEN DIGEST_LEN /* 20 */ -/* - * OR_CONN_HIGHWATER and OR_CONN_LOWWATER moved from connection_or.c so - * channeltls.c can see them too. - */ - -/** When adding cells to an OR connection's outbuf, keep adding until the - * outbuf is at least this long, or we run out of cells. */ -#define OR_CONN_HIGHWATER (32*1024) - -/** Add cells to an OR connection's outbuf whenever the outbuf's data length - * drops below this size. */ -#define OR_CONN_LOWWATER (16*1024) typedef struct connection_t connection_t; typedef struct control_connection_t control_connection_t; diff --git a/src/feature/nodelist/networkstatus.c b/src/feature/nodelist/networkstatus.c index 0138dff033..6867d8c98e 100644 --- a/src/feature/nodelist/networkstatus.c +++ b/src/feature/nodelist/networkstatus.c @@ -45,6 +45,7 @@ #include "core/or/channel.h" #include "core/or/channelpadding.h" #include "core/or/circuitpadding.h" +#include "core/or/congestion_control_common.h" #include "core/or/congestion_control_flow.h" #include "core/or/circuitmux.h" #include "core/or/circuitmux_ewma.h" @@ -1700,6 +1701,7 @@ notify_after_networkstatus_changes(void) channelpadding_new_consensus_params(c); circpad_new_consensus_params(c); router_new_consensus_params(c); + congestion_control_new_consensus_params(c); flow_control_new_consensus_params(c); /* Maintenance of our L2 guard list */ diff --git a/src/test/test_channeltls.c b/src/test/test_channeltls.c index 5219c86097..ca7fee2c53 100644 --- a/src/test/test_channeltls.c +++ b/src/test/test_channeltls.c @@ -20,6 +20,7 @@ #include "lib/tls/tortls.h" #include "core/or/or_connection_st.h" +#include "core/or/congestion_control_common.h" /* Test suite stuff */ #include "test/test.h" @@ -155,7 +156,7 @@ test_channeltls_num_bytes_queued(void *arg) * - 2 cells. */ n = ch->num_cells_writeable(ch); - tt_int_op(n, OP_EQ, CEIL_DIV(OR_CONN_HIGHWATER, 512) - 2); + tt_int_op(n, OP_EQ, CEIL_DIV(or_conn_highwatermark(), 512) - 2); UNMOCK(buf_datalen); tlschan_buf_datalen_mock_target = NULL; tlschan_buf_datalen_mock_size = 0;

1 0

[tor/main] Add a max cwnd consensus parameter and clamp.
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit e9038dc5f2fe6a62af1a32e1e4c51d8b894998e1 Author: Mike Perry <mikeperry-git(a)torproject.org> Date: Tue Sep 28 15:17:34 2021 +0000 Add a max cwnd consensus parameter and clamp. --- src/core/or/congestion_control_common.c | 29 +++++++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/src/core/or/congestion_control_common.c b/src/core/or/congestion_control_common.c index 393b79459d..4449d9bfd5 100644 --- a/src/core/or/congestion_control_common.c +++ b/src/core/or/congestion_control_common.c @@ -33,6 +33,7 @@ #define SENDME_INC_DFLT (50) #define CWND_MIN_DFLT (MAX(100, SENDME_INC_DFLT)) +#define CWND_MAX_DFLT (INT32_MAX) #define CWND_INC_DFLT (50) @@ -82,6 +83,14 @@ congestion_control_new_consensus_params(const networkstatus_t *ns) OR_CONN_LOWWATER_DFLT, OR_CONN_LOWWATER_MIN, OR_CONN_LOWWATER_MAX); + +#define CWND_MAX_MIN 500 +#define CWND_MAX_MAX (INT32_MAX) + cwnd_max = + networkstatus_get_param(NULL, "cc_cwnd_max", + CWND_MAX_DFLT, + CWND_MAX_MIN, + CWND_MAX_MAX); } /** @@ -963,20 +972,32 @@ congestion_control_dispatch_cc_alg(congestion_control_t *cc, const circuit_t *circ, const crypt_path_t *layer_hint) { + int ret = -END_CIRC_REASON_INTERNAL; switch (cc->cc_alg) { case CC_ALG_WESTWOOD: - return congestion_control_westwood_process_sendme(cc, circ, layer_hint); + ret = congestion_control_westwood_process_sendme(cc, circ, layer_hint); + break; case CC_ALG_VEGAS: - return congestion_control_vegas_process_sendme(cc, circ, layer_hint); + ret = congestion_control_vegas_process_sendme(cc, circ, layer_hint); + break; case CC_ALG_NOLA: - return congestion_control_nola_process_sendme(cc, circ, layer_hint); + ret = congestion_control_nola_process_sendme(cc, circ, layer_hint); + break; case CC_ALG_SENDME: default: tor_assert(0); } - return -END_CIRC_REASON_INTERNAL; + if (cc->cwnd > cwnd_max) { + static ratelim_t cwnd_limit = RATELIM_INIT(60); + log_fn_ratelim(&cwnd_limit, LOG_NOTICE, LD_CIRC, + "Congestion control cwnd %"PRIu64" exceeds max %d, clamping.", + cc->cwnd, cwnd_max); + cc->cwnd = cwnd_max; + } + + return ret; }

1 0

[tor/main] changes: Add file for ticket 40450 (prop324)
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit 7005046bd27968069dd56cad3fc66d644c7f517b Author: David Goulet <dgoulet(a)torproject.org> Date: Mon Oct 4 10:40:18 2021 -0400 changes: Add file for ticket 40450 (prop324) Closes #40450 Signed-off-by: David Goulet <dgoulet(a)torproject.org> --- changes/ticket40450 | 3 +++ 1 file changed, 3 insertions(+) diff --git a/changes/ticket40450 b/changes/ticket40450 new file mode 100644 index 0000000000..6753bd04f5 --- /dev/null +++ b/changes/ticket40450 @@ -0,0 +1,3 @@ + o Major features (congestion control): + - Implement support for flow control over congestion controlled circuits. + This work comes from proposal 324. Closes ticket 40450.

1 0

[tor/main] cc: Add comments and clean up some syntax
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit bfd69655afb2700bc9bf56333fdfe20b0a9b974b Author: David Goulet <dgoulet(a)torproject.org> Date: Mon Oct 4 10:32:59 2021 -0400 cc: Add comments and clean up some syntax Signed-off-by: David Goulet <dgoulet(a)torproject.org> --- src/core/or/congestion_control_common.c | 45 ++++++++++++++++++++++----------- src/core/or/congestion_control_flow.c | 41 ++++++++++++++++++++---------- 2 files changed, 57 insertions(+), 29 deletions(-) diff --git a/src/core/or/congestion_control_common.c b/src/core/or/congestion_control_common.c index be25258249..0919f037db 100644 --- a/src/core/or/congestion_control_common.c +++ b/src/core/or/congestion_control_common.c @@ -26,30 +26,45 @@ #include "lib/time/compat_time.h" #include "feature/nodelist/networkstatus.h" -/* Consensus parameter defaults */ +/* Consensus parameter defaults. + * + * More details for each of the parameters can be found in proposal 324, + * section 6.5 including tuning notes. */ #define CIRCWINDOW_INIT (500) - -#define CWND_INC_PCT_SS_DFLT (100) - -#define SENDME_INC_DFLT (50) -#define CWND_MIN_DFLT (MAX(100, SENDME_INC_DFLT)) -#define CWND_MAX_DFLT (INT32_MAX) +#define SENDME_INC_DFLT (50) #define CWND_INC_DFLT (50) - +#define CWND_INC_PCT_SS_DFLT (100) #define CWND_INC_RATE_DFLT (1) +#define CWND_MAX_DFLT (INT32_MAX) +#define CWND_MIN_DFLT (MAX(100, SENDME_INC_DFLT)) + +#define BWE_SENDME_MIN_DFLT (5) +#define EWMA_CWND_COUNT_DFLT (2) +/* BDP algorithms for each congestion control algorithms use the piecewise + * estimattor. See section 3.1.4 of proposal 324. */ #define WESTWOOD_BDP_ALG BDP_ALG_PIECEWISE #define VEGAS_BDP_MIX_ALG BDP_ALG_PIECEWISE #define NOLA_BDP_ALG BDP_ALG_PIECEWISE -#define EWMA_CWND_COUNT_DFLT 2 - -#define BWE_SENDME_MIN_DFLT 5 - +/* Indicate OR connection buffer limitations used to stop or start accepting + * cells in its outbuf. + * + * These watermarks are historical to tor in a sense that they've been used + * almost from the genesis point. And were likely defined to fit the bounds of + * TLS records of 16KB which would be around 32 cells. + * + * These are defaults of the consensus parameter "orconn_high" and "orconn_low" + * values. */ #define OR_CONN_HIGHWATER_DFLT (32*1024) #define OR_CONN_LOWWATER_DFLT (16*1024) +/* Low and high values of circuit cell queue sizes. They are used to tell when + * to start or stop reading on the streams attached on the circuit. + * + * These are defaults of the consensus parameters "cellq_high" and "cellq_low". + */ #define CELL_QUEUE_LOW_DFLT (10) #define CELL_QUEUE_HIGH_DFLT (256) @@ -60,12 +75,12 @@ static bool congestion_control_update_circuit_bdp(congestion_control_t *, const crypt_path_t *, uint64_t, uint64_t); +/* Consensus parameters cached. The non static ones are extern. */ static uint32_t cwnd_max = CWND_MAX_DFLT; -uint32_t or_conn_highwater = OR_CONN_HIGHWATER_DFLT; -uint32_t or_conn_lowwater = OR_CONN_LOWWATER_DFLT; - int32_t cell_queue_high = CELL_QUEUE_HIGH_DFLT; int32_t cell_queue_low = CELL_QUEUE_LOW_DFLT; +uint32_t or_conn_highwater = OR_CONN_HIGHWATER_DFLT; +uint32_t or_conn_lowwater = OR_CONN_LOWWATER_DFLT; /** * Update global congestion control related consensus parameter values, diff --git a/src/core/or/congestion_control_flow.c b/src/core/or/congestion_control_flow.c index 6742bb38bb..0c2baa96e2 100644 --- a/src/core/or/congestion_control_flow.c +++ b/src/core/or/congestion_control_flow.c @@ -32,28 +32,34 @@ static uint32_t xoff_client; static uint32_t xoff_exit; static uint32_t xon_change_pct; -static uint32_t xon_rate_bytes; static uint32_t xon_ewma_cnt; +static uint32_t xon_rate_bytes; -/** In normal operation, we can get a burst of up to 32 cells before - * returning to libevent to flush the outbuf. This is a heuristic from - * hardcoded values and strange logic in connection_bucket_get_share(). */ +/* In normal operation, we can get a burst of up to 32 cells before returning + * to libevent to flush the outbuf. This is a heuristic from hardcoded values + * and strange logic in connection_bucket_get_share(). */ #define MAX_EXPECTED_CELL_BURST 32 -/** - * The following three are for dropmark rate limiting. They define when - * we scale down our XON, XOFF, and xmit byte counts. Early scaling - * is beneficial because it limits the ability of spurious XON/XOFF - * to be sent after large amounts of data without XON/XOFF. At these - * limits, after 10MB of data (or more), an adversary can only inject - * (log2(10MB)-log2(200*500))*100 ~= 1000 cells of fake XOFF/XON before - * the xmit byte * count will be halved enough to triggering a limit. */ +/* The following three are for dropmark rate limiting. They define when we + * scale down our XON, XOFF, and xmit byte counts. Early scaling is beneficial + * because it limits the ability of spurious XON/XOFF to be sent after large + * amounts of data without XON/XOFF. At these limits, after 10MB of data (or + * more), an adversary can only inject (log2(10MB)-log2(200*500))*100 ~= 1000 + * cells of fake XOFF/XON before the xmit byte count will be halved enough to + * triggering a limit. */ #define XON_COUNT_SCALE_AT 200 #define XOFF_COUNT_SCALE_AT 200 #define ONE_MEGABYTE (UINT64_C(1) << 20) -#define TOTAL_XMIT_SCALE_AT 10*ONE_MEGABYTE +#define TOTAL_XMIT_SCALE_AT (10 * ONE_MEGABYTE) -static const congestion_control_t * +/** + * Return the congestion control object of the given edge connection. + * + * Returns NULL if the edge connection doesn't have a cpath_layer or not + * attached to a circuit. But also if the cpath_layer or circuit doesn't have a + * congestion control object. + */ +static inline const congestion_control_t * edge_get_ccontrol(const edge_connection_t *edge) { if (edge->cpath_layer) @@ -64,6 +70,13 @@ edge_get_ccontrol(const edge_connection_t *edge) return NULL; } +/** + * Update global congestion control related consensus parameter values, every + * consensus update. + * + * More details for each of the parameters can be found in proposal 324, + * section 6.5 including tuning notes. + */ void flow_control_new_consensus_params(const networkstatus_t *ns) {

1 0

[tor/main] Turn cell queue watermark limits into consensus params.
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit 322f213210d64de3db4ff0486e391378062e5a61 Author: Mike Perry <mikeperry-git(a)torproject.org> Date: Wed Sep 29 02:12:56 2021 +0000 Turn cell queue watermark limits into consensus params. This allows us to have quicker reaction to blocked orconns as a congestion signal. --- src/core/or/congestion_control_common.c | 20 ++++++++++++++++++++ src/core/or/congestion_control_common.h | 20 +++++++++++++++++++- src/core/or/relay.c | 15 ++++----------- 3 files changed, 43 insertions(+), 12 deletions(-) diff --git a/src/core/or/congestion_control_common.c b/src/core/or/congestion_control_common.c index 4449d9bfd5..be25258249 100644 --- a/src/core/or/congestion_control_common.c +++ b/src/core/or/congestion_control_common.c @@ -50,6 +50,9 @@ #define OR_CONN_HIGHWATER_DFLT (32*1024) #define OR_CONN_LOWWATER_DFLT (16*1024) +#define CELL_QUEUE_LOW_DFLT (10) +#define CELL_QUEUE_HIGH_DFLT (256) + static uint64_t congestion_control_update_circuit_rtt(congestion_control_t *, uint64_t); static bool congestion_control_update_circuit_bdp(congestion_control_t *, @@ -61,6 +64,9 @@ static uint32_t cwnd_max = CWND_MAX_DFLT; uint32_t or_conn_highwater = OR_CONN_HIGHWATER_DFLT; uint32_t or_conn_lowwater = OR_CONN_LOWWATER_DFLT; +int32_t cell_queue_high = CELL_QUEUE_HIGH_DFLT; +int32_t cell_queue_low = CELL_QUEUE_LOW_DFLT; + /** * Update global congestion control related consensus parameter values, * every consensus update. @@ -68,6 +74,20 @@ uint32_t or_conn_lowwater = OR_CONN_LOWWATER_DFLT; void congestion_control_new_consensus_params(const networkstatus_t *ns) { +#define CELL_QUEUE_HIGH_MIN (1) +#define CELL_QUEUE_HIGH_MAX (1000) + cell_queue_high = networkstatus_get_param(ns, "cellq_high", + CELL_QUEUE_HIGH_DFLT, + CELL_QUEUE_HIGH_MIN, + CELL_QUEUE_HIGH_MAX); + +#define CELL_QUEUE_LOW_MIN (1) +#define CELL_QUEUE_LOW_MAX (1000) + cell_queue_low = networkstatus_get_param(ns, "cellq_low", + CELL_QUEUE_LOW_DFLT, + CELL_QUEUE_LOW_MIN, + CELL_QUEUE_LOW_MAX); + #define OR_CONN_HIGHWATER_MIN (CELL_PAYLOAD_SIZE) #define OR_CONN_HIGHWATER_MAX (INT32_MAX) or_conn_highwater = diff --git a/src/core/or/congestion_control_common.h b/src/core/or/congestion_control_common.h index 02fee78013..01dbc1ceb4 100644 --- a/src/core/or/congestion_control_common.h +++ b/src/core/or/congestion_control_common.h @@ -43,10 +43,12 @@ bool is_monotime_clock_reliable(void); void congestion_control_new_consensus_params(const networkstatus_t *ns); -/* Ugh, C.. these two are private. Use the getter instead, when +/* Ugh, C.. these four are private. Use the getter instead, when * external to the congestion control code. */ extern uint32_t or_conn_highwater; extern uint32_t or_conn_lowwater; +extern int32_t cell_queue_high; +extern int32_t cell_queue_low; /** Stop writing on an orconn when its outbuf is this large */ static inline uint32_t @@ -62,6 +64,22 @@ or_conn_lowwatermark(void) return or_conn_lowwater; } +/** Stop reading on edge connections when we have this many cells + * waiting on the appropriate queue. */ +static inline int32_t +cell_queue_highwatermark(void) +{ + return cell_queue_high; +} + +/** Start reading from edge connections again when we get down to this many + * cells. */ +static inline int32_t +cell_queue_lowwatermark(void) +{ + return cell_queue_low; +} + /** * Compute an N-count EWMA, aka N-EWMA. N-EWMA is defined as: * EWMA = alpha*value + (1-alpha)*EWMA_prev diff --git a/src/core/or/relay.c b/src/core/or/relay.c index 0e889eb348..a2123f991c 100644 --- a/src/core/or/relay.c +++ b/src/core/or/relay.c @@ -117,13 +117,6 @@ static void adjust_exit_policy_from_exitpolicy_failure(origin_circuit_t *circ, node_t *node, const tor_addr_t *addr); -/** Stop reading on edge connections when we have this many cells - * waiting on the appropriate queue. */ -#define CELL_QUEUE_HIGHWATER_SIZE 256 -/** Start reading from edge connections again when we get down to this many - * cells. */ -#define CELL_QUEUE_LOWWATER_SIZE 64 - /** Stats: how many relay cells have originated at this hop, or have * been relayed onward (not recognized at this hop)? */ @@ -2400,8 +2393,8 @@ circuit_resume_edge_reading_helper(edge_connection_t *first_conn, or_circuit_t *or_circ = TO_OR_CIRCUIT(circ); cells_on_queue = or_circ->p_chan_cells.n; } - if (CELL_QUEUE_HIGHWATER_SIZE - cells_on_queue < max_to_package) - max_to_package = CELL_QUEUE_HIGHWATER_SIZE - cells_on_queue; + if (cell_queue_highwatermark() - cells_on_queue < max_to_package) + max_to_package = cell_queue_highwatermark() - cells_on_queue; /* Once we used to start listening on the streams in the order they * appeared in the linked list. That leads to starvation on the @@ -3121,7 +3114,7 @@ channel_flush_from_first_active_circuit, (channel_t *chan, int max)) /* Is the cell queue low enough to unblock all the streams that are waiting * to write to this circuit? */ - if (streams_blocked && queue->n <= CELL_QUEUE_LOWWATER_SIZE) + if (streams_blocked && queue->n <= cell_queue_lowwatermark()) set_streams_blocked_on_circ(circ, chan, 0, 0); /* unblock streams */ /* If n_flushed < max still, loop around and pick another circuit */ @@ -3239,7 +3232,7 @@ append_cell_to_circuit_queue(circuit_t *circ, channel_t *chan, /* If we have too many cells on the circuit, we should stop reading from * the edge streams for a while. */ - if (!streams_blocked && queue->n >= CELL_QUEUE_HIGHWATER_SIZE) + if (!streams_blocked && queue->n >= cell_queue_highwatermark()) set_streams_blocked_on_circ(circ, chan, 1, 0); /* block streams */ if (streams_blocked && fromstream) {

1 0

[tor/main] Merge branch 'tor-gitlab/mr/444'
by dgoulet＠torproject.org 04 Oct '21

04 Oct '21

commit 1873d4c14c62e8c737e24a60b0aca5a3fcd89eb4 Merge: 1a10948260 7005046bd2 Author: David Goulet <dgoulet(a)torproject.org> Date: Mon Oct 4 10:49:27 2021 -0400 Merge branch 'tor-gitlab/mr/444' changes/ticket40450 | 3 + src/app/main/main.c | 4 + src/core/mainloop/connection.c | 66 ++- src/core/mainloop/mainloop.c | 7 + src/core/or/channeltls.c | 3 +- src/core/or/circuitlist.c | 17 +- src/core/or/circuitmux_ewma.c | 26 +- src/core/or/congestion_control_common.c | 193 +++++++-- src/core/or/congestion_control_common.h | 58 +++ src/core/or/congestion_control_flow.c | 710 ++++++++++++++++++++++++++++++++ src/core/or/congestion_control_flow.h | 48 +++ src/core/or/connection_edge.c | 51 ++- src/core/or/connection_or.c | 7 +- src/core/or/edge_connection_st.h | 55 +++ src/core/or/half_edge_st.h | 12 + src/core/or/include.am | 5 + src/core/or/lttng_cc.inc | 166 ++++++++ src/core/or/or.h | 15 +- src/core/or/relay.c | 62 ++- src/core/or/sendme.c | 28 +- src/core/or/sendme.h | 2 +- src/core/or/trace_probes_cc.c | 33 ++ src/core/or/trace_probes_cc.h | 22 + src/feature/nodelist/networkstatus.c | 4 + src/test/test_channeltls.c | 3 +- src/trunnel/flow_control_cells.c | 382 +++++++++++++++++ src/trunnel/flow_control_cells.h | 120 ++++++ src/trunnel/flow_control_cells.trunnel | 20 + src/trunnel/include.am | 3 + 29 files changed, 2016 insertions(+), 109 deletions(-)

1 0

[snowflake/main] Update Snowflake client library documentation
by cohosh＠torproject.org 04 Oct '21

04 Oct '21

commit 638ec6c222327ca02950338b2919ce8b22e1f900 Author: Cecylia Bocovich <cohosh(a)torproject.org> Date: Thu Sep 9 16:01:38 2021 -0400 Update Snowflake client library documentation Follow best practices for documenting the exported pieces of the Snowflake client library. --- client/lib/interfaces.go | 17 +++++++------- client/lib/peers.go | 18 +++++++------- client/lib/rendezvous.go | 23 +++++++----------- client/lib/rendezvous_ampcache.go | 4 ++-- client/lib/rendezvous_http.go | 2 +- client/lib/rendezvous_test.go | 4 ++-- client/lib/snowflake.go | 49 ++++++++++++++++++++++++++++++++------- client/lib/webrtc.go | 16 ++++++++----- 8 files changed, 83 insertions(+), 50 deletions(-) diff --git a/client/lib/interfaces.go b/client/lib/interfaces.go index 0a871b8..66b9d56 100644 --- a/client/lib/interfaces.go +++ b/client/lib/interfaces.go @@ -1,23 +1,24 @@ package lib -// Interface for catching Snowflakes. (aka the remote dialer) +// Tongue is an interface for catching Snowflakes. (aka the remote dialer) type Tongue interface { + // Catch makes a connection to a new snowflake. Catch() (*WebRTCPeer, error) - // Get the maximum number of snowflakes + // GetMax returns the maximum number of snowflakes a client can have. GetMax() int } -// Interface for collecting some number of Snowflakes, for passing along -// ultimately to the SOCKS handler. +// SnowflakeCollector is an interface for managing a client's collection of snowflakes. type SnowflakeCollector interface { - // Add a Snowflake to the collection. - // Implementation should decide how to connect and maintain the webRTCConn. + // Collect adds a snowflake to the collection. + // The implementation of Collect should decide how to connect to and maintain + // the connection to the WebRTCPeer. Collect() (*WebRTCPeer, error) - // Remove and return the most available Snowflake from the collection. + // Pop removes and returns the most available snowflake from the collection. Pop() *WebRTCPeer - // Signal when the collector has stopped collecting. + // Melted returns a channel that will signal when the collector has stopped. Melted() <-chan struct{} } diff --git a/client/lib/peers.go b/client/lib/peers.go index 1abcd95..6bddbf2 100644 --- a/client/lib/peers.go +++ b/client/lib/peers.go @@ -8,7 +8,7 @@ import ( "sync" ) -// Container which keeps track of multiple WebRTC remote peers. +// Peers is a container that keeps track of multiple WebRTC remote peers. // Implements |SnowflakeCollector|. // // Maintaining a set of pre-connected Peers with fresh but inactive datachannels @@ -31,7 +31,7 @@ type Peers struct { collectLock sync.Mutex } -// Construct a fresh container of remote peers. +// NewPeers constructs a fresh container of remote peers. func NewPeers(tongue Tongue) (*Peers, error) { p := &Peers{} // Use buffered go channel to pass snowflakes onwards to the SOCKS handler. @@ -45,7 +45,7 @@ func NewPeers(tongue Tongue) (*Peers, error) { return p, nil } -// As part of |SnowflakeCollector| interface. +// Collect connects to and adds a new remote peer as part of |SnowflakeCollector| interface. func (p *Peers) Collect() (*WebRTCPeer, error) { // Engage the Snowflake Catching interface, which must be available. p.collectLock.Lock() @@ -76,8 +76,8 @@ func (p *Peers) Collect() (*WebRTCPeer, error) { return connection, nil } -// Pop blocks until an available, valid snowflake appears. Returns nil after End -// has been called. +// Pop blocks until an available, valid snowflake appears. +// Pop will return nil after End has been called. func (p *Peers) Pop() *WebRTCPeer { for { snowflake, ok := <-p.snowflakeChan @@ -93,12 +93,13 @@ func (p *Peers) Pop() *WebRTCPeer { } } -// As part of |SnowflakeCollector| interface. +// Melted returns a channel that will close when peers stop being collected. +// Melted is a necessary part of |SnowflakeCollector| interface. func (p *Peers) Melted() <-chan struct{} { return p.melt } -// Returns total available Snowflakes (including the active one) +// Count returns the total available Snowflakes (including the active ones) // The count only reduces when connections themselves close, rather than when // they are popped. func (p *Peers) Count() int { @@ -118,7 +119,8 @@ func (p *Peers) purgeClosedPeers() { } } -// Close all Peers contained here. +// End closes all active connections to Peers contained here, and stops the +// collection of future Peers. func (p *Peers) End() { close(p.melt) p.collectLock.Lock() diff --git a/client/lib/rendezvous.go b/client/lib/rendezvous.go index cf67e09..689e51c 100644 --- a/client/lib/rendezvous.go +++ b/client/lib/rendezvous.go @@ -1,10 +1,5 @@ // WebRTC rendezvous requires the exchange of SessionDescriptions between // peers in order to establish a PeerConnection. -// -// This file contains the one method currently available to Snowflake: -// -// - Domain-fronted HTTP signaling. The Broker automatically exchange offers -// and answers between this client and some remote WebRTC proxy. package lib @@ -22,7 +17,7 @@ import ( ) const ( - BrokerErrorUnexpected string = "Unexpected error, no answer." + brokerErrorUnexpected string = "Unexpected error, no answer." readLimit = 100000 //Maximum number of bytes to be read from an HTTP response ) @@ -55,7 +50,7 @@ func createBrokerTransport() http.RoundTripper { return transport } -// Construct a new BrokerChannel, where: +// NewBrokerChannel construct a new BrokerChannel, where: // |broker| is the full URL of the facilitating program which assigns proxies // to clients, and |front| is the option fronting domain. func NewBrokerChannel(broker, ampCache, front string, keepLocalAddresses bool) (*BrokerChannel, error) { @@ -85,10 +80,8 @@ func NewBrokerChannel(broker, ampCache, front string, keepLocalAddresses bool) ( }, nil } -// Roundtrip HTTP POST using WebRTC SessionDescriptions. -// -// Send an SDP offer to the broker, which assigns a proxy and responds -// with an SDP answer from a designated remote WebRTC peer. +// Negotiate uses a RendezvousMethod to send the client's WebRTC SDP offer +// and receive a snowflake proxy WebRTC SDP answer in return. func (bc *BrokerChannel) Negotiate(offer *webrtc.SessionDescription) ( *webrtc.SessionDescription, error) { // Ideally, we could specify an `RTCIceTransportPolicy` that would handle @@ -135,6 +128,7 @@ func (bc *BrokerChannel) Negotiate(offer *webrtc.SessionDescription) ( return util.DeserializeSessionDescription(resp.Answer) } +// SetNATType sets the NAT type of the client so we can send it to the WebRTC broker. func (bc *BrokerChannel) SetNATType(NATType string) { bc.lock.Lock() bc.natType = NATType @@ -142,13 +136,14 @@ func (bc *BrokerChannel) SetNATType(NATType string) { log.Printf("NAT Type: %s", NATType) } -// Implements the |Tongue| interface to catch snowflakes, using BrokerChannel. +// WebRTCDialer implements the |Tongue| interface to catch snowflakes, using BrokerChannel. type WebRTCDialer struct { *BrokerChannel webrtcConfig *webrtc.Configuration max int } +// NewWebRTCDialer constructs a new WebRTCDialer. func NewWebRTCDialer(broker *BrokerChannel, iceServers []webrtc.ICEServer, max int) *WebRTCDialer { config := webrtc.Configuration{ ICEServers: iceServers, @@ -161,14 +156,14 @@ func NewWebRTCDialer(broker *BrokerChannel, iceServers []webrtc.ICEServer, max i } } -// Initialize a WebRTC Connection by signaling through the broker. +// Catch initializes a WebRTC Connection by signaling through the BrokerChannel. func (w WebRTCDialer) Catch() (*WebRTCPeer, error) { // TODO: [#25591] Fetch ICE server information from Broker. // TODO: [#25596] Consider TURN servers here too. return NewWebRTCPeer(w.webrtcConfig, w.BrokerChannel) } -// Returns the maximum number of snowflakes to collect +// GetMax returns the maximum number of snowflakes to collect. func (w WebRTCDialer) GetMax() int { return w.max } diff --git a/client/lib/rendezvous_ampcache.go b/client/lib/rendezvous_ampcache.go index 2f1fb9f..6ac99b3 100644 --- a/client/lib/rendezvous_ampcache.go +++ b/client/lib/rendezvous_ampcache.go @@ -93,7 +93,7 @@ func (r *ampCacheRendezvous) Exchange(encPollReq []byte) ([]byte, error) { // * If the broker returns a 5xx status, the AMP cache // translates it to a 404. // https://amp.dev/documentation/guides-and-tutorials/learn/amp-caches-and-cor… - return nil, errors.New(BrokerErrorUnexpected) + return nil, errors.New(brokerErrorUnexpected) } if _, err := resp.Location(); err == nil { // The Google AMP Cache may return a "silent redirect" with @@ -103,7 +103,7 @@ func (r *ampCacheRendezvous) Exchange(encPollReq []byte) ([]byte, error) { // follow redirects nor execute JavaScript, but in any case we // cannot extract information from this response and can only // treat it as an error. - return nil, errors.New(BrokerErrorUnexpected) + return nil, errors.New(brokerErrorUnexpected) } lr := io.LimitReader(resp.Body, readLimit+1) diff --git a/client/lib/rendezvous_http.go b/client/lib/rendezvous_http.go index e020077..43ed075 100644 --- a/client/lib/rendezvous_http.go +++ b/client/lib/rendezvous_http.go @@ -60,7 +60,7 @@ func (r *httpRendezvous) Exchange(encPollReq []byte) ([]byte, error) { log.Printf("HTTP rendezvous response: %s", resp.Status) if resp.StatusCode != http.StatusOK { - return nil, errors.New(BrokerErrorUnexpected) + return nil, errors.New(brokerErrorUnexpected) } return limitedRead(resp.Body, readLimit) diff --git a/client/lib/rendezvous_test.go b/client/lib/rendezvous_test.go index 6a1a071..4bc5766 100644 --- a/client/lib/rendezvous_test.go +++ b/client/lib/rendezvous_test.go @@ -122,7 +122,7 @@ func TestHTTPRendezvous(t *testing.T) { answer, err := rend.Exchange(fakeEncPollReq) So(err, ShouldNotBeNil) So(answer, ShouldBeNil) - So(err.Error(), ShouldResemble, BrokerErrorUnexpected) + So(err.Error(), ShouldResemble, brokerErrorUnexpected) }) Convey("httpRendezvous.Exchange fails with error", func() { @@ -243,7 +243,7 @@ func TestAMPCacheRendezvous(t *testing.T) { answer, err := rend.Exchange(fakeEncPollReq) So(err, ShouldNotBeNil) So(answer, ShouldBeNil) - So(err.Error(), ShouldResemble, BrokerErrorUnexpected) + So(err.Error(), ShouldResemble, brokerErrorUnexpected) }) Convey("ampCacheRendezvous.Exchange fails with error", func() { diff --git a/client/lib/snowflake.go b/client/lib/snowflake.go index e0591a7..4b17f0b 100644 --- a/client/lib/snowflake.go +++ b/client/lib/snowflake.go @@ -17,12 +17,21 @@ import ( ) const ( + // ReconnectTimeout is the time a Snowflake client will wait before collecting + // more snowflakes. ReconnectTimeout = 10 * time.Second + // SnowflakeTimeout is the time a Snowflake client will wait before determining that + // a remote snowflake has been disconnected. If no new messages are sent or received + // in this time period, the client will terminate the connection with the remote + // peer and collect a new snowflake. SnowflakeTimeout = 20 * time.Second - // How long to wait for the OnOpen callback on a DataChannel. + // DataChannelTimeout is how long the client will wait for the OnOpen callback + // on a newly created DataChannel. DataChannelTimeout = 10 * time.Second + // WindowSize is the number of packets in the send and receive window of a KCP connection. WindowSize = 65535 + // StreamSize controls the maximum amount of in flight data between a client and server. StreamSize = 1048576 //1MB ) @@ -37,16 +46,31 @@ type Transport struct { dialer *WebRTCDialer } +// ClientConfig defines how the SnowflakeClient will connect to the broker and Snowflake proxies. type ClientConfig struct { - BrokerURL string - AmpCacheURL string - FrontDomain string - ICEAddresses []string + // BrokerURL is the full URL of the Snowflake broker that the client will connect to. + BrokerURL string + // AmpCacheURL is the full URL of a valid AMP cache. A nonzero value indicates + // that AMP cache will be used as the rendezvous method with the broker. + AmpCacheURL string + // FrontDomain is a the full URL of an optional front domain that can be used with either + // the AMP cache or HTTP domain fronting rendezvous method. + FrontDomain string + // ICEAddresses are a slice of ICE server URLs that will be used for NAT traversal and + // the creation of the client's WebRTC SDP offer. + ICEAddresses []string + // KeepLocalAddresses is an optional setting that will prevent the removal of local or + // invalid addresses from the client's SDP offer. This is useful for local deployments + // and testing. KeepLocalAddresses bool - Max int + // Max is the maximum number of snowflake proxy peers that the client should attempt to + // connect to. + Max int } -// Create a new Snowflake transport client that can spawn multiple Snowflake connections. +// NewSnowflakeClient creates a new Snowflake transport client that can spawn multiple +// Snowflake connections. +// // brokerURL and frontDomain are the urls for the broker host and domain fronting host // iceAddresses are the STUN/TURN urls needed for WebRTC negotiation // keepLocalAddresses is a flag to enable sending local network addresses (for testing purposes) @@ -82,8 +106,10 @@ func NewSnowflakeClient(config ClientConfig) (*Transport, error) { return transport, nil } -// Create a new Snowflake connection. Starts the collection of snowflakes and returns a -// smux Stream. +// Dial creates a new Snowflake connection. +// Dial starts the collection of snowflakes and returns a SnowflakeConn that is a +// wrapper around a smux.Stream that will reliably deliver data to a Snowflake +// server through one or more snowflake proxies. func (t *Transport) Dial() (net.Conn, error) { // Cleanup functions to run before returning, in case of an error. var cleanup []func() @@ -132,10 +158,12 @@ func (t *Transport) Dial() (net.Conn, error) { return &SnowflakeConn{Stream: stream, sess: sess, pconn: pconn, snowflakes: snowflakes}, nil } +// SetRendezvousMethod sets the rendezvous method to the Snowflake broker. func (t *Transport) SetRendezvousMethod(r RendezvousMethod) { t.dialer.Rendezvous = r } +// SnowflakeConn is a reliable connection to a snowflake server that implements net.Conn. type SnowflakeConn struct { *smux.Stream sess *smux.Session @@ -143,6 +171,9 @@ type SnowflakeConn struct { snowflakes *Peers } +// Close closes the connection. +// +// The collection of snowflake proxies for this connection is stopped. func (conn *SnowflakeConn) Close() error { log.Printf("---- SnowflakeConn: closed stream %v ---", conn.ID()) conn.Stream.Close() diff --git a/client/lib/webrtc.go b/client/lib/webrtc.go index 6fc24c0..538cc8b 100644 --- a/client/lib/webrtc.go +++ b/client/lib/webrtc.go @@ -12,10 +12,9 @@ import ( "github.com/pion/webrtc/v3" ) -// Remote WebRTC peer. +// WebRTCPeer represents a WebRTC connection to a remote snowflake proxy. // -// Handles preparation of go-webrtc PeerConnection. Only ever has -// one DataChannel. +// Each WebRTCPeer only ever has one DataChannel that is used as the peer's transport. type WebRTCPeer struct { id string pc *webrtc.PeerConnection @@ -35,7 +34,11 @@ type WebRTCPeer struct { bytesLogger bytesLogger } -// Construct a WebRTC PeerConnection. +// NewWebRTCPeer constructs a WebRTC PeerConnection to a snowflake proxy. +// +// The creation of the peer handles the signaling to the Snowflake broker, including +// the exchange of SDP information, the creation of a PeerConnection, and the establishment +// of a DataChannel to the Snowflake proxy. func NewWebRTCPeer(config *webrtc.Configuration, broker *BrokerChannel) (*WebRTCPeer, error) { connection := new(WebRTCPeer) @@ -79,7 +82,7 @@ func (c *WebRTCPeer) Write(b []byte) (int, error) { return len(b), nil } -//Returns a boolean indicated whether the peer is closed +// Closed returns a boolean indicated whether the peer is closed. func (c *WebRTCPeer) Closed() bool { select { case <-c.closed: @@ -89,6 +92,7 @@ func (c *WebRTCPeer) Closed() bool { return false } +// Close closes the connection the snowflake proxy. func (c *WebRTCPeer) Close() error { c.once.Do(func() { close(c.closed) @@ -225,7 +229,7 @@ func (c *WebRTCPeer) preparePeerConnection(config *webrtc.Configuration) error { return nil } -// Close all channels and transports +// cleanup closes all channels and transports func (c *WebRTCPeer) cleanup() { // Close this side of the SOCKS pipe. if c.writePipe != nil { // c.writePipe can be nil in tests.

1 0