diff --git a/les/balance.go b/les/balance.go index 51cef15c80..6f11daead2 100644 --- a/les/balance.go +++ b/les/balance.go @@ -17,24 +17,101 @@ package les import ( + "math" "sync" "time" "github.com/ethereum/go-ethereum/common/mclock" ) +const maxBalance = math.MaxInt64 + const ( balanceCallbackQueue = iota balanceCallbackZero balanceCallbackCount ) +// expirationController controls the exponential expiration of positive and negative +// balances +type expirationController interface { + posExpiration(mclock.AbsTime) uint64 + negExpiration(mclock.AbsTime) uint64 +} + +// expiredValue is a scalar value that is continuously expired (decreased exponentially) +// based on the logarithmic expiration offset value provided by the expirationController. +// The actual value can be calculated as base*2^(exp-logOffset/log2Multiplier). +// +// Note: this representation is basically a floating point value capable of handling +// 64 bit exponents. The operations are not general purpose floating point operations +// though because a monotonously increasing exponent is assumed and therefore if an +// operation is performed with two different exponents then simply the higher one is used. +type expiredValue struct { + base, exp uint64 +} + +// value calculates the value at the given moment +func (e expiredValue) value(logOffset uint64) uint64 { + return uint64(math.Exp(float64(int64(e.exp*log2Multiplier-logOffset))/logMultiplier) * float64(e.base)) +} + +// add adds a signed value at the given moment +func (e *expiredValue) add(amount int64, logOffset uint64) int64 { + addExp := logOffset / log2Multiplier + baseMul := math.Exp(float64(logOffset%log2Multiplier) / logMultiplier) + if addExp < e.exp { + baseMul /= math.Pow(2, float64(e.exp-addExp)) + } + if addExp > e.exp { + e.base >>= (addExp - e.exp) + e.exp = addExp + } + add := int64(float64(amount) * baseMul) + if add >= 0 || uint64(-add) <= e.base { + e.base += uint64(add) + return amount + } else { + e.base = 0 + return int64(-float64(e.base) / baseMul) + } +} + +// addExp adds another expiredValue +func (e *expiredValue) addExp(a expiredValue) { + if e.exp > a.exp { + a.base >>= (e.exp - a.exp) + } + if e.exp < a.exp { + e.base >>= (a.exp - e.exp) + e.exp = a.exp + } + e.base += a.base +} + +// subExp subtracts another expiredValue +func (e *expiredValue) subExp(a expiredValue) { + if e.exp > a.exp { + a.base >>= (e.exp - a.exp) + } + if e.exp < a.exp { + e.base >>= (a.exp - e.exp) + e.exp = a.exp + } + if e.base > a.base { + e.base -= a.base + } else { + e.base = 0 + } +} + // balanceTracker keeps track of the positive and negative balances of a connected // client and calculates actual and projected future priority values required by // prque.LazyQueue. type balanceTracker struct { lock sync.Mutex clock mclock.Clock + exp expirationController stopped bool capacity uint64 balance balance @@ -53,7 +130,7 @@ type balanceTracker struct { // balance represents a pair of positive and negative balances type balance struct { - pos, neg uint64 + pos, neg expiredValue } // balanceCallback represents a single callback that is activated when client priority @@ -65,6 +142,7 @@ type balanceCallback struct { } // init initializes balanceTracker +// Note: capacity should never be zero func (bt *balanceTracker) init(clock mclock.Clock, capacity uint64) { bt.clock = clock bt.initTime, bt.lastUpdate = clock.Now(), clock.Now() // Init timestamps @@ -95,10 +173,41 @@ func (bt *balanceTracker) stop(now mclock.AbsTime) { // first to disconnect. Positive balance translates to negative priority. If positive // balance is zero then negative balance translates to a positive priority. func (bt *balanceTracker) balanceToPriority(b balance) int64 { - if b.pos > 0 { - return ^int64(b.pos / bt.capacity) + if b.pos.base > 0 { + return -int64(b.pos.value(bt.exp.posExpiration(bt.clock.Now())) / bt.capacity) } - return int64(b.neg) + return int64(b.neg.value(bt.exp.negExpiration(bt.clock.Now()))) +} + +// posBalanceMissing calculates the missing amount of positive balance in order to +// connect at targetCapacity, stay connected for the given amount of time and then +// still have a priority of targetPriority +func (bt *balanceTracker) posBalanceMissing(targetPriority int64, targetCapacity uint64, after time.Duration) uint64 { + now := bt.clock.Now() + if targetPriority > 0 { + negPrice := uint64(float64(after) * bt.negTimeFactor) + negBalance := bt.balance.neg.value(bt.exp.negExpiration(now)) + if negPrice+negBalance < uint64(targetPriority) { + return 0 + } + if uint64(targetPriority) > negBalance && bt.negTimeFactor > 1e-100 { + if negTime := time.Duration(float64(uint64(targetPriority)-negBalance) / bt.negTimeFactor); negTime < after { + after -= negTime + } else { + after = 0 + } + } + targetPriority = 0 + } + posRequired := uint64(float64(-targetPriority)*float64(targetCapacity)+float64(after)*bt.timeFactor) + 1 + if posRequired >= maxBalance { + return math.MaxUint64 // target not reachable + } + posBalance := bt.balance.pos.value(bt.exp.posExpiration(now)) + if posRequired > posBalance { + return posRequired - posBalance + } + return 0 } // reducedBalance estimates the reduced balance at a given time in the fututre based @@ -106,20 +215,19 @@ func (bt *balanceTracker) balanceToPriority(b balance) int64 { func (bt *balanceTracker) reducedBalance(at mclock.AbsTime, avgReqCost float64) balance { dt := float64(at - bt.lastUpdate) b := bt.balance - if b.pos != 0 { + if b.pos.base != 0 { factor := bt.timeFactor + bt.requestFactor*avgReqCost - diff := uint64(dt * factor) - if diff <= b.pos { - b.pos -= diff + diff := -int64(dt * factor) + dd := b.pos.add(diff, bt.exp.posExpiration(at)) + if dd == diff { dt = 0 } else { - dt -= float64(b.pos) / factor - b.pos = 0 + dt += float64(dd) / factor } } - if dt != 0 { + if dt > 0 { factor := bt.negTimeFactor + bt.negRequestFactor*avgReqCost - b.neg += uint64(dt * factor) + b.neg.add(int64(dt*factor), bt.exp.negExpiration(at)) } return b } @@ -130,20 +238,22 @@ func (bt *balanceTracker) reducedBalance(at mclock.AbsTime, avgReqCost float64) // Note: the function assumes that the balance has been recently updated and // calculates the time starting from the last update. func (bt *balanceTracker) timeUntil(priority int64) (time.Duration, bool) { + now := bt.clock.Now() var dt float64 - if bt.balance.pos != 0 { + if bt.balance.pos.base != 0 { + posBalance := bt.balance.pos.value(bt.exp.posExpiration(now)) if bt.timeFactor < 1e-100 { return 0, false } if priority < 0 { - newBalance := uint64(^priority) * bt.capacity - if newBalance > bt.balance.pos { + newBalance := uint64(-priority) * bt.capacity + if newBalance > posBalance { return 0, false } - dt = float64(bt.balance.pos-newBalance) / bt.timeFactor + dt = float64(posBalance-newBalance) / bt.timeFactor return time.Duration(dt), true } else { - dt = float64(bt.balance.pos) / bt.timeFactor + dt = float64(posBalance) / bt.timeFactor } } else { if priority < 0 { @@ -151,16 +261,18 @@ func (bt *balanceTracker) timeUntil(priority int64) (time.Duration, bool) { } } // if we have a positive balance then dt equals the time needed to get it to zero - if uint64(priority) > bt.balance.neg { + negBalance := bt.balance.neg.value(bt.exp.negExpiration(now)) + if uint64(priority) > negBalance { if bt.negTimeFactor < 1e-100 { return 0, false } - dt += float64(uint64(priority)-bt.balance.neg) / bt.negTimeFactor + dt += float64(uint64(priority)-negBalance) / bt.negTimeFactor } return time.Duration(dt), true } // setCapacity updates the capacity value used for priority calculation +// Note: capacity should never be zero func (bt *balanceTracker) setCapacity(capacity uint64) { bt.lock.Lock() defer bt.lock.Unlock() @@ -262,26 +374,26 @@ func (bt *balanceTracker) updateAfter(dt time.Duration) { } // requestCost should be called after serving a request for the given peer -func (bt *balanceTracker) requestCost(cost uint64) { +func (bt *balanceTracker) requestCost(cost uint64) uint64 { bt.lock.Lock() defer bt.lock.Unlock() if bt.stopped { - return + return 0 } now := bt.clock.Now() bt.addBalance(now) fcost := float64(cost) - if bt.balance.pos != 0 { + posExp := bt.exp.posExpiration(now) + if bt.balance.pos.base != 0 { if bt.requestFactor != 0 { - c := uint64(fcost * bt.requestFactor) - if bt.balance.pos >= c { - bt.balance.pos -= c + c := -int64(fcost * bt.requestFactor) + cc := bt.balance.pos.add(c, posExp) + if c == cc { fcost = 0 } else { - fcost *= 1 - float64(bt.balance.pos)/float64(c) - bt.balance.pos = 0 + fcost *= 1 - float64(cc)/float64(c) } bt.checkCallbacks(now) } else { @@ -290,15 +402,16 @@ func (bt *balanceTracker) requestCost(cost uint64) { } if fcost > 0 { if bt.negRequestFactor != 0 { - bt.balance.neg += uint64(fcost * bt.negRequestFactor) + bt.balance.neg.add(int64(fcost*bt.negRequestFactor), bt.exp.negExpiration(now)) bt.checkCallbacks(now) } } bt.sumReqCost += cost + return bt.balance.pos.value(posExp) } // getBalance returns the current positive and negative balance -func (bt *balanceTracker) getBalance(now mclock.AbsTime) (uint64, uint64) { +func (bt *balanceTracker) getBalance(now mclock.AbsTime) (expiredValue, expiredValue) { bt.lock.Lock() defer bt.lock.Unlock() @@ -307,7 +420,7 @@ func (bt *balanceTracker) getBalance(now mclock.AbsTime) (uint64, uint64) { } // setBalance sets the positive and negative balance to the given values -func (bt *balanceTracker) setBalance(pos, neg uint64) error { +func (bt *balanceTracker) setBalance(pos, neg expiredValue) error { bt.lock.Lock() defer bt.lock.Unlock() diff --git a/les/balance_test.go b/les/balance_test.go index b571c2cc5c..91032fc714 100644 --- a/les/balance_test.go +++ b/les/balance_test.go @@ -23,18 +23,31 @@ import ( "github.com/ethereum/go-ethereum/common/mclock" ) +type zeroExpCtrl struct{} + +func (z zeroExpCtrl) posExpiration(mclock.AbsTime) uint64 { + return 0 +} + +func (z zeroExpCtrl) negExpiration(mclock.AbsTime) uint64 { + return 0 +} + +func expval(v uint64) expiredValue { + return expiredValue{base: v} +} + func TestSetBalance(t *testing.T) { var clock = &mclock.Simulated{} var inputs = []struct { - pos uint64 - neg uint64 + pos, neg expiredValue }{ - {1000, 0}, - {0, 1000}, - {1000, 1000}, + {expval(1000), expval(0)}, + {expval(0), expval(1000)}, + {expval(1000), expval(1000)}, } - tracker := balanceTracker{} + tracker := balanceTracker{exp: zeroExpCtrl{}} tracker.init(clock, 1000) defer tracker.stop(clock.Now()) @@ -53,14 +66,14 @@ func TestSetBalance(t *testing.T) { func TestBalanceTimeCost(t *testing.T) { var ( clock = &mclock.Simulated{} - tracker = balanceTracker{} + tracker = balanceTracker{exp: zeroExpCtrl{}} ) tracker.init(clock, 1000) defer tracker.stop(clock.Now()) tracker.setFactors(false, 1, 1) tracker.setFactors(true, 1, 1) - tracker.setBalance(uint64(time.Minute), 0) // 1 minute time allowance + tracker.setBalance(expval(uint64(time.Minute)), expval(0)) // 1 minute time allowance var inputs = []struct { runTime time.Duration @@ -74,21 +87,21 @@ func TestBalanceTimeCost(t *testing.T) { } for _, i := range inputs { clock.Run(i.runTime) - if pos, _ := tracker.getBalance(clock.Now()); pos != i.expPos { + if pos, _ := tracker.getBalance(clock.Now()); pos != expval(i.expPos) { t.Fatalf("Positive balance mismatch, want %v, got %v", i.expPos, pos) } - if _, neg := tracker.getBalance(clock.Now()); neg != i.expNeg { + if _, neg := tracker.getBalance(clock.Now()); neg != expval(i.expNeg) { t.Fatalf("Negative balance mismatch, want %v, got %v", i.expNeg, neg) } } - tracker.setBalance(uint64(time.Minute), 0) // Refill 1 minute time allowance + tracker.setBalance(expval(uint64(time.Minute)), expval(0)) // Refill 1 minute time allowance for _, i := range inputs { clock.Run(i.runTime) - if pos, _ := tracker.getBalance(clock.Now()); pos != i.expPos { + if pos, _ := tracker.getBalance(clock.Now()); pos != expval(i.expPos) { t.Fatalf("Positive balance mismatch, want %v, got %v", i.expPos, pos) } - if _, neg := tracker.getBalance(clock.Now()); neg != i.expNeg { + if _, neg := tracker.getBalance(clock.Now()); neg != expval(i.expNeg) { t.Fatalf("Negative balance mismatch, want %v, got %v", i.expNeg, neg) } } @@ -97,14 +110,14 @@ func TestBalanceTimeCost(t *testing.T) { func TestBalanceReqCost(t *testing.T) { var ( clock = &mclock.Simulated{} - tracker = balanceTracker{} + tracker = balanceTracker{exp: zeroExpCtrl{}} ) tracker.init(clock, 1000) defer tracker.stop(clock.Now()) tracker.setFactors(false, 1, 1) tracker.setFactors(true, 1, 1) - tracker.setBalance(uint64(time.Minute), 0) // 1 minute time serving time allowance + tracker.setBalance(expval(uint64(time.Minute)), expval(0)) // 1 minute time serving time allowance var inputs = []struct { reqCost uint64 expPos uint64 @@ -117,10 +130,10 @@ func TestBalanceReqCost(t *testing.T) { } for _, i := range inputs { tracker.requestCost(i.reqCost) - if pos, _ := tracker.getBalance(clock.Now()); pos != i.expPos { + if pos, _ := tracker.getBalance(clock.Now()); pos != expval(i.expPos) { t.Fatalf("Positive balance mismatch, want %v, got %v", i.expPos, pos) } - if _, neg := tracker.getBalance(clock.Now()); neg != i.expNeg { + if _, neg := tracker.getBalance(clock.Now()); neg != expval(i.expNeg) { t.Fatalf("Negative balance mismatch, want %v, got %v", i.expNeg, neg) } } @@ -129,7 +142,7 @@ func TestBalanceReqCost(t *testing.T) { func TestBalanceToPriority(t *testing.T) { var ( clock = &mclock.Simulated{} - tracker = balanceTracker{} + tracker = balanceTracker{exp: zeroExpCtrl{}} ) tracker.init(clock, 1000) // cap = 1000 defer tracker.stop(clock.Now()) @@ -141,13 +154,13 @@ func TestBalanceToPriority(t *testing.T) { neg uint64 priority int64 }{ - {1000, 0, ^int64(1)}, - {2000, 0, ^int64(2)}, // Higher balance, lower priority value + {1000, 0, -1}, + {2000, 0, -2}, // Higher balance, lower priority value {0, 0, 0}, {0, 1000, 1000}, } for _, i := range inputs { - tracker.setBalance(i.pos, i.neg) + tracker.setBalance(expval(i.pos), expval(i.neg)) priority := tracker.getPriority(clock.Now()) if priority != i.priority { t.Fatalf("Priority mismatch, want %v, got %v", i.priority, priority) @@ -158,30 +171,30 @@ func TestBalanceToPriority(t *testing.T) { func TestEstimatedPriority(t *testing.T) { var ( clock = &mclock.Simulated{} - tracker = balanceTracker{} + tracker = balanceTracker{exp: zeroExpCtrl{}} ) tracker.init(clock, 1000000000) // cap = 1000,000,000 defer tracker.stop(clock.Now()) tracker.setFactors(false, 1, 1) tracker.setFactors(true, 1, 1) - tracker.setBalance(uint64(time.Minute), 0) + tracker.setBalance(expval(uint64(time.Minute)), expval(0)) var inputs = []struct { runTime time.Duration // time cost futureTime time.Duration // diff of future time reqCost uint64 // single request cost priority int64 // expected estimated priority }{ - {time.Second, time.Second, 0, ^int64(58)}, - {0, time.Second, 0, ^int64(58)}, + {time.Second, time.Second, 0, -58}, + {0, time.Second, 0, -58}, // 2 seconds time cost, 1 second estimated time cost, 10^9 request cost, // 10^9 estimated request cost per second. - {time.Second, time.Second, 1000000000, ^int64(55)}, + {time.Second, time.Second, 1000000000, -55}, // 3 seconds time cost, 3 second estimated time cost, 10^9*2 request cost, // 4*10^9 estimated request cost. - {time.Second, 3 * time.Second, 1000000000, ^int64(48)}, + {time.Second, 3 * time.Second, 1000000000, -48}, // All positive balance is used up {time.Second * 55, 0, 0, 0}, @@ -202,7 +215,7 @@ func TestEstimatedPriority(t *testing.T) { func TestCallbackChecking(t *testing.T) { var ( clock = &mclock.Simulated{} - tracker = balanceTracker{} + tracker = balanceTracker{exp: zeroExpCtrl{}} ) tracker.init(clock, 1000000) // cap = 1000,000 defer tracker.stop(clock.Now()) @@ -213,11 +226,11 @@ func TestCallbackChecking(t *testing.T) { priority int64 expDiff time.Duration }{ - {^int64(500), time.Millisecond * 500}, + {-500, time.Millisecond * 500}, {0, time.Second}, {int64(time.Second), 2 * time.Second}, } - tracker.setBalance(uint64(time.Second), 0) + tracker.setBalance(expval(uint64(time.Second)), expval(0)) for _, i := range inputs { diff, _ := tracker.timeUntil(i.priority) if diff != i.expDiff { @@ -229,7 +242,7 @@ func TestCallbackChecking(t *testing.T) { func TestCallback(t *testing.T) { var ( clock = &mclock.Simulated{} - tracker = balanceTracker{} + tracker = balanceTracker{exp: zeroExpCtrl{}} ) tracker.init(clock, 1000) // cap = 1000 defer tracker.stop(clock.Now()) @@ -237,7 +250,7 @@ func TestCallback(t *testing.T) { tracker.setFactors(true, 1, 1) callCh := make(chan struct{}, 1) - tracker.setBalance(uint64(time.Minute), 0) + tracker.setBalance(expval(uint64(time.Minute)), expval(0)) tracker.addCallback(balanceCallbackZero, 0, func() { callCh <- struct{}{} }) clock.Run(time.Minute) @@ -247,7 +260,7 @@ func TestCallback(t *testing.T) { t.Fatalf("Callback hasn't been called yet") } - tracker.setBalance(uint64(time.Minute), 0) + tracker.setBalance(expval(uint64(time.Minute)), expval(0)) tracker.addCallback(balanceCallbackZero, 0, func() { callCh <- struct{}{} }) tracker.removeCallback(balanceCallbackZero) diff --git a/les/clientpool.go b/les/clientpool.go index b01c825a7a..760c24d1e5 100644 --- a/les/clientpool.go +++ b/les/clientpool.go @@ -36,21 +36,26 @@ import ( ) const ( - negBalanceExpTC = time.Hour // time constant for exponentially reducing negative balance - fixedPointMultiplier = 0x1000000 // constant to convert logarithms to fixed point format - lazyQueueRefresh = time.Second * 10 // refresh period of the connected queue - persistCumulativeTimeRefresh = time.Minute * 5 // refresh period of the cumulative running time persistence - posBalanceCacheLimit = 8192 // the maximum number of cached items in positive balance queue - negBalanceCacheLimit = 8192 // the maximum number of cached items in negative balance queue + defaultPosExpTC = 36000 // default time constant (in seconds) for exponentially reducing positive balance + defaultNegExpTC = 3600 // default time constant (in seconds) for exponentially reducing negative balance + logMultiplier = 24204406.323122971 // constant to convert natural logarithms to fixed point format + log2Multiplier = 0x1000000 // constant to convert 2-based logarithms to fixed point format + lazyQueueRefresh = time.Second * 10 // refresh period of the connected queue + tryActivatePeriod = time.Second * 5 // periodically check whether inactive clients can be activated + dropInactiveCycles = 2 // number of activation check periods after non-priority inactive peers are dropped + persistExpirationRefresh = time.Minute * 5 // refresh period of the token expiration persistence + posBalanceCacheLimit = 8192 // the maximum number of cached items in positive balance queue + negBalanceCacheLimit = 8192 // the maximum number of cached items in negative balance queue + freeRatioTC = time.Hour // time constant of token supply control based on free service availability - // connectedBias is applied to already connected clients So that + // activeBias is applied to already connected clients So that // already connected client won't be kicked out very soon and we // can ensure all connected clients can have enough time to request // or sync some data. // // todo(rjl493456442) make it configurable. It can be the option of // free trial time! - connectedBias = time.Minute * 3 + activeBias = time.Minute * 3 ) // clientPool implements a client database that assigns a priority to each client @@ -60,7 +65,7 @@ const ( // then negative balance is accumulated. // // Balance tracking and priority calculation for connected clients is done by -// balanceTracker. connectedQueue ensures that clients with the lowest positive or +// balanceTracker. activeQueue ensures that clients with the lowest positive or // highest negative balance get evicted when the total capacity allowance is full // and new clients with a better balance want to connect. // @@ -82,19 +87,31 @@ type clientPool struct { closed bool removePeer func(enode.ID) - connectedMap map[enode.ID]*clientInfo - connectedQueue *prque.LazyQueue + connectedMap map[enode.ID]*clientInfo + activeQueue *prque.LazyQueue + inactiveQueue *prque.Prque + dropInactivePeers map[uint64][]*clientInfo + dropInactiveCounter uint64 + + activeBalances, inactiveBalances expiredValue + lastConnectedBalanceUpdate mclock.AbsTime + freeRatio, averageFreeRatio float64 defaultPosFactors, defaultNegFactors priceFactors - connLimit int // The maximum number of connections that clientpool can support - 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 - 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. - disableBias bool // Disable connection bias(used in testing) + activeLimit int // The maximum number of connections that clientpool can support + capLimit uint64 // The maximum cumulative capacity that clientpool can support + activeCap uint64 // The sum of the capacity of the current clientpool connected + priorityActive 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 + disableBias bool // Disable connection bias(used in testing) + + // fields in this group are protected by expLock + expLock sync.RWMutex + posExp, negExp, posExpTC, negExpTC uint64 + posExpTCi, negExpTCi float64 // already inverted (logMultiplier/time) + freeRatioLastUpdate mclock.AbsTime } // clientPoolPeer represents a client peer in the pool. @@ -113,36 +130,38 @@ type clientPoolPeer interface { type clientInfo struct { address string id enode.ID + freeID string + active bool connectedAt mclock.AbsTime capacity uint64 priority bool pool *clientPool peer clientPoolPeer - queueIndex int // position in connectedQueue + queueIndex int // position in activeQueue balanceTracker balanceTracker posFactors, negFactors priceFactors balanceMetaInfo string } -// connSetIndex callback updates clientInfo item index in connectedQueue +// connSetIndex callback updates clientInfo item index in activeQueue func connSetIndex(a interface{}, index int) { a.(*clientInfo).queueIndex = index } -// connPriority callback returns actual priority of clientInfo item in connectedQueue +// connPriority callback returns actual priority of clientInfo item in activeQueue func connPriority(a interface{}, now mclock.AbsTime) int64 { c := a.(*clientInfo) return c.balanceTracker.getPriority(now) } -// connMaxPriority callback returns estimated maximum priority of clientInfo item in connectedQueue +// connMaxPriority callback returns estimated maximum priority of clientInfo item in activeQueue func connMaxPriority(a interface{}, until mclock.AbsTime) int64 { c := a.(*clientInfo) pri := c.balanceTracker.estimatedPriority(until, true) c.balanceTracker.addCallback(balanceCallbackQueue, pri+1, func() { c.pool.lock.Lock() - if c.queueIndex != -1 { - c.pool.connectedQueue.Update(c.queueIndex) + if c.active && c.queueIndex != -1 { + c.pool.activeQueue.Update(c.queueIndex) } c.pool.lock.Unlock() }) @@ -159,34 +178,94 @@ 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) + posExp, negExp := ndb.getExpiration() pool := &clientPool{ - ndb: ndb, - clock: clock, - connectedMap: make(map[enode.ID]*clientInfo), - connectedQueue: prque.NewLazyQueue(connSetIndex, connPriority, connMaxPriority, clock, lazyQueueRefresh), - freeClientCap: freeClientCap, - removePeer: removePeer, - startTime: clock.Now(), - cumulativeTime: ndb.getCumulativeTime(), - stopCh: make(chan struct{}), + ndb: ndb, + clock: clock, + connectedMap: make(map[enode.ID]*clientInfo), + activeQueue: prque.NewLazyQueue(connSetIndex, connPriority, connMaxPriority, clock, lazyQueueRefresh), + inactiveQueue: prque.New(connSetIndex), + dropInactivePeers: make(map[uint64][]*clientInfo), + minCap: minCap, + freeClientCap: freeClientCap, + removePeer: removePeer, + freeRatioLastUpdate: clock.Now(), + posExp: posExp, + negExp: negExp, + freeRatio: 1, + averageFreeRatio: 1, + stopCh: make(chan struct{}), + } + // set default expiration constants used by tests + // Note: server overwrites this if token sale is active + pool.setExpirationTCs(0, defaultNegExpTC) + // calculate total token balance amount + var start enode.ID + for { + ids := pool.ndb.getPosBalanceIDs(start, enode.ID{}, 1000) + var stop bool + l := len(ids) + if l == 1000 { + l-- + start = ids[l] + } else { + stop = true + } + for i := 0; i < l; i++ { + pool.inactiveBalances.addExp(pool.ndb.getOrNewPB(ids[i]).value) + } + if stop { + break + } } // If the negative balance of free client is even lower than 1, // delete this entry. ndb.nbEvictCallBack = func(now mclock.AbsTime, b negBalance) bool { - balance := math.Exp(float64(b.logValue-pool.logOffset(now)) / fixedPointMultiplier) - return balance <= 1 + return b.logValue <= pool.negExpiration(now) } go func() { for { select { case <-clock.After(lazyQueueRefresh): pool.lock.Lock() - pool.connectedQueue.Refresh() + pool.activeQueue.Refresh() + pool.lock.Unlock() + case <-pool.stopCh: + return + } + } + }() + go func() { + for { + select { + case <-clock.After(persistExpirationRefresh): + pool.lock.Lock() + now := pool.clock.Now() + posExp := pool.posExpiration(now) + negExp := pool.negExpiration(now) + pool.lock.Unlock() + pool.ndb.setExpiration(posExp, negExp) + case <-pool.stopCh: + return + } + } + }() + go func() { + for { + select { + case <-clock.After(tryActivatePeriod): + pool.lock.Lock() + pool.tryActivateClients() + for _, c := range pool.dropInactivePeers[pool.dropInactiveCounter] { + if _, ok := pool.connectedMap[c.id]; ok && !c.active && !c.priority { + pool.drop(c.peer, true) + } + } + delete(pool.dropInactivePeers, pool.dropInactiveCounter) + pool.dropInactiveCounter++ pool.lock.Unlock() - case <-clock.After(persistCumulativeTimeRefresh): - pool.ndb.setCumulativeTime(pool.logOffset(clock.Now())) case <-pool.stopCh: return } @@ -201,122 +280,213 @@ func (f *clientPool) stop() { f.lock.Lock() f.closed = true f.lock.Unlock() - f.ndb.setCumulativeTime(f.logOffset(f.clock.Now())) + now := f.clock.Now() + f.ndb.setExpiration(f.posExpiration(now), f.negExpiration(now)) f.ndb.close() } +// updateFreeRatio updates freeRatio, averageFreeRatio, posExp and negExp based +// on free service availability. Should be called after capLimit or priorityActive +// is changed. +func (f *clientPool) updateFreeRatio() { + f.freeRatio = 0 + if f.priorityActive < f.capLimit { + freeCap := f.capLimit - f.priorityActive + if freeCap > f.freeClientCap { + freeCapThreshold := f.capLimit / 4 + if freeCap > freeCapThreshold { + f.freeRatio = 1 + } else { + f.freeRatio = float64(freeCap-f.freeClientCap) / float64(freeCapThreshold-f.freeClientCap) + } + } + } + f.expLock.Lock() + now := f.clock.Now() + dt := now - f.freeRatioLastUpdate + if dt < 0 { + dt = 0 + } + f.averageFreeRatio -= (f.freeRatio - f.averageFreeRatio) * math.Expm1(-float64(dt)/float64(freeRatioTC)) + f.freeRatioLastUpdate = now + f.posExp += uint64(float64(dt) * f.posExpTCi * f.freeRatio) + f.negExp += uint64(float64(dt) * f.negExpTCi * f.freeRatio) + f.expLock.Unlock() +} + +// setExpirationTCs sets positive and negative token expiration time constants. +// Specified in seconds, 0 means infinite (no expiration). +func (f *clientPool) setExpirationTCs(pos, neg uint64) { + f.lock.Lock() + f.updateFreeRatio() + f.lock.Unlock() + + f.expLock.Lock() + f.posExpTC, f.negExpTC = pos, neg + if pos > 0 { + f.posExpTCi = logMultiplier / float64(pos*uint64(time.Second)) + } else { + f.posExpTCi = 0 + } + if neg > 0 { + f.negExpTCi = logMultiplier / float64(neg*uint64(time.Second)) + } else { + f.negExpTCi = 0 + } + f.expLock.Unlock() +} + +// getExpirationTCs returns the current positive and negative token expiration +// time constants +func (f *clientPool) getExpirationTCs() (pos, neg uint64) { + f.expLock.Lock() + defer f.expLock.Unlock() + + return f.posExpTC, f.negExpTC +} + +// posExpiration implements expirationController. Expiration happens only when +// free service is available. +func (f *clientPool) posExpiration(now mclock.AbsTime) uint64 { + f.expLock.RLock() + defer f.expLock.RUnlock() + + dt := now - f.freeRatioLastUpdate + if dt < 0 { + dt = 0 + } + return f.posExp + uint64(float64(dt)*f.posExpTCi*f.freeRatio) +} + +// negExpiration implements expirationController. Expiration happens only when +// free service is available. +func (f *clientPool) negExpiration(now mclock.AbsTime) uint64 { + f.expLock.RLock() + defer f.expLock.RUnlock() + + dt := now - f.freeRatioLastUpdate + if dt < 0 { + dt = 0 + } + return f.negExp + uint64(float64(dt)*f.negExpTCi*f.freeRatio) +} + +// totalTokenLimit returns the current token supply limit. Token prices are based +// on the ratio of total token amount and supply limit while the limit depends on +// averageFreeRatio, ensuring the availability of free service most of the time. +func (f *clientPool) totalTokenLimit() uint64 { + f.lock.Lock() + defer f.lock.Unlock() + + f.updateFreeRatio() + d := f.averageFreeRatio + if d > 0.5 { + d = -math.Log(0.5/d) * float64(freeRatioTC) + } else { + d = 0 + } + return uint64(d * float64(f.capLimit) * f.defaultPosFactors.capacityFactor) +} + +// totalTokenAmount returns the total amount of currently existing service tokens +func (f *clientPool) totalTokenAmount() uint64 { + f.lock.Lock() + defer f.lock.Unlock() + + now := f.clock.Now() + if now > f.lastConnectedBalanceUpdate+mclock.AbsTime(time.Second) { + f.activeBalances = expiredValue{} + for _, c := range f.connectedMap { + pos, _ := c.balanceTracker.getBalance(now) + f.activeBalances.addExp(pos) + } + f.lastConnectedBalanceUpdate = now + } + sum := f.activeBalances + sum.addExp(f.inactiveBalances) + return sum.value(f.posExpiration(now)) +} + // connect should be called after a successful handshake. If the connection was // rejected, there is no need to call disconnect. -func (f *clientPool) connect(peer clientPoolPeer, capacity uint64) bool { +func (f *clientPool) connect(peer clientPoolPeer, reqCapacity uint64) (uint64, error) { f.lock.Lock() defer f.lock.Unlock() // Short circuit if clientPool is already closed. if f.closed { - return false + return 0, fmt.Errorf("Client pool is already closed") } // Dedup connected peers. id, freeID := peer.ID(), peer.freeClientId() if _, ok := f.connectedMap[id]; ok { clientRejectedMeter.Mark(1) log.Debug("Client already connected", "address", freeID, "id", peerIdToString(id)) - return false + return 0, fmt.Errorf("Client already connected address = %s id = %s", freeID, peerIdToString(id)) } - // Create a clientInfo but do not add it yet - var ( - 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{ + capacity: reqCapacity, pool: f, peer: peer, address: freeID, queueIndex: -1, id: id, - connectedAt: now, - priority: posBalance != 0, + freeID: freeID, + connectedAt: f.clock.Now(), + priority: pb.value.base != 0, posFactors: f.defaultPosFactors, negFactors: f.defaultNegFactors, balanceMetaInfo: pb.meta, } - // If the client is a free client, assign with a low free capacity, - // Otherwise assign with the given value(priority client) - if !e.priority || capacity == 0 { - capacity = f.freeClientCap - } - e.capacity = 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. - // - // If the priority of the newly added client is lower than the priority of - // all connected clients, the client is rejected. - newCapacity := f.connectedCap + capacity - newCount := f.connectedQueue.Size() + 1 - if newCapacity > f.capLimit || newCount > f.connLimit { - var ( - kickList []*clientInfo - kickPriority int64 - ) - f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool { - c := data.(*clientInfo) - kickList = append(kickList, c) - kickPriority = priority - newCapacity -= c.capacity - newCount-- - return newCapacity > f.capLimit || newCount > f.connLimit - }) - bias := connectedBias - if f.disableBias { - bias = 0 - } - if newCapacity > f.capLimit || newCount > f.connLimit || (e.balanceTracker.estimatedPriority(now+mclock.AbsTime(bias), false)-kickPriority) > 0 { - for _, c := range kickList { - f.connectedQueue.Push(c) - } - clientRejectedMeter.Mark(1) - log.Debug("Client rejected", "address", freeID, "id", peerIdToString(id)) - return false - } - // accept new client, drop old ones - for _, c := range kickList { - f.dropClient(c, now, true) - } - } - - // Register new client to connection queue. + missing, capacity := f.capAvailable(id, freeID, reqCapacity, 0, true) f.connectedMap[id] = e - f.connectedQueue.Push(e) - f.connectedCap += e.capacity + if missing != 0 { + // capacity is not available, add client to inactive queue + f.initBalanceTracker(&e.balanceTracker, pb, nb, capacity, false) + f.inactiveQueue.Push(e, -connPriority(e, f.clock.Now())) + return 0, nil + } + // capacity is available, add client + e.active = true + e.capacity = capacity + f.initBalanceTracker(&e.balanceTracker, pb, nb, capacity, true) + // Register new client to connection queue. + f.inactiveBalances.subExp(pb.value) + f.activeBalances.addExp(pb.value) + f.activeQueue.Push(e) + f.activeCap += e.capacity // If the current client is a paid client, monitor the status of client, // downgrade it to normal client if positive balance is used up. if e.priority { - f.priorityConnected += capacity + f.priorityActive += capacity + f.updateFreeRatio() e.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) }) } - // If the capacity of client is not the default value(free capacity), notify - // it to update capacity. - if e.capacity != f.freeClientCap { - e.peer.updateCapacity(e.capacity) - } - totalConnectedGauge.Update(int64(f.connectedCap)) + totalConnectedGauge.Update(int64(f.activeCap)) clientConnectedMeter.Mark(1) log.Debug("Client accepted", "address", freeID) - return true + return e.capacity, nil +} + +// initBalanceTracker initializes the positive and negative balances and price factors +func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb posBalance, nb negBalance, capacity uint64, active bool) { + bt.exp = f + posBalance := pb.value + var negBalance expiredValue + if nb.logValue != 0 { + negBalance.exp = nb.logValue / log2Multiplier + negBalance.base = uint64(math.Exp(float64(nb.logValue%log2Multiplier)/logMultiplier) * float64(time.Second)) + } + bt.init(f.clock, capacity) + bt.setBalance(posBalance, negBalance) + if active { + updatePriceFactors(bt, f.defaultPosFactors, f.defaultNegFactors, capacity) + } else { + zeroPriceFactors(bt) + } } // disconnect should be called when a connection is terminated. If the disconnection @@ -326,17 +496,106 @@ func (f *clientPool) disconnect(p clientPoolPeer) { f.lock.Lock() defer f.lock.Unlock() + f.drop(p, false) +} + +// drop deactivates the peer if necessary and drops it from the inactive queue +func (f *clientPool) drop(p clientPeer, kicked bool) { // Short circuit if client pool is already closed. if f.closed { return } - // Short circuit if the peer hasn't been registered. - e := f.connectedMap[p.ID()] - if e == nil { + e, ok := f.connectedMap[p.ID()] + if !ok { log.Debug("Client not connected", "address", p.freeClientId(), "id", peerIdToString(p.ID())) return } - f.dropClient(e, f.clock.Now(), false) + tryActivate := e.active + if e.active { + f.deactivateClient(e, false) + } + f.finalizeBalance(e, f.clock.Now()) + f.inactiveQueue.Remove(e.queueIndex) + delete(f.connectedMap, e.id) + if kicked { + clientKickedMeter.Mark(1) + log.Debug("Client kicked out", "address", e.address) + } else { + clientDisconnectedMeter.Mark(1) + log.Debug("Client disconnected", "address", e.address) + } + if tryActivate { + f.tryActivateClients() + } +} + +// capAvailable checks whether the current priority level of the given client is enough to +// connect or change capacity to the requested level and then stay connected for at least +// the specified duration. If not then the additional required amount of positive balance is returned. +func (f *clientPool) capAvailable(id enode.ID, freeID string, capacity uint64, minConnTime time.Duration, kick bool) (uint64, uint64) { + var missing uint64 + if capacity == 0 { + capacity = f.freeClientCap + } + if capacity < f.minCap { + capacity = f.minCap + } + newCapacity := f.activeCap + capacity + newCount := f.activeQueue.Size() + 1 + client := f.connectedMap[id] + if client != nil && client.active { + newCapacity -= client.capacity + newCount-- + } + if newCapacity > f.capLimit || newCount > f.activeLimit { + var ( + popList []*clientInfo + targetPriority int64 + ) + f.activeQueue.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.activeLimit + }) + if newCapacity > f.capLimit || newCount > f.activeLimit { + 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, true) + } + if capacity != f.freeClientCap && targetPriority >= 0 { + targetPriority = -1 + } + bias := activeBias + if f.disableBias { + bias = 0 + } + if bias < minConnTime { + bias = minConnTime + } + missing = bt.posBalanceMissing(targetPriority, capacity, bias) + } + if missing != 0 { + kick = false + } + for _, c := range popList { + if kick && c != client { + f.deactivateClient(c, true) + } else { + f.activeQueue.Push(c) + } + } + } + return missing, capacity } // forClients iterates through a list of clients, calling the callback for each one. @@ -371,27 +630,61 @@ func (f *clientPool) setDefaultFactors(posFactors, negFactors priceFactors) { f.defaultNegFactors = negFactors } -// dropClient removes a client from the connected queue and finalizes its balance. -// If kick is true then it also initiates the disconnection. -func (f *clientPool) dropClient(e *clientInfo, now mclock.AbsTime, kick bool) { - if _, ok := f.connectedMap[e.id]; !ok { +// deactivateClient puts a client in inactive state +func (f *clientPool) deactivateClient(e *clientInfo, scheduleDrop bool) { + if _, ok := f.connectedMap[e.id]; !ok || !e.active { return } - f.finalizeBalance(e, now) - f.connectedQueue.Remove(e.queueIndex) - delete(f.connectedMap, e.id) - f.connectedCap -= e.capacity + f.activeQueue.Remove(e.queueIndex) + f.activeCap -= e.capacity if e.priority { - f.priorityConnected -= e.capacity + f.priorityActive -= e.capacity + f.updateFreeRatio() } - totalConnectedGauge.Update(int64(f.connectedCap)) - if kick { - clientKickedMeter.Mark(1) - log.Debug("Client kicked out", "address", e.address) - f.removePeer(e.id) - } else { - clientDisconnectedMeter.Mark(1) - log.Debug("Client disconnected", "address", e.address) + e.active = false + e.peer.updateCapacity(0) + totalConnectedGauge.Update(int64(f.activeCap)) + f.inactiveQueue.Push(e, -connPriority(e, f.clock.Now())) + if scheduleDrop { + f.dropInactivePeers[f.dropInactiveCounter+dropInactiveCycles] = append(f.dropInactivePeers[f.dropInactiveCounter+dropInactiveCycles], e) + } +} + +// tryActivateClients checks whether some inactive clients have enough priority now +// and activates them if possible +func (f *clientPool) tryActivateClients() { + now := f.clock.Now() + for f.inactiveQueue.Size() != 0 { + e := f.inactiveQueue.PopItem().(*clientInfo) + missing, capacity := f.capAvailable(e.id, e.freeID, e.capacity, 0, true) + if missing != 0 { + f.inactiveQueue.Push(e, -connPriority(e, now)) + return + } + // capacity is available, activate client + e.active = true + e.capacity = capacity + e.peer.updateCapacity(capacity) + balance, _ := e.balanceTracker.getBalance(now) + e.balanceTracker.setCapacity(capacity) + updatePriceFactors(&e.balanceTracker, f.defaultPosFactors, f.defaultNegFactors, capacity) + // Register activated client to connection queue. + f.inactiveBalances.subExp(balance) + f.activeBalances.addExp(balance) + f.activeQueue.Push(e) + f.activeCap += e.capacity + + // If the current client is a paid client, monitor the status of client, + // downgrade it to normal client if positive balance is used up. + if e.priority { + f.priorityActive += capacity + f.updateFreeRatio() + e.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(e.id) }) + } + e.peer.updateCapacity(e.capacity) + totalConnectedGauge.Update(int64(f.activeCap)) + clientConnectedMeter.Mark(1) + log.Debug("Client activated", "address", e.freeID) } } @@ -401,7 +694,7 @@ func (f *clientPool) capacityInfo() (uint64, uint64, uint64) { f.lock.Lock() defer f.lock.Unlock() - return f.capLimit, f.connectedCap, f.priorityConnected + return f.capLimit, f.activeCap, f.priorityActive } // finalizeBalance stops the balance tracker, retrieves the final balances and @@ -409,14 +702,19 @@ func (f *clientPool) capacityInfo() (uint64, uint64, uint64) { func (f *clientPool) finalizeBalance(c *clientInfo, now mclock.AbsTime) { c.balanceTracker.stop(now) pos, neg := c.balanceTracker.getBalance(now) + f.inactiveBalances.addExp(pos) + f.activeBalances.subExp(pos) pb, nb := f.ndb.getOrNewPB(c.id), f.ndb.getOrNewNB(c.address) pb.value = pos f.ndb.setPB(c.id, pb) - neg /= uint64(time.Second) // Convert the expanse to second level. - if neg > 1 { - nb.logValue = int64(math.Log(float64(neg))*fixedPointMultiplier) + f.logOffset(now) + var nbLog int64 + if neg.base > 0 { + nbLog = int64(math.Log(float64(neg.base)/float64(time.Second))*logMultiplier) + int64(neg.exp*log2Multiplier) + } + if nbLog > 0 { + nb.logValue = uint64(nbLog) f.ndb.setNB(c.address, nb) } else { f.ndb.delNB(c.address) // Negative balance is small enough, drop it directly. @@ -434,115 +732,104 @@ func (f *clientPool) balanceExhausted(id enode.ID) { return } if c.priority { - f.priorityConnected -= c.capacity + f.priorityActive -= c.capacity + f.updateFreeRatio() } c.priority = false if c.capacity != f.freeClientCap { - f.connectedCap += f.freeClientCap - c.capacity - totalConnectedGauge.Update(int64(f.connectedCap)) + f.activeCap += f.freeClientCap - c.capacity + totalConnectedGauge.Update(int64(f.activeCap)) c.capacity = f.freeClientCap c.balanceTracker.setCapacity(c.capacity) c.peer.updateCapacity(c.capacity) } pb := f.ndb.getOrNewPB(id) - pb.value = 0 + pb.value = expiredValue{} f.ndb.setPB(id, pb) } -// setConnLimit sets the maximum number and total capacity of connected clients, +// setactiveLimit sets the maximum number and total capacity of connected clients, // dropping some of them if necessary. func (f *clientPool) setLimits(totalConn int, totalCap uint64) { f.lock.Lock() defer f.lock.Unlock() - f.connLimit = totalConn + f.activeLimit = totalConn f.capLimit = totalCap - if f.connectedCap > f.capLimit || f.connectedQueue.Size() > f.connLimit { - f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool { - f.dropClient(data.(*clientInfo), mclock.Now(), true) - return f.connectedCap > f.capLimit || f.connectedQueue.Size() > f.connLimit + if f.activeCap > f.capLimit || f.activeQueue.Size() > f.activeLimit { + f.activeQueue.MultiPop(func(data interface{}, priority int64) bool { + f.deactivateClient(data.(*clientInfo), true) + return f.activeCap > f.capLimit || f.activeQueue.Size() > f.activeLimit }) + } else { + f.tryActivateClients() } + f.updateFreeRatio() } // setCapacity sets the assigned capacity of a connected client -func (f *clientPool) setCapacity(c *clientInfo, capacity uint64) error { - if f.connectedMap[c.id] != c { - return fmt.Errorf("client %064x is not connected", c.id[:]) - } - if c.capacity == capacity { - return nil - } - if !c.priority { - return errNoPriority - } - oldCapacity := c.capacity - c.capacity = capacity - f.connectedCap += capacity - oldCapacity - c.balanceTracker.setCapacity(capacity) - f.connectedQueue.Update(c.queueIndex) - if f.connectedCap > f.capLimit { - var kickList []*clientInfo - kick := true - f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool { - client := data.(*clientInfo) - kickList = append(kickList, client) - f.connectedCap -= client.capacity - if client == c { - kick = false - } - return kick && (f.connectedCap > f.capLimit) - }) - if kick { - now := mclock.Now() - for _, c := range kickList { - f.dropClient(c, now, true) - } - } else { - c.capacity = oldCapacity - c.balanceTracker.setCapacity(oldCapacity) - for _, c := range kickList { - f.connectedCap += c.capacity - f.connectedQueue.Push(c) - } - return errNoPriority +func (f *clientPool) setCapacity(id enode.ID, freeID string, capacity uint64, minConnTime time.Duration, setCap bool) (uint64, uint64, error) { + c := f.connectedMap[id] + if c != nil { + if c.capacity == capacity { + return 0, capacity, nil } } - totalConnectedGauge.Update(int64(f.connectedCap)) - f.priorityConnected += capacity - oldCapacity - c.updatePriceFactors() - c.peer.updateCapacity(c.capacity) - return nil + var missing uint64 + missing, capacity = f.capAvailable(id, freeID, capacity, 0, setCap && c != nil) + if missing != 0 { + return missing, capacity, errNoPriority + } + // capacity update is possible + if setCap { + if c == nil { + return 0, capacity, fmt.Errorf("client %064x is not connected", c.id[:]) + } + f.activeCap += capacity - c.capacity + f.priorityActive += capacity - c.capacity + f.updateFreeRatio() + c.capacity = capacity + c.balanceTracker.setCapacity(capacity) + f.activeQueue.Update(c.queueIndex) + totalConnectedGauge.Update(int64(f.activeCap)) + updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors, c.capacity) + c.peer.updateCapacity(c.capacity) + f.tryActivateClients() + } + return 0, capacity, nil } -// requestCost feeds request cost after serving a request from the given peer. -func (f *clientPool) requestCost(p *clientPeer, cost uint64) { +// setCapacityLocked is the equivalent of setCapacity used when f.lock is already locked +func (f *clientPool) setCapacityLocked(id enode.ID, freeID string, capacity uint64, minConnTime time.Duration, setCap bool) (uint64, uint64, error) { f.lock.Lock() defer f.lock.Unlock() - info, exist := f.connectedMap[p.ID()] - if !exist || f.closed { - return + return f.setCapacity(id, freeID, capacity, minConnTime, setCap) +} + +// requestCost feeds request cost after serving a request from the given peer and +// returns the remaining token balance +func (f *clientPool) requestCost(p *clientPeer, cost uint64) uint64 { + f.lock.Lock() + defer f.lock.Unlock() + + c := f.connectedMap[p.ID()] + if c == nil || f.closed { + return 0 } - info.balanceTracker.requestCost(cost) + return c.balanceTracker.requestCost(cost) } -// logOffset calculates the time-dependent offset for the logarithmic -// representation of negative balance -// -// From another point of view, the result returned by the function represents -// the total time that the clientpool is cumulatively running(total_hours/multiplier). -func (f *clientPool) logOffset(now mclock.AbsTime) int64 { - // Note: fixedPointMultiplier acts as a multiplier here; the reason for dividing the divisor - // is to avoid int64 overflow. We assume that int64(negBalanceExpTC) >> fixedPointMultiplier. - cumulativeTime := int64((time.Duration(now - f.startTime)) / (negBalanceExpTC / fixedPointMultiplier)) - return f.cumulativeTime + cumulativeTime +// 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) } -// 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) +// zeroPriceFactors sets the pricing factors to zero +func zeroPriceFactors(bt *balanceTracker) { + bt.setFactors(true, 0, 0) + bt.setFactors(false, 0, 0) } // getPosBalance retrieves a single positive balance entry from cache or the database @@ -550,7 +837,12 @@ func (f *clientPool) getPosBalance(id enode.ID) posBalance { f.lock.Lock() defer f.lock.Unlock() - return f.ndb.getOrNewPB(id) + if c := f.connectedMap[id]; c != nil { + pb, _ := c.balanceTracker.getBalance(f.clock.Now()) + return posBalance{value: pb, meta: c.balanceMetaInfo} + } else { + return f.ndb.getOrNewPB(id) + } } // addBalance updates the balance of a client (either overwrites it or adds to it). @@ -559,75 +851,87 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, f.lock.Lock() defer f.lock.Unlock() + now := f.clock.Now() pb := f.ndb.getOrNewPB(id) - var negBalance uint64 + var negBalance expiredValue c := f.connectedMap[id] if c != nil { - pb.value, negBalance = c.balanceTracker.getBalance(f.clock.Now()) + pb.value, negBalance = c.balanceTracker.getBalance(now) } oldBalance := pb.value - if amount > 0 { - if amount > maxBalance || pb.value > maxBalance-uint64(amount) { - return oldBalance, oldBalance, errBalanceOverflow - } - pb.value += uint64(amount) - } else { - if uint64(-amount) > pb.value { - pb.value = 0 - } else { - pb.value -= uint64(-amount) - } + posExp := f.posExpiration(now) + oldValue := oldBalance.value(posExp) + if amount > 0 && (amount > maxBalance || oldValue > maxBalance-uint64(amount)) { + return oldValue, oldValue, errBalanceOverflow } + pb.value.add(amount, posExp) pb.meta = meta f.ndb.setPB(id, pb) if c != nil { c.balanceTracker.setBalance(pb.value, negBalance) - if !c.priority && pb.value > 0 { - // The capacity should be adjusted based on the requirement, - // but we have no idea about the new capacity, need a second - // call to udpate it. - c.priority = true - f.priorityConnected += c.capacity - c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) }) + if c.active { + f.activeQueue.Update(c.queueIndex) + if !c.priority && pb.value.base > 0 { + // The capacity should be adjusted based on the requirement, + // but we have no idea about the new capacity, need a second + // call to udpate it. + f.priorityActive += c.capacity + f.updateFreeRatio() + c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) }) + } + c.balanceMetaInfo = meta + f.activeBalances.subExp(oldBalance) + f.activeBalances.addExp(pb.value) + } else { + f.inactiveQueue.Remove(c.queueIndex) + f.inactiveQueue.Push(c, -connPriority(c, f.clock.Now())) + f.inactiveBalances.subExp(oldBalance) + f.inactiveBalances.addExp(pb.value) } - // if balance is set to zero then reverting to non-priority status - // is handled by the balanceExhausted callback - c.balanceMetaInfo = meta + if pb.value.base > 0 { + c.priority = true + // if balance is set to zero then reverting to non-priority status + // is handled by the balanceExhausted callback + } + } else { + f.inactiveBalances.subExp(oldBalance) + f.inactiveBalances.addExp(pb.value) } - return oldBalance, pb.value, nil + f.tryActivateClients() + return oldValue, pb.value.value(posExp), nil } // posBalance represents a recently accessed positive balance entry type posBalance struct { - value uint64 + value expiredValue meta string } // EncodeRLP implements rlp.Encoder func (e *posBalance) EncodeRLP(w io.Writer) error { - return rlp.Encode(w, []interface{}{e.value, e.meta}) + return rlp.Encode(w, []interface{}{e.value.base, e.value.exp, e.meta}) } // DecodeRLP implements rlp.Decoder func (e *posBalance) DecodeRLP(s *rlp.Stream) error { var entry struct { - Value uint64 - Meta string + ValueBase, ValueExp uint64 + Meta string } if err := s.Decode(&entry); err != nil { return err } - e.value = entry.Value + e.value = expiredValue{base: entry.ValueBase, exp: entry.ValueExp} e.meta = entry.Meta return nil } // negBalance represents a negative balance entry of a disconnected client -type negBalance struct{ logValue int64 } +type negBalance struct{ logValue uint64 } // EncodeRLP implements rlp.Encoder func (e *negBalance) EncodeRLP(w io.Writer) error { - return rlp.Encode(w, []interface{}{uint64(e.logValue)}) + return rlp.Encode(w, []interface{}{e.logValue}) } // DecodeRLP implements rlp.Decoder @@ -638,7 +942,7 @@ func (e *negBalance) DecodeRLP(s *rlp.Stream) error { if err := s.Decode(&entry); err != nil { return err } - e.logValue = int64(entry.LogValue) + e.logValue = entry.LogValue return nil } @@ -654,9 +958,9 @@ const ( ) var ( - positiveBalancePrefix = []byte("pb:") // dbVersion(uint16 big endian) + positiveBalancePrefix + id -> balance - negativeBalancePrefix = []byte("nb:") // dbVersion(uint16 big endian) + negativeBalancePrefix + ip -> balance - cumulativeRunningTimeKey = []byte("cumulativeTime:") // dbVersion(uint16 big endian) + cumulativeRunningTimeKey -> cumulativeTime + positiveBalancePrefix = []byte("pb:") // dbVersion(uint16 big endian) + positiveBalancePrefix + id -> balance + negativeBalancePrefix = []byte("nb:") // dbVersion(uint16 big endian) + negativeBalancePrefix + ip -> balance + expirationKey = []byte("expiration:") // dbVersion(uint16 big endian) + expirationKey -> posExp, negExp ) type nodeDB struct { @@ -705,17 +1009,18 @@ func (db *nodeDB) key(id []byte, neg bool) []byte { return db.auxbuf[:len(prefix)+len(db.verbuf)+len(id)] } -func (db *nodeDB) getCumulativeTime() int64 { - blob, err := db.db.Get(append(cumulativeRunningTimeKey, db.verbuf[:]...)) - if err != nil || len(blob) == 0 { - return 0 +func (db *nodeDB) getExpiration() (uint64, uint64) { + blob, err := db.db.Get(append(expirationKey, db.verbuf[:]...)) + if err != nil || len(blob) != 16 { + return 0, 0 } - return int64(binary.BigEndian.Uint64(blob)) + return binary.BigEndian.Uint64(blob[:8]), binary.BigEndian.Uint64(blob[8:16]) } -func (db *nodeDB) setCumulativeTime(v int64) { - binary.BigEndian.PutUint64(db.auxbuf[:8], uint64(v)) - db.db.Put(append(cumulativeRunningTimeKey, db.verbuf[:]...), db.auxbuf[:8]) +func (db *nodeDB) setExpiration(pos, neg uint64) { + binary.BigEndian.PutUint64(db.auxbuf[:8], pos) + binary.BigEndian.PutUint64(db.auxbuf[8:16], neg) + db.db.Put(append(expirationKey, db.verbuf[:]...), db.auxbuf[:16]) } func (db *nodeDB) getOrNewPB(id enode.ID) posBalance { @@ -735,7 +1040,7 @@ func (db *nodeDB) getOrNewPB(id enode.ID) posBalance { } func (db *nodeDB) setPB(id enode.ID, b posBalance) { - if b.value == 0 && len(b.meta) == 0 { + if b.value.base == 0 && len(b.meta) == 0 { db.delPB(id) return } diff --git a/les/clientpool_test.go b/les/clientpool_test.go index 6308113fe7..dd0818c3a7 100644 --- a/les/clientpool_test.go +++ b/les/clientpool_test.go @@ -56,29 +56,34 @@ func TestClientPoolL100C300P20(t *testing.T) { const testClientPoolTicks = 100000 -type poolTestPeer int - -func (i poolTestPeer) ID() enode.ID { - return enode.ID{byte(i % 256), byte(i >> 8)} +type poolTestPeer struct { + index int + disconnCh chan int + cap uint64 } -func (i poolTestPeer) freeClientId() string { +func newPoolTestPeer(i int, disconnCh chan int) *poolTestPeer { + return &poolTestPeer{index: i, disconnCh: disconnCh} +} + +func (i *poolTestPeer) ID() enode.ID { + return enode.ID{byte(i.index % 256), byte(i.index >> 8)} +} + +func (i *poolTestPeer) freeClientId() string { return fmt.Sprintf("addr #%d", i) } -func (i poolTestPeer) updateCapacity(uint64) {} - -type poolTestPeerWithCap struct { - poolTestPeer - - cap uint64 +func (i *poolTestPeer) updateCapacity(cap uint64) { + i.cap = cap + if cap == 0 && i.disconnCh != nil { + i.disconnCh <- i.index + } } -func (i *poolTestPeerWithCap) updateCapacity(cap uint64) { i.cap = cap } +func (i *poolTestPeer) freezeClient() {} -func (i poolTestPeer) freezeClient() {} - -func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomDisconnect bool) { +func testClientPool(t *testing.T, activeLimit, clientCount, paidCount int, randomDisconnect bool) { rand.Seed(time.Now().UnixNano()) var ( clock mclock.Simulated @@ -89,15 +94,16 @@ 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)) + pool.setLimits(activeLimit, uint64(activeLimit)) pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) // pool should accept new peers up to its connected limit - for i := 0; i < connLimit; i++ { - if pool.connect(poolTestPeer(i), 0) { + for i := 0; i < activeLimit; i++ { + if cap, _ := pool.connect(newPoolTestPeer(i, disconnCh), 0); cap != 0 { connected[i] = true } else { t.Fatalf("Test peer #%d rejected", i) @@ -111,28 +117,30 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD // give a positive balance to some of the peers amount := testClientPoolTicks / 2 * int64(time.Second) // enough for half of the simulation period for i := 0; i < paidCount; i++ { - pool.addBalance(poolTestPeer(i).ID(), amount, "") + pool.addBalance(newPoolTestPeer(i, disconnCh).ID(), amount, "") } } i := rand.Intn(clientCount) if connected[i] { if randomDisconnect { - pool.disconnect(poolTestPeer(i)) + pool.disconnect(newPoolTestPeer(i, disconnCh)) connected[i] = false connTicks[i] += tickCounter } } else { - if pool.connect(poolTestPeer(i), 0) { + if cap, _ := pool.connect(newPoolTestPeer(i, disconnCh), 0); cap != 0 { connected[i] = true connTicks[i] -= tickCounter + } else { + pool.disconnect(newPoolTestPeer(i, disconnCh)) } } pollDisconnects: for { select { case i := <-disconnCh: - pool.disconnect(poolTestPeer(i)) + pool.disconnect(newPoolTestPeer(i, disconnCh)) if connected[i] { connTicks[i] += tickCounter connected[i] = false @@ -143,10 +151,10 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD } } - expTicks := testClientPoolTicks/2*connLimit/clientCount + testClientPoolTicks/2*(connLimit-paidCount)/(clientCount-paidCount) + expTicks := testClientPoolTicks/2*activeLimit/clientCount + testClientPoolTicks/2*(activeLimit-paidCount)/(clientCount-paidCount) expMin := expTicks - expTicks/5 expMax := expTicks + expTicks/5 - paidTicks := testClientPoolTicks/2*connLimit/clientCount + testClientPoolTicks/2 + paidTicks := testClientPoolTicks/2*activeLimit/clientCount + testClientPoolTicks/2 paidMin := paidTicks - paidTicks/5 paidMax := paidTicks + paidTicks/5 @@ -172,15 +180,15 @@ 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}) // Add balance for an external client and mark it as paid client - pool.addBalance(poolTestPeer(0).ID(), 1000, "") + pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000, "") - if !pool.connect(poolTestPeer(0), 10) { + if cap, _ := pool.connect(newPoolTestPeer(0, nil), 10); cap == 0 { t.Fatalf("Failed to connect paid client") } } @@ -190,16 +198,16 @@ 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}) // Add balance for an external client and mark it as paid client - pool.addBalance(poolTestPeer(0).ID(), 1000, "") + pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000, "") // Connect a fat paid client to pool, should reject it. - if pool.connect(poolTestPeer(0), 100) { + if cap, _ := pool.connect(newPoolTestPeer(0, nil), 100); cap != 0 { t.Fatalf("Connected fat paid client, should reject it") } } @@ -210,23 +218,23 @@ 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}) for i := 0; i < 10; i++ { - pool.addBalance(poolTestPeer(i).ID(), 1000000000, "") - pool.connect(poolTestPeer(i), 1) + pool.addBalance(newPoolTestPeer(i, nil).ID(), 1000000000, "") + pool.connect(newPoolTestPeer(i, nil), 1) } - pool.addBalance(poolTestPeer(11).ID(), 1000, "") // Add low balance to new paid client - if pool.connect(poolTestPeer(11), 1) { + pool.addBalance(newPoolTestPeer(11, nil).ID(), 1000, "") // Add low balance to new paid client + if cap, _ := pool.connect(newPoolTestPeer(11, nil), 1); cap != 0 { t.Fatalf("Low balance paid client should be rejected") } clock.Run(time.Second) - pool.addBalance(poolTestPeer(12).ID(), 1000000000*60*3, "") // Add high balance to new paid client - if !pool.connect(poolTestPeer(12), 1) { - t.Fatalf("High balance paid client should be accpected") + pool.addBalance(newPoolTestPeer(12, nil).ID(), 1000000000*60*3+1, "") // Add high balance to new paid client + if cap, _ := pool.connect(newPoolTestPeer(12, nil), 1); cap == 0 { + t.Fatalf("High balance paid client should be accepted") } } @@ -237,19 +245,19 @@ 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}) for i := 0; i < 10; i++ { - pool.addBalance(poolTestPeer(i).ID(), 1000000000, "") // 1 second allowance - pool.connect(poolTestPeer(i), 1) + pool.addBalance(newPoolTestPeer(i, kickedCh).ID(), 1000000000, "") // 1 second allowance + pool.connect(newPoolTestPeer(i, kickedCh), 1) clock.Run(time.Millisecond) } clock.Run(time.Second) - clock.Run(connectedBias) - if !pool.connect(poolTestPeer(11), 0) { + clock.Run(activeBias) + if cap, _ := pool.connect(newPoolTestPeer(11, kickedCh), 0); cap == 0 { t.Fatalf("Free client should be accectped") } select { @@ -267,11 +275,11 @@ 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}) - if !pool.connect(poolTestPeer(0), 10) { + if cap, _ := pool.connect(newPoolTestPeer(0, nil), 10); cap == 0 { t.Fatalf("Failed to connect free client") } } @@ -282,24 +290,24 @@ 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}) for i := 0; i < 10; i++ { - pool.connect(poolTestPeer(i), 1) + pool.connect(newPoolTestPeer(i, nil), 1) } - if pool.connect(poolTestPeer(11), 1) { + if cap, _ := pool.connect(newPoolTestPeer(11, nil), 1); cap != 0 { t.Fatalf("New free client should be rejected") } clock.Run(time.Minute) - if pool.connect(poolTestPeer(12), 1) { + if cap, _ := pool.connect(newPoolTestPeer(12, nil), 1); cap != 0 { t.Fatalf("New free client should be rejected") } clock.Run(time.Millisecond) clock.Run(4 * time.Minute) - if !pool.connect(poolTestPeer(13), 1) { + if cap, _ := pool.connect(newPoolTestPeer(13, nil), 1); cap == 0 { t.Fatalf("Old client connects more than 5min should be kicked") } } @@ -311,21 +319,22 @@ 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}) for i := 0; i < 10; i++ { - pool.connect(poolTestPeer(i), 1) + pool.connect(newPoolTestPeer(i, kicked), 1) clock.Run(time.Millisecond) } - if pool.connect(poolTestPeer(10), 1) { + if cap, _ := pool.connect(newPoolTestPeer(10, kicked), 1); cap != 0 { t.Fatalf("New free client should be rejected") } + pool.disconnect(newPoolTestPeer(10, kicked)) clock.Run(5 * time.Minute) for i := 0; i < 10; i++ { - pool.connect(poolTestPeer(i+10), 1) + pool.connect(newPoolTestPeer(i+10, kicked), 1) } for i := 0; i < 10; i++ { select { @@ -346,18 +355,18 @@ 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}) - pool.addBalance(poolTestPeer(0).ID(), int64(time.Minute*3), "") - pool.connect(poolTestPeer(0), 10) + pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute*3), "") + pool.connect(newPoolTestPeer(0, kicked), 10) clock.Run(time.Minute) - pool.disconnect(poolTestPeer(0)) - pb := pool.ndb.getOrNewPB(poolTestPeer(0).ID()) - if pb.value != uint64(time.Minute*2) { + pool.disconnect(newPoolTestPeer(0, kicked)) + pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID()) + if pb.value != expval(uint64(time.Minute*2)) { t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute*2), pb.value) } } @@ -369,16 +378,14 @@ 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}) - p := &poolTestPeerWithCap{ - poolTestPeer: poolTestPeer(0), - } + p := newPoolTestPeer(0, kicked) pool.addBalance(p.ID(), int64(time.Minute), "") - pool.connect(p, 10) + p.cap, _ = pool.connect(p, 10) if p.cap != 10 { t.Fatalf("The capcacity of priority peer hasn't been updated, got: %d", p.cap) } @@ -388,53 +395,51 @@ func TestDowngradePriorityClient(t *testing.T) { if p.cap != 1 { t.Fatalf("The capcacity of peer should be downgraded, got: %d", p.cap) } - pb := pool.ndb.getOrNewPB(poolTestPeer(0).ID()) - if pb.value != 0 { + pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID()) + if pb.value.base != 0 { t.Fatalf("Positive balance mismatch, want %v, got %v", 0, pb.value) } - pool.addBalance(poolTestPeer(0).ID(), int64(time.Minute), "") - pb = pool.ndb.getOrNewPB(poolTestPeer(0).ID()) - if pb.value != uint64(time.Minute) { + pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute), "") + pb = pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID()) + if pb.value != expval(uint64(time.Minute)) { t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute), pb.value) } } func TestNegativeBalanceCalculation(t *testing.T) { var ( - clock mclock.Simulated - db = rawdb.NewMemoryDatabase() - kicked = make(chan int, 10) + clock mclock.Simulated + db = rawdb.NewMemoryDatabase() ) - removeFn := func(id enode.ID) { kicked <- int(id[0]) } // Noop - pool := newClientPool(db, 1, &clock, removeFn) + 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}) for i := 0; i < 10; i++ { - pool.connect(poolTestPeer(i), 1) + pool.connect(newPoolTestPeer(i, nil), 1) } - clock.Run(time.Second) + clock.Run(time.Millisecond * 999) for i := 0; i < 10; i++ { - pool.disconnect(poolTestPeer(i)) - nb := pool.ndb.getOrNewNB(poolTestPeer(i).freeClientId()) + pool.disconnect(newPoolTestPeer(i, nil)) + nb := pool.ndb.getOrNewNB(newPoolTestPeer(i, nil).freeClientId()) if nb.logValue != 0 { t.Fatalf("Short connection shouldn't be recorded") } } for i := 0; i < 10; i++ { - pool.connect(poolTestPeer(i), 1) + pool.connect(newPoolTestPeer(i, nil), 1) } clock.Run(time.Minute) for i := 0; i < 10; i++ { - pool.disconnect(poolTestPeer(i)) - nb := pool.ndb.getOrNewNB(poolTestPeer(i).freeClientId()) - nb.logValue -= pool.logOffset(clock.Now()) - nb.logValue /= fixedPointMultiplier - if nb.logValue != int64(math.Log(float64(time.Minute/time.Second))) { + pool.disconnect(newPoolTestPeer(i, nil)) + nb := pool.ndb.getOrNewNB(newPoolTestPeer(i, nil).freeClientId()) + nb.logValue -= pool.negExpiration(clock.Now()) + nb.logValue = uint64(float64(nb.logValue) / logMultiplier) + if nb.logValue != uint64(math.Log(float64(time.Minute/time.Second))) { t.Fatalf("Negative balance mismatch, want %v, got %v", int64(math.Log(float64(time.Minute/time.Second))), nb.logValue) } } @@ -453,8 +458,8 @@ func TestNodeDB(t *testing.T) { balance interface{} positive bool }{ - {enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: 100}, true}, - {enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: 200}, true}, + {enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: expval(100)}, true}, + {enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: expval(200)}, true}, {enode.ID{}, "127.0.0.1", negBalance{logValue: 10}, false}, {enode.ID{}, "127.0.0.1", negBalance{logValue: 20}, false}, } @@ -484,9 +489,9 @@ func TestNodeDB(t *testing.T) { } } } - ndb.setCumulativeTime(100) - if ndb.getCumulativeTime() != 100 { - t.Fatalf("Cumulative time mismatch, want %v, got %v", 100, ndb.getCumulativeTime()) + ndb.setExpiration(100, 200) + if pos, neg := ndb.getExpiration(); pos != 100 || neg != 200 { + t.Fatalf("Expiration mismatch, want %v / %v, got %v / %v", 100, 200, pos, neg) } } @@ -541,3 +546,83 @@ func TestNodeDBExpiration(t *testing.T) { t.Fatalf("Failed to evict useless negative balances, want %v, got %d", 4, iterated) } } + +func TestInactiveClient(t *testing.T) { + var ( + clock mclock.Simulated + db = rawdb.NewMemoryDatabase() + ) + pool := newClientPool(db, 1, 1, &clock, nil) + defer pool.stop() + pool.setLimits(2, uint64(2)) // Total capacity limit is 10 + + p1 := newPoolTestPeer(1, nil) + p2 := newPoolTestPeer(2, nil) + p3 := newPoolTestPeer(3, nil) + pool.addBalance(p1.ID(), 1000, "") + pool.addBalance(p3.ID(), 2000, "") + // p1: 1000 p2: 0 p3: 2000 + p1.cap, _ = pool.connect(p1, 1) + if p1.cap != 1 { + t.Fatalf("Failed to connect peer #1") + } + p2.cap, _ = pool.connect(p2, 1) + if p2.cap != 1 { + t.Fatalf("Failed to connect peer #2") + } + p3.cap, _ = pool.connect(p3, 1) + if p3.cap != 1 { + t.Fatalf("Failed to connect peer #3") + } + if p2.cap != 0 { + t.Fatalf("Failed to deactivate peer #2") + } + pool.addBalance(p2.ID(), 3000, "") + // p1: 1000 p2: 3000 p3: 2000 + if p2.cap != 1 { + t.Fatalf("Failed to activate peer #2") + } + if p1.cap != 0 { + t.Fatalf("Failed to deactivate peer #1") + } + pool.addBalance(p2.ID(), -2500, "") + // p1: 1000 p2: 500 p3: 2000 + if p1.cap != 1 { + t.Fatalf("Failed to activate peer #1") + } + if p2.cap != 0 { + t.Fatalf("Failed to deactivate peer #2") + } + pool.setDefaultFactors(priceFactors{1e-9, 0, 0}, priceFactors{1e-9, 0, 0}) + p4 := newPoolTestPeer(4, nil) + pool.addBalance(p4.ID(), 1500, "") + // p1: 1000 p2: 500 p3: 2000 p4: 1500 + p4.cap, _ = pool.connect(p4, 1) + if p4.cap != 1 { + t.Fatalf("Failed to activate peer #4") + } + if p1.cap != 0 { + t.Fatalf("Failed to deactivate peer #1") + } + clock.Run(time.Second * 600) + // manually trigger a check to avoid a long real-time wait + pool.lock.Lock() + pool.tryActivateClients() + pool.lock.Unlock() + // p1: 1000 p2: 500 p3: 2000 p4: 900 + if p1.cap != 1 { + t.Fatalf("Failed to activate peer #1") + } + if p4.cap != 0 { + t.Fatalf("Failed to deactivate peer #4") + } + pool.disconnect(p2) + pool.disconnect(p4) + pool.addBalance(p1.ID(), -1000, "") + if p1.cap != 1 { + t.Fatalf("Should not deactivate peer #1") + } + if p2.cap != 0 { + t.Fatalf("Should not activate peer #2") + } +} diff --git a/les/costtracker.go b/les/costtracker.go index 81da045660..abf8618ffd 100644 --- a/les/costtracker.go +++ b/les/costtracker.go @@ -159,15 +159,9 @@ func newCostTracker(db ethdb.Database, config *eth.Config) (*costTracker, uint64 } ct.gfLoop() costList := ct.makeCostList(ct.globalFactor() * 1.25) - for _, c := range costList { - amount := minBufferReqAmount[c.MsgCode] - cost := c.BaseCost + amount*c.ReqCost - if cost > ct.minBufLimit { - ct.minBufLimit = cost - } - } - ct.minBufLimit *= uint64(minBufferMultiplier) - return ct, (ct.minBufLimit-1)/bufLimitRatio + 1 + var minRecharge uint64 + ct.minBufLimit, minRecharge = costList.decode(ProtocolLengths[ServerProtocolVersions[len(ServerProtocolVersions)-1]]).reqParams() + return ct, minRecharge } // stop stops the cost tracker and saves the cost factor statistics to the database @@ -480,6 +474,22 @@ func (table requestCostTable) getMaxCost(code, amount uint64) uint64 { return costs.baseCost + amount*costs.reqCost } +func (table requestCostTable) reqParams() (minRecharge, minBufLimit uint64) { + for code, c := range table { + amount := minBufferReqAmount[code] + cost := c.baseCost + amount*c.reqCost + if cost > minBufLimit { + minBufLimit = cost + } + } + minBufLimit *= uint64(minBufferMultiplier) + if minBufLimit < 1 { + minBufLimit = 1 + } + minRecharge = (minBufLimit-1)/bufLimitRatio + 1 + return +} + // decode converts a cost list to a cost table func (list RequestCostList) decode(protocolLength uint64) requestCostTable { table := make(requestCostTable) diff --git a/les/flowcontrol/control.go b/les/flowcontrol/control.go index 490013677c..1c40882902 100644 --- a/les/flowcontrol/control.go +++ b/les/flowcontrol/control.go @@ -185,6 +185,14 @@ func (node *ClientNode) UpdateParams(params ServerParams) { } } +// Params returns the current server parameters +func (node *ClientNode) Params() ServerParams { + node.lock.Lock() + defer node.lock.Unlock() + + return node.params +} + // updateParams updates the flow control parameters of the node func (node *ClientNode) updateParams(params ServerParams, now mclock.AbsTime) { diff := int64(params.BufLimit - node.params.BufLimit)