swarm/network/stream: RPC to request subscription info #18284
Conversation
holisticode
force-pushed
the
sync-monitor
branch
from
ffca572
to
c41721d
Compare
| //arbitrarily set to 16 | ||
| nodeCount := 16 | ||
| //set the syncUpdateDelay for sync registrations to start | ||
| syncUpdateDelay := 500 * time.Millisecond |
There was a problem hiding this comment.
Why delay and healthy-check?
There was a problem hiding this comment.
The health check is about the kademlia's health. It will be commented out/removed for now until we have it fixed.
The syncUpdateDelay is a streamer's internal delay time after which it starts actual syncing.
It is relevant to this test in so much as we need to wait at least this delay + the syncing duration time before we actually can check anything.
and
There was a problem hiding this comment.
Can't this also be done with messageevents?
There was a problem hiding this comment.
What do you mean?
There is no streamingStarted event or something like that, and the SyncUpdateDelay is part of the main code, so we need to set it to something and wait for it.
There was a problem hiding this comment.
Like also mentioned herer #18284 (comment) what I mean is then the streaming starts that streaming itself is actually devp2p messages being exchanged, no? Those messages could be inspected, I think;
- is it a syncSomethingType
- is it from a peer that hasn't sent this type of message yet
Not sure if I'm making myself clear...
There was a problem hiding this comment.
Yes I already updated the PR to wait for SubscribeMsg
There was a problem hiding this comment.
@nolash can you pls mark this as resolved if you agree or comment further if you think it still needs attention. Thanks
There was a problem hiding this comment.
The syncUpdateDelay is a streamer's internal delay time after which it starts actual syncing.
In such cases we should always get the value directly from the source where it's defined IMO.
There was a problem hiding this comment.
@nolash in this case no. If no SyncUpdateDelay is provided, the streamer will set it to 15 * time.Second, which is obviously too long for a test.
Thus we need to set it, which is why it is set to 500 * time.Millisecond for this test.
Check this #18284 (comment) comment please, that has the relevant snippet
| //arbitrarily set to 16 | ||
| nodeCount := 16 | ||
| //set the syncUpdateDelay for sync registrations to start | ||
| syncUpdateDelay := 500 * time.Millisecond |
There was a problem hiding this comment.
The syncUpdateDelay is a streamer's internal delay time after which it starts actual syncing.
In such cases we should always get the value directly from the source where it's defined IMO.
| msgs := sim.PeerEvents( | ||
| context.Background(), | ||
| sim.NodeIDs(), | ||
| simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("stream").MsgCode(4), //4 is SubscribeMsg |
There was a problem hiding this comment.
Ohh, dangerous with the literal 4.
In fact, if this is the only way to get a message type then the simulation framework has skipped the functionality enabled in the p2p/protocols layer where you can get the message code from the struct type.
There was a problem hiding this comment.
I changed it to get the code via spec.
As a side note though, our tests are littered with tests using hardcoded message codes (see this very same test file, all other codes are literals).
There was a problem hiding this comment.
Let's make an issue for this!
| } | ||
| log.Debug("Expected message count: ", "expectedMsgCount", expectedMsgCount) | ||
| //wait until all subscriptions are done | ||
| <-allSubscriptionsDone |
There was a problem hiding this comment.
catch the context timeout in a select here?
| t.Fatal(err) | ||
| } | ||
| //length of the subscriptions can not be smaller than number of peers | ||
| if len(pstreams) < len(registry.peers) { |
There was a problem hiding this comment.
Actually, are we absolutely sure here that the message is not only sent but the change also processed at this point, resulting in the change in the streamer state?
There was a problem hiding this comment.
At this point we are sure that the message has been received. There is no way currently to make sure that a message has been processed - I dare to say that none of our tests based on listening to message exchange can actually do that. To that end we would need a node-wide event propagation system.
There was a problem hiding this comment.
yes that or message acks after state change :)
holisticode
force-pushed
the
sync-monitor
branch
from
2ded8ce
to
577f8b7
Compare
fjl
changed the title
|
Reopened upstream as #18338 |
This PR adds to the debugging information available.
Over RPC, this code allows to request the
streamsyncing subscriptions and returns them as a map.It adds a simulation test which over a snapshot requests all subscriptions.
If provided with a
-printstats trueflag, it will print the results toSTDOUT