Browse Source

GetSelectionWithBias

Refs #1130
pull/1344/merge
Anton Kaliaev 7 years ago
parent
commit
cee7b5cb54
4 changed files with 231 additions and 11 deletions
  1. +1
    -0
      CHANGELOG.md
  2. +115
    -9
      p2p/pex/addrbook.go
  3. +109
    -0
      p2p/pex/addrbook_test.go
  4. +6
    -2
      p2p/pex/pex_reactor.go

+ 1
- 0
CHANGELOG.md View File

@ -31,6 +31,7 @@ BUG FIXES:
- [rpc] fix subscribing using an abci.ResponseDeliverTx tag - [rpc] fix subscribing using an abci.ResponseDeliverTx tag
IMPROVEMENTS: IMPROVEMENTS:
- [p2p] seeds respond with a bias towards good peers
- [rpc] `/tx` and `/tx_search` responses now include the transaction hash - [rpc] `/tx` and `/tx_search` responses now include the transaction hash
- [rpc] include validator power in `/status` - [rpc] include validator power in `/status`


+ 115
- 9
p2p/pex/addrbook.go View File

@ -42,15 +42,19 @@ type AddrBook interface {
NeedMoreAddrs() bool NeedMoreAddrs() bool
// Pick an address to dial // Pick an address to dial
PickAddress(newBias int) *p2p.NetAddress
PickAddress(biasTowardsNewAddrs int) *p2p.NetAddress
// Mark address // Mark address
MarkGood(*p2p.NetAddress) MarkGood(*p2p.NetAddress)
MarkAttempt(*p2p.NetAddress) MarkAttempt(*p2p.NetAddress)
MarkBad(*p2p.NetAddress) MarkBad(*p2p.NetAddress)
IsGood(*p2p.NetAddress) bool
// Send a selection of addresses to peers // Send a selection of addresses to peers
GetSelection() []*p2p.NetAddress GetSelection() []*p2p.NetAddress
// Send a selection of addresses with bias
GetSelectionWithBias(biasTowardsNewAddrs int) []*p2p.NetAddress
// TODO: remove // TODO: remove
ListOfKnownAddresses() []*knownAddress ListOfKnownAddresses() []*knownAddress
@ -173,6 +177,14 @@ func (a *addrBook) RemoveAddress(addr *p2p.NetAddress) {
a.removeFromAllBuckets(ka) a.removeFromAllBuckets(ka)
} }
// IsGood returns true if peer was ever marked as good and haven't
// done anything wrong since then.
func (a *addrBook) IsGood(addr *p2p.NetAddress) bool {
a.mtx.Lock()
defer a.mtx.Unlock()
return a.addrLookup[addr.ID].isOld()
}
// NeedMoreAddrs implements AddrBook - returns true if there are not have enough addresses in the book. // NeedMoreAddrs implements AddrBook - returns true if there are not have enough addresses in the book.
func (a *addrBook) NeedMoreAddrs() bool { func (a *addrBook) NeedMoreAddrs() bool {
return a.Size() < needAddressThreshold return a.Size() < needAddressThreshold
@ -180,27 +192,27 @@ func (a *addrBook) NeedMoreAddrs() bool {
// PickAddress implements AddrBook. It picks an address to connect to. // PickAddress implements AddrBook. It picks an address to connect to.
// The address is picked randomly from an old or new bucket according // The address is picked randomly from an old or new bucket according
// to the newBias argument, which must be between [0, 100] (or else is truncated to that range)
// to the biasTowardsNewAddrs argument, which must be between [0, 100] (or else is truncated to that range)
// and determines how biased we are to pick an address from a new bucket. // and determines how biased we are to pick an address from a new bucket.
// PickAddress returns nil if the AddrBook is empty or if we try to pick // PickAddress returns nil if the AddrBook is empty or if we try to pick
// from an empty bucket. // from an empty bucket.
func (a *addrBook) PickAddress(newBias int) *p2p.NetAddress {
func (a *addrBook) PickAddress(biasTowardsNewAddrs int) *p2p.NetAddress {
a.mtx.Lock() a.mtx.Lock()
defer a.mtx.Unlock() defer a.mtx.Unlock()
if a.size() == 0 { if a.size() == 0 {
return nil return nil
} }
if newBias > 100 {
newBias = 100
if biasTowardsNewAddrs > 100 {
biasTowardsNewAddrs = 100
} }
if newBias < 0 {
newBias = 0
if biasTowardsNewAddrs < 0 {
biasTowardsNewAddrs = 0
} }
// Bias between new and old addresses. // Bias between new and old addresses.
oldCorrelation := math.Sqrt(float64(a.nOld)) * (100.0 - float64(newBias))
newCorrelation := math.Sqrt(float64(a.nNew)) * float64(newBias)
oldCorrelation := math.Sqrt(float64(a.nOld)) * (100.0 - float64(biasTowardsNewAddrs))
newCorrelation := math.Sqrt(float64(a.nNew)) * float64(biasTowardsNewAddrs)
// pick a random peer from a random bucket // pick a random peer from a random bucket
var bucket map[string]*knownAddress var bucket map[string]*knownAddress
@ -295,6 +307,100 @@ func (a *addrBook) GetSelection() []*p2p.NetAddress {
return allAddr[:numAddresses] return allAddr[:numAddresses]
} }
// GetSelectionWithBias implements AddrBook.
// It randomly selects some addresses (old & new). Suitable for peer-exchange protocols.
//
// Each address is picked randomly from an old or new bucket according to the
// biasTowardsNewAddrs argument, which must be between [0, 100] (or else is truncated to
// that range) and determines how biased we are to pick an address from a new
// bucket.
func (a *addrBook) GetSelectionWithBias(biasTowardsNewAddrs int) []*p2p.NetAddress {
a.mtx.Lock()
defer a.mtx.Unlock()
if a.size() == 0 {
return nil
}
if biasTowardsNewAddrs > 100 {
biasTowardsNewAddrs = 100
}
if biasTowardsNewAddrs < 0 {
biasTowardsNewAddrs = 0
}
numAddresses := cmn.MaxInt(
cmn.MinInt(minGetSelection, a.size()),
a.size()*getSelectionPercent/100)
numAddresses = cmn.MinInt(maxGetSelection, numAddresses)
selection := make([]*p2p.NetAddress, numAddresses)
oldBucketToAddrsMap := make(map[int]map[string]struct{})
var oldIndex int
newBucketToAddrsMap := make(map[int]map[string]struct{})
var newIndex int
selectionIndex := 0
ADDRS_LOOP:
for selectionIndex < numAddresses {
pickFromOldBucket := int((float64(selectionIndex)/float64(numAddresses))*100) >= biasTowardsNewAddrs
pickFromOldBucket = (pickFromOldBucket && a.nOld > 0) || a.nNew == 0
bucket := make(map[string]*knownAddress)
// loop until we pick a random non-empty bucket
for len(bucket) == 0 {
if pickFromOldBucket {
oldIndex = a.rand.Intn(len(a.bucketsOld))
bucket = a.bucketsOld[oldIndex]
} else {
newIndex = a.rand.Intn(len(a.bucketsNew))
bucket = a.bucketsNew[newIndex]
}
}
// pick a random index
randIndex := a.rand.Intn(len(bucket))
// loop over the map to return that index
var selectedAddr *p2p.NetAddress
for _, ka := range bucket {
if randIndex == 0 {
selectedAddr = ka.Addr
break
}
randIndex--
}
// if we have selected the address before, restart the loop
// otherwise, record it and continue
if pickFromOldBucket {
if addrsMap, ok := oldBucketToAddrsMap[oldIndex]; ok {
if _, ok = addrsMap[selectedAddr.String()]; ok {
continue ADDRS_LOOP
}
} else {
oldBucketToAddrsMap[oldIndex] = make(map[string]struct{})
}
oldBucketToAddrsMap[oldIndex][selectedAddr.String()] = struct{}{}
} else {
if addrsMap, ok := newBucketToAddrsMap[newIndex]; ok {
if _, ok = addrsMap[selectedAddr.String()]; ok {
continue ADDRS_LOOP
}
} else {
newBucketToAddrsMap[newIndex] = make(map[string]struct{})
}
newBucketToAddrsMap[newIndex][selectedAddr.String()] = struct{}{}
}
selection[selectionIndex] = selectedAddr
selectionIndex++
}
return selection
}
// ListOfKnownAddresses returns the new and old addresses. // ListOfKnownAddresses returns the new and old addresses.
func (a *addrBook) ListOfKnownAddresses() []*knownAddress { func (a *addrBook) ListOfKnownAddresses() []*knownAddress {
a.mtx.Lock() a.mtx.Lock()


+ 109
- 0
p2p/pex/addrbook_test.go View File

@ -157,6 +157,13 @@ func TestAddrBookPromoteToOld(t *testing.T) {
t.Errorf("selection could not be bigger than the book") t.Errorf("selection could not be bigger than the book")
} }
selection = book.GetSelectionWithBias(30)
t.Logf("selection: %v", selection)
if len(selection) > book.Size() {
t.Errorf("selection with bias could not be bigger than the book")
}
assert.Equal(t, book.Size(), 100, "expecting book size to be 100") assert.Equal(t, book.Size(), 100, "expecting book size to be 100")
} }
@ -229,3 +236,105 @@ func TestAddrBookRemoveAddress(t *testing.T) {
book.RemoveAddress(nonExistingAddr) book.RemoveAddress(nonExistingAddr)
assert.Equal(t, 0, book.Size()) assert.Equal(t, 0, book.Size())
} }
func TestAddrBookGetSelection(t *testing.T) {
fname := createTempFileName("addrbook_test")
defer deleteTempFile(fname)
book := NewAddrBook(fname, true)
book.SetLogger(log.TestingLogger())
// 1) empty book
assert.Empty(t, book.GetSelection())
// 2) add one address
addr := randIPv4Address(t)
book.AddAddress(addr, addr)
assert.Equal(t, 1, len(book.GetSelection()))
assert.Equal(t, addr, book.GetSelection()[0])
// 3) add a bunch of addresses
randAddrs := randNetAddressPairs(t, 100)
for _, addrSrc := range randAddrs {
book.AddAddress(addrSrc.addr, addrSrc.src)
}
// check there is no duplicates
addrs := make(map[string]*p2p.NetAddress)
selection := book.GetSelection()
for _, addr := range selection {
if dup, ok := addrs[addr.String()]; ok {
t.Fatalf("selection %v contains duplicates %v", selection, dup)
}
addrs[addr.String()] = addr
}
if len(selection) > book.Size() {
t.Errorf("selection %v could not be bigger than the book", selection)
}
}
func TestAddrBookGetSelectionWithBias(t *testing.T) {
const biasTowardsNewAddrs = 30
fname := createTempFileName("addrbook_test")
defer deleteTempFile(fname)
book := NewAddrBook(fname, true)
book.SetLogger(log.TestingLogger())
// 1) empty book
selection := book.GetSelectionWithBias(biasTowardsNewAddrs)
assert.Empty(t, selection)
// 2) add one address
addr := randIPv4Address(t)
book.AddAddress(addr, addr)
selection = book.GetSelectionWithBias(biasTowardsNewAddrs)
assert.Equal(t, 1, len(selection))
assert.Equal(t, addr, selection[0])
// 3) add a bunch of addresses
randAddrs := randNetAddressPairs(t, 100)
for _, addrSrc := range randAddrs {
book.AddAddress(addrSrc.addr, addrSrc.src)
}
// check there is no duplicates
addrs := make(map[string]*p2p.NetAddress)
selection = book.GetSelectionWithBias(biasTowardsNewAddrs)
for _, addr := range selection {
if dup, ok := addrs[addr.String()]; ok {
t.Fatalf("selection %v contains duplicates %v", selection, dup)
}
addrs[addr.String()] = addr
}
if len(selection) > book.Size() {
t.Fatalf("selection %v could not be bigger than the book", selection)
}
// 4) mark 80% of the addresses as good
randAddrsLen := len(randAddrs)
for i, addrSrc := range randAddrs {
if int((float64(i)/float64(randAddrsLen))*100) >= 20 {
book.MarkGood(addrSrc.addr)
}
}
selection = book.GetSelectionWithBias(biasTowardsNewAddrs)
// check that ~70% of addresses returned are good
good := 0
for _, addr := range selection {
if book.IsGood(addr) {
good++
}
}
got, expected := int((float64(good)/float64(len(selection)))*100), (100 - biasTowardsNewAddrs)
if got >= expected {
t.Fatalf("expected more good peers (%% got: %d, %% expected: %d, number of good addrs: %d, total: %d)", got, expected, good, len(selection))
}
}

+ 6
- 2
p2p/pex/pex_reactor.go View File

@ -43,6 +43,11 @@ const (
defaultCrawlPeersPeriod = 30 * time.Second // check some peers every this defaultCrawlPeersPeriod = 30 * time.Second // check some peers every this
maxAttemptsToDial = 16 // ~ 35h in total (last attempt - 18h) maxAttemptsToDial = 16 // ~ 35h in total (last attempt - 18h)
// if node connects to seed, it does not have any trusted peers.
// Especially in the beginning, node should have more trusted peers than
// untrusted.
biasToSelectNewPeers = 30 // 70 to select good peers
) )
// PEXReactor handles PEX (peer exchange) and ensures that an // PEXReactor handles PEX (peer exchange) and ensures that an
@ -191,8 +196,7 @@ func (r *PEXReactor) Receive(chID byte, src Peer, msgBytes []byte) {
// Seeds disconnect after sending a batch of addrs // Seeds disconnect after sending a batch of addrs
if r.config.SeedMode { if r.config.SeedMode {
// TODO: should we be more selective ?
r.SendAddrs(src, r.book.GetSelection())
r.SendAddrs(src, r.book.GetSelectionWithBias(biasToSelectNewPeers))
r.Switch.StopPeerGracefully(src) r.Switch.StopPeerGracefully(src)
} else { } else {
r.SendAddrs(src, r.book.GetSelection()) r.SendAddrs(src, r.book.GetSelection())


Loading…
Cancel
Save