rust: Add subscription manager #281
Conversation
gasmith
force-pushed
the
gasmith/fg-10743-subscr
branch
from
1cec7f5 to
83dbb3c
Compare
gasmith
force-pushed
the
gasmith/fg-10745-ns-arc
branch
2 times, most recently
from
3ee1893 to
a10dd95
Compare
gasmith
force-pushed
the
gasmith/fg-10743-subscr
branch
4 times, most recently
from
38136ba to
66663f1
Compare
gasmith
force-pushed
the
gasmith/fg-10745-ns-arc
branch
from
a10dd95 to
763aeba
Compare
gasmith
force-pushed
the
gasmith/fg-10743-subscr
branch
from
66663f1 to
5939b94
Compare
gasmith
force-pushed
the
gasmith/fg-10745-ns-arc
branch
from
763aeba to
1f40c6b
Compare
gasmith
force-pushed
the
gasmith/fg-10743-subscr
branch
from
5939b94 to
9bd1af0
Compare
There was a problem hiding this comment.
A little hard to fully grasp how this is going to play out as we use it in practice, but seems probably ok to move forward and evolve this over time. None of my comments are really blocking but I will be curious to read your rebuttals 😄 Overall I felt like the doc comments could use a bit more fleshing out to make it clearer what problems the classes are solving (e.g. "tracking and querying channel subscriptions" feels a bit light on context)
rust/foxglove/src/subscription.rs
Outdated
| /// Current subscriptions. | ||
| subscriptions: Mutex<SubscriptionMap<K, V>>, | ||
| /// Cached map from channel to interested subscribers. | ||
| subscribers: ArcSwap<SubscriberMap<V>>, |
There was a problem hiding this comment.
It seems like it would be incorrect/racy to ever perform a store on subscribers when not holding the subscriptions mutex. Is there a way to encode that better in the type system?
Relatedly, I am a bit skeptical of where we may end up using has_subscribers() and whether (because of the lack of locking) it will be easy to accidentally introduce races by assuming that a subscriber doesn't sneak in between calls to has_subscribers() and other functions...
There was a problem hiding this comment.
It seems like it would be incorrect/racy to ever perform a store on
subscriberswhen not holding the subscriptions mutex. Is there a way to encode that better in the type system?
Nothing springs to mind.
Relatedly, I am a bit skeptical of where we may end up using
has_subscribers()and whether (because of the lack of locking) it will be easy to accidentally introduce races by assuming that a subscriber doesn't sneak in between calls tohas_subscribers()and other functions...
I'll add some comments to describe the intended use for has_subscribers(). The basic idea is to use it as a guard around channel.log, so that you don't waste time serializing messages when nobody is listening. In that situation, if has_subscribers() returns false, and a subscriber sneaks in on another thread, it's not a big deal. Those things weren't synchronized to begin with.
There's perhaps a bit more to think about in the other direction. Consider the proposed changes to Channel::log, and suppose that a sink is unregistered between the call to get_subscribers() and log_to_sinks(). We'll deliver a message to a sink that isn't expecting one.
gasmith
force-pushed
the
gasmith/fg-10743-subscr
branch
2 times, most recently
from
5a973fa to
03557f1
Compare
gasmith
force-pushed
the
gasmith/fg-10742-sink-ctx
branch
from
4b8d8de to
39471b2
Compare
gasmith
force-pushed
the
gasmith/fg-10743-subscr
branch
from
03557f1 to
4033ec7
Compare
gasmith
force-pushed
the
gasmith/fg-10743-subscr
branch
from
0be7ee2 to
2901071
Compare
|
I've simplified this change quite a bit. I realized that we probably want to do our synchronization elsewhere (like on |
| /// Has no effect if the subscriber has a global subscription. | ||
| pub fn subscribe_channels( | ||
| &mut self, | ||
| sink: Arc<dyn Sink>, |
There was a problem hiding this comment.
micro-nit: curious if there are common conventions here - it seems like taking the arc by value is requiring an extra clone on the caller's side, when we are already planning to clone internally. Would taking &Arc be a common alternative? On the other hand I think clone should be quite cheap for Arc – which raises the question of why Arc isn't Copy 🤔 🤷 These are just musings from a rust noob – feel free to ignore
There was a problem hiding this comment.
Right, that could totally be an &Arc. Let's do that.
which raises the question of why Arc isn't Copy
My guess: Even though the clone is cheap, it's not a trivial copy, because it involves incrementing a refcount. And likewise, there's a non-trivial Drop to decrement the refcount.
| } | ||
|
|
||
| /// Remove all global and per-channel subscriptions for a particular subscriber. | ||
| pub fn remove_subscriber(&mut self, sink_id: SinkId) -> bool { |
There was a problem hiding this comment.
naming nit: maybe consider whether this should be called something like unsubscribe_all? I'm also fine with remove_subscriber if you prefer it
There was a problem hiding this comment.
I think I mildly prefer remove_subscriber; I feel like unsubscribe_all might be confused for "remove a global subscription".
| self.by_channel.retain(|_, subs| { | ||
| removed |= subs.remove(&sink_id).is_some(); | ||
| !subs.is_empty() | ||
| }); |
There was a problem hiding this comment.
small optimization: if removed is already true, you can probably skip this, right? (since a sink with a global subscription shouldn't also have channel subscriptions)
| /// | ||
| /// We use a [`SmallVec`] to avoid heap allocations for lookups, when the number of concurrent | ||
| /// subscribers is reasonably small. | ||
| pub(crate) type SubscriberVec = SmallVec<[Arc<dyn Sink>; 6]>; |
| #[derive(Default)] | ||
| pub(crate) struct Subscriptions { | ||
| /// Global subscriptions (all channels). | ||
| global: HashMap<SinkId, Arc<dyn Sink>>, |
There was a problem hiding this comment.
Good idea splitting these out
Changelog
None
Related changes
Description
Updates:
Contextwith anRwLockaround channels, sinks, and subs together.ArcSwapcache - pre-computing the SmallVec doesn't make much of a difference since it's small and we have to clone it anyway.crate::context::subscriptions.This change adds a new type for tracking subscriptions of sinks to channels, but doesn't make use of it yet. That will come in a later PR.
We're planning to support two kinds of subscriptions: static subscriptions to all channels (e.g., for mcap), and dynamic subscriptions to particular channels (e.g., for websockets). We maintain these subscriptions as two separate sets. When subscriptions change, the
Contextwill be responsible for querying the subscription manager in order to regenerate theLogSinkSetfor affected channels.Part of: FG-10743