[tor-commits] [tor/master] Rework origin circuit tracking to use pubsub

commit 0bce0c339d5f4c6ddeb90c21502a9716fb6cd17a Author: Taylor Yu catalyst@torproject.org Date: Thu Mar 28 17:45:49 2019 -0500

Rework origin circuit tracking to use pubsub

Part of ticket 29976. --- src/app/main/main.c | 1 + src/core/or/circuitbuild.c | 11 ++-- src/core/or/circuitlist.c | 33 +++++------ src/core/or/ocirc_event.c | 112 ++++++++++++++++++++++++----------- src/core/or/ocirc_event.h | 35 +++-------- src/feature/control/btrack.c | 9 ++- src/feature/control/btrack_circuit.c | 52 ++++++++-------- src/feature/control/btrack_circuit.h | 3 + src/feature/control/btrack_orconn.c | 19 +++--- src/test/test_btrack.c | 24 +++++--- src/test/test_circuitstats.c | 2 +- src/test/test_controller_events.c | 11 ++-- src/test/test_helpers.c | 2 + 13 files changed, 178 insertions(+), 136 deletions(-)

diff --git a/src/app/main/main.c b/src/app/main/main.c index 54f2afaa0..04a0cec19 100644 --- a/src/app/main/main.c +++ b/src/app/main/main.c @@ -1257,6 +1257,7 @@ pubsub_connect(void) * this XXXX point, using tor_mainloop_set_delivery_strategy(). */ tor_mainloop_set_delivery_strategy("orconn", DELIV_IMMEDIATE); + tor_mainloop_set_delivery_strategy("ocirc", DELIV_IMMEDIATE); } }

diff --git a/src/core/or/circuitbuild.c b/src/core/or/circuitbuild.c index 3a4e72942..ff809c01c 100644 --- a/src/core/or/circuitbuild.c +++ b/src/core/or/circuitbuild.c @@ -522,14 +522,13 @@ origin_circuit_get_guard_state(origin_circuit_t *circ) static void circuit_chan_publish(const origin_circuit_t *circ, const channel_t *chan) { - ocirc_event_msg_t msg; + ocirc_chan_msg_t *msg = tor_malloc(sizeof(*msg));

- msg.type = OCIRC_MSGTYPE_CHAN; - msg.u.chan.gid = circ->global_identifier; - msg.u.chan.chan = chan->global_identifier; - msg.u.chan.onehop = circ->build_state->onehop_tunnel; + msg->gid = circ->global_identifier; + msg->chan = chan->global_identifier; + msg->onehop = circ->build_state->onehop_tunnel;

- ocirc_event_publish(&msg); + ocirc_chan_publish(msg); }

/** Start establishing the first hop of our circuit. Figure out what diff --git a/src/core/or/circuitlist.c b/src/core/or/circuitlist.c index 72952a8a5..55f04a77b 100644 --- a/src/core/or/circuitlist.c +++ b/src/core/or/circuitlist.c @@ -496,17 +496,16 @@ int circuit_event_status(origin_circuit_t *circ, circuit_status_event_t tp, int reason_code) { - ocirc_event_msg_t msg; + ocirc_cevent_msg_t *msg = tor_malloc(sizeof(*msg));

tor_assert(circ);

- msg.type = OCIRC_MSGTYPE_CEVENT; - msg.u.cevent.gid = circ->global_identifier; - msg.u.cevent.evtype = tp; - msg.u.cevent.reason = reason_code; - msg.u.cevent.onehop = circ->build_state->onehop_tunnel; + msg->gid = circ->global_identifier; + msg->evtype = tp; + msg->reason = reason_code; + msg->onehop = circ->build_state->onehop_tunnel;

- ocirc_event_publish(&msg); + ocirc_cevent_publish(msg); return control_event_circuit_status(circ, tp, reason_code); }

@@ -514,26 +513,25 @@ circuit_event_status(origin_circuit_t *circ, circuit_status_event_t tp, * Helper function to publish a state change message * * circuit_set_state() calls this to notify subscribers about a change - * of the state of an origin circuit. + * of the state of an origin circuit. @a circ must be an origin + * circuit. **/ static void circuit_state_publish(const circuit_t *circ) { - ocirc_event_msg_t msg; + ocirc_state_msg_t *msg = tor_malloc(sizeof(*msg)); const origin_circuit_t *ocirc;

- if (!CIRCUIT_IS_ORIGIN(circ)) - return; + tor_assert(CIRCUIT_IS_ORIGIN(circ)); ocirc = CONST_TO_ORIGIN_CIRCUIT(circ); /* Only inbound OR circuits can be in this state, not origin circuits. */ tor_assert(circ->state != CIRCUIT_STATE_ONIONSKIN_PENDING);

- msg.type = OCIRC_MSGTYPE_STATE; - msg.u.state.gid = ocirc->global_identifier; - msg.u.state.state = circ->state; - msg.u.state.onehop = ocirc->build_state->onehop_tunnel; + msg->gid = ocirc->global_identifier; + msg->state = circ->state; + msg->onehop = ocirc->build_state->onehop_tunnel;

- ocirc_event_publish(&msg); + ocirc_state_publish(msg); }

