les: implement clientPool.balanceMissing

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 {

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 {

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 (
+	now := f.clock.Now()
-		posBalance uint64
-		negBalance uint64
-		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
-
+	f.initBalanceTracker(&e.balanceTracker, pb, nb, capacity)
-	// Starts a balance tracker
-	e.balanceTracker.init(f.clock, capacity)
-	e.balanceTracker.setBalance(posBalance, negBalance)
-	e.updatePriceFactors()
 
 	// 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
+// updatePriceFactors sets the pricing factors for an individual connected client
-func (c *clientInfo) updatePriceFactors() {
+func updatePriceFactors(bt *balanceTracker, posFactors, negFactors priceFactors, capacity uint64) {
-	c.balanceTracker.setFactors(true, c.negFactors.timeFactor+float64(c.capacity)*c.negFactors.capacityFactor/1000000, c.negFactors.requestFactor)
+	bt.setFactors(true, negFactors.timeFactor+float64(capacity)*negFactors.capacityFactor/1000000, negFactors.requestFactor)
-	c.balanceTracker.setFactors(false, c.posFactors.timeFactor+float64(c.capacity)*c.posFactors.capacityFactor/1000000, c.posFactors.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

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})

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()

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 {

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 {