p2p/discv5: search and lookup improvement

This commit is contained in:
Zsolt Felfoldi 2016-11-27 13:12:11 +01:00
parent 3e617f3cd6
commit a6d3bf6fc3
2 changed files with 99 additions and 64 deletions

View File

@ -127,7 +127,14 @@ type topicRegisterReq struct {
type topicSearchReq struct { type topicSearchReq struct {
topic Topic topic Topic
found chan<- string found chan<- *Node
lookup chan<- bool
delay time.Duration
}
type topicSearchResult struct {
target lookupInfo
nodes []*Node
} }
type timeoutEvent struct { type timeoutEvent struct {
@ -263,7 +270,9 @@ func (net *Network) lookup(target common.Hash, stopOnMatch bool) []*Node {
break break
} }
// Wait for the next reply. // Wait for the next reply.
for _, n := range <-reply { select {
case nodes := <-reply:
for _, n := range nodes {
if n != nil && !seen[n.ID] { if n != nil && !seen[n.ID] {
seen[n.ID] = true seen[n.ID] = true
result.push(n, bucketSize) result.push(n, bucketSize)
@ -273,6 +282,11 @@ func (net *Network) lookup(target common.Hash, stopOnMatch bool) []*Node {
} }
} }
pendingQueries-- pendingQueries--
case <-time.After(respTimeout):
// forget all pending requests, start new ones
pendingQueries = 0
reply = make(chan []*Node, alpha)
}
} }
return result.entries return result.entries
} }
@ -293,18 +307,20 @@ func (net *Network) RegisterTopic(topic Topic, stop <-chan struct{}) {
} }
} }
func (net *Network) SearchTopic(topic Topic, stop <-chan struct{}, found chan<- string) { func (net *Network) SearchTopic(topic Topic, setPeriod <-chan time.Duration, found chan<- *Node, lookup chan<- bool) {
for {
select { select {
case net.topicSearchReq <- topicSearchReq{topic, found}: case <-net.closed:
return
case delay, ok := <-setPeriod:
select {
case net.topicSearchReq <- topicSearchReq{topic: topic, found: found, lookup: lookup, delay: delay}:
case <-net.closed: case <-net.closed:
return return
} }
select { if !ok {
case <-net.closed: return
case <-stop: }
select {
case net.topicSearchReq <- topicSearchReq{topic, nil}:
case <-net.closed:
} }
} }
} }
@ -347,6 +363,13 @@ func (net *Network) reqTableOp(f func()) (called bool) {
// TODO: external address handling. // TODO: external address handling.
type topicSearchInfo struct {
lookupChn chan<- bool
period time.Duration
}
const maxSearchCount = 5
func (net *Network) loop() { func (net *Network) loop() {
var ( var (
refreshTimer = time.NewTicker(autoRefreshInterval) refreshTimer = time.NewTicker(autoRefreshInterval)
@ -385,10 +408,12 @@ func (net *Network) loop() {
topicRegisterLookupTarget lookupInfo topicRegisterLookupTarget lookupInfo
topicRegisterLookupDone chan []*Node topicRegisterLookupDone chan []*Node
topicRegisterLookupTick = time.NewTimer(0) topicRegisterLookupTick = time.NewTimer(0)
topicSearchLookupTarget lookupInfo
searchReqWhenRefreshDone []topicSearchReq searchReqWhenRefreshDone []topicSearchReq
searchInfo = make(map[Topic]topicSearchInfo)
activeSearchCount int
) )
topicSearchLookupDone := make(chan []*Node, 1) topicSearchLookupDone := make(chan topicSearchResult, 100)
topicSearch := make(chan Topic, 100)
<-topicRegisterLookupTick.C <-topicRegisterLookupTick.C
statsDump := time.NewTicker(10 * time.Second) statsDump := time.NewTicker(10 * time.Second)
@ -504,21 +529,52 @@ loop:
case req := <-net.topicSearchReq: case req := <-net.topicSearchReq:
if refreshDone == nil { if refreshDone == nil {
debugLog("<-net.topicSearchReq") debugLog("<-net.topicSearchReq")
if req.found == nil { info, ok := searchInfo[req.topic]
if ok {
if req.delay == time.Duration(0) {
delete(searchInfo, req.topic)
net.ticketStore.removeSearchTopic(req.topic) net.ticketStore.removeSearchTopic(req.topic)
} else {
info.period = req.delay
searchInfo[req.topic] = info
}
continue continue
} }
if req.delay != time.Duration(0) {
var info topicSearchInfo
info.period = req.delay
info.lookupChn = req.lookup
searchInfo[req.topic] = info
net.ticketStore.addSearchTopic(req.topic, req.found) net.ticketStore.addSearchTopic(req.topic, req.found)
if (topicSearchLookupTarget.target == common.Hash{}) { topicSearch <- req.topic
topicSearchLookupDone <- nil
} }
} else { } else {
searchReqWhenRefreshDone = append(searchReqWhenRefreshDone, req) searchReqWhenRefreshDone = append(searchReqWhenRefreshDone, req)
} }
case nodes := <-topicSearchLookupDone: case topic := <-topicSearch:
debugLog("<-topicSearchLookupDone") if activeSearchCount < maxSearchCount {
net.ticketStore.searchLookupDone(topicSearchLookupTarget, nodes, func(n *Node) []byte { activeSearchCount++
target := net.ticketStore.nextSearchLookup(topic)
go func() {
nodes := net.lookup(target.target, false)
topicSearchLookupDone <- topicSearchResult{target: target, nodes: nodes}
}()
}
period := searchInfo[topic].period
if period != time.Duration(0) {
go func() {
time.Sleep(period)
topicSearch <- topic
}()
}
case res := <-topicSearchLookupDone:
activeSearchCount--
if lookupChn := searchInfo[res.target.topic].lookupChn; lookupChn != nil {
lookupChn <- net.ticketStore.radius[res.target.topic].converged
}
net.ticketStore.searchLookupDone(res.target, res.nodes, func(n *Node) []byte {
net.ping(n, n.addr()) net.ping(n, n.addr())
return n.pingEcho return n.pingEcho
}, func(n *Node, topic Topic) []byte { }, func(n *Node, topic Topic) []byte {
@ -531,11 +587,6 @@ loop:
return nil return nil
} }
}) })
topicSearchLookupTarget = net.ticketStore.nextSearchLookup()
target := topicSearchLookupTarget.target
if (target != common.Hash{}) {
go func() { topicSearchLookupDone <- net.lookup(target, false) }()
}
case <-statsDump.C: case <-statsDump.C:
debugLog("<-statsDump.C") debugLog("<-statsDump.C")

View File

@ -138,16 +138,12 @@ type ticketStore struct {
nextTicketReg mclock.AbsTime nextTicketReg mclock.AbsTime
searchTopicMap map[Topic]searchTopic searchTopicMap map[Topic]searchTopic
searchTopicList []Topic
searchTopicPtr int
nextTopicQueryCleanup mclock.AbsTime nextTopicQueryCleanup mclock.AbsTime
queriesSent map[*Node]map[common.Hash]sentQuery queriesSent map[*Node]map[common.Hash]sentQuery
radiusLookupCnt int
} }
type searchTopic struct { type searchTopic struct {
foundChn chan<- string foundChn chan<- *Node
listIdx int
} }
type sentQuery struct { type sentQuery struct {
@ -183,23 +179,15 @@ func (s *ticketStore) addTopic(t Topic, register bool) {
} }
} }
func (s *ticketStore) addSearchTopic(t Topic, foundChn chan<- string) { func (s *ticketStore) addSearchTopic(t Topic, foundChn chan<- *Node) {
s.addTopic(t, false) s.addTopic(t, false)
if s.searchTopicMap[t].foundChn == nil { if s.searchTopicMap[t].foundChn == nil {
s.searchTopicList = append(s.searchTopicList, t) s.searchTopicMap[t] = searchTopic{foundChn: foundChn}
s.searchTopicMap[t] = searchTopic{foundChn: foundChn, listIdx: len(s.searchTopicList) - 1}
} }
} }
func (s *ticketStore) removeSearchTopic(t Topic) { func (s *ticketStore) removeSearchTopic(t Topic) {
if st := s.searchTopicMap[t]; st.foundChn != nil { if st := s.searchTopicMap[t]; st.foundChn != nil {
lastIdx := len(s.searchTopicList) - 1
lastTopic := s.searchTopicList[lastIdx]
s.searchTopicList[st.listIdx] = lastTopic
sl := s.searchTopicMap[lastTopic]
sl.listIdx = st.listIdx
s.searchTopicMap[lastTopic] = sl
s.searchTopicList = s.searchTopicList[:lastIdx]
delete(s.searchTopicMap, t) delete(s.searchTopicMap, t)
} }
} }
@ -247,20 +235,13 @@ func (s *ticketStore) nextRegisterLookup() (lookup lookupInfo, delay time.Durati
return lookupInfo{}, 40 * time.Second return lookupInfo{}, 40 * time.Second
} }
func (s *ticketStore) nextSearchLookup() lookupInfo { func (s *ticketStore) nextSearchLookup(topic Topic) lookupInfo {
if len(s.searchTopicList) == 0 { tr := s.radius[topic]
return lookupInfo{} target := tr.nextTarget(tr.radiusLookupCnt >= searchForceQuery)
}
if s.searchTopicPtr >= len(s.searchTopicList) {
s.searchTopicPtr = 0
}
topic := s.searchTopicList[s.searchTopicPtr]
s.searchTopicPtr++
target := s.radius[topic].nextTarget(s.radiusLookupCnt >= searchForceQuery)
if target.radiusLookup { if target.radiusLookup {
s.radiusLookupCnt++ tr.radiusLookupCnt++
} else { } else {
s.radiusLookupCnt = 0 tr.radiusLookupCnt = 0
} }
return target return target
} }
@ -662,9 +643,9 @@ func (s *ticketStore) gotTopicNodes(from *Node, hash common.Hash, nodes []rpcNod
if ip.IsUnspecified() || ip.IsLoopback() { if ip.IsUnspecified() || ip.IsLoopback() {
ip = from.IP ip = from.IP
} }
enode := NewNode(node.ID, ip, node.UDP-1, node.TCP-1).String() // subtract one from port while discv5 is running in test mode on UDPport+1 n := NewNode(node.ID, ip, node.UDP-1, node.TCP-1) // subtract one from port while discv5 is running in test mode on UDPport+1
select { select {
case chn <- enode: case chn <- n:
default: default:
return false return false
} }
@ -677,6 +658,8 @@ type topicRadius struct {
topicHashPrefix uint64 topicHashPrefix uint64
radius, minRadius uint64 radius, minRadius uint64
buckets []topicRadiusBucket buckets []topicRadiusBucket
converged bool
radiusLookupCnt int
} }
type topicRadiusEvent int type topicRadiusEvent int
@ -706,7 +689,7 @@ func (b *topicRadiusBucket) update(now mclock.AbsTime) {
b.lastTime = now b.lastTime = now
for target, tm := range b.lookupSent { for target, tm := range b.lookupSent {
if now-tm > mclock.AbsTime(pingTimeout) { if now-tm > mclock.AbsTime(respTimeout) {
b.weights[trNoAdjust] += 1 b.weights[trNoAdjust] += 1
delete(b.lookupSent, target) delete(b.lookupSent, target)
} }
@ -906,6 +889,7 @@ func (r *topicRadius) recalcRadius() (radius uint64, radiusLookup int) {
if radiusLookup == -1 { if radiusLookup == -1 {
// no more radius lookups needed at the moment, return a radius // no more radius lookups needed at the moment, return a radius
r.converged = true
rad := maxBucket rad := maxBucket
if minRadBucket < rad { if minRadBucket < rad {
rad = minRadBucket rad = minRadBucket