/** Change the state of <b>circ</b> to <b>state</b>, adding it to or removing @@ -565,7 +563,8 @@ circuit_set_state(circuit_t *circ, uint8_t state) if (state == CIRCUIT_STATE_GUARD_WAIT || state == CIRCUIT_STATE_OPEN) tor_assert(!circ->n_chan_create_cell); circ->state = state; - circuit_state_publish(circ); + if (CIRCUIT_IS_ORIGIN(circ)) + circuit_state_publish(circ); }

/** Append to <b>out</b> all circuits in state CHAN_WAIT waiting for diff --git a/src/core/or/ocirc_event.c b/src/core/or/ocirc_event.c index 4a6fc748c..3cb914713 100644 --- a/src/core/or/ocirc_event.c +++ b/src/core/or/ocirc_event.c @@ -26,59 +26,103 @@ #include "core/or/origin_circuit_st.h" #include "lib/subsys/subsys.h"

-/** List of subscribers */ -static smartlist_t *ocirc_event_rcvrs; +DECLARE_PUBLISH(ocirc_state); +DECLARE_PUBLISH(ocirc_chan); +DECLARE_PUBLISH(ocirc_cevent); + +static void +ocirc_event_free(msg_aux_data_t u) +{ + tor_free_(u.ptr); +} + +static char * +ocirc_state_fmt(msg_aux_data_t u) +{ + ocirc_state_msg_t *msg = (ocirc_state_msg_t *)u.ptr; + char *s = NULL; + + tor_asprintf(&s, "<gid=%"PRIu32" state=%d onehop=%d>", + msg->gid, msg->state, msg->onehop); + return s; +} + +static char * +ocirc_chan_fmt(msg_aux_data_t u) +{ + ocirc_chan_msg_t *msg = (ocirc_chan_msg_t *)u.ptr; + char *s = NULL; + + tor_asprintf(&s, "<gid=%"PRIu32" chan=%"PRIu64" onehop=%d>", + msg->gid, msg->chan, msg->onehop); + return s; +} + +static char * +ocirc_cevent_fmt(msg_aux_data_t u) +{ + ocirc_cevent_msg_t *msg = (ocirc_cevent_msg_t *)u.ptr; + char *s = NULL; + + tor_asprintf(&s, "<gid=%"PRIu32" evtype=%d reason=%d onehop=%d>", + msg->gid, msg->evtype, msg->reason, msg->onehop); + return s; +} + +static dispatch_typefns_t ocirc_state_fns = { + .free_fn = ocirc_event_free, + .fmt_fn = ocirc_state_fmt, +}; + +static dispatch_typefns_t ocirc_chan_fns = { + .free_fn = ocirc_event_free, + .fmt_fn = ocirc_chan_fmt, +}; + +static dispatch_typefns_t ocirc_cevent_fns = { + .free_fn = ocirc_event_free, + .fmt_fn = ocirc_cevent_fmt, +};

