nihilist/tor
1
0
Fork 0
mirror of https://gitlab.torproject.org/tpo/core/tor.git synced 2024-09-21 05:26:20 +02:00
Code Issues Packages Projects Releases Wiki Activity

Rework origin circuit tracking to use pubsub

Browse Source 
Part of ticket 29976.
This commit is contained in:
Taylor Yu 2019-03-28 17:45:49 -05:00 committed by David Goulet
parent a8c0f4ddfe
commit 0bce0c339d
13 changed files with 179 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/app/main/main.c
View File

@ -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);
}
}

11
src/core/or/circuitbuild.c
View File

@ -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

33
src/core/or/circuitlist.c
View File

@ -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

114
src/core/or/ocirc_event.c
View File

@ -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);
/** Initialize subscriber list */
static int
ocirc_event_init(void)
static void
ocirc_event_free(msg_aux_data_t u)
{
ocirc_event_rcvrs = smartlist_new();
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,
};
static int
ocirc_add_pubsub(struct pubsub_connector_t *connector)
{
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,
};

35
src/core/or/ocirc_event.h
View File

@ -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) */

9
src/feature/control/btrack.c
View File

@ -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 = {

52
src/feature/control/btrack_circuit.c
View File

@ -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;
}

3
src/feature/control/btrack_circuit.h
View File

@ -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) */

19
src/feature/control/btrack_orconn.c
View File

@ -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;
}

24
src/test/test_btrack.c
View File

@ -30,11 +30,20 @@ send_status(const orconn_status_msg_t *msg_in)
orconn_status_publish(msg);
}
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();

2
src/test/test_circuitstats.c
View File

@ -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),

11
src/test/test_controller_events.c
View File

@ -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

2
src/test/test_helpers.c
View File

@ -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;
}