Update PubSub and include topic filter #7496
Conversation
There was a problem hiding this comment.
Main concerns are that this PR introduces another state initialization watcher, and it is unclear why it is actually needed.
Ex: What we use now.
prysm/beacon-chain/p2p/service.go
Line 370 in 06d16a2
beacon-chain/p2p/pubsub_filter.go
Outdated
| } | ||
|
|
||
| // Monitor the state feed notifier for the state initialization event. | ||
| func (sf *subscriptionFilter) monitorStateInitialized() { |
There was a problem hiding this comment.
why does the filter need its own state initialization watcher ?
There was a problem hiding this comment.
We need genesis time and genesis validator root to determine the fork digest. I'll add a comment.
There was a problem hiding this comment.
Removed in favor of using the existing state init watcher
Resolved issues in the following files via DeepSource Autofix: 1. beacon-chain/p2p/pubsub_filter.go 2. beacon-chain/p2p/pubsub_filter_test.go
|
|
|
Ready for review, but please do not merge until I can run this for at least an hour at runtime. |
| msn.feedLock.Lock() | ||
| defer msn.feedLock.Unlock() |
There was a problem hiding this comment.
Added this to prevent potential race condition on accessing the state feed mock in test.
| @@ -69,6 +69,8 @@ func (s *Service) PublishToTopic(ctx context.Context, topic string, data []byte, | |||
|
|
|||
| // SubscribeToTopic joins (if necessary) and subscribes to PubSub topic. | |||
| func (s *Service) SubscribeToTopic(topic string, opts ...pubsub.SubOpt) (*pubsub.Subscription, error) { | |||
| s.awaitStateInitialized() // Genesis time and genesis validator root are required to subscribe. | |||
There was a problem hiding this comment.
This isn't the correct place, this will end up blocking subscription of new topics.
There was a problem hiding this comment.
https://github.com/prysmaticlabs/prysm/blob/master/beacon-chain/sync/service.go#L267
This will only call it after we receive the state initialized and synced events.
https://github.com/prysmaticlabs/prysm/blob/master/beacon-chain/p2p/service.go#L377
This will end up being blocked here as no more state initialized events are emmited now.
There was a problem hiding this comment.
This solves the race condition and yes it should block subscription of new topics until the state is initialized.
Note: If awaitStateInitialized() is called and the state is already initialized then it returns immediately.
|
Thanks @nisdas, I will refactor the pubsub filter to be a pointer receiver of the p2p service such that it can share the existing state initialization subscriber. Discussed offline that we should still keep this subscription check to awaitStateInitialization since subscribe will indiscriminately fail if the p2p service isn't aware of the genesis time or genesis validator root for fork digest. Link to discussion on discord: https://discordapp.com/channels/476244492043812875/483017808658169866/765763343029895178 |
| } | ||
| if parts[2] != fmt.Sprintf("%x", fd) { | ||
| return false | ||
| } |
There was a problem hiding this comment.
maybe a small comment on why we skip parts[3]
What type of PR is this?
Other
What does this PR do? Why is it needed?
This PR updates pubsub to the latest release and introduces a pubsub topic filter to improve pubsub peer management.
Which issues(s) does this PR fix?
Other notes for review
The basic idea of the pubsub topic filter is that our peers will announce what topics they are interested in or current subscribed to and we keep this information in memory in the libp2p pubsub implementation. A key benefit to the topic filter is that Prysm doesn't need to keep track of topics it will never subscribe since this in-memory storage of peer's topics is to facilitate topic subscriptions when Prysm does want to subscribe to some topic.