-/** Initialize subscriber list */ static int -ocirc_event_init(void) +ocirc_add_pubsub(struct pubsub_connector_t *connector) { - ocirc_event_rcvrs = smartlist_new(); + if (DISPATCH_REGISTER_TYPE(connector, ocirc_state, &ocirc_state_fns)) + return -1; + if (DISPATCH_REGISTER_TYPE(connector, ocirc_chan, &ocirc_chan_fns)) + return -1; + if (DISPATCH_REGISTER_TYPE(connector, ocirc_cevent, &ocirc_cevent_fns)) + return -1; + if (DISPATCH_ADD_PUB(connector, ocirc, ocirc_state)) + return -1; + if (DISPATCH_ADD_PUB(connector, ocirc, ocirc_chan)) + return -1; + if (DISPATCH_ADD_PUB(connector, ocirc, ocirc_cevent)) + return -1; return 0; }

-/** Free subscriber list */ -static void -ocirc_event_fini(void) +void +ocirc_state_publish(ocirc_state_msg_t *msg) { - smartlist_free(ocirc_event_rcvrs); + PUBLISH(ocirc_state, msg); }

-/** - * Subscribe to messages about origin circuit events - * - * Register a callback function to receive messages about origin - * circuits. The publisher calls this function synchronously. - **/ void -ocirc_event_subscribe(ocirc_event_rcvr_t fn) +ocirc_chan_publish(ocirc_chan_msg_t *msg) { - tor_assert(fn); - /* Don't duplicate subscriptions. */ - if (smartlist_contains(ocirc_event_rcvrs, fn)) - return; - - smartlist_add(ocirc_event_rcvrs, fn); + PUBLISH(ocirc_chan, msg); }

-/** - * Publish a message about OR connection events - * - * This calls the subscriber receiver function synchronously. - **/ void -ocirc_event_publish(const ocirc_event_msg_t *msg) +ocirc_cevent_publish(ocirc_cevent_msg_t *msg) { - SMARTLIST_FOREACH_BEGIN(ocirc_event_rcvrs, ocirc_event_rcvr_t, fn) { - tor_assert(fn); - (*fn)(msg); - } SMARTLIST_FOREACH_END(fn); + PUBLISH(ocirc_cevent, msg); }

const subsys_fns_t sys_ocirc_event = { .name = "ocirc_event", .supported = true, .level = -32, - .initialize = ocirc_event_init, - .shutdown = ocirc_event_fini, + .add_pubsub = ocirc_add_pubsub, }; diff --git a/src/core/or/ocirc_event.h b/src/core/or/ocirc_event.h index 0b125c289..ad6d69ffb 100644 --- a/src/core/or/ocirc_event.h +++ b/src/core/or/ocirc_event.h @@ -12,6 +12,7 @@ #include <stdbool.h>

#include "lib/cc/torint.h" +#include "lib/pubsub/pubsub.h"

/** Used to indicate the type of a circuit event passed to the controller. * The various types are defined in control-spec.txt */ @@ -30,6 +31,8 @@ typedef struct ocirc_state_msg_t { bool onehop; /**< one-hop circuit? */ } ocirc_state_msg_t;

+DECLARE_MESSAGE(ocirc_state, ocirc_state, ocirc_state_msg_t *); + /** * Message when a channel gets associated to a circuit. * @@ -44,6 +47,8 @@ typedef struct ocirc_chan_msg_t { bool onehop; /**< one-hop circuit? */ } ocirc_chan_msg_t;

+DECLARE_MESSAGE(ocirc_chan, ocirc_chan, ocirc_chan_msg_t *); + /** * Message for origin circuit status event * @@ -56,34 +61,12 @@ typedef struct ocirc_cevent_msg_t { bool onehop; /**< one-hop circuit? */ } ocirc_cevent_msg_t;

-/** Discriminant values for origin circuit event message */ -typedef enum ocirc_msgtype_t { - OCIRC_MSGTYPE_STATE, - OCIRC_MSGTYPE_CHAN, - OCIRC_MSGTYPE_CEVENT, -} ocirc_msgtype_t; - -/** Discriminated union for the actual message */ -typedef struct ocirc_event_msg_t { - int type; - union { - ocirc_state_msg_t state; - ocirc_chan_msg_t chan; - ocirc_cevent_msg_t cevent; - } u; -} ocirc_event_msg_t; - -/** - * Receiver function pointer for origin circuit subscribers - * - * This function gets called synchronously by the publisher. - **/ -typedef void (*ocirc_event_rcvr_t)(const ocirc_event_msg_t *); - -void ocirc_event_subscribe(ocirc_event_rcvr_t fn); +DECLARE_MESSAGE(ocirc_cevent, ocirc_cevent, ocirc_cevent_msg_t *);

