LBRYCommunity/reflector.go
1
0
Fork 0
Code Issues 7 Pull requests Projects Releases 10 Packages Wiki Activity

bucket splitting is solid

Browse source
This commit is contained in:
Alex Grintsvayg 2018-07-12 14:34:24 -04:00
parent bbe3bee3b0
commit 2e83654f1a
4 changed files with 225 additions and 153 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
Gopkg.lock generated
View file

@ -73,6 +73,27 @@
packages = ["."] packages = ["."]
revision = "3287d94d4c6a48a63e16fffaabf27ab20203af2a" revision = "3287d94d4c6a48a63e16fffaabf27ab20203af2a"
[[projects]]
name = "github.com/gorilla/context"
packages = ["."]
revision = "08b5f424b9271eedf6f9f0ce86cb9396ed337a42"
version = "v1.1.1"
[[projects]]
name = "github.com/gorilla/mux"
packages = ["."]
revision = "e3702bed27f0d39777b0b37b664b6280e8ef8fbf"
version = "v1.6.2"
[[projects]]
name = "github.com/gorilla/rpc"
packages = [
".",
"json"
]
revision = "22c016f3df3febe0c1f6727598b6389507e03a18"
version = "v1.1.0"
[[projects]] [[projects]]
name = "github.com/gorilla/websocket" name = "github.com/gorilla/websocket"
packages = ["."] packages = ["."]
@ -268,6 +289,6 @@
[solve-meta] [solve-meta]
analyzer-name = "dep" analyzer-name = "dep"
analyzer-version = 1 analyzer-version = 1
inputs-digest = "4dc432f7df1c1d59d5ee47417ab4f0fe187d26eb9e1f53fecdb6396b3bd1e6e0" inputs-digest = "6fac5a5bd6eb2f49d18558f8ed96b510e0852f95d7c746e301d53f5df92fffc4"
solver-name = "gps-cdcl" solver-name = "gps-cdcl"
solver-version = 1 solver-version = 1

32
dht/bootstrap.go
View file

