diff --git a/les/api.go b/les/api.go index 1e689e3582..f60a4fa57d 100644 --- a/les/api.go +++ b/les/api.go @@ -104,12 +104,16 @@ func (api *PrivateLightServerAPI) clientInfo(c *clientInfo, id enode.ID) map[str info["capacity"] = c.capacity pb, nb := c.balanceTracker.getBalance(now) info["pricing/balance"], info["pricing/negBalance"] = pb, nb - info["pricing/balanceMeta"] = c.balanceMetaInfo + + cb := api.server.clientPool.ndb.getCurrencyBalance(id) + info["pricing/currency"] = cb.amount info["priority"] = pb.base != 0 } else { info["isConnected"] = false - pb := api.server.clientPool.ndb.getOrNewPB(id) - info["pricing/balance"], info["pricing/balanceMeta"] = pb.value, pb.meta + pb := api.server.clientPool.ndb.getOrNewBalance(id.Bytes(), false) + + cb := api.server.clientPool.ndb.getCurrencyBalance(id) + info["pricing/balance"], info["pricing/currency"] = pb.value, cb.amount info["priority"] = pb.value.base != 0 } return info @@ -150,7 +154,7 @@ func (api *PrivateLightServerAPI) setParams(params map[string]interface{}, clien setFactor(&negFactors.requestFactor) case !defParams && name == "capacity": if capacity, ok := value.(float64); ok && uint64(capacity) >= api.server.minCapacity { - _, _, err = api.server.clientPool.setCapacity(client.id, client.freeID, uint64(capacity), 0, true) + _, _, err = api.server.clientPool.setCapacity(client.id, client.address, uint64(capacity), 0, true) // Don't have to call factor update explicitly. It's already done // in setCapacity function. } else { @@ -172,8 +176,8 @@ func (api *PrivateLightServerAPI) setParams(params map[string]interface{}, clien // AddBalance updates the balance of a client (either overwrites it or adds to it). // It also updates the balance meta info string. -func (api *PrivateLightServerAPI) AddBalance(id enode.ID, value int64, meta string) ([2]uint64, error) { - oldBalance, newBalance, err := api.server.clientPool.addBalance(id, value, meta) +func (api *PrivateLightServerAPI) AddBalance(id enode.ID, value int64) ([2]uint64, error) { + oldBalance, newBalance, err := api.server.clientPool.addBalance(id, value) return [2]uint64{oldBalance, newBalance}, err } @@ -184,7 +188,7 @@ func (api *PrivateLightServerAPI) SetClientParams(ids []enode.ID, params map[str if client != nil { update, err := api.setParams(params, client, nil, nil) if update { - updatePriceFactors(&client.balanceTracker, client.posFactors, client.negFactors, client.capacity) + updatePriceFactors(&client.balanceTracker, client.posFactors, client.negFactors) } return err } else { diff --git a/les/balance.go b/les/balance.go index 6f11daead2..97fc0d76ee 100644 --- a/les/balance.go +++ b/les/balance.go @@ -35,74 +35,25 @@ const ( // expirationController controls the exponential expiration of positive and negative // balances type expirationController interface { - posExpiration(mclock.AbsTime) uint64 - negExpiration(mclock.AbsTime) uint64 + posExpiration(mclock.AbsTime) fixed64 + negExpiration(mclock.AbsTime) fixed64 } -// 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 +// priceFactors determine the pricing policy (may apply either to positive or +// negative balances which may have different factors). +// - timeFactor is cost unit per nanosecond of connection time +// - capacityFactor is cost unit per nanosecond of connection time per 1000000 capacity +// - requestFactor is cost unit per request "realCost" unit +type priceFactors struct { + timeFactor, capacityFactor, requestFactor float64 } -// 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)) +func (p priceFactors) timePrice(cap uint64) float64 { + return p.timeFactor + float64(cap)*p.capacityFactor/1000000 } -// 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 - } +func (p priceFactors) reqPrice() float64 { + return p.requestFactor } // balanceTracker keeps track of the positive and negative balances of a connected @@ -115,8 +66,7 @@ type balanceTracker struct { stopped bool capacity uint64 balance balance - timeFactor, requestFactor float64 - negTimeFactor, negRequestFactor float64 + posFactor, negFactor priceFactors sumReqCost uint64 lastUpdate, nextUpdate, initTime mclock.AbsTime updateEvent mclock.Timer @@ -159,10 +109,8 @@ func (bt *balanceTracker) stop(now mclock.AbsTime) { bt.stopped = true bt.addBalance(now) - bt.negTimeFactor = 0 - bt.negRequestFactor = 0 - bt.timeFactor = 0 - bt.requestFactor = 0 + bt.posFactor = priceFactors{0, 0, 0} + bt.negFactor = priceFactors{0, 0, 0} if bt.updateEvent != nil { bt.updateEvent.Stop() bt.updateEvent = nil @@ -185,13 +133,14 @@ func (bt *balanceTracker) balanceToPriority(b balance) int64 { 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) + timePrice := bt.negFactor.timePrice(targetCapacity) + timeCost := uint64(float64(after) * timePrice) negBalance := bt.balance.neg.value(bt.exp.negExpiration(now)) - if negPrice+negBalance < uint64(targetPriority) { + if timeCost+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 { + if uint64(targetPriority) > negBalance && timePrice > 1e-100 { + if negTime := time.Duration(float64(uint64(targetPriority)-negBalance) / timePrice); negTime < after { after -= negTime } else { after = 0 @@ -199,7 +148,8 @@ func (bt *balanceTracker) posBalanceMissing(targetPriority int64, targetCapacity } targetPriority = 0 } - posRequired := uint64(float64(-targetPriority)*float64(targetCapacity)+float64(after)*bt.timeFactor) + 1 + timePrice := bt.posFactor.timePrice(targetCapacity) + posRequired := uint64(float64(-targetPriority)*float64(targetCapacity)+float64(after)*timePrice) + 1 if posRequired >= maxBalance { return math.MaxUint64 // target not reachable } @@ -216,7 +166,7 @@ func (bt *balanceTracker) reducedBalance(at mclock.AbsTime, avgReqCost float64) dt := float64(at - bt.lastUpdate) b := bt.balance if b.pos.base != 0 { - factor := bt.timeFactor + bt.requestFactor*avgReqCost + factor := bt.posFactor.timePrice(bt.capacity) + bt.posFactor.reqPrice()*avgReqCost diff := -int64(dt * factor) dd := b.pos.add(diff, bt.exp.posExpiration(at)) if dd == diff { @@ -226,7 +176,7 @@ func (bt *balanceTracker) reducedBalance(at mclock.AbsTime, avgReqCost float64) } } if dt > 0 { - factor := bt.negTimeFactor + bt.negRequestFactor*avgReqCost + factor := bt.negFactor.timePrice(bt.capacity) + bt.negFactor.reqPrice()*avgReqCost b.neg.add(int64(dt*factor), bt.exp.negExpiration(at)) } return b @@ -242,7 +192,8 @@ func (bt *balanceTracker) timeUntil(priority int64) (time.Duration, bool) { var dt float64 if bt.balance.pos.base != 0 { posBalance := bt.balance.pos.value(bt.exp.posExpiration(now)) - if bt.timeFactor < 1e-100 { + timePrice := bt.posFactor.timePrice(bt.capacity) + if timePrice < 1e-100 { return 0, false } if priority < 0 { @@ -250,10 +201,10 @@ func (bt *balanceTracker) timeUntil(priority int64) (time.Duration, bool) { if newBalance > posBalance { return 0, false } - dt = float64(posBalance-newBalance) / bt.timeFactor + dt = float64(posBalance-newBalance) / timePrice return time.Duration(dt), true } else { - dt = float64(posBalance) / bt.timeFactor + dt = float64(posBalance) / timePrice } } else { if priority < 0 { @@ -262,11 +213,12 @@ 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 negBalance := bt.balance.neg.value(bt.exp.negExpiration(now)) + timePrice := bt.negFactor.timePrice(bt.capacity) if uint64(priority) > negBalance { - if bt.negTimeFactor < 1e-100 { + if timePrice < 1e-100 { return 0, false } - dt += float64(uint64(priority)-negBalance) / bt.negTimeFactor + dt += float64(uint64(priority)-negBalance) / timePrice } return time.Duration(dt), true } @@ -387,8 +339,8 @@ func (bt *balanceTracker) requestCost(cost uint64) uint64 { posExp := bt.exp.posExpiration(now) if bt.balance.pos.base != 0 { - if bt.requestFactor != 0 { - c := -int64(fcost * bt.requestFactor) + if bt.posFactor.reqPrice() != 0 { + c := -int64(fcost * bt.posFactor.reqPrice()) cc := bt.balance.pos.add(c, posExp) if c == cc { fcost = 0 @@ -401,8 +353,8 @@ func (bt *balanceTracker) requestCost(cost uint64) uint64 { } } if fcost > 0 { - if bt.negRequestFactor != 0 { - bt.balance.neg.add(int64(fcost*bt.negRequestFactor), bt.exp.negExpiration(now)) + if bt.negFactor.reqPrice() != 0 { + bt.balance.neg.add(int64(fcost*bt.negFactor.reqPrice()), bt.exp.negExpiration(now)) bt.checkCallbacks(now) } } @@ -434,7 +386,7 @@ func (bt *balanceTracker) setBalance(pos, neg expiredValue) error { // setFactors sets the price factors. timeFactor is the price of a nanosecond of // connection while requestFactor is the price of a "realCost" unit. -func (bt *balanceTracker) setFactors(neg bool, timeFactor, requestFactor float64) { +func (bt *balanceTracker) setFactors(posFactor, negFactor priceFactors) { bt.lock.Lock() defer bt.lock.Unlock() @@ -443,13 +395,7 @@ func (bt *balanceTracker) setFactors(neg bool, timeFactor, requestFactor float64 } now := bt.clock.Now() bt.addBalance(now) - if neg { - bt.negTimeFactor = timeFactor - bt.negRequestFactor = requestFactor - } else { - bt.timeFactor = timeFactor - bt.requestFactor = requestFactor - } + bt.posFactor, bt.negFactor = posFactor, negFactor bt.checkCallbacks(now) } diff --git a/les/balance_test.go b/les/balance_test.go index 91032fc714..69d8caf95b 100644 --- a/les/balance_test.go +++ b/les/balance_test.go @@ -25,11 +25,11 @@ import ( type zeroExpCtrl struct{} -func (z zeroExpCtrl) posExpiration(mclock.AbsTime) uint64 { +func (z zeroExpCtrl) posExpiration(mclock.AbsTime) fixed64 { return 0 } -func (z zeroExpCtrl) negExpiration(mclock.AbsTime) uint64 { +func (z zeroExpCtrl) negExpiration(mclock.AbsTime) fixed64 { return 0 } @@ -70,8 +70,7 @@ func TestBalanceTimeCost(t *testing.T) { ) tracker.init(clock, 1000) defer tracker.stop(clock.Now()) - tracker.setFactors(false, 1, 1) - tracker.setFactors(true, 1, 1) + tracker.setFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) tracker.setBalance(expval(uint64(time.Minute)), expval(0)) // 1 minute time allowance @@ -114,8 +113,7 @@ func TestBalanceReqCost(t *testing.T) { ) tracker.init(clock, 1000) defer tracker.stop(clock.Now()) - tracker.setFactors(false, 1, 1) - tracker.setFactors(true, 1, 1) + tracker.setFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) tracker.setBalance(expval(uint64(time.Minute)), expval(0)) // 1 minute time serving time allowance var inputs = []struct { @@ -146,8 +144,7 @@ func TestBalanceToPriority(t *testing.T) { ) tracker.init(clock, 1000) // cap = 1000 defer tracker.stop(clock.Now()) - tracker.setFactors(false, 1, 1) - tracker.setFactors(true, 1, 1) + tracker.setFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) var inputs = []struct { pos uint64 @@ -175,8 +172,7 @@ func TestEstimatedPriority(t *testing.T) { ) tracker.init(clock, 1000000000) // cap = 1000,000,000 defer tracker.stop(clock.Now()) - tracker.setFactors(false, 1, 1) - tracker.setFactors(true, 1, 1) + tracker.setFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) tracker.setBalance(expval(uint64(time.Minute)), expval(0)) var inputs = []struct { @@ -219,8 +215,7 @@ func TestCallbackChecking(t *testing.T) { ) tracker.init(clock, 1000000) // cap = 1000,000 defer tracker.stop(clock.Now()) - tracker.setFactors(false, 1, 1) - tracker.setFactors(true, 1, 1) + tracker.setFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) var inputs = []struct { priority int64 @@ -246,8 +241,7 @@ func TestCallback(t *testing.T) { ) tracker.init(clock, 1000) // cap = 1000 defer tracker.stop(clock.Now()) - tracker.setFactors(false, 1, 1) - tracker.setFactors(true, 1, 1) + tracker.setFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) callCh := make(chan struct{}, 1) tracker.setBalance(expval(uint64(time.Minute)), expval(0)) diff --git a/les/clientdb.go b/les/clientdb.go new file mode 100644 index 0000000000..b1beeac915 --- /dev/null +++ b/les/clientdb.go @@ -0,0 +1,278 @@ +// Copyright 2020 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package les + +import ( + "bytes" + "encoding/binary" + "io" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/ethereum/go-ethereum/rlp" + lru "github.com/hashicorp/golang-lru" +) + +const balanceCacheLimit = 8192 // the maximum number of cached items in service token balance queue + +// tokenBalance is a wrapper of expiredValue which represents the service token +// balance of clients. The balance value will decay exponentially over time and +// can be deleted when the amount is small enough. +type tokenBalance struct { + value expiredValue +} + +// EncodeRLP implements rlp.Encoder +func (b *tokenBalance) EncodeRLP(w io.Writer) error { + return rlp.Encode(w, []interface{}{b.value.base, b.value.exp}) +} + +// DecodeRLP implements rlp.Decoder +func (b *tokenBalance) DecodeRLP(s *rlp.Stream) error { + var entry struct { + Base, Exp uint64 + } + if err := s.Decode(&entry); err != nil { + return err + } + b.value = expiredValue{base: entry.Base, exp: entry.Exp} + return nil +} + +// currencyBalance represents the client's currency balance. +type currencyBalance struct { + amount uint64 + typ string +} + +// EncodeRLP implements rlp.Encoder +func (b *currencyBalance) EncodeRLP(w io.Writer) error { + return rlp.Encode(w, []interface{}{b.amount, b.typ}) +} + +// DecodeRLP implements rlp.Decoder +func (b *currencyBalance) DecodeRLP(s *rlp.Stream) error { + var entry struct { + Amount uint64 + Type string + } + if err := s.Decode(&entry); err != nil { + return err + } + b.amount, b.typ = entry.Amount, entry.Type + return nil +} + +const ( + // nodeDBVersion is the version identifier of the node data in db + // + // Changelog: + // * Replace `lastTotal` with `meta` in positive balance: version 0=>1 + // * Rework balance, add currency balance: version 1=>2 + nodeDBVersion = 2 + + // dbCleanupCycle is the cycle of db for useless data cleanup + dbCleanupCycle = time.Hour +) + +var ( + posBalancePrefix = []byte("pb:") // dbVersion(uint16 big endian) + posBalancePrefix + id -> positive balance + negBalancePrefix = []byte("nb:") // dbVersion(uint16 big endian) + negBalancePrefix + ip -> negative balance + curBalancePrefix = []byte("cb:") // dbVersion(uint16 big endian) + curBalancePrefix + id -> currency balance + expirationKey = []byte("expiration:") // dbVersion(uint16 big endian) + expirationKey -> posExp, negExp +) + +type nodeDB struct { + db ethdb.Database + cache *lru.Cache + clock mclock.Clock + closeCh chan struct{} + evictCallBack func(mclock.AbsTime, bool, tokenBalance) bool // Callback to determine whether the balance can be evicted. + cleanupHook func() // Test hook used for testing +} + +func newNodeDB(db ethdb.Database, clock mclock.Clock) *nodeDB { + var buff [2]byte + binary.BigEndian.PutUint16(buff[:], uint16(nodeDBVersion)) + + cache, _ := lru.New(balanceCacheLimit) + ndb := &nodeDB{ + db: rawdb.NewTable(db, string(buff[:])), + cache: cache, + clock: clock, + closeCh: make(chan struct{}), + } + go ndb.expirer() + return ndb +} + +func (db *nodeDB) close() { + close(db.closeCh) +} + +func (db *nodeDB) key(id []byte, neg bool) []byte { + prefix := posBalancePrefix + if neg { + prefix = negBalancePrefix + } + return append(prefix, id...) +} + +func (db *nodeDB) getExpiration() (fixed64, fixed64) { + blob, err := db.db.Get(expirationKey) + if err != nil || len(blob) != 16 { + return 0, 0 + } + return fixed64(binary.BigEndian.Uint64(blob[:8])), fixed64(binary.BigEndian.Uint64(blob[8:16])) +} + +func (db *nodeDB) setExpiration(pos, neg fixed64) { + var buff [16]byte + binary.BigEndian.PutUint64(buff[:8], uint64(pos)) + binary.BigEndian.PutUint64(buff[8:16], uint64(neg)) + db.db.Put(expirationKey, buff[:16]) +} + +func (db *nodeDB) getCurrencyBalance(id enode.ID) currencyBalance { + var b currencyBalance + enc, err := db.db.Get(append(curBalancePrefix, id.Bytes()...)) + if err != nil || len(enc) == 0 { + return b + } + if err := rlp.DecodeBytes(enc, &b); err != nil { + log.Crit("Failed to decode positive balance", "err", err) + } + return b +} + +func (db *nodeDB) setCurrencyBalance(id enode.ID, b currencyBalance) { + enc, err := rlp.EncodeToBytes(&(b)) + if err != nil { + log.Crit("Failed to encode currency balance", "err", err) + } + db.db.Put(append(curBalancePrefix, id.Bytes()...), enc) +} + +func (db *nodeDB) getOrNewBalance(id []byte, neg bool) tokenBalance { + key := db.key(id, neg) + item, exist := db.cache.Get(string(key)) + if exist { + return item.(tokenBalance) + } + var b tokenBalance + enc, err := db.db.Get(key) + if err != nil || len(enc) == 0 { + return b + } + if err := rlp.DecodeBytes(enc, &b); err != nil { + log.Crit("Failed to decode positive balance", "err", err) + } + db.cache.Add(string(key), b) + return b +} + +func (db *nodeDB) setBalance(id []byte, neg bool, b tokenBalance) { + key := db.key(id, neg) + enc, err := rlp.EncodeToBytes(&(b)) + if err != nil { + log.Crit("Failed to encode positive balance", "err", err) + } + db.db.Put(key, enc) + db.cache.Add(string(key), b) +} + +func (db *nodeDB) delBalance(id []byte, neg bool) { + key := db.key(id, neg) + db.db.Delete(key) + db.cache.Remove(string(key)) +} + +// getPosBalanceIDs returns a lexicographically ordered list of IDs of accounts +// with a positive balance +func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID) { + if maxCount <= 0 { + return + } + it := db.db.NewIteratorWithStart(db.key(start.Bytes(), false)) + defer it.Release() + for i := len(stop[:]) - 1; i >= 0; i-- { + stop[i]-- + if stop[i] != 255 { + break + } + } + stopKey := db.key(stop.Bytes(), false) + keyLen := len(stopKey) + + for it.Next() { + var id enode.ID + if len(it.Key()) != keyLen || bytes.Compare(it.Key(), stopKey) == 1 { + return + } + copy(id[:], it.Key()[keyLen-len(id):]) + result = append(result, id) + if len(result) == maxCount { + return + } + } + return +} + +func (db *nodeDB) expirer() { + for { + select { + case <-db.clock.After(dbCleanupCycle): + db.expireNodes() + case <-db.closeCh: + return + } + } +} + +// expireNodes iterates the whole node db and checks whether the +// token balances can deleted. +func (db *nodeDB) expireNodes() { + var ( + visited int + deleted int + start = time.Now() + ) + for index, prefix := range [][]byte{posBalancePrefix, negBalancePrefix} { + iter := db.db.NewIteratorWithPrefix(prefix) + for iter.Next() { + visited += 1 + var balance tokenBalance + if err := rlp.DecodeBytes(iter.Value(), &balance); err != nil { + log.Crit("Failed to decode negative balance", "err", err) + } + if db.evictCallBack != nil && db.evictCallBack(db.clock.Now(), index != 0, balance) { + deleted += 1 + db.db.Delete(iter.Key()) + } + } + } + // Invoke testing hook if it's not nil. + if db.cleanupHook != nil { + db.cleanupHook() + } + log.Debug("Expire nodes", "visited", visited, "deleted", deleted, "elapsed", common.PrettyDuration(time.Since(start))) +} diff --git a/les/clientdb_test.go b/les/clientdb_test.go new file mode 100644 index 0000000000..ac2181fb66 --- /dev/null +++ b/les/clientdb_test.go @@ -0,0 +1,139 @@ +// Copyright 2020 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package les + +import ( + "reflect" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/p2p/enode" +) + +func TestNodeDB(t *testing.T) { + ndb := newNodeDB(rawdb.NewMemoryDatabase(), mclock.System{}) + defer ndb.close() + + var cases = []struct { + id enode.ID + ip string + balance tokenBalance + positive bool + }{ + {enode.ID{0x00, 0x01, 0x02}, "", tokenBalance{value: expval(100)}, true}, + {enode.ID{0x00, 0x01, 0x02}, "", tokenBalance{value: expval(200)}, true}, + {enode.ID{}, "127.0.0.1", tokenBalance{value: expval(100)}, false}, + {enode.ID{}, "127.0.0.1", tokenBalance{value: expval(200)}, false}, + } + for _, c := range cases { + if c.positive { + ndb.setBalance(c.id.Bytes(), false, c.balance) + if pb := ndb.getOrNewBalance(c.id.Bytes(), false); !reflect.DeepEqual(pb, c.balance) { + t.Fatalf("Positive balance mismatch, want %v, got %v", c.balance, pb) + } + } else { + ndb.setBalance([]byte(c.ip), true, c.balance) + if nb := ndb.getOrNewBalance([]byte(c.ip), true); !reflect.DeepEqual(nb, c.balance) { + t.Fatalf("Negative balance mismatch, want %v, got %v", c.balance, nb) + } + } + } + for _, c := range cases { + if c.positive { + ndb.delBalance(c.id.Bytes(), false) + if pb := ndb.getOrNewBalance(c.id.Bytes(), false); !reflect.DeepEqual(pb, tokenBalance{}) { + t.Fatalf("Positive balance mismatch, want %v, got %v", tokenBalance{}, pb) + } + } else { + ndb.delBalance([]byte(c.ip), true) + if nb := ndb.getOrNewBalance([]byte(c.ip), true); !reflect.DeepEqual(nb, tokenBalance{}) { + t.Fatalf("Negative balance mismatch, want %v, got %v", tokenBalance{}, nb) + } + } + } + posExp, negExp := fixed64(1000), fixed64(2000) + ndb.setExpiration(posExp, negExp) + if pos, neg := ndb.getExpiration(); pos != posExp || neg != negExp { + t.Fatalf("Expiration mismatch, want %v / %v, got %v / %v", posExp, negExp, pos, neg) + } + curBalance := currencyBalance{typ: "ETH", amount: 10000} + ndb.setCurrencyBalance(enode.ID{0x01, 0x02}, curBalance) + if got := ndb.getCurrencyBalance(enode.ID{0x01, 0x02}); !reflect.DeepEqual(got, curBalance) { + t.Fatalf("Currency balance mismatch, want %v, got %v", curBalance, got) + } +} + +func TestNodeDBExpiration(t *testing.T) { + var ( + iterated int + done = make(chan struct{}, 1) + ) + callback := func(now mclock.AbsTime, neg bool, b tokenBalance) bool { + iterated += 1 + return true + } + clock := &mclock.Simulated{} + ndb := newNodeDB(rawdb.NewMemoryDatabase(), clock) + defer ndb.close() + ndb.evictCallBack = callback + ndb.cleanupHook = func() { done <- struct{}{} } + + var cases = []struct { + id []byte + neg bool + balance tokenBalance + }{ + {[]byte{0x01, 0x02}, false, tokenBalance{value: expval(1)}}, + {[]byte{0x03, 0x04}, false, tokenBalance{value: expval(1)}}, + {[]byte{0x05, 0x06}, false, tokenBalance{value: expval(1)}}, + {[]byte{0x07, 0x08}, false, tokenBalance{value: expval(1)}}, + + {[]byte("127.0.0.1"), true, tokenBalance{value: expval(1)}}, + {[]byte("127.0.0.2"), true, tokenBalance{value: expval(1)}}, + {[]byte("127.0.0.3"), true, tokenBalance{value: expval(1)}}, + {[]byte("127.0.0.4"), true, tokenBalance{value: expval(1)}}, + } + for _, c := range cases { + ndb.setBalance(c.id, c.neg, c.balance) + } + clock.WaitForTimers(1) + clock.Run(time.Hour + time.Minute) + select { + case <-done: + case <-time.NewTimer(time.Second).C: + t.Fatalf("timeout") + } + if iterated != 8 { + t.Fatalf("Failed to evict useless balances, want %v, got %d", 8, iterated) + } + + for _, c := range cases { + ndb.setBalance(c.id, c.neg, c.balance) + } + clock.WaitForTimers(1) + clock.Run(time.Hour + time.Minute) + select { + case <-done: + case <-time.NewTimer(time.Second).C: + t.Fatalf("timeout") + } + if iterated != 16 { + t.Fatalf("Failed to evict useless balances, want %v, got %d", 16, iterated) + } +} diff --git a/les/clientpool.go b/les/clientpool.go index 80915317ca..ebdddd3a6a 100644 --- a/les/clientpool.go +++ b/les/clientpool.go @@ -17,36 +17,26 @@ package les import ( - "bytes" - "encoding/binary" "fmt" - "io" "math" "sync" "time" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/common/prque" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/enode" - "github.com/ethereum/go-ethereum/rlp" - lru "github.com/hashicorp/golang-lru" ) const ( - 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 + defaultPosExpTC = 36000 // default time constant (in seconds) for exponentially reducing positive balance + defaultNegExpTC = 3600 // default time constant (in seconds) for exponentially reducing negative balance + 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 + freeRatioTC = time.Hour // time constant of token supply control based on free service availability // activeBias is applied to already connected clients So that // already connected client won't be kicked out very soon and we @@ -74,11 +64,8 @@ const ( // each client can have several minutes of connection time. // // Balances of disconnected clients are stored in nodeDB including positive balance -// and negative banalce. Negative balance is transformed into a logarithmic form -// with a constantly shifting linear offset in order to implement an exponential -// decrease. Besides nodeDB will have a background thread to check the negative -// balance of disconnected client. If the balance is low enough, then the record -// will be dropped. +// and negative banalce. Boeth positive balance and negative balance will decrease +// exponentially. If the balance is low enough, then the record will be dropped. type clientPool struct { ndb *nodeDB lock sync.Mutex @@ -108,10 +95,11 @@ type clientPool struct { 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 + expLock sync.RWMutex + posExpTC, negExpTC uint64 + posExp, negExp fixed64 + posExpTCi, negExpTCi float64 // already inverted (logMultiplier/time) + freeRatioLastUpdate mclock.AbsTime } // clientPoolPeer represents a client peer in the pool. @@ -126,21 +114,20 @@ type clientPoolPeer interface { freeze() } -// clientInfo represents a connected client +// clientInfo defines all information required by clientpool. 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 activeQueue - balanceTracker balanceTracker - posFactors, negFactors priceFactors - balanceMetaInfo string + id enode.ID + address string + active bool + capacity uint64 + priority bool + pool *clientPool + peer clientPoolPeer + connectedAt mclock.AbsTime + queueIndex int + balanceTracker balanceTracker + posFactors priceFactors + negFactors priceFactors } // connSetIndex callback updates clientInfo item index in activeQueue @@ -168,15 +155,6 @@ func connMaxPriority(a interface{}, until mclock.AbsTime) int64 { return pri } -// priceFactors determine the pricing policy (may apply either to positive or -// negative balances which may have different factors). -// - timeFactor is cost unit per nanosecond of connection time -// - capacityFactor is cost unit per nanosecond of connection time per 1000000 capacity -// - requestFactor is cost unit per request "realCost" unit -type priceFactors struct { - timeFactor, capacityFactor, requestFactor float64 -} - // newClientPool creates a new client pool func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock.Clock, removePeer func(enode.ID)) *clientPool { ndb := newNodeDB(db, clock) @@ -214,16 +192,23 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock stop = true } for i := 0; i < l; i++ { - pool.inactiveBalances.addExp(pool.ndb.getOrNewPB(ids[i]).value) + pool.inactiveBalances.addExp(pool.ndb.getOrNewBalance(ids[i].Bytes(), false).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 { - return b.logValue <= pool.negExpiration(now) + // The positive and negative balances of clients are stored in database + // and both of these decay exponentially over time. Delete them if the + // value is small enough. + ndb.evictCallBack = func(now mclock.AbsTime, neg bool, b tokenBalance) bool { + var expiration fixed64 + if neg { + expiration = pool.negExpiration(now) + } else { + expiration = pool.posExpiration(now) + } + return b.value.value(expiration) <= uint64(time.Second) } go func() { for { @@ -241,11 +226,9 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock 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 @@ -309,8 +292,9 @@ func (f *clientPool) updateFreeRatio() { } 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.posExp += fixed64(float64(dt) * f.posExpTCi * f.freeRatio) + f.negExp += fixed64(float64(dt) * f.negExpTCi * f.freeRatio) f.expLock.Unlock() } @@ -324,12 +308,12 @@ func (f *clientPool) setExpirationTCs(pos, neg uint64) { f.expLock.Lock() f.posExpTC, f.negExpTC = pos, neg if pos > 0 { - f.posExpTCi = logMultiplier / float64(pos*uint64(time.Second)) + f.posExpTCi = fixedFactor / float64(pos*uint64(time.Second)) } else { f.posExpTCi = 0 } if neg > 0 { - f.negExpTCi = logMultiplier / float64(neg*uint64(time.Second)) + f.negExpTCi = fixedFactor / float64(neg*uint64(time.Second)) } else { f.negExpTCi = 0 } @@ -347,28 +331,34 @@ func (f *clientPool) getExpirationTCs() (pos, neg uint64) { // posExpiration implements expirationController. Expiration happens only when // free service is available. -func (f *clientPool) posExpiration(now mclock.AbsTime) uint64 { +func (f *clientPool) posExpiration(now mclock.AbsTime) fixed64 { f.expLock.RLock() defer f.expLock.RUnlock() + if f.posExpTC == 0 { + return f.posExp + } dt := now - f.freeRatioLastUpdate if dt < 0 { dt = 0 } - return f.posExp + uint64(float64(dt)*f.posExpTCi*f.freeRatio) + return f.posExp + fixed64(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 { +func (f *clientPool) negExpiration(now mclock.AbsTime) fixed64 { f.expLock.RLock() defer f.expLock.RUnlock() + if f.negExpTC == 0 { + return f.negExp + } dt := now - f.freeRatioLastUpdate if dt < 0 { dt = 0 } - return f.negExp + uint64(float64(dt)*f.negExpTCi*f.freeRatio) + return f.negExp + fixed64(float64(dt)*f.negExpTCi*f.freeRatio) } // totalTokenLimit returns the current token supply limit. Token prices are based @@ -422,23 +412,21 @@ func (f *clientPool) connect(peer clientPoolPeer, reqCapacity uint64) (uint64, e if _, ok := f.connectedMap[id]; ok { clientRejectedMeter.Mark(1) log.Debug("Client already connected", "address", freeID, "id", peerIdToString(id)) - return 0, fmt.Errorf("Client already connected address = %s id = %s", freeID, peerIdToString(id)) + return 0, fmt.Errorf("Client already connected address=%s id=%s", freeID, peerIdToString(id)) } - pb := f.ndb.getOrNewPB(id) - nb := f.ndb.getOrNewNB(freeID) + pb := f.ndb.getOrNewBalance(id.Bytes(), false) + nb := f.ndb.getOrNewBalance([]byte(freeID), true) e := &clientInfo{ - capacity: reqCapacity, - pool: f, - peer: peer, - address: freeID, - queueIndex: -1, - id: id, - freeID: freeID, - connectedAt: f.clock.Now(), - priority: pb.value.base != 0, - posFactors: f.defaultPosFactors, - negFactors: f.defaultNegFactors, - balanceMetaInfo: pb.meta, + id: id, + address: freeID, + capacity: reqCapacity, + pool: f, + peer: peer, + queueIndex: -1, + connectedAt: f.clock.Now(), + priority: pb.value.base != 0, + posFactors: f.defaultPosFactors, + negFactors: f.defaultNegFactors, } missing, capacity := f.capAvailable(id, freeID, reqCapacity, 0, true) f.connectedMap[id] = e @@ -472,18 +460,12 @@ func (f *clientPool) connect(peer clientPoolPeer, reqCapacity uint64) (uint64, e } // initBalanceTracker initializes the positive and negative balances and price factors -func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb posBalance, nb negBalance, capacity uint64, active bool) { +func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb tokenBalance, nb tokenBalance, 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) + bt.setBalance(pb.value, nb.value) if active { - updatePriceFactors(bt, f.defaultPosFactors, f.defaultNegFactors, capacity) + updatePriceFactors(bt, f.defaultPosFactors, f.defaultNegFactors) } else { zeroPriceFactors(bt) } @@ -570,7 +552,7 @@ func (f *clientPool) capAvailable(id enode.ID, freeID string, capacity uint64, m bt = &client.balanceTracker } else { bt = &balanceTracker{} - f.initBalanceTracker(bt, f.ndb.getOrNewPB(id), f.ndb.getOrNewNB(freeID), capacity, true) + f.initBalanceTracker(bt, f.ndb.getOrNewBalance(id.Bytes(), false), f.ndb.getOrNewBalance([]byte(freeID), true), capacity, true) } if capacity != f.freeClientCap && targetPriority >= 0 { targetPriority = -1 @@ -656,7 +638,7 @@ 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) + missing, capacity := f.capAvailable(e.id, e.address, e.capacity, 0, true) if missing != 0 { f.inactiveQueue.Push(e, -connPriority(e, now)) return @@ -667,7 +649,7 @@ func (f *clientPool) tryActivateClients() { e.peer.updateCapacity(capacity) balance, _ := e.balanceTracker.getBalance(now) e.balanceTracker.setCapacity(capacity) - updatePriceFactors(&e.balanceTracker, f.defaultPosFactors, f.defaultNegFactors, capacity) + updatePriceFactors(&e.balanceTracker, f.defaultPosFactors, f.defaultNegFactors) // Register activated client to connection queue. f.inactiveBalances.subExp(balance) f.activeBalances.addExp(balance) @@ -684,7 +666,7 @@ func (f *clientPool) tryActivateClients() { e.peer.updateCapacity(e.capacity) totalConnectedGauge.Update(int64(f.activeCap)) clientConnectedMeter.Mark(1) - log.Debug("Client activated", "address", e.freeID) + log.Debug("Client activated", "address", e.address) } } @@ -705,19 +687,24 @@ func (f *clientPool) finalizeBalance(c *clientInfo, now mclock.AbsTime) { 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) - - 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. + for index, value := range []expiredValue{pos, neg} { + var ( + id []byte + expiration fixed64 + ) + neg := index == 1 + if !neg { + id = c.id.Bytes() + expiration = f.posExpiration(f.clock.Now()) + } else { + id = []byte(c.address) + expiration = f.negExpiration(f.clock.Now()) + } + if value.value(expiration) > uint64(time.Second) { + f.ndb.setBalance(id, neg, tokenBalance{value: value}) + } else { + f.ndb.delBalance(id, neg) // balance is small enough, drop it directly. + } } } @@ -743,9 +730,7 @@ func (f *clientPool) balanceExhausted(id enode.ID) { c.balanceTracker.setCapacity(c.capacity) c.peer.updateCapacity(c.capacity) } - pb := f.ndb.getOrNewPB(id) - pb.value = expiredValue{} - f.ndb.setPB(id, pb) + f.ndb.delBalance(id.Bytes(), false) } // setactiveLimit sets the maximum number and total capacity of connected clients, @@ -792,7 +777,7 @@ func (f *clientPool) setCapacity(id enode.ID, freeID string, capacity uint64, mi c.balanceTracker.setCapacity(capacity) f.activeQueue.Update(c.queueIndex) totalConnectedGauge.Update(int64(f.activeCap)) - updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors, c.capacity) + updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors) c.peer.updateCapacity(c.capacity) f.tryActivateClients() } @@ -821,38 +806,36 @@ func (f *clientPool) requestCost(p *clientPeer, cost uint64) uint64 { } // 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) +func updatePriceFactors(bt *balanceTracker, posFactors, negFactors priceFactors) { + bt.setFactors(posFactors, negFactors) } // zeroPriceFactors sets the pricing factors to zero func zeroPriceFactors(bt *balanceTracker) { - bt.setFactors(true, 0, 0) - bt.setFactors(false, 0, 0) + bt.setFactors(priceFactors{0, 0, 0}, priceFactors{0, 0, 0}) } // getPosBalance retrieves a single positive balance entry from cache or the database -func (f *clientPool) getPosBalance(id enode.ID) posBalance { +func (f *clientPool) getPosBalance(id enode.ID) tokenBalance { f.lock.Lock() defer f.lock.Unlock() if c := f.connectedMap[id]; c != nil { - pb, _ := c.balanceTracker.getBalance(f.clock.Now()) - return posBalance{value: pb, meta: c.balanceMetaInfo} + value, _ := c.balanceTracker.getBalance(f.clock.Now()) + return tokenBalance{value: value} } else { - return f.ndb.getOrNewPB(id) + return f.ndb.getOrNewBalance(id.Bytes(), false) } } // addBalance updates the balance of a client (either overwrites it or adds to it). // It also updates the balance meta info string. -func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, uint64, error) { +func (f *clientPool) addBalance(id enode.ID, amount int64) (uint64, uint64, error) { f.lock.Lock() defer f.lock.Unlock() now := f.clock.Now() - pb := f.ndb.getOrNewPB(id) + pb := f.ndb.getOrNewBalance(id.Bytes(), false) var negBalance expiredValue c := f.connectedMap[id] if c != nil { @@ -865,8 +848,7 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, return oldValue, oldValue, errBalanceOverflow } pb.value.add(amount, posExp) - pb.meta = meta - f.ndb.setPB(id, pb) + f.ndb.setBalance(id.Bytes(), false, pb) if c != nil { c.balanceTracker.setBalance(pb.value, negBalance) if c.active { @@ -879,7 +861,6 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, f.updateFreeRatio() c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) }) } - c.balanceMetaInfo = meta f.activeBalances.subExp(oldBalance) f.activeBalances.addExp(pb.value) } else { @@ -900,268 +881,3 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, f.tryActivateClients() return oldValue, pb.value.value(posExp), nil } - -// posBalance represents a recently accessed positive balance entry -type posBalance struct { - value expiredValue - meta string -} - -// EncodeRLP implements rlp.Encoder -func (e *posBalance) EncodeRLP(w io.Writer) error { - 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 { - ValueBase, ValueExp uint64 - Meta string - } - if err := s.Decode(&entry); err != nil { - return err - } - 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 uint64 } - -// EncodeRLP implements rlp.Encoder -func (e *negBalance) EncodeRLP(w io.Writer) error { - return rlp.Encode(w, []interface{}{e.logValue}) -} - -// DecodeRLP implements rlp.Decoder -func (e *negBalance) DecodeRLP(s *rlp.Stream) error { - var entry struct { - LogValue uint64 - } - if err := s.Decode(&entry); err != nil { - return err - } - e.logValue = entry.LogValue - return nil -} - -const ( - // nodeDBVersion is the version identifier of the node data in db - // - // Changelog: - // * Replace `lastTotal` with `meta` in positive balance: version 0=>1 - nodeDBVersion = 1 - - // dbCleanupCycle is the cycle of db for useless data cleanup - dbCleanupCycle = time.Hour -) - -var ( - 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 { - db ethdb.Database - pcache *lru.Cache - ncache *lru.Cache - auxbuf []byte // 37-byte auxiliary buffer for key encoding - verbuf [2]byte // 2-byte auxiliary buffer for db version - nbEvictCallBack func(mclock.AbsTime, negBalance) bool // Callback to determine whether the negative balance can be evicted. - clock mclock.Clock - closeCh chan struct{} - cleanupHook func() // Test hook used for testing -} - -func newNodeDB(db ethdb.Database, clock mclock.Clock) *nodeDB { - pcache, _ := lru.New(posBalanceCacheLimit) - ncache, _ := lru.New(negBalanceCacheLimit) - ndb := &nodeDB{ - db: db, - pcache: pcache, - ncache: ncache, - auxbuf: make([]byte, 37), - clock: clock, - closeCh: make(chan struct{}), - } - binary.BigEndian.PutUint16(ndb.verbuf[:], uint16(nodeDBVersion)) - go ndb.expirer() - return ndb -} - -func (db *nodeDB) close() { - close(db.closeCh) -} - -func (db *nodeDB) key(id []byte, neg bool) []byte { - prefix := positiveBalancePrefix - if neg { - prefix = negativeBalancePrefix - } - if len(prefix)+len(db.verbuf)+len(id) > len(db.auxbuf) { - db.auxbuf = append(db.auxbuf, make([]byte, len(prefix)+len(db.verbuf)+len(id)-len(db.auxbuf))...) - } - copy(db.auxbuf[:len(db.verbuf)], db.verbuf[:]) - copy(db.auxbuf[len(db.verbuf):len(db.verbuf)+len(prefix)], prefix) - copy(db.auxbuf[len(prefix)+len(db.verbuf):len(prefix)+len(db.verbuf)+len(id)], id) - return db.auxbuf[:len(prefix)+len(db.verbuf)+len(id)] -} - -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 binary.BigEndian.Uint64(blob[:8]), binary.BigEndian.Uint64(blob[8:16]) -} - -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 { - key := db.key(id.Bytes(), false) - item, exist := db.pcache.Get(string(key)) - if exist { - return item.(posBalance) - } - var balance posBalance - if enc, err := db.db.Get(key); err == nil { - if err := rlp.DecodeBytes(enc, &balance); err != nil { - log.Error("Failed to decode positive balance", "err", err) - } - } - db.pcache.Add(string(key), balance) - return balance -} - -func (db *nodeDB) setPB(id enode.ID, b posBalance) { - if b.value.base == 0 && len(b.meta) == 0 { - db.delPB(id) - return - } - key := db.key(id.Bytes(), false) - enc, err := rlp.EncodeToBytes(&(b)) - if err != nil { - log.Error("Failed to encode positive balance", "err", err) - return - } - db.db.Put(key, enc) - db.pcache.Add(string(key), b) -} - -func (db *nodeDB) delPB(id enode.ID) { - key := db.key(id.Bytes(), false) - db.db.Delete(key) - db.pcache.Remove(string(key)) -} - -// getPosBalanceIDs returns a lexicographically ordered list of IDs of accounts -// with a positive balance -func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID) { - if maxCount <= 0 { - return - } - it := db.db.NewIteratorWithStart(db.key(start.Bytes(), false)) - defer it.Release() - for i := len(stop[:]) - 1; i >= 0; i-- { - stop[i]-- - if stop[i] != 255 { - break - } - } - stopKey := db.key(stop.Bytes(), false) - keyLen := len(stopKey) - - for it.Next() { - var id enode.ID - if len(it.Key()) != keyLen || bytes.Compare(it.Key(), stopKey) == 1 { - return - } - copy(id[:], it.Key()[keyLen-len(id):]) - result = append(result, id) - if len(result) == maxCount { - return - } - } - return -} - -func (db *nodeDB) getOrNewNB(id string) negBalance { - key := db.key([]byte(id), true) - item, exist := db.ncache.Get(string(key)) - if exist { - return item.(negBalance) - } - var balance negBalance - if enc, err := db.db.Get(key); err == nil { - if err := rlp.DecodeBytes(enc, &balance); err != nil { - log.Error("Failed to decode negative balance", "err", err) - } - } - db.ncache.Add(string(key), balance) - return balance -} - -func (db *nodeDB) setNB(id string, b negBalance) { - key := db.key([]byte(id), true) - enc, err := rlp.EncodeToBytes(&(b)) - if err != nil { - log.Error("Failed to encode negative balance", "err", err) - return - } - db.db.Put(key, enc) - db.ncache.Add(string(key), b) -} - -func (db *nodeDB) delNB(id string) { - key := db.key([]byte(id), true) - db.db.Delete(key) - db.ncache.Remove(string(key)) -} - -func (db *nodeDB) expirer() { - for { - select { - case <-db.clock.After(dbCleanupCycle): - db.expireNodes() - case <-db.closeCh: - return - } - } -} - -// expireNodes iterates the whole node db and checks whether the negative balance -// entry can deleted. -// -// The rationale behind this is: server doesn't need to keep the negative balance -// records if they are low enough. -func (db *nodeDB) expireNodes() { - var ( - visited int - deleted int - start = time.Now() - ) - iter := db.db.NewIteratorWithPrefix(append(db.verbuf[:], negativeBalancePrefix...)) - for iter.Next() { - visited += 1 - var balance negBalance - if err := rlp.DecodeBytes(iter.Value(), &balance); err != nil { - log.Error("Failed to decode negative balance", "err", err) - continue - } - if db.nbEvictCallBack != nil && db.nbEvictCallBack(db.clock.Now(), balance) { - deleted += 1 - db.db.Delete(iter.Key()) - } - } - // Invoke testing hook if it's not nil. - if db.cleanupHook != nil { - db.cleanupHook() - } - log.Debug("Expire nodes", "visited", visited, "deleted", deleted, "elapsed", common.PrettyDuration(time.Since(start))) -} diff --git a/les/clientpool_test.go b/les/clientpool_test.go index 94ccb33209..a34deff1b8 100644 --- a/les/clientpool_test.go +++ b/les/clientpool_test.go @@ -17,11 +17,8 @@ package les import ( - "bytes" "fmt" - "math" "math/rand" - "reflect" "testing" "time" @@ -117,7 +114,7 @@ func testClientPool(t *testing.T, activeLimit, clientCount, paidCount int, rando // 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(newPoolTestPeer(i, disconnCh).ID(), amount, "") + pool.addBalance(newPoolTestPeer(i, disconnCh).ID(), amount) } } @@ -186,7 +183,7 @@ func TestConnectPaidClient(t *testing.T) { pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) // Add balance for an external client and mark it as paid client - pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000, "") + pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000) if cap, _ := pool.connect(newPoolTestPeer(0, nil), 10); cap == 0 { t.Fatalf("Failed to connect paid client") @@ -204,7 +201,7 @@ func TestConnectPaidClientToSmallPool(t *testing.T) { pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) // Add balance for an external client and mark it as paid client - pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000, "") + pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000) // Connect a fat paid client to pool, should reject it. if cap, _ := pool.connect(newPoolTestPeer(0, nil), 100); cap != 0 { @@ -224,15 +221,15 @@ func TestConnectPaidClientToFullPool(t *testing.T) { pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) for i := 0; i < 10; i++ { - pool.addBalance(newPoolTestPeer(i, nil).ID(), 1000000000, "") + pool.addBalance(newPoolTestPeer(i, nil).ID(), int64(time.Second)) pool.connect(newPoolTestPeer(i, nil), 1) } - pool.addBalance(newPoolTestPeer(11, nil).ID(), 1000, "") // Add low balance to new paid client + 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(newPoolTestPeer(12, nil).ID(), 1000000000*60*3+1, "") // Add high balance to new paid client + pool.addBalance(newPoolTestPeer(12, nil).ID(), int64(time.Minute*5)) // 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") } @@ -251,7 +248,7 @@ func TestPaidClientKickedOut(t *testing.T) { pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) for i := 0; i < 10; i++ { - pool.addBalance(newPoolTestPeer(i, kickedCh).ID(), 1000000000, "") // 1 second allowance + pool.addBalance(newPoolTestPeer(i, kickedCh).ID(), 1000000000) // 1 second allowance pool.connect(newPoolTestPeer(i, kickedCh), 1) clock.Run(time.Millisecond) } @@ -360,12 +357,12 @@ func TestPositiveBalanceCalculation(t *testing.T) { pool.setLimits(10, uint64(10)) // Total capacity limit is 10 pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) - pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute*3), "") + pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute*3)) pool.connect(newPoolTestPeer(0, kicked), 10) clock.Run(time.Minute) pool.disconnect(newPoolTestPeer(0, kicked)) - pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID()) + pb := pool.ndb.getOrNewBalance(newPoolTestPeer(0, kicked).ID().Bytes(), false) if pb.value != expval(uint64(time.Minute*2)) { t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute*2), pb.value) } @@ -384,7 +381,7 @@ func TestDowngradePriorityClient(t *testing.T) { pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) p := newPoolTestPeer(0, kicked) - pool.addBalance(p.ID(), int64(time.Minute), "") + pool.addBalance(p.ID(), int64(time.Minute)) 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) @@ -395,13 +392,13 @@ 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(newPoolTestPeer(0, kicked).ID()) + pb := pool.ndb.getOrNewBalance(newPoolTestPeer(0, kicked).ID().Bytes(), false) if pb.value.base != 0 { t.Fatalf("Positive balance mismatch, want %v, got %v", 0, pb.value) } - pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute), "") - pb = pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID()) + pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute)) + pb = pool.ndb.getOrNewBalance(newPoolTestPeer(0, kicked).ID().Bytes(), false) if pb.value != expval(uint64(time.Minute)) { t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute), pb.value) } @@ -420,133 +417,29 @@ func TestNegativeBalanceCalculation(t *testing.T) { for i := 0; i < 10; i++ { pool.connect(newPoolTestPeer(i, nil), 1) } - clock.Run(time.Millisecond * 999) + clock.Run(time.Second) for i := 0; i < 10; i++ { pool.disconnect(newPoolTestPeer(i, nil)) - nb := pool.ndb.getOrNewNB(newPoolTestPeer(i, nil).freeClientId()) - if nb.logValue != 0 { + nb := pool.ndb.getOrNewBalance([]byte(newPoolTestPeer(i, nil).freeClientId()), true) + if nb.value.base != 0 { t.Fatalf("Short connection shouldn't be recorded") } } - for i := 0; i < 10; i++ { pool.connect(newPoolTestPeer(i, nil), 1) } clock.Run(time.Minute) for i := 0; i < 10; i++ { 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) + nb := pool.ndb.getOrNewBalance([]byte(newPoolTestPeer(i, nil).freeClientId()), true) + value := nb.value.value(pool.negExpiration(clock.Now())) + if value != uint64(time.Minute) { + t.Fatalf("Negative balance mismatch, want %v, got %v", time.Minute, value) } } } -func TestNodeDB(t *testing.T) { - ndb := newNodeDB(rawdb.NewMemoryDatabase(), mclock.System{}) - defer ndb.close() - - if !bytes.Equal(ndb.verbuf[:], []byte{0x00, nodeDBVersion}) { - t.Fatalf("version buffer mismatch, want %v, got %v", []byte{0x00, nodeDBVersion}, ndb.verbuf) - } - var cases = []struct { - id enode.ID - ip string - balance interface{} - positive bool - }{ - {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}, - } - for _, c := range cases { - if c.positive { - ndb.setPB(c.id, c.balance.(posBalance)) - if pb := ndb.getOrNewPB(c.id); !reflect.DeepEqual(pb, c.balance.(posBalance)) { - t.Fatalf("Positive balance mismatch, want %v, got %v", c.balance.(posBalance), pb) - } - } else { - ndb.setNB(c.ip, c.balance.(negBalance)) - if nb := ndb.getOrNewNB(c.ip); !reflect.DeepEqual(nb, c.balance.(negBalance)) { - t.Fatalf("Negative balance mismatch, want %v, got %v", c.balance.(negBalance), nb) - } - } - } - for _, c := range cases { - if c.positive { - ndb.delPB(c.id) - if pb := ndb.getOrNewPB(c.id); !reflect.DeepEqual(pb, posBalance{}) { - t.Fatalf("Positive balance mismatch, want %v, got %v", posBalance{}, pb) - } - } else { - ndb.delNB(c.ip) - if nb := ndb.getOrNewNB(c.ip); !reflect.DeepEqual(nb, negBalance{}) { - t.Fatalf("Negative balance mismatch, want %v, got %v", negBalance{}, nb) - } - } - } - 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) - } -} - -func TestNodeDBExpiration(t *testing.T) { - var ( - iterated int - done = make(chan struct{}, 1) - ) - callback := func(now mclock.AbsTime, b negBalance) bool { - iterated += 1 - return true - } - clock := &mclock.Simulated{} - ndb := newNodeDB(rawdb.NewMemoryDatabase(), clock) - defer ndb.close() - ndb.nbEvictCallBack = callback - ndb.cleanupHook = func() { done <- struct{}{} } - - var cases = []struct { - ip string - balance negBalance - }{ - {"127.0.0.1", negBalance{logValue: 1}}, - {"127.0.0.2", negBalance{logValue: 1}}, - {"127.0.0.3", negBalance{logValue: 1}}, - {"127.0.0.4", negBalance{logValue: 1}}, - } - for _, c := range cases { - ndb.setNB(c.ip, c.balance) - } - clock.WaitForTimers(1) - clock.Run(time.Hour + time.Minute) - select { - case <-done: - case <-time.NewTimer(time.Second).C: - t.Fatalf("timeout") - } - if iterated != 4 { - t.Fatalf("Failed to evict useless negative balances, want %v, got %d", 4, iterated) - } - clock.WaitForTimers(1) - for _, c := range cases { - ndb.setNB(c.ip, c.balance) - } - clock.Run(time.Hour + time.Minute) - select { - case <-done: - case <-time.NewTimer(time.Second).C: - t.Fatalf("timeout") - } - if iterated != 8 { - t.Fatalf("Failed to evict useless negative balances, want %v, got %d", 4, iterated) - } -} - func TestInactiveClient(t *testing.T) { var ( clock mclock.Simulated @@ -559,8 +452,8 @@ func TestInactiveClient(t *testing.T) { p1 := newPoolTestPeer(1, nil) p2 := newPoolTestPeer(2, nil) p3 := newPoolTestPeer(3, nil) - pool.addBalance(p1.ID(), 1000, "") - pool.addBalance(p3.ID(), 2000, "") + 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 { @@ -577,7 +470,7 @@ func TestInactiveClient(t *testing.T) { if p2.cap != 0 { t.Fatalf("Failed to deactivate peer #2") } - pool.addBalance(p2.ID(), 3000, "") + pool.addBalance(p2.ID(), 3000) // p1: 1000 p2: 3000 p3: 2000 if p2.cap != 1 { t.Fatalf("Failed to activate peer #2") @@ -585,7 +478,7 @@ func TestInactiveClient(t *testing.T) { if p1.cap != 0 { t.Fatalf("Failed to deactivate peer #1") } - pool.addBalance(p2.ID(), -2500, "") + pool.addBalance(p2.ID(), -2500) // p1: 1000 p2: 500 p3: 2000 if p1.cap != 1 { t.Fatalf("Failed to activate peer #1") @@ -595,7 +488,7 @@ func TestInactiveClient(t *testing.T) { } pool.setDefaultFactors(priceFactors{1e-9, 0, 0}, priceFactors{1e-9, 0, 0}) p4 := newPoolTestPeer(4, nil) - pool.addBalance(p4.ID(), 1500, "") + pool.addBalance(p4.ID(), 1500) // p1: 1000 p2: 500 p3: 2000 p4: 1500 p4.cap, _ = pool.connect(p4, 1) if p4.cap != 1 { @@ -618,7 +511,7 @@ func TestInactiveClient(t *testing.T) { } pool.disconnect(p2) pool.disconnect(p4) - pool.addBalance(p1.ID(), -1000, "") + pool.addBalance(p1.ID(), -1000) if p1.cap != 1 { t.Fatalf("Should not deactivate peer #1") } diff --git a/les/expiredvalue.go b/les/expiredvalue.go new file mode 100644 index 0000000000..6ce3638c25 --- /dev/null +++ b/les/expiredvalue.go @@ -0,0 +1,128 @@ +// Copyright 2020 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package les + +import "math" + +// expiredValue is a scalar value that is continuously expired (decreased +// exponentially) based on the provided logarithmic expiration offset value. +// +// The formula for value calculation is: base*2^(exp-logOffset). In order to +// simplify the calculation of expiredValue, its value is expressed in the form +// of an exponent with a base of 2. +// +// Also here is a trick to reduce a lot of calculations. In theory, when a value X +// decays over time and then a new value Y is added, the final result should be +// X*2^(exp-logOffset)+Y. However it's very hard to represent in memory. +// So the trick is using the idea of inflation instead of exponential decay. At this +// moment the temporary value becomes: X*2^exp+Y*2^logOffset_1, apply the exponential +// decay when we actually want to calculate the value. +// +// e.g. +// t0: V = 100 +// t1: add 30, inflationary value is: 100 + 30/0.3, 0.3 is the decay coefficient +// t2: get value, decay coefficient is 0.2 now, final result is: 200*0.2 = 40 +type expiredValue struct { + base, exp uint64 +} + +// value calculates the value at the given moment. +func (e expiredValue) value(logOffset fixed64) uint64 { + offset := uint64ToFixed64(e.exp) - logOffset + return uint64(float64(e.base) * offset.pow2Fixed()) +} + +// add adds a signed value at the given moment +func (e *expiredValue) add(amount int64, logOffset fixed64) int64 { + integer, frac := logOffset.toUint64(), logOffset.fraction() + factor := frac.pow2Fixed() + base := factor * float64(amount) + if integer < e.exp { + base /= math.Pow(2, float64(e.exp-integer)) + } + if integer > e.exp { + e.base >>= (integer - e.exp) + e.exp = integer + } + if base >= 0 || uint64(-base) <= e.base { + e.base += uint64(base) + return amount + } + net := int64(-float64(e.base) / factor) + e.base = 0 + return net +} + +// 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 + } +} + +// fixedFactor is the factor used by fixed64 +const fixedFactor = 0x1000000 + +// fixed64 is a float64 wrapper that uses integer arithmetic +// to avoid precision loss in floating-point arithmetic. +type fixed64 int64 + +// uint64ToFixed64 converts uint64 integer to fixed64 format. +func uint64ToFixed64(f uint64) fixed64 { + return fixed64(f * fixedFactor) +} + +// float64ToFixed64 converts float64 to fixed64 format. +func float64ToFixed64(f float64) fixed64 { + return fixed64(f * fixedFactor) +} + +// toUint64 converts fixed64 format to uint64. +func (f64 fixed64) toUint64() uint64 { + return uint64(f64) / fixedFactor +} + +// fraction returns the fraction of the fixed64. +func (f64 fixed64) fraction() fixed64 { + return f64 % fixedFactor +} + +// pow2Fixed returns the 2 based pow of the fixed value. +func (f64 fixed64) pow2Fixed() float64 { + return math.Pow(2, float64(f64)/fixedFactor) +} diff --git a/les/expiredvalue_test.go b/les/expiredvalue_test.go new file mode 100644 index 0000000000..a59510a702 --- /dev/null +++ b/les/expiredvalue_test.go @@ -0,0 +1,116 @@ +// Copyright 2020 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package les + +import "testing" + +func TestValueExpiration(t *testing.T) { + var cases = []struct { + input expiredValue + timeOffset fixed64 + expect uint64 + }{ + {expiredValue{base: 128, exp: 0}, uint64ToFixed64(0), 128}, + {expiredValue{base: 128, exp: 0}, uint64ToFixed64(1), 64}, + {expiredValue{base: 128, exp: 0}, uint64ToFixed64(2), 32}, + {expiredValue{base: 128, exp: 2}, uint64ToFixed64(2), 128}, + {expiredValue{base: 128, exp: 2}, uint64ToFixed64(3), 64}, + } + for _, c := range cases { + if got := c.input.value(c.timeOffset); got != c.expect { + t.Fatalf("Value mismatch, want=%d, got=%d", c.expect, got) + } + } +} + +func TestValueAddition(t *testing.T) { + var cases = []struct { + input expiredValue + addend int64 + timeOffset fixed64 + expect uint64 + expectNet int64 + }{ + // Addition + {expiredValue{base: 128, exp: 0}, 128, uint64ToFixed64(0), 256, 128}, + {expiredValue{base: 128, exp: 2}, 128, uint64ToFixed64(0), 640, 128}, + + // Addition with offset + {expiredValue{base: 128, exp: 0}, 128, uint64ToFixed64(1), 192, 128}, + {expiredValue{base: 128, exp: 2}, 128, uint64ToFixed64(1), 384, 128}, + {expiredValue{base: 128, exp: 2}, 128, uint64ToFixed64(3), 192, 128}, + + // Subtraction + {expiredValue{base: 128, exp: 0}, -64, uint64ToFixed64(0), 64, -64}, + {expiredValue{base: 128, exp: 0}, -128, uint64ToFixed64(0), 0, -128}, + {expiredValue{base: 128, exp: 0}, -192, uint64ToFixed64(0), 0, -128}, + + // Subtraction with offset + {expiredValue{base: 128, exp: 0}, -64, uint64ToFixed64(1), 0, -64}, + {expiredValue{base: 128, exp: 0}, -128, uint64ToFixed64(1), 0, -64}, + {expiredValue{base: 128, exp: 2}, -128, uint64ToFixed64(1), 128, -128}, + {expiredValue{base: 128, exp: 2}, -128, uint64ToFixed64(2), 0, -128}, + } + for _, c := range cases { + if net := c.input.add(c.addend, c.timeOffset); net != c.expectNet { + t.Fatalf("Net amount mismatch, want=%d, got=%d", c.expectNet, net) + } + if got := c.input.value(c.timeOffset); got != c.expect { + t.Fatalf("Value mismatch, want=%d, got=%d", c.expect, got) + } + } +} + +func TestExpiredValueAddition(t *testing.T) { + var cases = []struct { + input expiredValue + another expiredValue + timeOffset fixed64 + expect uint64 + }{ + {expiredValue{base: 128, exp: 0}, expiredValue{base: 128, exp: 0}, uint64ToFixed64(0), 256}, + {expiredValue{base: 128, exp: 1}, expiredValue{base: 128, exp: 0}, uint64ToFixed64(0), 384}, + {expiredValue{base: 128, exp: 0}, expiredValue{base: 128, exp: 1}, uint64ToFixed64(0), 384}, + {expiredValue{base: 128, exp: 0}, expiredValue{base: 128, exp: 0}, uint64ToFixed64(1), 128}, + } + for _, c := range cases { + c.input.addExp(c.another) + if got := c.input.value(c.timeOffset); got != c.expect { + t.Fatalf("Value mismatch, want=%d, got=%d", c.expect, got) + } + } +} + +func TestExpiredValueSubtraction(t *testing.T) { + var cases = []struct { + input expiredValue + another expiredValue + timeOffset fixed64 + expect uint64 + }{ + {expiredValue{base: 128, exp: 0}, expiredValue{base: 128, exp: 0}, uint64ToFixed64(0), 0}, + {expiredValue{base: 128, exp: 0}, expiredValue{base: 128, exp: 1}, uint64ToFixed64(0), 0}, + {expiredValue{base: 128, exp: 1}, expiredValue{base: 128, exp: 0}, uint64ToFixed64(0), 128}, + {expiredValue{base: 128, exp: 1}, expiredValue{base: 128, exp: 0}, uint64ToFixed64(1), 64}, + } + for _, c := range cases { + c.input.subExp(c.another) + if got := c.input.value(c.timeOffset); got != c.expect { + t.Fatalf("Value mismatch, want=%d, got=%d", c.expect, got) + } + } +} diff --git a/les/tokensale.go b/les/tokensale.go index 2e4ed40b87..b00fcfa776 100644 --- a/les/tokensale.go +++ b/les/tokensale.go @@ -19,7 +19,6 @@ package les import ( "encoding/binary" "fmt" - "io" "math" "strconv" "sync" @@ -41,8 +40,8 @@ const ( // payment technologies type paymentReceiver interface { info() keyValueList - receivePayment(from enode.ID, proofOfPayment, oldMeta []byte) (value uint64, newMeta []byte, err error) - requestPayment(from enode.ID, value uint64, meta []byte) uint64 + receivePayment(from enode.ID, proofOfPayment []byte) (value uint64, err error) + requestPayment(from enode.ID, value uint64) uint64 } // tokenSale handles client balance deposits, conversion to and from service tokens @@ -401,10 +400,8 @@ func (t *tokenSale) connection(id enode.ID, freeID string, requestedCapacity uin tokensMissing, availableCapacity, err = t.clientPool.setCapacityLocked(id, freeID, requestedCapacity, stayConnected, setCap) pb := t.clientPool.getPosBalance(id) tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now())) - var meta tokenSaleMeta - if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil { - pcBalance = meta.pcBalance - } + cb := t.clientPool.ndb.getCurrencyBalance(id) + pcBalance = cb.amount if tokensMissing == 0 { return } @@ -440,7 +437,7 @@ func (t *tokenSale) connection(id enode.ID, freeID string, requestedCapacity uin if rec, ok := t.receivers[recID]; !ok || pcMissing == math.MaxUint64 { paymentRequired[i] = math.MaxUint64 } else { - paymentRequired[i] = rec.requestPayment(id, pcMissing, meta.receiverMeta[recID]) + paymentRequired[i] = rec.requestPayment(id, pcMissing) } } return @@ -451,24 +448,19 @@ func (t *tokenSale) deposit(id enode.ID, paymentModule string, proofOfPayment [] t.lock.Lock() defer t.lock.Unlock() - pb := t.clientPool.getPosBalance(id) - var meta tokenSaleMeta - if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil { - pcBalance = meta.pcBalance - } - + cb := t.clientPool.ndb.getCurrencyBalance(id) + pcBalance = cb.amount pm := t.receivers[paymentModule] if pm == nil { return 0, pcBalance, fmt.Errorf("Unknown payment receiver '%s'", paymentModule) } - pcValue, meta.receiverMeta[paymentModule], err = pm.receivePayment(id, proofOfPayment, meta.receiverMeta[paymentModule]) + pcValue, err = pm.receivePayment(id, proofOfPayment) if err != nil { return 0, pcBalance, err } pcBalance += pcValue - meta.pcBalance = pcBalance - metaEnc, _ := rlp.EncodeToBytes(&meta) - t.clientPool.addBalance(id, 0, string(metaEnc)) + cb.amount = pcBalance + t.clientPool.ndb.setCurrencyBalance(id, cb) return } @@ -491,10 +483,8 @@ func (t *tokenSale) buyTokens(id enode.ID, maxSpend, minReceive uint64, relative pb := t.clientPool.getPosBalance(id) tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now())) - var meta tokenSaleMeta - if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil { - pcBalance = meta.pcBalance - } + cb := t.clientPool.ndb.getCurrencyBalance(id) + pcBalance = cb.amount if relative { if pcBalance > maxSpend { maxSpend = pcBalance - maxSpend @@ -526,10 +516,10 @@ func (t *tokenSale) buyTokens(id enode.ID, maxSpend, minReceive uint64, relative } if success { pcBalance -= spend + cb.amount = pcBalance tokenBalance += receive - meta.pcBalance = pcBalance - metaEnc, _ := rlp.EncodeToBytes(&meta) - t.clientPool.addBalance(id, int64(receive), string(metaEnc)) + t.clientPool.ndb.setCurrencyBalance(id, cb) + t.clientPool.addBalance(id, int64(receive)) } return } @@ -542,10 +532,8 @@ func (t *tokenSale) sellTokens(id enode.ID, maxSell, minRefund uint64, relative, pb := t.clientPool.getPosBalance(id) tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now())) - var meta tokenSaleMeta - if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil { - pcBalance = meta.pcBalance - } + cb := t.clientPool.ndb.getCurrencyBalance(id) + pcBalance = cb.amount if relative { if pcBalance < minRefund { minRefund -= pcBalance @@ -580,9 +568,9 @@ func (t *tokenSale) sellTokens(id enode.ID, maxSell, minRefund uint64, relative, if success { pcBalance += refund tokenBalance -= sell - meta.pcBalance = pcBalance - metaEnc, _ := rlp.EncodeToBytes(&meta) - t.clientPool.addBalance(id, -int64(sell), string(metaEnc)) + cb.amount = pcBalance + t.clientPool.ndb.setCurrencyBalance(id, cb) + t.clientPool.addBalance(id, -int64(sell)) } return } @@ -593,12 +581,8 @@ func (t *tokenSale) getBalance(id enode.ID) (pcBalance, tokenBalance uint64) { defer t.lock.Unlock() pb := t.clientPool.getPosBalance(id) - tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now())) - var meta tokenSaleMeta - if err := rlp.DecodeBytes([]byte(pb.meta), &meta); err == nil { - pcBalance = meta.pcBalance - } - return + cb := t.clientPool.ndb.getCurrencyBalance(id) + return pb.value.value(t.clientPool.posExpiration(mclock.Now())), cb.amount } // info returns general information about the server, including version info of the @@ -626,65 +610,6 @@ func (t *tokenSale) receiverInfo(receiverIDs []string) []keyValueList { return res } -// tokenSaleMeta is the "meta" field used by the lespay token sale module. It is -// attached to token balances and it includes the permanent balance of the client -// nominated in the server's preferred currency and the meta fields provided by -// the used payment receivers. -type tokenSaleMeta struct { - pcBalance uint64 - receiverMeta map[string][]byte -} - -// receiverMetaEnc is used for easy RLP encoding/decoding -type receiverMetaEnc struct { - Id string - Meta []byte -} - -// tokenSaleMetaEnc is used for easy RLP encoding/decoding -type tokenSaleMetaEnc struct { - Id string - Version uint - PcBalance uint64 - Receivers []receiverMetaEnc -} - -// EncodeRLP implements rlp.Encoder -func (t *tokenSaleMeta) EncodeRLP(w io.Writer) error { - receivers := make([]receiverMetaEnc, len(t.receiverMeta)) - i := 0 - for id, meta := range t.receiverMeta { - receivers[i] = receiverMetaEnc{id, meta} - i++ - } - return rlp.Encode(w, tokenSaleMetaEnc{ - Id: "tokenSale", - Version: 1, - PcBalance: t.pcBalance, - Receivers: receivers, - }) -} - -// DecodeRLP implements rlp.Decoder -func (t *tokenSaleMeta) DecodeRLP(s *rlp.Stream) error { - if t.receiverMeta == nil { - t.receiverMeta = make(map[string][]byte) - } - var e tokenSaleMetaEnc - if err := s.Decode(&e); err != nil { - return err - } - if e.Id != "tokenSale" || e.Version != 1 { - return fmt.Errorf("Unknown balance meta format '%s' version %d", e.Id, e.Version) - } - t.receiverMeta = make(map[string][]byte) - t.pcBalance = e.PcBalance - for _, r := range e.Receivers { - t.receiverMeta[r.Id] = r.Meta - } - return nil -} - const ( tsInfo = iota tsReceiverInfo @@ -853,7 +778,7 @@ func (t testReceiver) info() keyValueList { // receivePayment implements paymentReceiver. proofOfPayment is a base 10 ascii number // which is credited to the sender's account without any further conditions. -func (t testReceiver) receivePayment(from enode.ID, proofOfPayment, oldMeta []byte) (value uint64, newMeta []byte, err error) { +func (t testReceiver) receivePayment(from enode.ID, proofOfPayment []byte) (value uint64, err error) { if len(proofOfPayment) > 8 { err = fmt.Errorf("proof of payment is too long; max 8 bytes long big endian integer expected") return @@ -865,6 +790,6 @@ func (t testReceiver) receivePayment(from enode.ID, proofOfPayment, oldMeta []by } // requestPayment implements paymentReceiver -func (t testReceiver) requestPayment(from enode.ID, value uint64, meta []byte) uint64 { +func (t testReceiver) requestPayment(from enode.ID, value uint64) uint64 { return value }