#ifdef OCIRC_EVENT_PRIVATE -void ocirc_event_publish(const ocirc_event_msg_t *msg); +void ocirc_state_publish(ocirc_state_msg_t *msg); +void ocirc_chan_publish(ocirc_chan_msg_t *msg); +void ocirc_cevent_publish(ocirc_cevent_msg_t *msg); #endif

#endif /* defined(TOR_OCIRC_EVENT_STATE_H) */ diff --git a/src/feature/control/btrack.c b/src/feature/control/btrack.c index 3a6ae0788..3ce97dc85 100644 --- a/src/feature/control/btrack.c +++ b/src/feature/control/btrack.c @@ -32,8 +32,6 @@ btrack_init(void) { if (btrack_orconn_init()) return -1; - if (btrack_circ_init()) - return -1;

return 0; } @@ -48,7 +46,12 @@ btrack_fini(void) static int btrack_add_pubsub(pubsub_connector_t *connector) { - return btrack_orconn_add_pubsub(connector); + if (btrack_orconn_add_pubsub(connector)) + return -1; + if (btrack_circ_add_pubsub(connector)) + return -1; + + return 0; }

const subsys_fns_t sys_btrack = { diff --git a/src/feature/control/btrack_circuit.c b/src/feature/control/btrack_circuit.c index dcee9e460..2980c77dd 100644 --- a/src/feature/control/btrack_circuit.c +++ b/src/feature/control/btrack_circuit.c @@ -109,51 +109,53 @@ btc_update_evtype(const ocirc_cevent_msg_t *msg, btc_best_t *best, return false; }

+DECLARE_SUBSCRIBE(ocirc_state, btc_state_rcvr); +DECLARE_SUBSCRIBE(ocirc_cevent, btc_cevent_rcvr); +DECLARE_SUBSCRIBE(ocirc_chan, btc_chan_rcvr); + static void -btc_state_rcvr(const ocirc_state_msg_t *msg) +btc_state_rcvr(const msg_t *msg, const ocirc_state_msg_t *arg) { + (void)msg; log_debug(LD_BTRACK, "CIRC gid=%"PRIu32" state=%d onehop=%d", - msg->gid, msg->state, msg->onehop); + arg->gid, arg->state, arg->onehop);

- btc_update_state(msg, &best_any_state, "ANY"); - if (msg->onehop) + btc_update_state(arg, &best_any_state, "ANY"); + if (arg->onehop) return; - btc_update_state(msg, &best_ap_state, "AP"); + btc_update_state(arg, &best_ap_state, "AP"); }

static void -btc_cevent_rcvr(const ocirc_cevent_msg_t *msg) +btc_cevent_rcvr(const msg_t *msg, const ocirc_cevent_msg_t *arg) { + (void)msg; log_debug(LD_BTRACK, "CIRC gid=%"PRIu32" evtype=%d reason=%d onehop=%d", - msg->gid, msg->evtype, msg->reason, msg->onehop); + arg->gid, arg->evtype, arg->reason, arg->onehop);

- btc_update_evtype(msg, &best_any_evtype, "ANY"); - if (msg->onehop) + btc_update_evtype(arg, &best_any_evtype, "ANY"); + if (arg->onehop) return; - btc_update_evtype(msg, &best_ap_evtype, "AP"); + btc_update_evtype(arg, &best_ap_evtype, "AP"); }

