forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 09:23:48 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

common, les: move WeightedRandomSelect to common package

Browse source
This commit is contained in:
Zsolt Felfoldi 2017-11-24 22:33:05 +01:00
parent 97472e2540
commit ff832be9c5
4 changed files with 51 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
les/randselect.go → common/randselect.go
View file

@ -14,44 +14,43 @@
// You should have received a copy of the GNU Lesser General Public License // You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol. package common
package les
import ( import (
"math/rand" "math/rand"
) )
// wrsItem interface should be implemented by any entries that are to be selected from // wrsItem interface should be implemented by any entries that are to be selected from
// a weightedRandomSelect set. Note that recalculating monotonously decreasing item // a WeightedRandomSelect set. Note that recalculating monotonously decreasing item
// weights on-demand (without constantly calling update) is allowed // weights on-demand (without constantly calling Update) is allowed
type wrsItem interface { type wrsItem interface {
Weight() int64 Weight() int64
} }
// weightedRandomSelect is capable of weighted random selection from a set of items // WeightedRandomSelect is capable of weighted random selection from a set of items
type weightedRandomSelect struct { type WeightedRandomSelect struct {
root *wrsNode root *wrsNode
idx map[wrsItem]int idx map[wrsItem]int
} }
// newWeightedRandomSelect returns a new weightedRandomSelect structure // newWeightedRandomSelect returns a new WeightedRandomSelect structure
func newWeightedRandomSelect() *weightedRandomSelect { func NewWeightedRandomSelect() *WeightedRandomSelect {
return &weightedRandomSelect{root: &wrsNode{maxItems: wrsBranches}, idx: make(map[wrsItem]int)} 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. // 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()) w.setWeight(item, item.Weight())
} }
// remove removes an item from the set // Remove removes an item from the set
func (w *weightedRandomSelect) remove(item wrsItem) { func (w *WeightedRandomSelect) Remove(item wrsItem) {
w.setWeight(item, 0) w.setWeight(item, 0)
} }
// setWeight sets an item's weight to a specific value (removes it if zero) // 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] idx, ok := w.idx[item]
if ok { if ok {
w.root.setWeight(idx, weight) 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 // 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 // last stored value, returns it with a newWeight/oldWeight chance, otherwise just
// updates its weight and selects another one // updates its weight and selects another one
func (w *weightedRandomSelect) choose() wrsItem { func (w *WeightedRandomSelect) Choose() wrsItem {
for { for {
if w.root.sumWeight == 0 { if w.root.sumWeight == 0 {
return nil return nil
} }
val := rand.Int63n(w.root.sumWeight) val := rand.Int63n(w.root.sumWeight)
choice, lastWeight := w.root.choose(val) choice, lastWeight := w.root.Choose(val)
weight := choice.Weight() weight := choice.Weight()
if weight != lastWeight { if weight != lastWeight {
w.setWeight(choice, weight) w.setWeight(choice, weight)
@ -156,14 +155,14 @@ func (n *wrsNode) setWeight(idx int, weight int64) int64 {
return diff return diff
} }
// choose recursively selects an item from the tree and returns it along with its weight // Choose recursively selects an item from the tree and returns it along with its weight
func (n *wrsNode) choose(val int64) (wrsItem, int64) { func (n *wrsNode) Choose(val int64) (wrsItem, int64) {
for i, w := range n.weights { for i, w := range n.weights {
if val < w { if val < w {
if n.level == 0 { if n.level == 0 {
return n.items[i].(wrsItem), n.weights[i] return n.items[i].(wrsItem), n.weights[i]
} else { } else {
return n.items[i].(*wrsNode).choose(val) return n.items[i].(*wrsNode).Choose(val)
} }
} else { } else {
val -= w val -= w

10
les/randselect_test.go → common/randselect_test.go
View file

@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package les package common
import ( import (
"math/rand" "math/rand"
@ -36,15 +36,15 @@ func (t *testWrsItem) Weight() int64 {
func TestWeightedRandomSelect(t *testing.T) { func TestWeightedRandomSelect(t *testing.T) {
testFn := func(cnt int) { testFn := func(cnt int) {
s := newWeightedRandomSelect() s := NewWeightedRandomSelect()
w := -1 w := -1
list := make([]testWrsItem, cnt) list := make([]testWrsItem, cnt)
for i := range list { for i := range list {
list[i] = testWrsItem{idx: i, widx: &w} list[i] = testWrsItem{idx: i, widx: &w}
s.update(&list[i]) s.Update(&list[i])
} }
w = rand.Intn(cnt) w = rand.Intn(cnt)
c := s.choose() c := s.Choose()
if c == nil { if c == nil {
t.Errorf("expected item, got nil") t.Errorf("expected item, got nil")
} else { } else {
@ -53,7 +53,7 @@ func TestWeightedRandomSelect(t *testing.T) {
} }
} }
w = -2 w = -2
if s.choose() != nil { if s.Choose() != nil {
t.Errorf("expected nil, got item") t.Errorf("expected nil, got item")
} }
} }

12
les/distributor.go
View file

@ -23,6 +23,8 @@ import (
"errors" "errors"
"sync" "sync"
"time" "time"
"github.com/ethereum/go-ethereum/common"
) )
// ErrNoPeers is returned if no peers capable of serving a queued request are available // 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 { type selectPeerItem struct {
peer distPeer peer distPeer
req *distReq req *distReq
@ -182,7 +184,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) {
bestPeer distPeer bestPeer distPeer
bestReq *distReq bestReq *distReq
bestWait time.Duration bestWait time.Duration
sel *weightedRandomSelect sel *common.WeightedRandomSelect
) )
d.peerLock.RLock() d.peerLock.RLock()
@ -198,9 +200,9 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) {
wait, bufRemain := peer.waitBefore(cost) wait, bufRemain := peer.waitBefore(cost)
if wait == 0 { if wait == 0 {
if sel == nil { 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 { } else {
if bestReq == nil || wait < bestWait { if bestReq == nil || wait < bestWait {
bestPeer = peer bestPeer = peer
@ -220,7 +222,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) {
} }
if sel != nil { if sel != nil {
c := sel.choose().(selectPeerItem) c := sel.Choose().(selectPeerItem)
return c.peer, c.req, 0 return c.peer, c.req, 0
} }
return bestPeer, bestReq, bestWait return bestPeer, bestReq, bestWait

39
les/serverpool.go
View file

@ -27,6 +27,7 @@ import (
"sync" "sync"
"time" "time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
@ -114,7 +115,7 @@ type serverPool struct {
adjustStats chan poolStatAdjust adjustStats chan poolStatAdjust
knownQueue, newQueue poolEntryQueue knownQueue, newQueue poolEntryQueue
knownSelect, newSelect *weightedRandomSelect knownSelect, newSelect *common.WeightedRandomSelect
knownSelected, newSelected int knownSelected, newSelected int
fastDiscover bool fastDiscover bool
} }
@ -129,8 +130,8 @@ func newServerPool(db ethdb.Database, quit chan struct{}, wg *sync.WaitGroup) *s
timeout: make(chan *poolEntry, 1), timeout: make(chan *poolEntry, 1),
adjustStats: make(chan poolStatAdjust, 100), adjustStats: make(chan poolStatAdjust, 100),
enableRetry: make(chan *poolEntry, 1), enableRetry: make(chan *poolEntry, 1),
knownSelect: newWeightedRandomSelect(), knownSelect: common.NewWeightedRandomSelect(),
newSelect: newWeightedRandomSelect(), newSelect: common.NewWeightedRandomSelect(),
fastDiscover: true, fastDiscover: true,
} }
pool.knownQueue = newPoolEntryQueue(maxKnownEntries, pool.removeEntry) 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.lastConnected = addr
entry.addr = make(map[string]*poolEntryAddress) entry.addr = make(map[string]*poolEntryAddress)
entry.addr[addr.strKey()] = addr entry.addr[addr.strKey()] = addr
entry.addrSelect = *newWeightedRandomSelect() entry.addrSelect = *common.NewWeightedRandomSelect()
entry.addrSelect.update(addr) entry.addrSelect.Update(addr)
return entry return entry
} }
@ -352,7 +353,7 @@ func (pool *serverPool) findOrNewNode(id discover.NodeID, ip net.IP, port uint16
entry = &poolEntry{ entry = &poolEntry{
id: id, id: id,
addr: make(map[string]*poolEntryAddress), addr: make(map[string]*poolEntryAddress),
addrSelect: *newWeightedRandomSelect(), addrSelect: *common.NewWeightedRandomSelect(),
shortRetry: shortRetryCnt, shortRetry: shortRetryCnt,
} }
pool.entries[id] = entry 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 entry.addr[addr.strKey()] = addr
} }
addr.lastSeen = now addr.lastSeen = now
entry.addrSelect.update(addr) entry.addrSelect.Update(addr)
if !entry.known { if !entry.known {
pool.newQueue.setLatest(entry) pool.newQueue.setLatest(entry)
} }
@ -400,7 +401,7 @@ func (pool *serverPool) loadNodes() {
"timeout", fmt.Sprintf("%v/%v", e.timeoutStats.avg, e.timeoutStats.weight)) "timeout", fmt.Sprintf("%v/%v", e.timeoutStats.avg, e.timeoutStats.weight))
pool.entries[e.id] = e pool.entries[e.id] = e
pool.knownQueue.setLatest(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 // 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. // been removed so removing it from the queues is not necessary.
func (pool *serverPool) removeEntry(entry *poolEntry) { func (pool *serverPool) removeEntry(entry *poolEntry) {
pool.newSelect.remove((*discoveredEntry)(entry)) pool.newSelect.Remove((*discoveredEntry)(entry))
pool.knownSelect.remove((*knownEntry)(entry)) pool.knownSelect.Remove((*knownEntry)(entry))
entry.removed = true entry.removed = true
delete(pool.entries, entry.id) 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 // 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. // its selection weights and checks if new dials can/should be made.
func (pool *serverPool) updateCheckDial(entry *poolEntry) { func (pool *serverPool) updateCheckDial(entry *poolEntry) {
pool.newSelect.update((*discoveredEntry)(entry)) pool.newSelect.Update((*discoveredEntry)(entry))
pool.knownSelect.update((*knownEntry)(entry)) pool.knownSelect.Update((*knownEntry)(entry))
pool.checkDial() pool.checkDial()
} }
@ -461,7 +462,7 @@ func (pool *serverPool) updateCheckDial(entry *poolEntry) {
func (pool *serverPool) checkDial() { func (pool *serverPool) checkDial() {
fillWithKnownSelects := !pool.fastDiscover fillWithKnownSelects := !pool.fastDiscover
for pool.knownSelected < targetKnownSelect { for pool.knownSelected < targetKnownSelect {
entry := pool.knownSelect.choose() entry := pool.knownSelect.Choose()
if entry == nil { if entry == nil {
fillWithKnownSelects = false fillWithKnownSelects = false
break break
@ -469,7 +470,7 @@ func (pool *serverPool) checkDial() {
pool.dial((*poolEntry)(entry.(*knownEntry)), true) pool.dial((*poolEntry)(entry.(*knownEntry)), true)
} }
for pool.knownSelected+pool.newSelected < targetServerCount { for pool.knownSelected+pool.newSelected < targetServerCount {
entry := pool.newSelect.choose() entry := pool.newSelect.Choose()
if entry == nil { if entry == nil {
break break
} }
@ -480,7 +481,7 @@ func (pool *serverPool) checkDial() {
// is over, we probably won't find more in the near future so select more // is over, we probably won't find more in the near future so select more
// known entries if possible // known entries if possible
for pool.knownSelected < targetServerCount { for pool.knownSelected < targetServerCount {
entry := pool.knownSelect.choose() entry := pool.knownSelect.Choose()
if entry == nil { if entry == nil {
break break
} }
@ -501,7 +502,7 @@ func (pool *serverPool) dial(entry *poolEntry, knownSelected bool) {
} else { } else {
pool.newSelected++ 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) log.Debug("Dialing new peer", "lesaddr", entry.id.String()+"@"+addr.strKey(), "set", len(entry.addr), "known", knownSelected)
entry.dialed = addr entry.dialed = addr
go func() { go func() {
@ -548,7 +549,7 @@ type poolEntry struct {
id discover.NodeID id discover.NodeID
addr map[string]*poolEntryAddress addr map[string]*poolEntryAddress
lastConnected, dialed *poolEntryAddress lastConnected, dialed *poolEntryAddress
addrSelect weightedRandomSelect addrSelect common.WeightedRandomSelect
lastDiscovered mclock.AbsTime lastDiscovered mclock.AbsTime
known, knownSelected bool known, knownSelected bool
@ -582,8 +583,8 @@ func (e *poolEntry) DecodeRLP(s *rlp.Stream) error {
e.id = entry.ID e.id = entry.ID
e.addr = make(map[string]*poolEntryAddress) e.addr = make(map[string]*poolEntryAddress)
e.addr[addr.strKey()] = addr e.addr[addr.strKey()] = addr
e.addrSelect = *newWeightedRandomSelect() e.addrSelect = *common.NewWeightedRandomSelect()
e.addrSelect.update(addr) e.addrSelect.Update(addr)
e.lastConnected = addr e.lastConnected = addr
e.connectStats = entry.CStat e.connectStats = entry.CStat
e.delayStats = entry.DStat e.delayStats = entry.DStat