From 2d11d7455b897f8952280fd7959bd13d097bb57a Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Fri, 15 Nov 2019 18:44:54 +0100 Subject: [PATCH] les: implement clientPool.balanceMissing --- les/api.go | 5 +- les/balance.go | 28 ++++++++++ les/clientpool.go | 116 +++++++++++++++++++++++++++++++++-------- les/clientpool_test.go | 22 ++++---- les/server.go | 2 +- les/test_helper.go | 2 +- p2p/discv5/net.go | 6 ++- 7 files changed, 140 insertions(+), 41 deletions(-) diff --git a/les/api.go b/les/api.go index ad511c9d6b..d40846cf10 100644 --- a/les/api.go +++ b/les/api.go @@ -19,7 +19,6 @@ package les import ( "errors" "fmt" - "math" "time" "github.com/ethereum/go-ethereum/common/hexutil" @@ -35,8 +34,6 @@ var ( errNoPriority = errors.New("priority too low to raise capacity") ) -const maxBalance = math.MaxInt64 - // PrivateLightServerAPI provides an API to access the LES light server. type PrivateLightServerAPI struct { server *LesServer @@ -184,7 +181,7 @@ func (api *PrivateLightServerAPI) SetClientParams(ids []enode.ID, params map[str if client != nil { update, err := api.setParams(params, client, nil, nil) if update { - client.updatePriceFactors() + updatePriceFactors(&client.balanceTracker, client.posFactors, client.negFactors, client.capacity) } return err } else { diff --git a/les/balance.go b/les/balance.go index 51cef15c80..5cbd2e3d48 100644 --- a/les/balance.go +++ b/les/balance.go @@ -17,12 +17,15 @@ package les import ( + "math" "sync" "time" "github.com/ethereum/go-ethereum/common/mclock" ) +const maxBalance = math.MaxInt64 + const ( balanceCallbackQueue = iota balanceCallbackZero @@ -101,6 +104,31 @@ func (bt *balanceTracker) balanceToPriority(b balance) int64 { return int64(b.neg) } +func (bt *balanceTracker) posBalanceMissing(targetPriority int64, after time.Duration) uint64 { + if targetPriority > 0 { + negPrice := uint64(float64(after) * bt.negTimeFactor) + if negPrice+bt.balance.neg <= uint64(targetPriority) { + return 0 + } + if uint64(targetPriority) > bt.balance.neg && bt.negTimeFactor > 1e-100 { + if negTime := time.Duration(float64(uint64(targetPriority)-bt.balance.neg) / bt.negTimeFactor); negTime < after { + after -= negTime + } else { + after = 0 + } + } + targetPriority = 0 + } + posRequired := uint64(float64(^targetPriority)*float64(bt.capacity) + float64(after)*bt.timeFactor) + if posRequired >= maxBalance { + return math.MaxUint64 // target not reachable + } + if posRequired > bt.balance.pos { + return posRequired - bt.balance.pos + } + return 0 +} + // reducedBalance estimates the reduced balance at a given time in the fututre based // on the current balance, the time factor and an estimated average request cost per time ratio func (bt *balanceTracker) reducedBalance(at mclock.AbsTime, avgReqCost float64) balance { diff --git a/les/clientpool.go b/les/clientpool.go index da76f08b91..59f707cef1 100644 --- a/les/clientpool.go +++ b/les/clientpool.go @@ -91,6 +91,7 @@ type clientPool struct { capLimit uint64 // The maximum cumulative capacity that clientpool can support connectedCap uint64 // The sum of the capacity of the current clientpool connected priorityConnected uint64 // The sum of the capacity of currently connected priority clients + minCap uint64 // The minimal capacity value allowed for any client freeClientCap uint64 // The capacity value of each free client startTime mclock.AbsTime // The timestamp at which the clientpool started running cumulativeTime int64 // The cumulative running time of clientpool at the start point. @@ -159,13 +160,14 @@ type priceFactors struct { } // newClientPool creates a new client pool -func newClientPool(db ethdb.Database, freeClientCap uint64, clock mclock.Clock, removePeer func(enode.ID)) *clientPool { +func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock.Clock, removePeer func(enode.ID)) *clientPool { ndb := newNodeDB(db, clock) pool := &clientPool{ ndb: ndb, clock: clock, connectedMap: make(map[enode.ID]*clientInfo), connectedQueue: prque.NewLazyQueue(connSetIndex, connPriority, connMaxPriority, clock, lazyQueueRefresh), + minCap: minCap, freeClientCap: freeClientCap, removePeer: removePeer, startTime: clock.Now(), @@ -223,18 +225,10 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool { return false } // Create a clientInfo but do not add it yet - var ( - posBalance uint64 - negBalance uint64 - now = f.clock.Now() - ) + now := f.clock.Now() pb := f.ndb.getOrNewPB(id) - posBalance = pb.value - nb := f.ndb.getOrNewNB(freeID) - if nb.logValue != 0 { - negBalance = uint64(math.Exp(float64(nb.logValue-f.logOffset(now))/fixedPointMultiplier) * float64(time.Second)) - } + e := &clientInfo{ pool: f, peer: peer, @@ -242,7 +236,7 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool { queueIndex: -1, id: id, connectedAt: now, - priority: posBalance != 0, + priority: pb.value != 0, posFactors: f.defaultPosFactors, negFactors: f.defaultNegFactors, balanceMetaInfo: pb.meta, @@ -252,12 +246,11 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool { if !e.priority || capacity == 0 { capacity = f.freeClientCap } + if capacity < f.minCap { + capacity = f.minCap + } e.capacity = capacity - - // Starts a balance tracker - e.balanceTracker.init(f.clock, capacity) - e.balanceTracker.setBalance(posBalance, negBalance) - e.updatePriceFactors() + f.initBalanceTracker(&e.balanceTracker, pb, nb, capacity) // If the number of clients already connected in the clientpool exceeds its // capacity, evict some clients with lowest priority. @@ -319,6 +312,17 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool { return true } +func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb posBalance, nb negBalance, capacity uint64) { + posBalance := pb.value + var negBalance uint64 + if nb.logValue != 0 { + negBalance = uint64(math.Exp(float64(nb.logValue-f.logOffset(f.clock.Now()))/fixedPointMultiplier) * float64(time.Second)) + } + bt.init(f.clock, capacity) + bt.setBalance(posBalance, negBalance) + updatePriceFactors(bt, f.defaultPosFactors, f.defaultNegFactors, capacity) +} + // disconnect should be called when a connection is terminated. If the disconnection // was initiated by the pool itself using disconnectFn then calling disconnect is // not necessary but permitted. @@ -339,6 +343,68 @@ func (f *clientPool) disconnect(p clientPeer) { f.dropClient(e, f.clock.Now(), false) } +func (f *clientPool) balanceMissing(id enode.ID, freeID string, capacity uint64, minConnTime time.Duration) (uint64, uint64) { + f.lock.Lock() + defer f.lock.Unlock() + + var missing uint64 + if capacity == 0 { + capacity = f.freeClientCap + } + if capacity < f.minCap { + capacity = f.minCap + } + newCapacity := f.connectedCap + capacity + newCount := f.connectedQueue.Size() + 1 + client := f.connectedMap[id] + if client != nil { + newCapacity -= client.capacity + newCount-- + } + if newCapacity > f.capLimit || newCount > f.connLimit { + var ( + popList []*clientInfo + targetPriority int64 + ) + f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool { + c := data.(*clientInfo) + popList = append(popList, c) + if c != client { + targetPriority = priority + newCapacity -= c.capacity + newCount-- + } + return newCapacity > f.capLimit || newCount > f.connLimit + }) + if newCapacity > f.capLimit || newCount > f.connLimit { + missing = math.MaxUint64 + } else { + var bt *balanceTracker + if client != nil { + bt = &client.balanceTracker + } else { + bt := &balanceTracker{} + f.initBalanceTracker(bt, f.ndb.getOrNewPB(id), f.ndb.getOrNewNB(freeID), capacity) + } + if capacity != f.freeClientCap && targetPriority >= -1 { + targetPriority = -2 + } + bias := connectedBias + if f.disableBias { + bias = 0 + } + if bias < minConnTime { + bias = minConnTime + } + missing = bt.posBalanceMissing(targetPriority, bias) + } + for _, c := range popList { + f.connectedQueue.Push(c) + } + } + return missing, capacity +} + // forClients iterates through a list of clients, calling the callback for each one. // If a client is not connected then clientInfo is nil. If the specified list is empty // then the callback is called for all connected clients. @@ -467,6 +533,12 @@ func (f *clientPool) setLimits(totalConn int, totalCap uint64) { // setCapacity sets the assigned capacity of a connected client func (f *clientPool) setCapacity(c *clientInfo, capacity uint64) error { + if capacity == 0 { + capacity = f.freeClientCap + } + if capacity < f.minCap { + capacity = f.minCap + } if f.connectedMap[c.id] != c { return fmt.Errorf("client %064x is not connected", c.id[:]) } @@ -510,7 +582,7 @@ func (f *clientPool) setCapacity(c *clientInfo, capacity uint64) error { } totalConnectedGauge.Update(int64(f.connectedCap)) f.priorityConnected += capacity - oldCapacity - c.updatePriceFactors() + updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors, c.capacity) c.peer.updateCapacity(c.capacity) return nil } @@ -539,10 +611,10 @@ func (f *clientPool) logOffset(now mclock.AbsTime) int64 { return f.cumulativeTime + cumulativeTime } -// setClientPriceFactors sets the pricing factors for an individual connected client -func (c *clientInfo) updatePriceFactors() { - c.balanceTracker.setFactors(true, c.negFactors.timeFactor+float64(c.capacity)*c.negFactors.capacityFactor/1000000, c.negFactors.requestFactor) - c.balanceTracker.setFactors(false, c.posFactors.timeFactor+float64(c.capacity)*c.posFactors.capacityFactor/1000000, c.posFactors.requestFactor) +// updatePriceFactors sets the pricing factors for an individual connected client +func updatePriceFactors(bt *balanceTracker, posFactors, negFactors priceFactors, capacity uint64) { + bt.setFactors(true, negFactors.timeFactor+float64(capacity)*negFactors.capacityFactor/1000000, negFactors.requestFactor) + bt.setFactors(false, posFactors.timeFactor+float64(capacity)*posFactors.capacityFactor/1000000, posFactors.requestFactor) } // getPosBalance retrieves a single positive balance entry from cache or the database diff --git a/les/clientpool_test.go b/les/clientpool_test.go index 06f782ac96..23564d4052 100644 --- a/les/clientpool_test.go +++ b/les/clientpool_test.go @@ -89,7 +89,7 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD disconnFn = func(id enode.ID) { disconnCh <- int(id[0]) + int(id[1])<<8 } - pool = newClientPool(db, 1, &clock, disconnFn) + pool = newClientPool(db, 1, 1, &clock, disconnFn) ) pool.disableBias = true pool.setLimits(connLimit, uint64(connLimit)) @@ -172,7 +172,7 @@ func TestConnectPaidClient(t *testing.T) { clock mclock.Simulated db = rawdb.NewMemoryDatabase() ) - pool := newClientPool(db, 1, &clock, nil) + pool := newClientPool(db, 1, 1, &clock, nil) defer pool.stop() pool.setLimits(10, uint64(10)) pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -190,7 +190,7 @@ func TestConnectPaidClientToSmallPool(t *testing.T) { clock mclock.Simulated db = rawdb.NewMemoryDatabase() ) - pool := newClientPool(db, 1, &clock, nil) + pool := newClientPool(db, 1, 1, &clock, nil) defer pool.stop() pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -210,7 +210,7 @@ func TestConnectPaidClientToFullPool(t *testing.T) { db = rawdb.NewMemoryDatabase() ) removeFn := func(enode.ID) {} // Noop - pool := newClientPool(db, 1, &clock, removeFn) + pool := newClientPool(db, 1, 1, &clock, removeFn) defer pool.stop() pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -237,7 +237,7 @@ func TestPaidClientKickedOut(t *testing.T) { kickedCh = make(chan int, 1) ) removeFn := func(id enode.ID) { kickedCh <- int(id[0]) } - pool := newClientPool(db, 1, &clock, removeFn) + pool := newClientPool(db, 1, 1, &clock, removeFn) defer pool.stop() pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -267,7 +267,7 @@ func TestConnectFreeClient(t *testing.T) { clock mclock.Simulated db = rawdb.NewMemoryDatabase() ) - pool := newClientPool(db, 1, &clock, nil) + pool := newClientPool(db, 1, 1, &clock, nil) defer pool.stop() pool.setLimits(10, uint64(10)) pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -282,7 +282,7 @@ func TestConnectFreeClientToFullPool(t *testing.T) { db = rawdb.NewMemoryDatabase() ) removeFn := func(enode.ID) {} // Noop - pool := newClientPool(db, 1, &clock, removeFn) + pool := newClientPool(db, 1, 1, &clock, removeFn) defer pool.stop() pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -311,7 +311,7 @@ func TestFreeClientKickedOut(t *testing.T) { kicked = make(chan int, 10) ) removeFn := func(id enode.ID) { kicked <- int(id[0]) } - pool := newClientPool(db, 1, &clock, removeFn) + pool := newClientPool(db, 1, 1, &clock, removeFn) defer pool.stop() pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -346,7 +346,7 @@ func TestPositiveBalanceCalculation(t *testing.T) { kicked = make(chan int, 10) ) removeFn := func(id enode.ID) { kicked <- int(id[0]) } // Noop - pool := newClientPool(db, 1, &clock, removeFn) + pool := newClientPool(db, 1, 1, &clock, removeFn) defer pool.stop() pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -369,7 +369,7 @@ func TestDowngradePriorityClient(t *testing.T) { kicked = make(chan int, 10) ) removeFn := func(id enode.ID) { kicked <- int(id[0]) } // Noop - pool := newClientPool(db, 1, &clock, removeFn) + pool := newClientPool(db, 1, 1, &clock, removeFn) defer pool.stop() pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) @@ -407,7 +407,7 @@ func TestNegativeBalanceCalculation(t *testing.T) { kicked = make(chan int, 10) ) removeFn := func(id enode.ID) { kicked <- int(id[0]) } // Noop - pool := newClientPool(db, 1, &clock, removeFn) + pool := newClientPool(db, 1, 1, &clock, removeFn) defer pool.stop() pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) diff --git a/les/server.go b/les/server.go index e68903dd81..429e866cb8 100644 --- a/les/server.go +++ b/les/server.go @@ -114,7 +114,7 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) { srv.maxCapacity = totalRecharge } srv.fcManager.SetCapacityLimits(srv.freeCapacity, srv.maxCapacity, srv.freeCapacity*2) - srv.clientPool = newClientPool(srv.chainDb, srv.freeCapacity, mclock.System{}, func(id enode.ID) { go srv.peers.Unregister(peerIdToString(id)) }) + srv.clientPool = newClientPool(srv.chainDb, srv.minCapacity, srv.freeCapacity, mclock.System{}, func(id enode.ID) { go srv.peers.Unregister(peerIdToString(id)) }) srv.clientPool.setDefaultFactors(priceFactors{0, 1, 1}, priceFactors{0, 1, 1}) checkpoint := srv.latestLocalCheckpoint() diff --git a/les/test_helper.go b/les/test_helper.go index ee3d7a32e1..544c3d9e59 100644 --- a/les/test_helper.go +++ b/les/test_helper.go @@ -280,7 +280,7 @@ func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Da } server.costTracker, server.freeCapacity = newCostTracker(db, server.config) server.costTracker.testCostList = testCostList(0) // Disable flow control mechanism. - server.clientPool = newClientPool(db, 1, clock, nil) + server.clientPool = newClientPool(db, 1, 1, clock, nil) server.clientPool.setLimits(10000, 10000) // Assign enough capacity for clientpool server.handler = newServerHandler(server, simulation.Blockchain(), db, txpool, func() bool { return true }) if server.oracle != nil { diff --git a/p2p/discv5/net.go b/p2p/discv5/net.go index d278429716..adbc45bb8b 100644 --- a/p2p/discv5/net.go +++ b/p2p/discv5/net.go @@ -1330,10 +1330,12 @@ func (net *Network) RegisterTalkHandler(talkID string, handler TalkRequestHandle } func (net *Network) SendTalkRequest(to *enode.Node, talkID string, payload rlp.RawValue, handler TalkResponseHandler) func() bool { - node := NewNode(to.ID(), to.IP(), to.UDP(), to.TCP()) + var nodeID NodeID + copy(nodeID[:], crypto.FromECDSAPub(to.Pubkey())[1:]) + node := NewNode(nodeID, to.IP(), uint16(to.UDP()), uint16(to.TCP())) net.talkResponseSubLock.Lock() hash := net.conn.send(node, talkRequestPacket, talkRequest{TalkID: []byte(talkID), Payload: payload}) - key := string(to.sha[:]) + string(hash[:]) + key := string(node.sha[:]) + string(hash[:]) net.talkResponseSubs[key] = handler net.talkResponseSubLock.Unlock() return func() bool {