static void -btc_event_rcvr(const ocirc_event_msg_t *msg) +btc_chan_rcvr(const msg_t *msg, const ocirc_chan_msg_t *arg) { - switch (msg->type) { - case OCIRC_MSGTYPE_STATE: - return btc_state_rcvr(&msg->u.state); - case OCIRC_MSGTYPE_CHAN: - log_debug(LD_BTRACK, "CIRC gid=%"PRIu32" chan=%"PRIu64" onehop=%d", - msg->u.chan.gid, msg->u.chan.chan, msg->u.chan.onehop); - break; - case OCIRC_MSGTYPE_CEVENT: - return btc_cevent_rcvr(&msg->u.cevent); - default: - break; - } + (void)msg; + log_debug(LD_BTRACK, "CIRC gid=%"PRIu32" chan=%"PRIu64" onehop=%d", + arg->gid, arg->chan, arg->onehop); }

int -btrack_circ_init(void) +btrack_circ_add_pubsub(pubsub_connector_t *connector) { - ocirc_event_subscribe(btc_event_rcvr); + if (DISPATCH_ADD_SUB(connector, ocirc, ocirc_chan)) + return -1; + if (DISPATCH_ADD_SUB(connector, ocirc, ocirc_cevent)) + return -1; + if (DISPATCH_ADD_SUB(connector, ocirc, ocirc_state)) + return -1; return 0; }

diff --git a/src/feature/control/btrack_circuit.h b/src/feature/control/btrack_circuit.h index c40822f1f..b2ae6484f 100644 --- a/src/feature/control/btrack_circuit.h +++ b/src/feature/control/btrack_circuit.h @@ -9,7 +9,10 @@ #ifndef TOR_BTRACK_CIRCUIT_H #define TOR_BTRACK_CIRCUIT_H

+#include "lib/pubsub/pubsub.h" + int btrack_circ_init(void); void btrack_circ_fini(void); +int btrack_circ_add_pubsub(pubsub_connector_t *);

#endif /* defined(TOR_BTRACK_CIRCUIT_H) */ diff --git a/src/feature/control/btrack_orconn.c b/src/feature/control/btrack_orconn.c index cbeb7b4ff..922b542a0 100644 --- a/src/feature/control/btrack_orconn.c +++ b/src/feature/control/btrack_orconn.c @@ -49,6 +49,7 @@

DECLARE_SUBSCRIBE(orconn_state, bto_state_rcvr); DECLARE_SUBSCRIBE(orconn_status, bto_status_rcvr); +DECLARE_SUBSCRIBE(ocirc_chan, bto_chan_rcvr);

/** Pair of a best ORCONN GID and with its state */ typedef struct bto_best_t { @@ -155,21 +156,18 @@ bto_status_rcvr(const msg_t *msg, const orconn_status_msg_t *arg) * and whether it's a one-hop circuit. **/ static void -bto_chan_rcvr(const ocirc_event_msg_t *msg) +bto_chan_rcvr(const msg_t *msg, const ocirc_chan_msg_t *arg) { bt_orconn_t *bto;

- /* Ignore other kinds of origin circuit events; we don't need them */ - if (msg->type != OCIRC_MSGTYPE_CHAN) - return; - - bto = bto_find_or_new(0, msg->u.chan.chan); - if (!bto->is_orig || (bto->is_onehop && !msg->u.chan.onehop)) { + (void)msg; + bto = bto_find_or_new(0, arg->chan); + if (!bto->is_orig || (bto->is_onehop && !arg->onehop)) { log_debug(LD_BTRACK, "ORCONN LAUNCH chan=%"PRIu64" onehop=%d", - msg->u.chan.chan, msg->u.chan.onehop); + arg->chan, arg->onehop); } bto->is_orig = true; - if (!msg->u.chan.onehop) + if (!arg->onehop) bto->is_onehop = false; bto_update_bests(bto); } @@ -182,7 +180,6 @@ int btrack_orconn_init(void) { bto_init_maps(); - ocirc_event_subscribe(bto_chan_rcvr);

return 0; } @@ -194,6 +191,8 @@ btrack_orconn_add_pubsub(pubsub_connector_t *connector) return -1; if (DISPATCH_ADD_SUB(connector, orconn, orconn_status)) return -1; + if (DISPATCH_ADD_SUB(connector, ocirc, ocirc_chan)) + return -1; return 0; }

