Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Merged by Bors] - fetch: check if hash peer is connected before using it #4939

Closed
wants to merge 6 commits into from
Closed

[Merged by Bors] - fetch: check if hash peer is connected before using it #4939

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1d4c8e5
use connected peer
countvonzero Aug 31, 2023
2add05e
break if found
countvonzero Aug 31, 2023
265f827
check connection instead
countvonzero Aug 31, 2023
6f16791
add test
countvonzero Aug 31, 2023
58648df
Merge branch 'develop' into fetch-use-connected-peer
countvonzero Aug 31, 2023
f1ee5cd
add CHANGELOG
countvonzero Aug 31, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 7 additions & 15 deletions fetch/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"sync"

lru "github.com/hashicorp/golang-lru/v2"
"golang.org/x/exp/maps"

"github.com/spacemeshos/go-spacemesh/common/types"
"github.com/spacemeshos/go-spacemesh/datastore"
Expand Down Expand Up @@ -71,24 +72,15 @@ func (hpc *HashPeersCache) Add(hash types.Hash32, peer p2p.Peer) {
hpc.add(hash, peer)
}

// GetRandom returns a random peer for a given hash.
func (hpc *HashPeersCache) GetRandom(hash types.Hash32, hint datastore.Hint, rng *rand.Rand) (p2p.Peer, bool) {
// GetRandom returns a randomized list of peers for a given hash.
func (hpc *HashPeersCache) GetRandom(hash types.Hash32, hint datastore.Hint, rng *rand.Rand) []p2p.Peer {
hpc.mu.Lock()
defer hpc.mu.Unlock()

hashPeersMap, exists := hpc.getWithStats(hash, hint)
if !exists {
return p2p.NoPeer, false
}
n := rng.Intn(len(hashPeersMap)) + 1
i := 0
for peer := range hashPeersMap {
i++
if i == n {
return peer, true
}
}
return p2p.NoPeer, false
pm, _ := hpc.getWithStats(hash, hint)
peers := maps.Keys(pm)
rng.Shuffle(len(peers), func(i, j int) { peers[i], peers[j] = peers[j], peers[i] })
Comment on lines +81 to +82
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maps.Keys extracts the keys from a map by ranging over it. range for maps doesn't produce a deterministic order:

https://go.dev/play/p/nImOnOek6c3

So shuffling might not be necessary.

return peers
}

// RegisterPeerHashes registers provided peer for a list of hashes.
Expand Down
15 changes: 6 additions & 9 deletions fetch/cache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,9 +94,8 @@ func TestGetRandom(t *testing.T) {
}()
wg.Wait()
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
peer, exists := cache.GetRandom(hash, datastore.TXDB, rng)
require.Equal(t, true, exists)
require.Contains(t, []p2p.Peer{peer1, peer2, peer3}, peer)
peers := cache.GetRandom(hash, datastore.TXDB, rng)
require.ElementsMatch(t, []p2p.Peer{peer1, peer2, peer3}, peers)
})
t.Run("2Hashes1Peer", func(t *testing.T) {
cache := NewHashPeersCache(10)
Expand All @@ -115,12 +114,10 @@ func TestGetRandom(t *testing.T) {
}()
wg.Wait()
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
randomPeer, exists := cache.GetRandom(hash1, datastore.TXDB, rng)
require.Equal(t, true, exists)
require.Equal(t, peer, randomPeer)
randomPeer, exists = cache.GetRandom(hash2, datastore.TXDB, rng)
require.Equal(t, true, exists)
require.Equal(t, peer, randomPeer)
randomPeers := cache.GetRandom(hash1, datastore.TXDB, rng)
require.Equal(t, []p2p.Peer{peer}, randomPeers)
randomPeers = cache.GetRandom(hash2, datastore.TXDB, rng)
require.Equal(t, []p2p.Peer{peer}, randomPeers)
})
}

Expand Down
24 changes: 17 additions & 7 deletions fetch/fetch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -482,18 +482,28 @@ func (f *Fetch) organizeRequests(requests []RequestMessage) map[p2p.Peer][][]Req
}
return nil
}
pm := map[p2p.Peer]struct{}{}
for _, p := range peers {
pm[p] = struct{}{}
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think it is better to ask host.Connected or something like that. creating a map from peer list on every request is not ideal

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks. will add a method that checks fh.Network().Connectedness(p) == network.Connected


for _, req := range requests {
p, exists := f.hashToPeers.GetRandom(req.Hash, req.Hint, rng)
if !exists {
p = randomPeer(peers)
target := p2p.NoPeer
hashPeers := f.hashToPeers.GetRandom(req.Hash, req.Hint, rng)
for _, p := range hashPeers {
if _, ok := pm[p]; ok {
target = p
break
}
}

_, ok := peer2requests[p]
if target == p2p.NoPeer {
target = randomPeer(peers)
}
_, ok := peer2requests[target]
if !ok {
peer2requests[p] = []RequestMessage{req}
peer2requests[target] = []RequestMessage{req}
} else {
peer2requests[p] = append(peer2requests[p], req)
peer2requests[target] = append(peer2requests[target], req)
}
}

Expand Down
35 changes: 35 additions & 0 deletions fetch/fetch_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,41 @@ func TestFetch_GetHash(t *testing.T) {
require.NotEqual(t, p1.completed, p2.completed)
}

func TestFetch_GetHashPeerNotConnected(t *testing.T) {
f := createFetch(t)
f.cfg.MaxRetriesForRequest = 0
f.cfg.MaxRetriesForPeer = 0
peer := p2p.Peer("buddy")
f.mh.EXPECT().GetPeers().Return([]p2p.Peer{peer})
f.mh.EXPECT().ID().Return(p2p.Peer("self"))
hsh := types.RandomHash()
f.RegisterPeerHashes("notConnected", []types.Hash32{hsh})

res := ResponseMessage{
Hash: hsh,
Data: []byte("a"),
}
f.mHashS.EXPECT().Request(gomock.Any(), peer, gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
func(_ context.Context, _ p2p.Peer, req []byte, okFunc func([]byte), _ func(error)) error {
var rb RequestBatch
err := codec.Decode(req, &rb)
require.NoError(t, err)
resBatch := ResponseBatch{
ID: rb.ID,
Responses: []ResponseMessage{res},
}
bts, err := codec.Encode(&resBatch)
require.NoError(t, err)
okFunc(bts)
return nil
})

p, err := f.getHash(context.TODO(), hsh, datastore.BlockDB, goodReceiver)
require.NoError(t, err)
f.requestHashBatchFromPeers()
<-p.completed
}

func TestFetch_RequestHashBatchFromPeers(t *testing.T) {
tt := []struct {
name string
Expand Down