@ -21,7 +21,7 @@ type BootstrapNode struct {
checkInterval time.Duration checkInterval time.Duration
nlock *sync.RWMutex nlock *sync.RWMutex
nodes map[bits.Bitmap]*peer peers map[bits.Bitmap]*peer
nodeIDs []bits.Bitmap // necessary for efficient random ID selection nodeIDs []bits.Bitmap // necessary for efficient random ID selection
} }
@ -34,7 +34,7 @@ func NewBootstrapNode(id bits.Bitmap, initialPingInterval, rePingInterval time.D
checkInterval: rePingInterval, checkInterval: rePingInterval,
nlock: &sync.RWMutex{}, nlock: &sync.RWMutex{},
nodes: make(map[bits.Bitmap]*peer), peers: make(map[bits.Bitmap]*peer),
nodeIDs: make([]bits.Bitmap, 0), nodeIDs: make([]bits.Bitmap, 0),
} }
@ -77,14 +77,14 @@ func (b *BootstrapNode) upsert(c Contact) {
b.nlock.Lock() b.nlock.Lock()
defer b.nlock.Unlock() defer b.nlock.Unlock()
if node, exists := b.nodes[c.ID]; exists { if peer, exists := b.peers[c.ID]; exists {
log.Debugf("[%s] bootstrap: touching contact %s", b.id.HexShort(), node.Contact.ID.HexShort()) log.Debugf("[%s] bootstrap: touching contact %s", b.id.HexShort(), peer.Contact.ID.HexShort())
node.Touch() peer.Touch()
return return
} }
log.Debugf("[%s] bootstrap: adding new contact %s", b.id.HexShort(), c.ID.HexShort()) log.Debugf("[%s] bootstrap: adding new contact %s", b.id.HexShort(), c.ID.HexShort())
b.nodes[c.ID] = &peer{c, time.Now(), 0} b.peers[c.ID] = &peer{c, b.id.Xor(c.ID), time.Now(), 0}
b.nodeIDs = append(b.nodeIDs, c.ID) b.nodeIDs = append(b.nodeIDs, c.ID)
} }
@ -93,13 +93,13 @@ func (b *BootstrapNode) remove(c Contact) {
b.nlock.Lock() b.nlock.Lock()
defer b.nlock.Unlock() defer b.nlock.Unlock()
_, exists := b.nodes[c.ID] _, exists := b.peers[c.ID]
if !exists { if !exists {
return return
} }
log.Debugf("[%s] bootstrap: removing contact %s", b.id.HexShort(), c.ID.HexShort()) log.Debugf("[%s] bootstrap: removing contact %s", b.id.HexShort(), c.ID.HexShort())
delete(b.nodes, c.ID) delete(b.peers, c.ID)
for i := range b.nodeIDs { for i := range b.nodeIDs {
if b.nodeIDs[i].Equals(c.ID) { if b.nodeIDs[i].Equals(c.ID) {
b.nodeIDs = append(b.nodeIDs[:i], b.nodeIDs[i+1:]...) b.nodeIDs = append(b.nodeIDs[:i], b.nodeIDs[i+1:]...)
@ -113,13 +113,13 @@ func (b *BootstrapNode) get(limit int) []Contact {
b.nlock.RLock() b.nlock.RLock()
defer b.nlock.RUnlock() defer b.nlock.RUnlock()
if len(b.nodes) < limit { if len(b.peers) < limit {
limit = len(b.nodes) limit = len(b.peers)
} }
ret := make([]Contact, limit) ret := make([]Contact, limit)
for i, k := range randKeys(len(b.nodeIDs))[:limit] { for i, k := range randKeys(len(b.nodeIDs))[:limit] {
ret[i] = b.nodes[b.nodeIDs[k]].Contact ret[i] = b.peers[b.nodeIDs[k]].Contact
} }
return ret return ret
@ -152,9 +152,9 @@ func (b *BootstrapNode) check() {
b.nlock.RLock() b.nlock.RLock()
defer b.nlock.RUnlock() defer b.nlock.RUnlock()
for i := range b.nodes { for i := range b.peers {
if !b.nodes[i].ActiveInLast(b.checkInterval) { if !b.peers[i].ActiveInLast(b.checkInterval) {
go b.ping(b.nodes[i].Contact) go b.ping(b.peers[i].Contact)
} }
} }
} }
@ -185,13 +185,13 @@ func (b *BootstrapNode) handleRequest(addr *net.UDPAddr, request Request) {
go func() { go func() {
b.nlock.RLock() b.nlock.RLock()
_, exists := b.nodes[request.NodeID] _, exists := b.peers[request.NodeID]
b.nlock.RUnlock() b.nlock.RUnlock()
if !exists { if !exists {
log.Debugf("[%s] bootstrap: queuing %s to ping", b.id.HexShort(), request.NodeID.HexShort()) log.Debugf("[%s] bootstrap: queuing %s to ping", b.id.HexShort(), request.NodeID.HexShort())
<-time.After(b.initialPingInterval) <-time.After(b.initialPingInterval)
b.nlock.RLock() b.nlock.RLock()
_, exists = b.nodes[request.NodeID] _, exists = b.peers[request.NodeID]
b.nlock.RUnlock() b.nlock.RUnlock()
if !exists { if !exists {
b.ping(Contact{ID: request.NodeID, IP: addr.IP, Port: addr.Port}) b.ping(Contact{ID: request.NodeID, IP: addr.IP, Port: addr.Port})

105
dht/routing_table.go
View file

@ -10,7 +10,6 @@ import (
"sync" "sync"
"time" "time"
"github.com/davecgh/go-spew/spew"
"github.com/lbryio/lbry.go/errors" "github.com/lbryio/lbry.go/errors"
"github.com/lbryio/lbry.go/stop" "github.com/lbryio/lbry.go/stop"
"github.com/lbryio/reflector.go/dht/bits" "github.com/lbryio/reflector.go/dht/bits"
@ -21,9 +20,10 @@ import (
// TODO: use a tree with bucket splitting instead of a fixed bucket list. include jack's optimization (see link in commit mesg) // TODO: use a tree with bucket splitting instead of a fixed bucket list. include jack's optimization (see link in commit mesg)
// https://github.com/lbryio/lbry/pull/1211/commits/341b27b6d21ac027671d42458826d02735aaae41 // https://github.com/lbryio/lbry/pull/1211/commits/341b27b6d21ac027671d42458826d02735aaae41
// peer is a contact with extra freshness information // peer is a contact with extra information
type peer struct { type peer struct {
Contact Contact Contact Contact
Distance bits.Bitmap
LastActivity time.Time LastActivity time.Time
// LastReplied time.Time // LastReplied time.Time
// LastRequested time.Time // LastRequested time.Time
@ -79,7 +79,7 @@ func (b bucket) Len() int {
return len(b.peers) return len(b.peers)
} }
func (b bucket) Contains(c Contact) bool { func (b bucket) Has(c Contact) bool {
b.lock.RLock() b.lock.RLock()
defer b.lock.RUnlock() defer b.lock.RUnlock()
for _, p := range b.peers { for _, p := range b.peers {
@ -101,30 +101,27 @@ func (b bucket) Contacts() []Contact {
return contacts return contacts
} }
// UpdateContact marks a contact as having been successfully contacted. if insertIfNew and the contact is does not exist yet, it is inserted // UpdatePeer marks a contact as having been successfully contacted. if insertIfNew and the contact is does not exist yet, it is inserted
func (b *bucket) UpdateContact(c Contact, insertIfNew bool) { func (b *bucket) UpdatePeer(p peer, insertIfNew bool) error {
b.lock.Lock() b.lock.Lock()
defer b.lock.Unlock() defer b.lock.Unlock()
fmt.Printf("updating contact %s\n", c.ID)
// TODO: verify the peer is in the bucket key range if !b.Range.Contains(p.Distance) {
return errors.Err("this bucket range does not cover this peer")
}
peerIndex := find(c.ID, b.peers) peerIndex := find(p.Contact.ID, b.peers)
if peerIndex >= 0 { if peerIndex >= 0 {
fmt.Println("exists, moving to back")
b.lastUpdate = time.Now() b.lastUpdate = time.Now()
b.peers[peerIndex].Touch() b.peers[peerIndex].Touch()
moveToBack(b.peers, peerIndex) moveToBack(b.peers, peerIndex)
} else if insertIfNew { } else if insertIfNew {
fmt.Println("inserting new")
hasRoom := true hasRoom := true
if len(b.peers) >= bucketSize { if len(b.peers) >= bucketSize {
fmt.Println("no room")
hasRoom = false hasRoom = false
for i := range b.peers { for i := range b.peers {
if b.peers[i].IsBad(maxPeerFails) { if b.peers[i].IsBad(maxPeerFails) {
fmt.Println("dropping bad peer to make room")
// TODO: Ping contact first. Only remove if it does not respond // TODO: Ping contact first. Only remove if it does not respond
b.peers = append(b.peers[:i], b.peers[i+1:]...) b.peers = append(b.peers[:i], b.peers[i+1:]...)
hasRoom = true hasRoom = true
@ -134,15 +131,13 @@ func (b *bucket) UpdateContact(c Contact, insertIfNew bool) {
} }
if hasRoom { if hasRoom {
fmt.Println("actually adding")
b.lastUpdate = time.Now() b.lastUpdate = time.Now()
peer := peer{Contact: c} p.Touch()
peer.Touch() b.peers = append(b.peers, p)
b.peers = append(b.peers, peer)
} else {
fmt.Println("no room, dropping")
} }
} }
return nil
} }
// FailContact marks a contact as having failed, and removes it if it failed too many times // FailContact marks a contact as having failed, and removes it if it failed too many times
@ -183,19 +178,21 @@ func (b *bucket) Split() (*bucket, *bucket) {
right.lastUpdate = b.lastUpdate right.lastUpdate = b.lastUpdate
for _, p := range b.peers { for _, p := range b.peers {
if left.Range.Contains(p.Contact.ID) { if left.Range.Contains(p.Distance) {
left.peers = append(left.peers, p) left.peers = append(left.peers, p)
} else { } else {
right.peers = append(right.peers, p) right.peers = append(right.peers, p)
} }
} }
if len(left.peers) == 0 { if len(b.peers) > 1 {
left, right = right.Split() if len(left.peers) == 0 {
left.Range.Start = b.Range.Start left, right = right.Split()
} else if len(right.peers) == 0 { left.Range.Start = b.Range.Start
left, right = left.Split() } else if len(right.peers) == 0 {
right.Range.End = b.Range.End left, right = left.Split()
right.Range.End = b.Range.End
}
} }
return left, right return left, right
@ -248,22 +245,33 @@ func (rt *routingTable) Update(c Contact) {
rt.mu.Lock() // write lock, because updates may cause bucket splits rt.mu.Lock() // write lock, because updates may cause bucket splits
defer rt.mu.Unlock() defer rt.mu.Unlock()
if rt.shouldSplit(c) { b := rt.bucketFor(c.ID)
spew.Dump("splitting")
i := rt.bucketNumFor(c.ID) if rt.shouldSplit(b, c) {
left, right := rt.buckets[i].Split() left, right := b.Split()
rt.buckets = append(rt.buckets[:i], append([]*bucket{left, right}, rt.buckets[i+1:]...)...)
} else { for i := range rt.buckets {
spew.Dump("no split") if rt.buckets[i].Range.Start.Equals(left.Range.Start) {
rt.buckets = append(rt.buckets[:i], append([]*bucket{left, right}, rt.buckets[i+1:]...)...)
break
}
}
if left.Range.Contains(c.ID) {
b = left
} else {
b = right
}
} }
rt.buckets[rt.bucketNumFor(c.ID)].UpdateContact(c, true)
b.UpdatePeer(peer{Contact: c, Distance: rt.id.Xor(c.ID)}, true)
} }
// Fresh refreshes a contact if its already in the routing table // Fresh refreshes a contact if its already in the routing table
func (rt *routingTable) Fresh(c Contact) { func (rt *routingTable) Fresh(c Contact) {
rt.mu.RLock() rt.mu.RLock()
defer rt.mu.RUnlock() defer rt.mu.RUnlock()
rt.bucketFor(c.ID).UpdateContact(c, false) rt.bucketFor(c.ID).UpdatePeer(peer{Contact: c, Distance: rt.id.Xor(c.ID)}, false)
} }
// FailContact marks a contact as having failed, and removes it if it failed too many times // FailContact marks a contact as having failed, and removes it if it failed too many times
@ -319,38 +327,21 @@ func (rt *routingTable) Len() int {
return len(rt.buckets) return len(rt.buckets)
} }
// BucketRanges returns a slice of ranges, where the `start` of each range is the smallest id that can func (rt *routingTable) bucketFor(target bits.Bitmap) *bucket {
// go in that bucket, and the `end` is the largest id
func (rt *routingTable) BucketRanges() []bits.Range {
rt.mu.RLock()
defer rt.mu.RUnlock()
ranges := make([]bits.Range, len(rt.buckets))
for i, b := range rt.buckets {
ranges[i] = b.Range
}
return ranges
}
func (rt *routingTable) bucketNumFor(target bits.Bitmap) int {
if rt.id.Equals(target) { if rt.id.Equals(target) {
panic("routing table does not have a bucket for its own id") panic("routing table does not have a bucket for its own id")
} }
distance := target.Xor(rt.id) distance := target.Xor(rt.id)
for i, b := range rt.buckets { for _, b := range rt.buckets {
if b.Range.Contains(distance) { if b.Range.Contains(distance) {
return i return b
} }
} }
panic("target value overflows the key space") panic("target is not contained in any buckets")
} }
func (rt *routingTable) bucketFor(target bits.Bitmap) *bucket { func (rt *routingTable) shouldSplit(b *bucket, c Contact) bool {
return rt.buckets[rt.bucketNumFor(target)] if b.Has(c) {
}
func (rt *routingTable) shouldSplit(c Contact) bool {
b := rt.bucketFor(c.ID)
if b.Contains(c) {
return false return false
} }
if b.Len() >= bucketSize { if b.Len() >= bucketSize {

218
dht/routing_table_test.go
View file

@ -2,7 +2,6 @@ package dht
import ( import (
"encoding/json" "encoding/json"
"fmt"
"math/big" "math/big"
"net" "net"
"strconv" "strconv"
@ -13,37 +12,7 @@ import (
"github.com/sebdah/goldie" "github.com/sebdah/goldie"
) )
func checkBucketCount(t *testing.T, rt *routingTable, correctSize, correctCount, testCaseIndex int) { func TestBucket_Split(t *testing.T) {
if len(rt.buckets) != correctSize {
t.Errorf("failed test case %d. there should be %d buckets, got %d", testCaseIndex+1, correctSize, len(rt.buckets))
}
if rt.Count() != correctCount {
t.Errorf("failed test case %d. there should be %d contacts, got %d", testCaseIndex+1, correctCount, rt.Count())
}
}
func checkRangeContinuity(t *testing.T, rt *routingTable) {
position := big.NewInt(0)
for i, bucket := range rt.buckets {
bucketStart := bucket.Range.Start.Big()
if bucketStart.Cmp(position) != 0 {
t.Errorf("invalid start of bucket range: %s vs %s", position.String(), bucketStart.String())
}
if bucketStart.Cmp(bucket.Range.End.Big()) != -1 {
t.Error("range start is not less than bucket end")
}
position = bucket.Range.End.Big()
if i != len(rt.buckets)-1 {
position.Add(position, big.NewInt(1))
}
}
if position.Cmp(bits.MaxP().Big()) != 0 {
t.Errorf("range does not cover the whole keyspace, %s vs %s", bits.FromBigP(position).String(), bits.MaxP().String())
}
}
func TestSplitBuckets(t *testing.T) {
rt := newRoutingTable(bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000")) rt := newRoutingTable(bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"))
if len(rt.buckets) != 1 { if len(rt.buckets) != 1 {
t.Errorf("there should only be one bucket so far") t.Errorf("there should only be one bucket so far")
@ -53,53 +22,51 @@ func TestSplitBuckets(t *testing.T) {
} }
var tests = []struct { var tests = []struct {
name string
id bits.Bitmap id bits.Bitmap
expectedBucketCount int expectedBucketCount int
expectedTotalContacts int expectedTotalContacts int
}{ }{
//fill first bucket //fill first bucket
{bits.FromHexP("F00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 1, 1}, {"b1-one", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000100"), 1, 1},
{bits.FromHexP("FF0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 1, 2}, {"b1-two", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000200"), 1, 2},
{bits.FromHexP("FFF000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 1, 3}, {"b1-three", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000300"), 1, 3},
{bits.FromHexP("FFFF00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 1, 4}, {"b1-four", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000400"), 1, 4},
{bits.FromHexP("FFFFF0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 1, 5}, {"b1-five", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000500"), 1, 5},
{bits.FromHexP("FFFFFF000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 1, 6}, {"b1-six", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000600"), 1, 6},
{bits.FromHexP("FFFFFFF00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 1, 7}, {"b1-seven", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000700"), 1, 7},
{bits.FromHexP("FFFFFFFF0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 1, 8}, {"b1-eight", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000800"), 1, 8},
// fill second bucket // split off second bucket and fill it
{bits.FromHexP("FFFFFFFFF000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 9}, {"b2-one", bits.FromHexP("001000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 9},
{bits.FromHexP("FFFFFFFFFF00000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 10}, {"b2-two", bits.FromHexP("002000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 10},
{bits.FromHexP("FFFFFFFFFFF0000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 11}, {"b2-three", bits.FromHexP("003000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 11},
{bits.FromHexP("FFFFFFFFFFFF000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 12}, {"b2-four", bits.FromHexP("004000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 12},
{bits.FromHexP("FFFFFFFFFFFFF00000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 13}, {"b2-five", bits.FromHexP("005000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 13},
{bits.FromHexP("FFFFFFFFFFFFFF0000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 14}, {"b2-six", bits.FromHexP("006000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 14},
{bits.FromHexP("FFFFFFFFFFFFFFF000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 15}, {"b2-seven", bits.FromHexP("007000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 15},
{bits.FromHexP("FFFFFFFFFFFFFFFF00000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 16},
// this should be skipped (no split should occur) // at this point there are two buckets. the first has 7 contacts, the second has 8
{bits.FromHexP("FFFFFFFFFFFFFFFFF0000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 16},
{bits.FromHexP("100000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 3, 17}, // inserts into the second bucket should be skipped
{bits.FromHexP("200000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 3, 18}, {"dont-split", bits.FromHexP("009000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2, 15},
{bits.FromHexP("300000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 3, 19},
{bits.FromHexP("400000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 4, 20}, // ... unless the ID is closer than the kth-closest contact
{bits.FromHexP("500000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 4, 21}, {"split-kth-closest", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001"), 2, 16},
{bits.FromHexP("600000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 4, 22},
{bits.FromHexP("700000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 4, 23}, {"b3-two", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002"), 3, 17},
{bits.FromHexP("800000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 4, 24}, {"b3-three", bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003"), 3, 18},
{bits.FromHexP("900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 4, 25},
{bits.FromHexP("A00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 4, 26},
{bits.FromHexP("B00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 4, 27},
} }
for i, testCase := range tests { for i, testCase := range tests {
fmt.Printf("\n\n\ncase %d\n", i)
rt.Update(Contact{testCase.id, net.ParseIP("127.0.0.1"), 8000 + i}) rt.Update(Contact{testCase.id, net.ParseIP("127.0.0.1"), 8000 + i})
//spew.Dump(rt.buckets)
checkBucketCount(t, rt, testCase.expectedBucketCount, testCase.expectedTotalContacts, i) if len(rt.buckets) != testCase.expectedBucketCount {
checkRangeContinuity(t, rt) t.Errorf("failed test case %s. there should be %d buckets, got %d", testCase.name, testCase.expectedBucketCount, len(rt.buckets))
}
if rt.Count() != testCase.expectedTotalContacts {
t.Errorf("failed test case %s. there should be %d contacts, got %d", testCase.name, testCase.expectedTotalContacts, rt.Count())
}
} }
var testRanges = []struct { var testRanges = []struct {
@ -108,20 +75,115 @@ func TestSplitBuckets(t *testing.T) {
}{ }{
{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001"), 0}, {bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001"), 0},
{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000005"), 0}, {bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000005"), 0},
{bits.FromHexP("200000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010"), 1}, {bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000410"), 1},
{bits.FromHexP("380000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), 2}, {bits.FromHexP("0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000007f0"), 1},
{bits.FromHexP("F00000000000000000000000000000000000000000000000000F00000000000000000000000000000000000000000000"), 3}, {bits.FromHexP("F00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000800"), 2},
{bits.FromHexP("F0000000000000000000000000000000F0000000000000000000000000F0000000000000000000000000000000000000"), 3}, {bits.FromHexP("F00000000000000000000000000000000000000000000000000F00000000000000000000000000000000000000000000"), 2},
{bits.FromHexP("F0000000000000000000000000000000F0000000000000000000000000F0000000000000000000000000000000000000"), 2},
} }
for _, tt := range testRanges { for _, tt := range testRanges {
bucket := rt.bucketNumFor(tt.id) bucket := bucketNumFor(rt, tt.id)
if bucket != tt.expected { if bucket != tt.expected {
t.Errorf("bucketFor(%s, %s) => %d, want %d", tt.id.Hex(), rt.id.Hex(), bucket, tt.expected) t.Errorf("bucketFor(%s, %s) => got %d, expected %d", tt.id.Hex(), rt.id.Hex(), bucket, tt.expected)
} }
} }
}
rt.printBucketInfo() func bucketNumFor(rt *routingTable, target bits.Bitmap) int {
if rt.id.Equals(target) {
panic("routing table does not have a bucket for its own id")
}
distance := target.Xor(rt.id)
for i := range rt.buckets {
if rt.buckets[i].Range.Contains(distance) {
return i
}
}
panic("target is not contained in any buckets")
}
func TestBucket_Split_Continuous(t *testing.T) {
b := newBucket(bits.MaxRange())
left, right := b.Split()
if !left.Range.Start.Equals(b.Range.Start) {
t.Errorf("left bucket start does not align with original bucket start. got %s, expected %s", left.Range.Start, b.Range.Start)
}
if !right.Range.End.Equals(b.Range.End) {
t.Errorf("right bucket end does not align with original bucket end. got %s, expected %s", right.Range.End, b.Range.End)
}
leftEndNext := (&big.Int{}).Add(left.Range.End.Big(), big.NewInt(1))
if !bits.FromBigP(leftEndNext).Equals(right.Range.Start) {
t.Errorf("there's a gap between left bucket end and right bucket start. end is %s, start is %s", left.Range.End, right.Range.Start)
}
}
func TestBucket_Split_KthClosest_DoSplit(t *testing.T) {
rt := newRoutingTable(bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"))
// add 4 low IDs
rt.Update(Contact{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001"), net.ParseIP("127.0.0.1"), 8001})
rt.Update(Contact{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002"), net.ParseIP("127.0.0.1"), 8002})
rt.Update(Contact{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003"), net.ParseIP("127.0.0.1"), 8003})
rt.Update(Contact{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000004"), net.ParseIP("127.0.0.1"), 8004})
// add 4 high IDs
rt.Update(Contact{bits.FromHexP("800000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8001})
rt.Update(Contact{bits.FromHexP("900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8002})
rt.Update(Contact{bits.FromHexP("a00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8003})
rt.Update(Contact{bits.FromHexP("b00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8004})
// split the bucket and fill the high bucket
rt.Update(Contact{bits.FromHexP("c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8005})
rt.Update(Contact{bits.FromHexP("d00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8006})
rt.Update(Contact{bits.FromHexP("e00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8007})
rt.Update(Contact{bits.FromHexP("f00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8008})
// add a high ID. it should split because the high ID is closer than the Kth closest ID
rt.Update(Contact{bits.FromHexP("910000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.1"), 8009})
if len(rt.buckets) != 3 {
t.Errorf("expected 3 buckets, got %d", len(rt.buckets))
}
if rt.Count() != 13 {
t.Errorf("expected 13 contacts, got %d", rt.Count())
}
}
func TestBucket_Split_KthClosest_DontSplit(t *testing.T) {
rt := newRoutingTable(bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"))
// add 4 low IDs
rt.Update(Contact{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001"), net.ParseIP("127.0.0.1"), 8001})
rt.Update(Contact{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002"), net.ParseIP("127.0.0.1"), 8002})
rt.Update(Contact{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003"), net.ParseIP("127.0.0.1"), 8003})
rt.Update(Contact{bits.FromHexP("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000004"), net.ParseIP("127.0.0.1"), 8004})
// add 4 high IDs
rt.Update(Contact{bits.FromHexP("800000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8001})
rt.Update(Contact{bits.FromHexP("900000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8002})
rt.Update(Contact{bits.FromHexP("a00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8003})
rt.Update(Contact{bits.FromHexP("b00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8004})
// split the bucket and fill the high bucket
rt.Update(Contact{bits.FromHexP("c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8005})
rt.Update(Contact{bits.FromHexP("d00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8006})
rt.Update(Contact{bits.FromHexP("e00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8007})
rt.Update(Contact{bits.FromHexP("f00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.2"), 8008})
// add a really high ID. this should not split because its not closer than the Kth closest ID
rt.Update(Contact{bits.FromHexP("ffff00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"), net.ParseIP("127.0.0.1"), 8009})
if len(rt.buckets) != 2 {
t.Errorf("expected 2 buckets, got %d", len(rt.buckets))
}
if rt.Count() != 12 {
t.Errorf("expected 12 contacts, got %d", rt.Count())
}
} }
func TestRoutingTable_GetClosest(t *testing.T) { func TestRoutingTable_GetClosest(t *testing.T) {
@ -213,14 +275,12 @@ func TestRoutingTable_Save(t *testing.T) {
id := bits.FromHexP("1c8aff71b99462464d9eeac639595ab99664be3482cb91a29d87467515c7d9158fe72aa1f1582dab07d8f8b5db277f41") id := bits.FromHexP("1c8aff71b99462464d9eeac639595ab99664be3482cb91a29d87467515c7d9158fe72aa1f1582dab07d8f8b5db277f41")
rt := newRoutingTable(id) rt := newRoutingTable(id)
ranges := rt.BucketRanges() for i, b := range rt.buckets {
for i, r := range ranges {
for j := 0; j < bucketSize; j++ { for j := 0; j < bucketSize; j++ {
toAdd := r.Start.Add(bits.FromShortHexP(strconv.Itoa(j))) toAdd := b.Range.Start.Add(bits.FromShortHexP(strconv.Itoa(j)))
if toAdd.Cmp(r.End) <= 0 { if toAdd.Cmp(b.Range.End) <= 0 {
rt.Update(Contact{ rt.Update(Contact{
ID: r.Start.Add(bits.FromShortHexP(strconv.Itoa(j))), ID: b.Range.Start.Add(bits.FromShortHexP(strconv.Itoa(j))),
IP: net.ParseIP("1.2.3." + strconv.Itoa(j)), IP: net.ParseIP("1.2.3." + strconv.Itoa(j)),
Port: 1 + i*bucketSize + j, Port: 1 + i*bucketSize + j,
}) })