diff --git a/src/test/test_btrack.c b/src/test/test_btrack.c index fef1da484..9e5d0d072 100644 --- a/src/test/test_btrack.c +++ b/src/test/test_btrack.c @@ -31,10 +31,19 @@ send_status(const orconn_status_msg_t *msg_in) }

static void +send_chan(const ocirc_chan_msg_t *msg_in) +{ + ocirc_chan_msg_t *msg = tor_malloc(sizeof(*msg)); + + *msg = *msg_in; + ocirc_chan_publish(msg); +} + +static void test_btrack_launch(void *arg) { orconn_state_msg_t conn; - ocirc_event_msg_t circ; + ocirc_chan_msg_t circ;

(void)arg; conn.gid = 1; @@ -48,12 +57,11 @@ test_btrack_launch(void *arg) expect_no_log_msg_containing("ORCONN BEST_"); teardown_capture_of_logs();

- circ.type = OCIRC_MSGTYPE_CHAN; - circ.u.chan.chan = 1; - circ.u.chan.onehop = true; + circ.chan = 1; + circ.onehop = true;

setup_full_capture_of_logs(LOG_DEBUG); - ocirc_event_publish(&circ); + send_chan(&circ); expect_log_msg_containing("ORCONN LAUNCH chan=1 onehop=1"); expect_log_msg_containing("ORCONN BEST_ANY state -1->1 gid=1"); teardown_capture_of_logs(); @@ -67,11 +75,11 @@ test_btrack_launch(void *arg) expect_no_log_msg_containing("ORCONN BEST_"); teardown_capture_of_logs();

- circ.u.chan.chan = 2; - circ.u.chan.onehop = false; + circ.chan = 2; + circ.onehop = false;

setup_full_capture_of_logs(LOG_DEBUG); - ocirc_event_publish(&circ); + send_chan(&circ); expect_log_msg_containing("ORCONN LAUNCH chan=2 onehop=0"); expect_log_msg_containing("ORCONN BEST_AP state -1->1 gid=2"); teardown_capture_of_logs(); diff --git a/src/test/test_circuitstats.c b/src/test/test_circuitstats.c index 2a09622f0..9bfaabeb2 100644 --- a/src/test/test_circuitstats.c +++ b/src/test/test_circuitstats.c @@ -197,7 +197,7 @@ test_circuitstats_hoplen(void *arg) }

#define TEST_CIRCUITSTATS(name, flags) \ - { #name, test_##name, (flags), NULL, NULL } + { #name, test_##name, (flags), &helper_pubsub_setup, NULL }

struct testcase_t circuitstats_tests[] = { TEST_CIRCUITSTATS(circuitstats_hoplen, TT_FORK), diff --git a/src/test/test_controller_events.c b/src/test/test_controller_events.c index 14fe4fd66..a8967bba5 100644 --- a/src/test/test_controller_events.c +++ b/src/test/test_controller_events.c @@ -417,13 +417,12 @@ send_orconn_state(const orconn_state_msg_t *msg_in, uint8_t state) static void send_ocirc_chan(uint32_t gid, uint64_t chan, bool onehop) { - ocirc_event_msg_t msg; + ocirc_chan_msg_t *msg = tor_malloc(sizeof(*msg));

- msg.type = OCIRC_MSGTYPE_CHAN; - msg.u.chan.gid = gid; - msg.u.chan.chan = chan; - msg.u.chan.onehop = onehop; - ocirc_event_publish(&msg); + msg->gid = gid; + msg->chan = chan; + msg->onehop = onehop; + ocirc_chan_publish(msg); }

static void diff --git a/src/test/test_helpers.c b/src/test/test_helpers.c index e856dc6cc..b4389f2d1 100644 --- a/src/test/test_helpers.c +++ b/src/test/test_helpers.c @@ -338,6 +338,8 @@ helper_setup_pubsub(const struct testcase_t *testcase) dispatcher = pubsub_builder_finalize(builder, NULL); tor_assert(dispatcher); dispatch_set_alert_fn(dispatcher, chan, alertfn_immediate, NULL); + chan = get_channel_id("ocirc"); + dispatch_set_alert_fn(dispatcher, chan, alertfn_immediate, NULL); return dispatcher; }