From ff832be9c55d000d71bd2dcddd4e842949f6eccd Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Fri, 24 Nov 2017 22:33:05 +0100 Subject: [PATCH] common, les: move WeightedRandomSelect to common package --- {les => common}/randselect.go | 39 +++++++++++++++--------------- {les => common}/randselect_test.go | 10 ++++---- les/distributor.go | 12 +++++---- les/serverpool.go | 39 +++++++++++++++--------------- 4 files changed, 51 insertions(+), 49 deletions(-) rename {les => common}/randselect.go (80%) rename {les => common}/randselect_test.go (93%) diff --git a/les/randselect.go b/common/randselect.go similarity index 80% rename from les/randselect.go rename to common/randselect.go index 1a9d0695bd..3b1ef8c9c5 100644 --- a/les/randselect.go +++ b/common/randselect.go @@ -14,44 +14,43 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -// Package les implements the Light Ethereum Subprotocol. -package les +package common import ( "math/rand" ) // wrsItem interface should be implemented by any entries that are to be selected from -// a weightedRandomSelect set. Note that recalculating monotonously decreasing item -// weights on-demand (without constantly calling update) is allowed +// a WeightedRandomSelect set. Note that recalculating monotonously decreasing item +// weights on-demand (without constantly calling Update) is allowed type wrsItem interface { Weight() int64 } -// weightedRandomSelect is capable of weighted random selection from a set of items -type weightedRandomSelect struct { +// WeightedRandomSelect is capable of weighted random selection from a set of items +type WeightedRandomSelect struct { root *wrsNode idx map[wrsItem]int } -// newWeightedRandomSelect returns a new weightedRandomSelect structure -func newWeightedRandomSelect() *weightedRandomSelect { - return &weightedRandomSelect{root: &wrsNode{maxItems: wrsBranches}, idx: make(map[wrsItem]int)} +// newWeightedRandomSelect returns a new WeightedRandomSelect structure +func NewWeightedRandomSelect() *WeightedRandomSelect { + return &WeightedRandomSelect{root: &wrsNode{maxItems: wrsBranches}, idx: make(map[wrsItem]int)} } -// update updates an item's weight, adds it if it was non-existent or removes it if +// Update updates an item's weight, adds it if it was non-existent or removes it if // the new weight is zero. Note that explicitly updating decreasing weights is not necessary. -func (w *weightedRandomSelect) update(item wrsItem) { +func (w *WeightedRandomSelect) Update(item wrsItem) { w.setWeight(item, item.Weight()) } -// remove removes an item from the set -func (w *weightedRandomSelect) remove(item wrsItem) { +// Remove removes an item from the set +func (w *WeightedRandomSelect) Remove(item wrsItem) { w.setWeight(item, 0) } // setWeight sets an item's weight to a specific value (removes it if zero) -func (w *weightedRandomSelect) setWeight(item wrsItem, weight int64) { +func (w *WeightedRandomSelect) setWeight(item wrsItem, weight int64) { idx, ok := w.idx[item] if ok { w.root.setWeight(idx, weight) @@ -72,17 +71,17 @@ func (w *weightedRandomSelect) setWeight(item wrsItem, weight int64) { } } -// choose randomly selects an item from the set, with a chance proportional to its +// Choose randomly selects an item from the set, with a chance proportional to its // current weight. If the weight of the chosen element has been decreased since the // last stored value, returns it with a newWeight/oldWeight chance, otherwise just // updates its weight and selects another one -func (w *weightedRandomSelect) choose() wrsItem { +func (w *WeightedRandomSelect) Choose() wrsItem { for { if w.root.sumWeight == 0 { return nil } val := rand.Int63n(w.root.sumWeight) - choice, lastWeight := w.root.choose(val) + choice, lastWeight := w.root.Choose(val) weight := choice.Weight() if weight != lastWeight { w.setWeight(choice, weight) @@ -156,14 +155,14 @@ func (n *wrsNode) setWeight(idx int, weight int64) int64 { return diff } -// choose recursively selects an item from the tree and returns it along with its weight -func (n *wrsNode) choose(val int64) (wrsItem, int64) { +// Choose recursively selects an item from the tree and returns it along with its weight +func (n *wrsNode) Choose(val int64) (wrsItem, int64) { for i, w := range n.weights { if val < w { if n.level == 0 { return n.items[i].(wrsItem), n.weights[i] } else { - return n.items[i].(*wrsNode).choose(val) + return n.items[i].(*wrsNode).Choose(val) } } else { val -= w diff --git a/les/randselect_test.go b/common/randselect_test.go similarity index 93% rename from les/randselect_test.go rename to common/randselect_test.go index 9ae7726ddd..5b23faa4df 100644 --- a/les/randselect_test.go +++ b/common/randselect_test.go @@ -14,7 +14,7 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -package les +package common import ( "math/rand" @@ -36,15 +36,15 @@ func (t *testWrsItem) Weight() int64 { func TestWeightedRandomSelect(t *testing.T) { testFn := func(cnt int) { - s := newWeightedRandomSelect() + s := NewWeightedRandomSelect() w := -1 list := make([]testWrsItem, cnt) for i := range list { list[i] = testWrsItem{idx: i, widx: &w} - s.update(&list[i]) + s.Update(&list[i]) } w = rand.Intn(cnt) - c := s.choose() + c := s.Choose() if c == nil { t.Errorf("expected item, got nil") } else { @@ -53,7 +53,7 @@ func TestWeightedRandomSelect(t *testing.T) { } } w = -2 - if s.choose() != nil { + if s.Choose() != nil { t.Errorf("expected nil, got item") } } diff --git a/les/distributor.go b/les/distributor.go index 159fa4c73f..eeb42418ad 100644 --- a/les/distributor.go +++ b/les/distributor.go @@ -23,6 +23,8 @@ import ( "errors" "sync" "time" + + "github.com/ethereum/go-ethereum/common" ) // ErrNoPeers is returned if no peers capable of serving a queued request are available @@ -161,7 +163,7 @@ func (d *requestDistributor) loop() { } } -// selectPeerItem represents a peer to be selected for a request by weightedRandomSelect +// selectPeerItem represents a peer to be selected for a request by WeightedRandomSelect type selectPeerItem struct { peer distPeer req *distReq @@ -182,7 +184,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { bestPeer distPeer bestReq *distReq bestWait time.Duration - sel *weightedRandomSelect + sel *common.WeightedRandomSelect ) d.peerLock.RLock() @@ -198,9 +200,9 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { wait, bufRemain := peer.waitBefore(cost) if wait == 0 { if sel == nil { - sel = newWeightedRandomSelect() + sel = common.NewWeightedRandomSelect() } - sel.update(selectPeerItem{peer: peer, req: req, weight: int64(bufRemain*1000000) + 1}) + sel.Update(selectPeerItem{peer: peer, req: req, weight: int64(bufRemain*1000000) + 1}) } else { if bestReq == nil || wait < bestWait { bestPeer = peer @@ -220,7 +222,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { } if sel != nil { - c := sel.choose().(selectPeerItem) + c := sel.Choose().(selectPeerItem) return c.peer, c.req, 0 } return bestPeer, bestReq, bestWait diff --git a/les/serverpool.go b/les/serverpool.go index dc1ea6bf02..c7bc6cf131 100644 --- a/les/serverpool.go +++ b/les/serverpool.go @@ -27,6 +27,7 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/log" @@ -114,7 +115,7 @@ type serverPool struct { adjustStats chan poolStatAdjust knownQueue, newQueue poolEntryQueue - knownSelect, newSelect *weightedRandomSelect + knownSelect, newSelect *common.WeightedRandomSelect knownSelected, newSelected int fastDiscover bool } @@ -129,8 +130,8 @@ func newServerPool(db ethdb.Database, quit chan struct{}, wg *sync.WaitGroup) *s timeout: make(chan *poolEntry, 1), adjustStats: make(chan poolStatAdjust, 100), enableRetry: make(chan *poolEntry, 1), - knownSelect: newWeightedRandomSelect(), - newSelect: newWeightedRandomSelect(), + knownSelect: common.NewWeightedRandomSelect(), + newSelect: common.NewWeightedRandomSelect(), fastDiscover: true, } pool.knownQueue = newPoolEntryQueue(maxKnownEntries, pool.removeEntry) @@ -183,8 +184,8 @@ func (pool *serverPool) connect(p *peer, ip net.IP, port uint16) *poolEntry { entry.lastConnected = addr entry.addr = make(map[string]*poolEntryAddress) entry.addr[addr.strKey()] = addr - entry.addrSelect = *newWeightedRandomSelect() - entry.addrSelect.update(addr) + entry.addrSelect = *common.NewWeightedRandomSelect() + entry.addrSelect.Update(addr) return entry } @@ -352,7 +353,7 @@ func (pool *serverPool) findOrNewNode(id discover.NodeID, ip net.IP, port uint16 entry = &poolEntry{ id: id, addr: make(map[string]*poolEntryAddress), - addrSelect: *newWeightedRandomSelect(), + addrSelect: *common.NewWeightedRandomSelect(), shortRetry: shortRetryCnt, } pool.entries[id] = entry @@ -373,7 +374,7 @@ func (pool *serverPool) findOrNewNode(id discover.NodeID, ip net.IP, port uint16 entry.addr[addr.strKey()] = addr } addr.lastSeen = now - entry.addrSelect.update(addr) + entry.addrSelect.Update(addr) if !entry.known { pool.newQueue.setLatest(entry) } @@ -400,7 +401,7 @@ func (pool *serverPool) loadNodes() { "timeout", fmt.Sprintf("%v/%v", e.timeoutStats.avg, e.timeoutStats.weight)) pool.entries[e.id] = e pool.knownQueue.setLatest(e) - pool.knownSelect.update((*knownEntry)(e)) + pool.knownSelect.Update((*knownEntry)(e)) } } @@ -421,8 +422,8 @@ func (pool *serverPool) saveNodes() { // Note that it is called by the new/known queues from which the entry has already // been removed so removing it from the queues is not necessary. func (pool *serverPool) removeEntry(entry *poolEntry) { - pool.newSelect.remove((*discoveredEntry)(entry)) - pool.knownSelect.remove((*knownEntry)(entry)) + pool.newSelect.Remove((*discoveredEntry)(entry)) + pool.knownSelect.Remove((*knownEntry)(entry)) entry.removed = true delete(pool.entries, entry.id) } @@ -451,8 +452,8 @@ func (pool *serverPool) setRetryDial(entry *poolEntry) { // updateCheckDial is called when an entry can potentially be dialed again. It updates // its selection weights and checks if new dials can/should be made. func (pool *serverPool) updateCheckDial(entry *poolEntry) { - pool.newSelect.update((*discoveredEntry)(entry)) - pool.knownSelect.update((*knownEntry)(entry)) + pool.newSelect.Update((*discoveredEntry)(entry)) + pool.knownSelect.Update((*knownEntry)(entry)) pool.checkDial() } @@ -461,7 +462,7 @@ func (pool *serverPool) updateCheckDial(entry *poolEntry) { func (pool *serverPool) checkDial() { fillWithKnownSelects := !pool.fastDiscover for pool.knownSelected < targetKnownSelect { - entry := pool.knownSelect.choose() + entry := pool.knownSelect.Choose() if entry == nil { fillWithKnownSelects = false break @@ -469,7 +470,7 @@ func (pool *serverPool) checkDial() { pool.dial((*poolEntry)(entry.(*knownEntry)), true) } for pool.knownSelected+pool.newSelected < targetServerCount { - entry := pool.newSelect.choose() + entry := pool.newSelect.Choose() if entry == nil { break } @@ -480,7 +481,7 @@ func (pool *serverPool) checkDial() { // is over, we probably won't find more in the near future so select more // known entries if possible for pool.knownSelected < targetServerCount { - entry := pool.knownSelect.choose() + entry := pool.knownSelect.Choose() if entry == nil { break } @@ -501,7 +502,7 @@ func (pool *serverPool) dial(entry *poolEntry, knownSelected bool) { } else { pool.newSelected++ } - addr := entry.addrSelect.choose().(*poolEntryAddress) + addr := entry.addrSelect.Choose().(*poolEntryAddress) log.Debug("Dialing new peer", "lesaddr", entry.id.String()+"@"+addr.strKey(), "set", len(entry.addr), "known", knownSelected) entry.dialed = addr go func() { @@ -548,7 +549,7 @@ type poolEntry struct { id discover.NodeID addr map[string]*poolEntryAddress lastConnected, dialed *poolEntryAddress - addrSelect weightedRandomSelect + addrSelect common.WeightedRandomSelect lastDiscovered mclock.AbsTime known, knownSelected bool @@ -582,8 +583,8 @@ func (e *poolEntry) DecodeRLP(s *rlp.Stream) error { e.id = entry.ID e.addr = make(map[string]*poolEntryAddress) e.addr[addr.strKey()] = addr - e.addrSelect = *newWeightedRandomSelect() - e.addrSelect.update(addr) + e.addrSelect = *common.NewWeightedRandomSelect() + e.addrSelect.Update(addr) e.lastConnected = addr e.connectStats = entry.CStat e.delayStats = entry.DStat