diff --git a/swarm/network/hive.go b/swarm/network/hive.go index dcf666d0b9..bb01365a09 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -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()) diff --git a/swarm/network/kademlia/kademlia.go b/swarm/network/kademlia/kademlia.go index e635c042c2..6df255e4a1 100644 --- a/swarm/network/kademlia/kademlia.go +++ b/swarm/network/kademlia/kademlia.go @@ -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) } diff --git a/swarm/network/kademlia/kademlia_test.go b/swarm/network/kademlia/kademlia_test.go index ca05e1ded5..de906cabe6 100644 --- a/swarm/network/kademlia/kademlia_test.go +++ b/swarm/network/kademlia/kademlia_test.go @@ -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 } diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index 80893336b7..f778773001 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -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