Bucketizing for addrmanager

Implement the bucketing by source group and group using essentially the
same algorithm as the address maanger in bitcoind.

Fix up the saving of peer.json to do so in a json format that keeps bucket
metadata.

If we fail to load the some of the data we asssume that we have
incomplete information, so we nuke the existing file and reinitialise so
we have a clean slate.
This commit is contained in:
Owain G. Ainsworth 2013-09-17 17:24:15 +01:00
parent d9214030e0
commit 850420055f
2 changed files with 427 additions and 116 deletions

View file

@ -5,9 +5,14 @@
package main package main
import ( import (
"bytes"
"container/list" "container/list"
crand "crypto/rand" // for seeding
"encoding/binary"
"encoding/json" "encoding/json"
"fmt"
"github.com/conformal/btcwire" "github.com/conformal/btcwire"
"io"
"math" "math"
"math/rand" "math/rand"
"net" "net"
@ -37,10 +42,30 @@ const (
// tried address bucket. // tried address bucket.
triedBucketSize = 64 triedBucketSize = 64
// triedBucketCount is the number of buckets we split tried
// addresses over.
triedBucketCount = 64
// newBucketSize is the maximum number of addresses in each new address // newBucketSize is the maximum number of addresses in each new address
// bucket. // bucket.
newBucketSize = 64 newBucketSize = 64
// newBucketCount is the number of buckets taht we spread new addresses
// over.
newBucketCount = 256
// triedBucketsPerGroup is the number of trieed buckets over which an
// address group will be spread.
triedBucketsPerGroup = 4
// newBucketsPerGroup is the number of new buckets over which an
// source address group will be spread.
newBucketsPerGroup = 32
// newBucketsPerAddress is the number of buckets a frequently seen new
// address may end up in.
newBucketsPerAddress = 4
// numMissingDays is the number of days before which we assume an // numMissingDays is the number of days before which we assume an
// address has vanished if we have not seen it announced in that long. // address has vanished if we have not seen it announced in that long.
numMissingDays = 30 numMissingDays = 30
@ -56,17 +81,37 @@ const (
// minBadDays is the number of days since the last success before we // minBadDays is the number of days since the last success before we
// will consider evicting an address. // will consider evicting an address.
minBadDays = 7 minBadDays = 7
// getAddrMax is the most addresses that we will send in response
// to a getAddr (in practise the most addresses we will return from a
// call to AddressCache()).
getAddrMax = 2500
// getAddrPercent is the percentage of total addresses known that we
// will share with a call to AddressCache.
getAddrPercent = 23
// serialisationVersion is the current version of the on-disk format.
serialisationVersion = 1
) )
// updateAddress is a helper function to either update an address already known // updateAddress is a helper function to either update an address already known
// to the address manager, or to add the address if not already known. // to the address manager, or to add the address if not already known.
func (a *AddrManager) updateAddress(netAddr, srcAddr *btcwire.NetAddress) { func (a *AddrManager) updateAddress(netAddr, srcAddr *btcwire.NetAddress) {
// Filter out non-routable addresses. Note that non-routable
// also includes invalid and local addresses.
if !Routable(netAddr) {
return
}
// Protect concurrent access. // Protect concurrent access.
a.mtx.Lock() a.mtx.Lock()
defer a.mtx.Unlock() defer a.mtx.Unlock()
addr := NetAddressKey(netAddr)
ka := a.find(netAddr) ka := a.find(netAddr)
if ka != nil { if ka != nil {
// TODO(oga) only update adresses periodically.
// Update the last seen time. // Update the last seen time.
if netAddr.Timestamp.After(ka.na.Timestamp) { if netAddr.Timestamp.After(ka.na.Timestamp) {
ka.na.Timestamp = netAddr.Timestamp ka.na.Timestamp = netAddr.Timestamp
@ -75,28 +120,48 @@ func (a *AddrManager) updateAddress(netAddr, srcAddr *btcwire.NetAddress) {
// Update services. // Update services.
ka.na.AddService(netAddr.Services) ka.na.AddService(netAddr.Services)
log.Tracef("[AMGR] Updated address manager address %s", // If already in tried, we have nothing to do here.
NetAddressKey(netAddr)) if ka.tried {
return
}
// Already at our max?
if ka.refs == newBucketsPerAddress {
return
}
// The more entries we have, the less likely we are to add more.
// likelyhood is 2N.
factor := int32(2 * ka.refs)
if a.rand.Int31n(factor) != 0 {
return
}
} else {
ka = &knownAddress{na: netAddr, srcAddr: srcAddr}
a.addrIndex[addr] = ka
a.nNew++
// XXX time penalty?
}
bucket := a.getNewBucket(netAddr, srcAddr)
// Already exists?
if _, ok := a.addrNew[bucket][addr]; ok {
return return
} }
// Enforce max addresses. // Enforce max addresses.
if len(a.addrNew) > newBucketSize { if len(a.addrNew[bucket]) > newBucketSize {
log.Tracef("[AMGR] new bucket is full, expiring old ") log.Tracef("[AMGR] new bucket is full, expiring old ")
a.expireNew() a.expireNew(bucket)
} }
addr := NetAddressKey(netAddr)
ka = &knownAddress{na: netAddr}
// Fill in index.
a.addrIndex[addr] = ka
// Add to new bucket. // Add to new bucket.
a.addrNew[addr] = ka ka.refs++
a.addrNew[bucket][addr] = ka
log.Tracef("[AMGR] Added new address %s for a total of %d addresses", log.Tracef("[AMGR] Added new address %s for a total of %d addresses",
addr, len(a.addrNew)+a.addrTried.Len()) addr, a.nTried+a.nNew)
} }
// bad returns true if the address in question has not been tried in the last // bad returns true if the address in question has not been tried in the last
@ -110,7 +175,6 @@ func (a *AddrManager) updateAddress(netAddr, srcAddr *btcwire.NetAddress) {
func bad(ka *knownAddress) bool { func bad(ka *knownAddress) bool {
if ka.lastattempt.After(time.Now().Add(-1 * time.Minute)) { if ka.lastattempt.After(time.Now().Add(-1 * time.Minute)) {
return false return false
} }
// From the future? // From the future?
@ -177,7 +241,7 @@ func chance(ka *knownAddress) float64 {
// Failed attempts deprioritise. // Failed attempts deprioritise.
if ka.attempts > 0 { if ka.attempts > 0 {
c /= (float64(ka.attempts) * 1.5) c /= float64(ka.attempts) * 1.5
} }
return c return c
@ -185,18 +249,22 @@ func chance(ka *knownAddress) float64 {
// expireNew makes space in the new buckets by expiring the really bad entries. // expireNew makes space in the new buckets by expiring the really bad entries.
// If no bad entries are available we look at a few and remove the oldest. // If no bad entries are available we look at a few and remove the oldest.
func (a *AddrManager) expireNew() { func (a *AddrManager) expireNew(bucket int) {
// First see if there are any entries that are so bad we can just throw // First see if there are any entries that are so bad we can just throw
// them away. otherwise we throw away the oldest entry in the cache. // them away. otherwise we throw away the oldest entry in the cache.
// Bitcoind here chooses four random and just throws the oldest of // Bitcoind here chooses four random and just throws the oldest of
// those away, but we keep track of oldest in the initial traversal and // those away, but we keep track of oldest in the initial traversal and
// use that information instead // use that information instead.
var oldest *knownAddress var oldest *knownAddress
for k, v := range a.addrNew { for k, v := range a.addrNew[bucket] {
if bad(v) { if bad(v) {
log.Tracef("[AMGR] expiring bad address %v", k) log.Tracef("[AMGR] expiring bad address %v", k)
delete(a.addrNew[bucket], k)
a.nNew--
v.refs--
if v.refs == 0 {
delete(a.addrIndex, k) delete(a.addrIndex, k)
delete(a.addrNew, k) }
return return
} }
if oldest == nil { if oldest == nil {
@ -210,17 +278,22 @@ func (a *AddrManager) expireNew() {
key := NetAddressKey(oldest.na) key := NetAddressKey(oldest.na)
log.Tracef("[AMGR] expiring oldest address %v", key) log.Tracef("[AMGR] expiring oldest address %v", key)
delete(a.addrNew[bucket], key)
a.nNew--
oldest.refs--
if oldest.refs == 0 {
delete(a.addrIndex, key) delete(a.addrIndex, key)
delete(a.addrNew, key) }
} }
} }
// pickTried selects an address from the tried bucket to be evicted. // pickTried selects an address from the tried bucket to be evicted.
// We just choose the eldest. // We just choose the eldest. Bitcoind selects 4 random entries and throws away
func (a *AddrManager) pickTried() *list.Element { // the older of them.
func (a *AddrManager) pickTried(bucket int) *list.Element {
var oldest *knownAddress var oldest *knownAddress
var oldestElem *list.Element var oldestElem *list.Element
for e := a.addrTried.Front(); e != nil; e = e.Next() { for e := a.addrTried[bucket].Front(); e != nil; e = e.Next() {
ka := e.Value.(*knownAddress) ka := e.Value.(*knownAddress)
if oldest == nil || oldest.na.Timestamp.After(ka.na.Timestamp) { if oldest == nil || oldest.na.Timestamp.After(ka.na.Timestamp) {
oldestElem = e oldestElem = e
@ -235,11 +308,12 @@ func (a *AddrManager) pickTried() *list.Element {
// to determine how viable an address is. // to determine how viable an address is.
type knownAddress struct { type knownAddress struct {
na *btcwire.NetAddress na *btcwire.NetAddress
srcAddr *btcwire.NetAddress
attempts int attempts int
lastattempt time.Time lastattempt time.Time
lastsuccess time.Time lastsuccess time.Time
time time.Time
tried bool tried bool
refs int // reference count of new buckets
} }
// AddrManager provides a concurrency safe address manager for caching potential // AddrManager provides a concurrency safe address manager for caching potential
@ -247,17 +321,58 @@ type knownAddress struct {
type AddrManager struct { type AddrManager struct {
mtx sync.Mutex mtx sync.Mutex
rand *rand.Rand rand *rand.Rand
key [32]byte
addrIndex map[string]*knownAddress // address key to ka for all addrs. addrIndex map[string]*knownAddress // address key to ka for all addrs.
addrNew map[string]*knownAddress addrNew [newBucketCount]map[string]*knownAddress
addrTried *list.List addrTried [triedBucketCount]*list.List
started bool started bool
shutdown bool shutdown bool
wg sync.WaitGroup wg sync.WaitGroup
quit chan bool quit chan bool
nTried int
nNew int
} }
type JsonSave struct { func (a *AddrManager) getNewBucket(netAddr, srcAddr *btcwire.NetAddress) int {
AddrList []string // bitcoind:
// doublesha256(key + sourcegroup + int64(doublesha256(key + group + sourcegroup))%bucket_per_source_group) % num_new_buckes
data1 := []byte{}
data1 = append(data1, a.key[:]...)
data1 = append(data1, []byte(GroupKey(netAddr))...)
data1 = append(data1, []byte(GroupKey(srcAddr))...)
hash1 := btcwire.DoubleSha256(data1)
hash64 := binary.LittleEndian.Uint64(hash1)
hash64 %= newBucketsPerGroup
hashbuf := new(bytes.Buffer)
binary.Write(hashbuf, binary.LittleEndian, hash64)
data2 := []byte{}
data2 = append(data2, a.key[:]...)
data2 = append(data2, GroupKey(srcAddr)...)
data2 = append(data2, hashbuf.Bytes()...)
hash2 := btcwire.DoubleSha256(data2)
return int(binary.LittleEndian.Uint64(hash2) % newBucketCount)
}
func (a *AddrManager) getTriedBucket(netAddr *btcwire.NetAddress) int {
// bitcoind hashes this as:
// doublesha256(key + group + truncate_to_64bits(doublesha256(key)) % buckets_per_group) % num_buckets
data1 := []byte{}
data1 = append(data1, a.key[:]...)
data1 = append(data1, []byte(NetAddressKey(netAddr))...)
hash1 := btcwire.DoubleSha256(data1)
hash64 := binary.LittleEndian.Uint64(hash1)
hash64 %= triedBucketsPerGroup
hashbuf := new(bytes.Buffer)
binary.Write(hashbuf, binary.LittleEndian, hash64)
data2 := []byte{}
data2 = append(data2, a.key[:]...)
data2 = append(data2, GroupKey(netAddr)...)
data2 = append(data2, hashbuf.Bytes()...)
hash2 := btcwire.DoubleSha256(data2)
return int(binary.LittleEndian.Uint64(hash2) % triedBucketCount)
} }
// addressHandler is the main handler for the address manager. It must be run // addressHandler is the main handler for the address manager. It must be run
@ -282,25 +397,78 @@ out:
log.Trace("[AMGR] Address handler done") log.Trace("[AMGR] Address handler done")
} }
type serialisedKnownAddress struct {
Addr string
Src string
Attempts int
TimeStamp int64
LastAttempt int64
LastSuccess int64
// no refcount or tried, that is available from context.
}
type serialisedAddrManager struct {
Version int
Key [32]byte
Addresses []*serialisedKnownAddress
NewBuckets [newBucketCount][]string // string is NetAddressKey
TriedBuckets [triedBucketCount][]string
}
// savePeers saves all the known addresses to a file so they can be read back // savePeers saves all the known addresses to a file so they can be read back
// in at next run. // in at next run.
func (a *AddrManager) savePeers() { func (a *AddrManager) savePeers() {
// First we make a serialisable datastructure so we can encode it to
// json.
sam := new(serialisedAddrManager)
sam.Version = serialisationVersion
copy(sam.Key[:], a.key[:])
sam.Addresses = make([]*serialisedKnownAddress, len(a.addrIndex))
i := 0
for k, v := range a.addrIndex {
ska := new(serialisedKnownAddress)
ska.Addr = k
ska.TimeStamp = v.na.Timestamp.Unix()
ska.Src = NetAddressKey(v.srcAddr)
ska.Attempts = v.attempts
ska.LastAttempt = v.lastattempt.Unix()
ska.LastSuccess = v.lastsuccess.Unix()
// Tried and refs are implicit in the rest of the structure
// and will be worked out from context on unserialisation.
sam.Addresses[i] = ska
i++
}
for i := range a.addrNew {
sam.NewBuckets[i] = make([]string, len(a.addrNew[i]))
j := 0
for k := range a.addrNew[i] {
sam.NewBuckets[i][j] = k
j++
}
}
for i := range a.addrTried {
sam.TriedBuckets[i] = make([]string, a.addrTried[i].Len())
j := 0
for e := a.addrTried[i].Front(); e != nil; e = e.Next() {
ka := e.Value.(*knownAddress)
sam.TriedBuckets[i][j] = NetAddressKey(ka.na)
j++
}
}
// May give some way to specify this later. // May give some way to specify this later.
filename := "peers.json" filename := "peers.json"
filePath := filepath.Join(cfg.DataDir, filename) filePath := filepath.Join(cfg.DataDir, filename)
var toSave JsonSave
list := a.AddressCacheFlat()
toSave.AddrList = list
w, err := os.Create(filePath) w, err := os.Create(filePath)
if err != nil { if err != nil {
log.Error("[AMGR] Error opening file: ", filePath, err) log.Error("Error opening file: ", filePath, err)
} }
enc := json.NewEncoder(w) enc := json.NewEncoder(w)
defer w.Close() defer w.Close()
enc.Encode(&toSave) enc.Encode(&sam)
} }
// loadPeers loads the known address from the saved file. If empty, missing, or // loadPeers loads the known address from the saved file. If empty, missing, or
@ -310,31 +478,124 @@ func (a *AddrManager) loadPeers() {
filename := "peers.json" filename := "peers.json"
filePath := filepath.Join(cfg.DataDir, filename) filePath := filepath.Join(cfg.DataDir, filename)
err := a.deserialisePeers(filePath)
if err != nil {
log.Errorf("[AMGR] Failed to parse %s: %v", filePath,
err)
// if it is invalid we nuke the old one unconditionally.
err = os.Remove(filePath)
if err != nil {
log.Warn("Failed to remove corrupt peers "+
"file: ", err)
}
a.reset()
return
}
log.Infof("[AMGR] Successfuly loaded %d addresses from %s",
a.NumAddresses(), filePath)
}
func (a *AddrManager) deserialisePeers(filePath string) error {
_, err := os.Stat(filePath) _, err := os.Stat(filePath)
if os.IsNotExist(err) { if os.IsNotExist(err) {
log.Debugf("[AMGR] %s does not exist.", filePath) return fmt.Errorf("%s does not exist.\n", filePath)
} else { }
r, err := os.Open(filePath) r, err := os.Open(filePath)
if err != nil { if err != nil {
log.Error("[AMGR] Error opening file: ", filePath, err) return fmt.Errorf("%s error opening file: ", filePath, err)
return
} }
defer r.Close() defer r.Close()
var inList JsonSave var sam serialisedAddrManager
dec := json.NewDecoder(r) dec := json.NewDecoder(r)
err = dec.Decode(&inList) err = dec.Decode(&sam)
if err != nil { if err != nil {
log.Error("[AMGR] Error reading:", filePath, err) return fmt.Errorf("error reading %s: %v", filePath, err)
return
} }
log.Debug("[AMGR] Adding ", len(inList.AddrList), " saved peers.")
if len(inList.AddrList) > 0 { if sam.Version != serialisationVersion {
for _, ip := range inList.AddrList { return fmt.Errorf("unknown version %v in serialised "+
a.AddAddressByIP(ip) "addrmanager", sam.Version)
}
copy(a.key[:], sam.Key[:])
for _, v := range sam.Addresses {
ka := new(knownAddress)
ka.na, err = deserialiseNetAddress(v.Addr)
if err != nil {
return fmt.Errorf("failed to deserialise netaddress "+
"%s: %v", v.Addr, err)
}
ka.srcAddr, err = deserialiseNetAddress(v.Src)
if err != nil {
return fmt.Errorf("failed to deserialise netaddress "+
"%s: %v", v.Src, err)
}
ka.attempts = v.Attempts
ka.lastattempt = time.Unix(v.LastAttempt, 0)
ka.lastsuccess = time.Unix(v.LastSuccess, 0)
a.addrIndex[NetAddressKey(ka.na)] = ka
}
for i := range sam.NewBuckets {
for _, val := range sam.NewBuckets[i] {
ka, ok := a.addrIndex[val]
if !ok {
return fmt.Errorf("newbucket contains %s but "+
"none in address list", val)
}
if ka.refs == 0 {
a.nNew++
}
ka.refs++
a.addrNew[i][val] = ka
} }
} }
for i := range sam.TriedBuckets {
for _, val := range sam.TriedBuckets[i] {
ka, ok := a.addrIndex[val]
if !ok {
return fmt.Errorf("Newbucket contains %s but "+
"none in address list", val)
} }
ka.tried = true
a.nTried++
a.addrTried[i].PushBack(ka)
}
}
// Sanity checking.
for k, v := range a.addrIndex {
if v.refs == 0 && !v.tried {
return fmt.Errorf("address %s after serialisation "+
"with no references", k)
}
if v.refs > 0 && v.tried {
return fmt.Errorf("address %s after serialisation "+
"which is both new and tried!", k)
}
}
return nil
}
func deserialiseNetAddress(addr string) (*btcwire.NetAddress, error) {
host, portStr, err := net.SplitHostPort(addr)
if err != nil {
return nil, err
}
ip := net.ParseIP(host)
port, err := strconv.ParseUint(portStr, 10, 16)
if err != nil {
return nil, err
}
na := btcwire.NewNetAddressIPPort(ip, uint16(port),
btcwire.SFNodeNetwork)
return na, nil
} }
// Start begins the core address handler which manages a pool of known // Start begins the core address handler which manages a pool of known
@ -377,13 +638,9 @@ func (a *AddrManager) Stop() error {
func (a *AddrManager) AddAddresses(addrs []*btcwire.NetAddress, func (a *AddrManager) AddAddresses(addrs []*btcwire.NetAddress,
srcAddr *btcwire.NetAddress) { srcAddr *btcwire.NetAddress) {
for _, na := range addrs { for _, na := range addrs {
// Filter out non-routable addresses. Note that non-routable
// also includes invalid and local addresses.
if Routable(na) {
a.updateAddress(na, srcAddr) a.updateAddress(na, srcAddr)
} }
} }
}
// AddAddress adds a new address to the address manager. It enforces a max // AddAddress adds a new address to the address manager. It enforces a max
// number of addresses and silently ignores duplicate addresses. It is // number of addresses and silently ignores duplicate addresses. It is
@ -433,45 +690,54 @@ func (a *AddrManager) NumAddresses() int {
a.mtx.Lock() a.mtx.Lock()
defer a.mtx.Unlock() defer a.mtx.Unlock()
return len(a.addrNew) + a.addrTried.Len() return a.nTried + a.nNew
} }
// AddressCache returns the current address cache. It must be treated as // AddressCache returns the current address cache. It must be treated as
// read-only (but since it is a copy now, this is not as dangerous). // read-only (but since it is a copy now, this is not as dangerous).
func (a *AddrManager) AddressCache() map[string]*btcwire.NetAddress { func (a *AddrManager) AddressCache() []*btcwire.NetAddress {
allAddr := make(map[string]*btcwire.NetAddress)
a.mtx.Lock() a.mtx.Lock()
defer a.mtx.Unlock() defer a.mtx.Unlock()
for k, v := range a.addrNew { if a.nNew+a.nTried == 0 {
allAddr[k] = v.na return nil
} }
for e := a.addrTried.Front(); e != nil; e = e.Next() { allAddr := make([]*btcwire.NetAddress, a.nNew+a.nTried)
ka := e.Value.(*knownAddress) i := 0
allAddr[NetAddressKey(ka.na)] = ka.na // Iteration order is undefined here, but we randomise it anyway.
for _, v := range a.addrIndex {
allAddr[i] = v.na
i++
}
// Fisher-Yates shuffle the array
for i := range allAddr {
j := rand.Intn(i + 1)
allAddr[i], allAddr[j] = allAddr[j], allAddr[i]
} }
return allAddr numAddresses := len(allAddr) * getAddrPercent / 100
if numAddresses > getAddrMax {
numAddresses = getAddrMax
} }
// AddressCacheFlat returns a flat list of strings with the current address // slice off the limit we are willing to share.
// cache. Just a copy, so one can do whatever they want to it. return allAddr[:numAddresses]
func (a *AddrManager) AddressCacheFlat() []string {
var allAddr []string
a.mtx.Lock()
defer a.mtx.Unlock()
for k := range a.addrNew {
allAddr = append(allAddr, k)
} }
for e := a.addrTried.Front(); e != nil; e = e.Next() { // reset resets the address manager by reinitialising the random source
ka := e.Value.(*knownAddress) // and allocating fresh empty bucket storage.
allAddr = append(allAddr, NetAddressKey(ka.na)) func (a *AddrManager) reset() {
}
return allAddr a.addrIndex = make(map[string]*knownAddress)
// fill key with bytes from a good random source.
io.ReadFull(crand.Reader, a.key[:])
for i := range a.addrNew {
a.addrNew[i] = make(map[string]*knownAddress)
}
for i := range a.addrTried {
a.addrTried[i] = list.New()
}
} }
// New returns a new bitcoin address manager. // New returns a new bitcoin address manager.
@ -479,11 +745,9 @@ func (a *AddrManager) AddressCacheFlat() []string {
func NewAddrManager() *AddrManager { func NewAddrManager() *AddrManager {
am := AddrManager{ am := AddrManager{
rand: rand.New(rand.NewSource(time.Now().UnixNano())), rand: rand.New(rand.NewSource(time.Now().UnixNano())),
addrIndex: make(map[string]*knownAddress),
addrNew: make(map[string]*knownAddress),
addrTried: list.New(),
quit: make(chan bool), quit: make(chan bool),
} }
am.reset()
return &am return &am
} }
@ -511,10 +775,10 @@ func (a *AddrManager) GetAddress(class string, newBias int) *knownAddress {
newBias = 0 newBias = 0
} }
// Bias 50% for now between new and tried. // Bias between new and tried addresses.
triedCorrelation := math.Sqrt(float64(a.addrTried.Len())) * triedCorrelation := math.Sqrt(float64(a.nTried)) *
(100.0 - float64(newBias)) (100.0 - float64(newBias))
newCorrelation := math.Sqrt(float64(len(a.addrNew))) * float64(newBias) newCorrelation := math.Sqrt(float64(a.nNew)) * float64(newBias)
if (newCorrelation+triedCorrelation)*a.rand.Float64() < if (newCorrelation+triedCorrelation)*a.rand.Float64() <
triedCorrelation { triedCorrelation {
@ -522,9 +786,16 @@ func (a *AddrManager) GetAddress(class string, newBias int) *knownAddress {
large := 1 << 30 large := 1 << 30
factor := 1.0 factor := 1.0
for { for {
// pick a random bucket.
bucket := a.rand.Intn(len(a.addrTried))
if a.addrTried[bucket].Len() == 0 {
continue
}
// Pick a random entry in the list // Pick a random entry in the list
e := a.addrTried.Front() e := a.addrTried[bucket].Front()
for i := a.rand.Int63n(int64(a.addrTried.Len())); i > 0; i-- { for i :=
a.rand.Int63n(int64(a.addrTried[bucket].Len())); i > 0; i-- {
e = e.Next() e = e.Next()
} }
ka := e.Value.(*knownAddress) ka := e.Value.(*knownAddress)
@ -539,15 +810,23 @@ func (a *AddrManager) GetAddress(class string, newBias int) *knownAddress {
} else { } else {
// new node. // new node.
// XXX use a closure/function to avoid repeating this. // XXX use a closure/function to avoid repeating this.
keyList := []string{}
for key := range a.addrNew {
keyList = append(keyList, key)
}
large := 1 << 30 large := 1 << 30
factor := 1.0 factor := 1.0
for { for {
testKey := keyList[a.rand.Int63n(int64(len(keyList)))] // Pick a random bucket.
ka := a.addrNew[testKey] bucket := a.rand.Intn(len(a.addrNew))
if len(a.addrNew[bucket]) == 0 {
continue
}
// Then, a random entry in it.
var ka *knownAddress
nth := a.rand.Intn(len(a.addrNew[bucket]))
for _, value := range a.addrNew[bucket] {
if nth == 0 {
ka = value
}
nth--
}
randval := a.rand.Intn(large) randval := a.rand.Intn(large)
if float64(randval) < (factor * chance(ka) * float64(large)) { if float64(randval) < (factor * chance(ka) * float64(large)) {
log.Tracef("[AMGR] Selected %v from new bucket", log.Tracef("[AMGR] Selected %v from new bucket",
@ -628,35 +907,67 @@ func (a *AddrManager) Good(addr *btcwire.NetAddress) {
// ok, need to move it to tried. // ok, need to move it to tried.
// remove from new buckets. // remove from all new buckets.
// record one of the buckets in question and call it the `first'
addrKey := NetAddressKey(addr) addrKey := NetAddressKey(addr)
delete(a.addrNew, addrKey) oldBucket := -1
for i := range a.addrNew {
// we check for existance so we can record the first one
if _, ok := a.addrNew[i][addrKey]; ok {
delete(a.addrNew[i], addrKey)
ka.refs--
if oldBucket == -1 {
oldBucket = i
}
}
}
a.nNew--
// is tried full? or is it ok? if oldBucket == -1 {
if a.addrTried.Len() < triedBucketSize { // What? wasn't in a bucket after all.... Panic?
a.addrTried.PushBack(ka) return
}
bucket := a.getTriedBucket(ka.na)
// Room in this tried bucket?
if a.addrTried[bucket].Len() < triedBucketSize {
ka.tried = true
a.addrTried[bucket].PushBack(ka)
a.nTried++
return return
} }
// No room, we have to evict something else. // No room, we have to evict something else.
entry := a.pickTried(bucket)
// pick another one to throw out
entry := a.pickTried()
rmka := entry.Value.(*knownAddress) rmka := entry.Value.(*knownAddress)
rmkey := NetAddressKey(rmka.na) // First bucket it would have been put in.
newBucket := a.getNewBucket(rmka.na, rmka.srcAddr)
// replace with ka. // If no room in the original bucket, we put it in a bucket we just
// freed up a space in.
if len(a.addrNew[newBucket]) >= newBucketSize {
newBucket = oldBucket
}
// replace with ka in list.
ka.tried = true
entry.Value = ka entry.Value = ka
rmka.tried = false rmka.tried = false
rmka.refs++
// We don't touch a.nTried here since the number of tried stays the same
// but we decemented new above, raise it again since we're putting
// something back.
a.nNew++
rmkey := NetAddressKey(rmka.na)
log.Tracef("[AMGR] replacing %s with %s in tried", rmkey, addrKey) log.Tracef("[AMGR] replacing %s with %s in tried", rmkey, addrKey)
// We know there is space for it since we just moved out of new. // We made sure there is space here just above.
// TODO(oga) when we move to multiple buckets then we will need to a.addrNew[newBucket][rmkey] = rmka
// check for size and consider putting it elsewhere.
a.addrNew[rmkey] = rmka
} }
// RFC1918: IPv4 Private networks (10.0.0.0/8, 192.168.0.0/16, 172.16.0.0/12) // RFC1918: IPv4 Private networks (10.0.0.0/8, 192.168.0.0/16, 172.16.0.0/12)

View file

@ -261,9 +261,7 @@ func (p *peer) handleVersionMsg(msg *btcwire.MsgVersion) {
p.Disconnect() p.Disconnect()
return return
} }
addresses := map[string]*btcwire.NetAddress{ addresses := []*btcwire.NetAddress{na}
NetAddressKey(na): na,
}
p.pushAddrMsg(addresses) p.pushAddrMsg(addresses)
} }
@ -640,7 +638,7 @@ func (p *peer) handleGetAddrMsg(msg *btcwire.MsgGetAddr) {
// pushAddrMsg sends one, or more, addr message(s) to the connected peer using // pushAddrMsg sends one, or more, addr message(s) to the connected peer using
// the provided addresses. // the provided addresses.
func (p *peer) pushAddrMsg(addresses map[string]*btcwire.NetAddress) error { func (p *peer) pushAddrMsg(addresses []*btcwire.NetAddress) error {
// Nothing to send. // Nothing to send.
if len(addresses) == 0 { if len(addresses) == 0 {
return nil return nil
@ -712,6 +710,8 @@ func (p *peer) handleAddrMsg(msg *btcwire.MsgAddr) {
// Add addresses to server address manager. The address manager handles // Add addresses to server address manager. The address manager handles
// the details of things such as preventing duplicate addresses, max // the details of things such as preventing duplicate addresses, max
// addresses, and last seen updates. // addresses, and last seen updates.
// XXX bitcoind gives a 2 hour time penalty here, do we want to do the
// same?
p.server.addrManager.AddAddresses(msg.AddrList, p.na) p.server.addrManager.AddAddresses(msg.AddrList, p.na)
} }