[Merged by Bors] - Simplify code in fetch and syncer pkgs
#5453
Conversation
poszu
force-pushed
the
refactor-fetch-request
branch
from
59dfb09 to
1eb03ff
Compare
poszu
force-pushed
the
refactor-fetch-request
branch
from
1eb03ff to
6ab4d1a
Compare
Codecov ReportAttention:
Additional details and impacted files@@ Coverage Diff @@
## develop #5453 +/- ##
========================================
Coverage 79.5% 79.6%
========================================
Files 268 268
Lines 26591 26445 -146
========================================
- Hits 21145 21054 -91
+ Misses 3917 3875 -42
+ Partials 1529 1516 -13 ☔ View full report in Codecov by Sentry. |
syncer/interface.go
Outdated
| GetMaliciousIDs(context.Context, []p2p.Peer) <-chan fetch.Result | ||
| GetLayerData(context.Context, []p2p.Peer, types.LayerID) (<-chan fetch.Result, error) | ||
| GetLayerOpinions(context.Context, []p2p.Peer, types.LayerID) (<-chan fetch.Result, error) |
There was a problem hiding this comment.
I think it would be easier to have these methods only take 1 peer as parameter and return the data of the response directly instead of a fetch.Result. This completely gets rid of poll and makes the caller side also simpler. As an example PollLayerOpinions becomes:
func (d *DataFetch) PollLayerOpinions(
ctx context.Context,
lid types.LayerID,
needCert bool,
peers []p2p.Peer,
) ([]*fetch.LayerOpinion, []*types.Certificate, error) {
logger := d.logger.WithContext(ctx).WithFields(lid)
opinions := make([]*fetch.LayerOpinion, 0, len(peers))
success := false
var combinedErr error
for _, peer := range peers {
resp, err := d.fetcher.GetLayerOpinion(ctx, peer, lid)
if err != nil {
opnsPeerError.Inc()
logger.With().Debug("received peer error for layer opinions", log.Err(err))
if !success {
combinedErr = errors.Join(combinedErr, err)
}
continue
}
logger.Debug("received layer opinions from peer")
var lo fetch.LayerOpinion
if err := codec.Decode(resp, &lo); err != nil {
logger.With().Debug("error decoding LayerOpinion", log.Err(err))
if !success {
combinedErr = errors.Join(combinedErr, err)
}
continue
}
lo.SetPeer(peer)
opinions = append(opinions, &lo)
success = true
combinedErr = nil
}
if !needCert {
return opinions, nil, combinedErr
}
certs := make([]*types.Certificate, 0, len(opinions))
peerCerts := map[types.BlockID][]p2p.Peer{}
for _, opinion := range opinions {
if opinion.Certified == nil {
continue
}
if _, ok := peerCerts[*opinion.Certified]; !ok {
peerCerts[*opinion.Certified] = []p2p.Peer{}
}
peerCerts[*opinion.Certified] = append(peerCerts[*opinion.Certified], opinion.Peer())
// note that we want to fetch block certificate for types.EmptyBlockID as well,
// but we don't need to register hash for the actual block fetching
if *opinion.Certified != types.EmptyBlockID {
d.fetcher.RegisterPeerHashes(
opinion.Peer(),
[]types.Hash32{opinion.Certified.AsHash32()},
)
}
}
for bid, bidPeers := range peerCerts {
cert, err := d.fetcher.GetCert(ctx, lid, bid, bidPeers)
if err != nil {
certPeerError.Inc()
continue
}
certs = append(certs, cert)
}
return opinions, certs, combinedErr
}It removes the need to process data from the channel. And now since all the control flow is in PollLayerOpinions we could re-add an errgroup to process all concurrently that we want to process concurrently. opinions and peerCerts could for instance become channels that are being read from while layer opinions are still being fetched from peers.
There was a problem hiding this comment.
The code from your comment is oversimplified as it makes the queries serially. In reality, you would need to add an errgroup and collect results on a channel. But it still could be simpler then the current approach, let me test it out 👍 .
There was a problem hiding this comment.
That's what I meant by
we could re-add an
errgroupto process all concurrently that we want to process concurrently.opinionsandpeerCertscould for instance become channels that are being read from while layer opinions are still being fetched from peers.
🙂
There was a problem hiding this comment.
@fasmat Updated as suggested.
|
my hope was that someone would rewrite it to use streaming both for reads and writes, rather then doing style improvements. it can be done even in backward compatible manner. for some reason it is getting avoided though. |
|
@dshulyak I plan to switch to full streaming with no intermediate buffers as a part of syncv2 work (starting with chunked streaming, but I intend to skip intermediate buffering altogether in the final version). But I'm not sure it's worth trying to switch current fetcher to "full streaming" before that's done, as many parts of the existing fetcher will no longer be necessary, and it's possible to do a simple interim solution to avoid deadline errors. Also, the |
@dshulyak I spotted a few places where code was unnecessarily convoluted or inefficient in the process of trying to understand how it works and decided to make it simpler. These are not merely "style improvements" - the code should be more efficient now. |
poszu
changed the title
fetch::requester interfacefetch and syncer pkgs
|
bors merge |
## Motivation The `requester::Request(...)` interface from the `fetch` package seems unnecessarily complicated with its callback methods. It can be made blocking instead, simplifying the code using this interface. ## Changes - make `fetch::requester::Request(...)` blocking and remove the callbacks. It returns the result directly now, - make `GetMaliciousIDs`, `GetLayerData` and `GetLayerOpinions` from `syncer::fetcher` interface blocking, - remove `fetch::poll(...)` - refactor the code and tests to use the new approach ## Test Plan existing tests pass
|
This PR was included in a batch that was canceled, it will be automatically retried |
## Motivation The `requester::Request(...)` interface from the `fetch` package seems unnecessarily complicated with its callback methods. It can be made blocking instead, simplifying the code using this interface. ## Changes - make `fetch::requester::Request(...)` blocking and remove the callbacks. It returns the result directly now, - make `GetMaliciousIDs`, `GetLayerData` and `GetLayerOpinions` from `syncer::fetcher` interface blocking, - remove `fetch::poll(...)` - refactor the code and tests to use the new approach ## Test Plan existing tests pass
|
Build failed: |
|
bors merge |
## Motivation The `requester::Request(...)` interface from the `fetch` package seems unnecessarily complicated with its callback methods. It can be made blocking instead, simplifying the code using this interface. ## Changes - make `fetch::requester::Request(...)` blocking and remove the callbacks. It returns the result directly now, - make `GetMaliciousIDs`, `GetLayerData` and `GetLayerOpinions` from `syncer::fetcher` interface blocking, - remove `fetch::poll(...)` - refactor the code and tests to use the new approach ## Test Plan existing tests pass
|
Build failed (retrying...):
|
## Motivation The `requester::Request(...)` interface from the `fetch` package seems unnecessarily complicated with its callback methods. It can be made blocking instead, simplifying the code using this interface. ## Changes - make `fetch::requester::Request(...)` blocking and remove the callbacks. It returns the result directly now, - make `GetMaliciousIDs`, `GetLayerData` and `GetLayerOpinions` from `syncer::fetcher` interface blocking, - remove `fetch::poll(...)` - refactor the code and tests to use the new approach ## Test Plan existing tests pass
|
Pull request successfully merged into develop. Build succeeded: |
spacemesh-bors
bot
changed the title
fetch and syncer pkgsfetch and syncer pkgs
## Motivation The `requester::Request(...)` interface from the `fetch` package seems unnecessarily complicated with its callback methods. It can be made blocking instead, simplifying the code using this interface. ## Changes - make `fetch::requester::Request(...)` blocking and remove the callbacks. It returns the result directly now, - make `GetMaliciousIDs`, `GetLayerData` and `GetLayerOpinions` from `syncer::fetcher` interface blocking, - remove `fetch::poll(...)` - refactor the code and tests to use the new approach ## Test Plan existing tests pass
## Motivation The `requester::Request(...)` interface from the `fetch` package seems unnecessarily complicated with its callback methods. It can be made blocking instead, simplifying the code using this interface. ## Changes - make `fetch::requester::Request(...)` blocking and remove the callbacks. It returns the result directly now, - make `GetMaliciousIDs`, `GetLayerData` and `GetLayerOpinions` from `syncer::fetcher` interface blocking, - remove `fetch::poll(...)` - refactor the code and tests to use the new approach ## Test Plan existing tests pass
## Motivation The `requester::Request(...)` interface from the `fetch` package seems unnecessarily complicated with its callback methods. It can be made blocking instead, simplifying the code using this interface. ## Changes - make `fetch::requester::Request(...)` blocking and remove the callbacks. It returns the result directly now, - make `GetMaliciousIDs`, `GetLayerData` and `GetLayerOpinions` from `syncer::fetcher` interface blocking, - remove `fetch::poll(...)` - refactor the code and tests to use the new approach ## Test Plan existing tests pass
Motivation
The
requester::Request(...)interface from thefetchpackage seems unnecessarily complicated with its callback methods. It can be made blocking instead, simplifying the code using this interface.Changes
fetch::requester::Request(...)blocking and remove the callbacks. It returns the result directly now,GetMaliciousIDs,GetLayerDataandGetLayerOpinionsfromsyncer::fetcherinterface blocking,fetch::poll(...)Test Plan
existing tests pass