diff --git a/src/app/main/main.c b/src/app/main/main.c
index 54f2afaa01..04a0cec19d 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 3a4e729429..ff809c01cf 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 72952a8a52..55f04a77b9 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 circ to state, 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 out 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 4a6fc748c9..3cb9147134 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);
-/** 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, "",
+ 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, "",
+ 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, "",
+ 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, ô_state_fns))
+ return -1;
+ if (DISPATCH_REGISTER_TYPE(connector, ocirc_chan, ô_chan_fns))
+ return -1;
+ if (DISPATCH_REGISTER_TYPE(connector, ocirc_cevent, ô_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 0b125c2898..ad6d69ffbe 100644
--- a/src/core/or/ocirc_event.h
+++ b/src/core/or/ocirc_event.h
@@ -12,6 +12,7 @@
#include
#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 3a6ae07881..3ce97dc855 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 dcee9e460e..2980c77ddc 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 c40822f1f1..b2ae6484f0 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 cbeb7b4ff1..922b542a0c 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 fef1da4842..9e5d0d0723 100644
--- a/src/test/test_btrack.c
+++ b/src/test/test_btrack.c
@@ -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();
diff --git a/src/test/test_circuitstats.c b/src/test/test_circuitstats.c
index 2a09622f09..9bfaabeb2f 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 14fe4fd661..a8967bba50 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 e856dc6cca..b4389f2d17 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;
}