From 26c625d8e61cc18a16d2d38b13140faca0c55b85 Mon Sep 17 00:00:00 2001 From: zelig Date: Fri, 9 Jan 2015 12:15:31 +0000 Subject: [PATCH] cademlia improvements - parameter are Cademlia public fields - compulsory Hash, verifiable HashBits, Hash, RowLength, adjustable MaxAge, PurgeInterval, ProxSize - they default to package level constants and vars - row initialisation - adjustProx, updateProx to 'update depth' - proper global locking thanks to Prox - map from entries to peerInfo for GetPeers - call disconnect on the worst peer expunged when inserting - locking for purge - extract self.Distance <- self.DistanceTo (made public method) - call disconnect when purging --- p2p/cademlia.go | 140 +++++++++++++++++++++++++++++++++++------------- 1 file changed, 104 insertions(+), 36 deletions(-) diff --git a/p2p/cademlia.go b/p2p/cademlia.go index a89023aa02..21f0c98a3d 100644 --- a/p2p/cademlia.go +++ b/p2p/cademlia.go @@ -13,31 +13,33 @@ var cadlogger = ethlogger.NewLogger("CAD") const ( hashBits = 160 rowLength = 10 - maxAge = 1 ) +var maxAge = 180 * time.Nanosecond +var purgeInterval = 300 * time.Second + type Cademlia struct { - hash []byte - hashBits int - rowLength int - rows [hashBits]*row + Hash []byte + HashBits int + RowLength int - depth int + MaxProxSize int - maxAge time.Duration - purgeInterval time.Duration + MaxAge time.Duration + PurgeInterval time.Duration + + proxLimit int + proxSize int + + rows []*row lock sync.RWMutex quitC chan bool } -func newCademlia(hash []byte) *Cademlia { +func NewCademlia(hash []byte) *Cademlia { return &Cademlia{ - hash: hash, - hashBits: hashBits, - rowLength: rowLength, - maxAge: maxAge * time.Second, - rows: [hashBits]*row{}, + Hash: hash, // compulsory fields without default } } @@ -47,6 +49,28 @@ func (self *Cademlia) Start() error { if self.quitC != nil { return nil } + // these + self.Hash can and should be checked against the + // saved file/db + if self.HashBits == 0 { + self.HashBits = hashBits + } + if self.RowLength == 0 { + self.RowLength = rowLength + } + // runtime parameters + if self.MaxProxSize == 0 { + self.MaxProxSize = self.RowLength + } + if self.MaxAge == time.Duration(0) { + self.MaxAge = maxAge + } + if self.PurgeInterval == time.Duration(0) { + self.PurgeInterval = purgeInterval + } + self.rows = make([]*row, self.HashBits) + for i, _ := range self.rows { + self.rows[i] = &row{} // will initialise row{int(0),[]*entry(nil),sync.Mutex} + } self.quitC = make(chan bool) go self.purgeLoop() return nil @@ -63,12 +87,14 @@ func (self *Cademlia) Stop() { } func (self *Cademlia) AddPeer(peer peerInfo) (err error) { - index := self.commonPrefixLength(peer.Hash()) + self.lock.Lock() + defer self.lock.Unlock() + index := self.DistanceTo(peer.Hash()) row := self.rows[index] - needed := row.insert(&entry{peer: peer}) - if needed { - if index >= self.depth { - go self.updateDepth() + added := row.insert(&entry{peer: peer}) + if added { + if index >= self.proxLimit { + go self.adjustProx(index, 1) } cadlogger.Infof("accept peer %x...", peer.Hash()[:8]) } else { @@ -78,16 +104,43 @@ func (self *Cademlia) AddPeer(peer peerInfo) (err error) { return } +func (self *Cademlia) adjustProx(r int, add int) { + if self.proxSize+add > self.MaxProxSize && + self.rows[r].len() > 0 { + self.proxLimit-- + } else { + self.proxSize += add + } +} + +func (self *Cademlia) updateProx() { + var sum, proxSize int + for _, r := range self.rows { + sum += r.len() + if sum <= self.MaxProxSize || r.len() == 0 { + proxSize = sum + } + } + self.lock.Lock() + self.proxSize = proxSize + self.lock.Unlock() +} + func (self *Cademlia) GetPeers(target []byte) (peers []peerInfo) { - index := self.commonPrefixLength(target) + self.lock.RLock() + defer self.lock.RUnlock() + index := self.DistanceTo(target) var entries []*entry - if index >= self.depth { - for i := self.depth; i < self.hashBits; i++ { + if index >= self.proxLimit { + for i := self.proxLimit; i < self.HashBits; i++ { entries = append(entries, self.rows[i].row...) } } else { entries = self.rows[index].row } + for _, entry := range entries { + peers = append(peers, entry.peer) + } return } @@ -102,16 +155,23 @@ type row struct { lock sync.RWMutex } -func (self *row) insert(entry *entry) (ok bool) { +func (self *row) len() int { + self.lock.RLock() + defer self.lock.RUnlock() + return self.length +} + +func (self *row) insert(entry *entry) (added bool) { self.lock.Lock() defer self.lock.Unlock() - if len(self.row) >= self.length { + if len(self.row) >= self.length { // >= allows us to add peers beyond the Rowlength limitation worst := self.worst() - // err = diconnectF(self.row[worst]) + self.row[worst].peer.Disconnect(DiscSubprotocolError) self.row[worst] = entry } else { self.row = append(self.row, entry) - ok = true + added = true + self.length++ } return } @@ -128,33 +188,37 @@ func (self *row) worst() (index int) { } func (self *row) purge(recently time.Time) { + self.lock.Lock() var newRow []*entry for _, entry := range self.row { if !entry.peer.LastActive().Before(recently) { newRow = append(newRow, entry) } else { - // self.DisconnectF(entry.peer) + entry.peer.Disconnect(DiscSubprotocolError) } } + self.row = newRow + self.length = len(newRow) + self.lock.Unlock() } func Hash(key []byte) []byte { return key } -func (self *Cademlia) updateDepth() { -} - func (self *Cademlia) purgeLoop() { - ticker := time.Tick(self.purgeInterval) + ticker := time.Tick(self.PurgeInterval) for { select { case <-self.quitC: return case <-ticker: + self.lock.Lock() for _, r := range self.rows { - r.purge(time.Now().Add(-self.maxAge)) + r.purge(time.Now().Add(-self.MaxAge)) } + self.updateProx() + self.lock.Unlock() } } } @@ -166,14 +230,18 @@ func Xor(one, other []byte) (xor []byte) { return } -func (self *Cademlia) commonPrefixLength(other []byte) (ret int) { - xor := Xor(self.hash, other) - for i := 0; i < self.hashBits; i++ { +func (self *Cademlia) DistanceTo(other []byte) (ret int) { + return self.Distance(self.Hash, other) +} + +func (self *Cademlia) Distance(one, other []byte) (ret int) { + xor := Xor(one, other) + for i := 0; i < self.HashBits; i++ { for j := 0; j < 8; j++ { if (xor[i]>>uint8(7-j))&0x1 != 0 { return i*8 + j } } } - return self.hashBits*8 - 1 + return self.HashBits*8 - 1 }