improve logging and minor reorg in kademlia

This commit is contained in:
zelig 2016-06-22 14:44:13 +02:00
parent babb3a7526
commit 0e23886184
4 changed files with 34 additions and 29 deletions

View file

@ -123,11 +123,10 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee
// to attempt to write to more (remove Peer when shutting down)
return
}
node, need, proxLimit := self.kad.FindBest()
node, need, proxLimit := self.kad.Suggest()
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: select candidate peer")
if node != nil && len(node.Url) > 0 {
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: call for bee %v", node.Url)
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: call known bee %v", node.Url)
// enode or any lower level connection address is unnecessary in future
// discovery table is used to look it up.
connectPeer(node.Url)
@ -141,15 +140,15 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee
req := &retrieveRequestMsgData{
Key: storage.Key(randAddr[:]),
}
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: call any bee in area %v messenger bee %v", randAddr, peers[0])
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: call any bee near %v (PO%03d) - messenger bee: %v", randAddr, proxLimit, peers[0])
peers[0].(*peer).retrieve(req)
} else {
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: no peer")
glog.V(logger.Warn).Infof("[BZZ] KΛÐΞMLIΛ hive: no peer")
}
self.toggle <- true
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: buzz kept alive")
glog.V(logger.Detail).Infof("[BZZ] KΛÐΞMLIΛ hive: buzz kept alive")
} else {
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: no need for more bees")
glog.V(logger.Info).Infof("[BZZ] KΛÐΞMLIΛ hive: no need for more bees")
self.toggle <- false
}
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: queen's address: %v, population: %d (%d)", self.addr, self.kad.Count(), self.kad.DBCount())

View file

@ -279,7 +279,7 @@ func (self *Kademlia) binsize(p int) int {
return len(b.nodes)
}
func (self *Kademlia) FindBest() (*NodeRecord, bool, int) {
func (self *Kademlia) Suggest() (*NodeRecord, bool, int) {
return self.db.findBest(self.BucketSize, self.binsize)
}

View file

@ -82,7 +82,7 @@ func TestBootstrap(t *testing.T) {
t.Fatalf("backend not accepting node: %v", err)
}
record, need, _ := kad.FindBest()
record, need, _ := kad.Suggest()
if !need {
break
}

View file

@ -298,6 +298,7 @@ func (self *syncer) sync() {
if state.LastSeenAt < state.SessionAt {
state.Last = state.SessionAt
glog.V(logger.Debug).Infof("[BZZ] syncer[%v]: start syncronising history since last disconnect at %v up until session start at %v: %v", self.key.Log(), state.LastSeenAt, state.SessionAt, state)
// blocks until state syncing is finished
self.syncState(state)
}
glog.V(logger.Info).Infof("[BZZ] syncer[%v]: syncing all history complete", self.key.Log())
@ -358,19 +359,18 @@ func (self *syncer) syncHistory(state *syncState) chan interface{} {
IT:
for {
key := it.Next()
if key != nil {
select {
// blocking until history channel is read from
case history <- storage.Key(key):
n++
glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: history: %v (%v keys)", self.key.Log(), key.Log(), n)
state.Latest = key
case <-self.quit:
return
}
} else {
if key == nil {
break IT
}
select {
// blocking until history channel is read from
case history <- storage.Key(key):
n++
glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: history: %v (%v keys)", self.key.Log(), key.Log(), n)
state.Latest = key
case <-self.quit:
return
}
}
glog.V(logger.Debug).Infof("[BZZ] syncer[%v]: finished syncing history between %v - %v for chunk addresses %v - %v (at %v) (chunks = %v)", self.key.Log(), state.First, state.Last, state.Start, state.Stop, state.Latest, n)
}()
@ -416,25 +416,31 @@ LOOP:
// are checked first - integrity can only be guaranteed if writing
// is locked while selecting
if priority != High || len(keys) == 0 {
// selection is not needed if the High priority queue has items
keys = nil
PRIORITIES:
for priority = High; priority >= 0; priority-- {
// the first priority channel that is non-empty will be assigned to keys
if len(self.keys[priority]) > 0 {
glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: reading request with priority %v", self.key.Log(), priority)
keys = self.keys[priority]
break
break PRIORITIES
}
glog.V(logger.Debug).Infof("[BZZ] syncer[%v/%v]: queue: [%v, %v, %v]", self.key.Log(), priority, len(self.keys[High]), len(self.keys[High]), len(self.keys[High]))
// if the input queue is empty on this level, resort to history if there is any
if uint(priority) == histPrior && history != nil {
glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: reading history for %v", self.key.Log(), self.key)
keys = history
break
break PRIORITIES
}
}
// if peer ready to receive but nothing to send
if keys == nil && deliveryRequest == nil {
// if no items left and switch to waiting mode
glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: buffers consumed. Waiting", self.key.Log())
newUnsyncedKeys = self.newUnsyncedKeys
}
}
// if peer ready to receive but nothing to send
if keys == nil && deliveryRequest == nil {
// if no items left and switch to waiting mode
glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: buffers consumed. Waiting", self.key.Log())
newUnsyncedKeys = self.newUnsyncedKeys
}
// send msg iff
@ -447,7 +453,7 @@ LOOP:
len(unsynced) > 0 && keys == nil ||
len(unsynced) == int(self.SyncBatchSize)) {
justSynced = false
// listen to requests again
// listen to requests
deliveryRequest = self.deliveryRequest
newUnsyncedKeys = nil // not care about data until next req comes in
// set sync to current counter