From 923fc541907cdde9e87560af574ad910380faf51 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Sat, 23 Nov 2019 23:50:27 +0100 Subject: [PATCH] les: implement token sale --- les/api.go | 2 +- les/clientpool.go | 125 +++++++++++++++++++++++++++++++++++------- les/server.go | 1 + les/server_handler.go | 11 +++- 4 files changed, 117 insertions(+), 22 deletions(-) diff --git a/les/api.go b/les/api.go index cd390c791d..0d944e7dcd 100644 --- a/les/api.go +++ b/les/api.go @@ -147,7 +147,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, uint64(capacity)) + _, _, err = api.server.clientPool.setCapacity(client.id, client.freeID, uint64(capacity), 0, true) // Don't have to call factor update explicitly. It's already done // in setCapacity function. } else { diff --git a/les/clientpool.go b/les/clientpool.go index eb61978564..a5eb0e8a95 100644 --- a/les/clientpool.go +++ b/les/clientpool.go @@ -42,6 +42,7 @@ const ( 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 + fullRatioTC = time.Hour // connectedBias is applied to already connected clients So that // already connected client won't be kicked out very soon and we @@ -85,6 +86,10 @@ type clientPool struct { connectedMap map[enode.ID]*clientInfo connectedQueue *prque.LazyQueue + connectedBalances, disconnectedBalances uint64 + lastConnectedBalanceUpdate, fullRatioLastUpdate mclock.AbsTime + fullRatio float64 + defaultPosFactors, defaultNegFactors priceFactors connLimit int // The maximum number of connections that clientpool can support @@ -175,6 +180,23 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock cumulativeTime: ndb.getCumulativeTime(), stopCh: make(chan struct{}), } + // 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 { + stop = true + l-- + } + for i := 0; i < l; i++ { + pool.disconnectedBalances += 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 { @@ -208,6 +230,59 @@ func (f *clientPool) stop() { f.ndb.close() } +func (f *clientPool) updateFullRatio() { + full := float64(1) + if f.priorityConnected < f.capLimit { + freeCap := f.capLimit - f.priorityConnected + if freeCap > f.freeClientCap { + freeCapThreshold := f.capLimit / 4 + if freeCap > freeCapThreshold { + full = 0 + } else { + full = float64(freeCapThreshold-freeCap) / float64(freeCapThreshold-f.freeClientCap) + } + } + } + now := f.clock.Now() + dt := now - f.fullRatioLastUpdate + f.fullRatioLastUpdate = now + if dt < 0 { + dt = 0 + } + d := math.Exp(-float64(dt) / float64(fullRatioTC)) + f.fullRatio = full - (full-f.fullRatio)*d +} + +func (f *clientPool) totalTokenLimit() uint64 { + f.lock.Lock() + defer f.lock.Unlock() + + f.updateFullRatio() + d := 1 - f.fullRatio + if d > 0.5 { + d = -math.Log(0.5/d) * float64(fullRatioTC) + } else { + d = 0 + } + return uint64(d * float64(f.capLimit) * f.defaultPosFactors.capacityFactor) +} + +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.connectedBalances = 0 + for _, c := range f.connectedMap { + pos, _ := c.balanceTracker.getBalance(now) + f.connectedBalances += pos + } + f.lastConnectedBalanceUpdate = now + } + return f.connectedBalances + f.disconnectedBalances +} + // 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 clientPeer, capacity uint64) bool { @@ -249,6 +324,8 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool { } f.initBalanceTracker(&e.balanceTracker, pb, nb, capacity) // Register new client to connection queue. + f.disconnectedBalances -= pb.value + f.connectedBalances += pb.value f.connectedMap[id] = e f.connectedQueue.Push(e) f.connectedCap += e.capacity @@ -256,6 +333,7 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool { // 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.updateFullRatio() f.priorityConnected += capacity e.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) }) } @@ -301,13 +379,6 @@ func (f *clientPool) disconnect(p clientPeer) { f.dropClient(e, f.clock.Now(), false) } -func (f *clientPool) balanceMissing(id enode.ID, freeID string, capacity uint64, minConnTime time.Duration) (uint64, uint64) { - f.lock.Lock() - defer f.lock.Unlock() - - return f.capAvailable(id, freeID, capacity, minConnTime, false) -} - // 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. @@ -420,6 +491,7 @@ func (f *clientPool) dropClient(e *clientInfo, now mclock.AbsTime, kick bool) { delete(f.connectedMap, e.id) f.connectedCap -= e.capacity if e.priority { + f.updateFullRatio() f.priorityConnected -= e.capacity } totalConnectedGauge.Update(int64(f.connectedCap)) @@ -447,6 +519,8 @@ 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.disconnectedBalances += pos + f.connectedBalances -= pos pb, nb := f.ndb.getOrNewPB(c.id), f.ndb.getOrNewNB(c.address) pb.value = pos @@ -472,6 +546,7 @@ func (f *clientPool) balanceExhausted(id enode.ID) { return } if c.priority { + f.updateFullRatio() f.priorityConnected -= c.capacity } c.priority = false @@ -493,6 +568,7 @@ func (f *clientPool) setLimits(totalConn int, totalCap uint64) { f.lock.Lock() defer f.lock.Unlock() + f.updateFullRatio() f.connLimit = totalConn f.capLimit = totalCap if f.connectedCap > f.capLimit || f.connectedQueue.Size() > f.connLimit { @@ -504,26 +580,24 @@ func (f *clientPool) setLimits(totalConn int, totalCap uint64) { } // setCapacity sets the assigned capacity of a connected client -func (f *clientPool) setCapacity(c *clientInfo, capacity uint64) (uint64, uint64, error) { - if capacity == 0 { - capacity = f.freeClientCap - } - if capacity < f.minCap { - capacity = f.minCap - } - if f.connectedMap[c.id] != c { - return 0, capacity, fmt.Errorf("client %064x is not connected", c.id[:]) - } - if c.capacity == capacity { - return 0, capacity, nil +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 + } } var missing uint64 - missing, capacity = f.capAvailable(c.id, c.freeID, capacity, 0, true) + missing, capacity = f.capAvailable(id, freeID, capacity, 0, setCap && c != nil) if missing != 0 { return missing, capacity, errNoPriority } + if setCap && c == nil { + return missing, capacity, fmt.Errorf("client %064x is not connected", c.id[:]) + } // capacity update is possible f.connectedCap += capacity - c.capacity + f.updateFullRatio() f.priorityConnected += capacity - c.capacity c.capacity = capacity c.balanceTracker.setCapacity(capacity) @@ -534,6 +608,13 @@ func (f *clientPool) setCapacity(c *clientInfo, capacity uint64) (uint64, uint64 return 0, capacity, nil } +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() + + return f.setCapacity(id, freeID, capacity, minConnTime, setCap) +} + // requestCost feeds request cost after serving a request from the given peer. func (f *clientPool) requestCost(p *peer, cost uint64) { f.lock.Lock() @@ -605,6 +686,7 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, // 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.updateFullRatio() c.priority = true f.priorityConnected += c.capacity c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) }) @@ -612,6 +694,9 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, // if balance is set to zero then reverting to non-priority status // is handled by the balanceExhausted callback c.balanceMetaInfo = meta + f.connectedBalances += pb.value - oldBalance + } else { + f.disconnectedBalances += pb.value - oldBalance } return oldBalance, pb.value, nil } diff --git a/les/server.go b/les/server.go index ccae237c68..d5885a3162 100644 --- a/les/server.go +++ b/les/server.go @@ -50,6 +50,7 @@ type LesServer struct { defParams flowcontrol.ServerParams servingQueue *servingQueue clientPool *clientPool + tokenSale *tokenSale minCapacity, maxCapacity, freeCapacity uint64 threadsIdle int // Request serving threads count when system is idle. diff --git a/les/server_handler.go b/les/server_handler.go index 3755547771..24e096dffa 100644 --- a/les/server_handler.go +++ b/les/server_handler.go @@ -957,5 +957,14 @@ func (h *serverHandler) broadcastHeaders() { } func (h *serverHandler) talkRequestHandler(id enode.ID, addr *net.UDPAddr, payload []byte) ([]byte, bool) { - return payload, true // dummy handler, just returns the same payload + if h.server.tokenSale == nil { + return nil, false + } + var cmds [][]byte + if err := rlp.DecodeBytes(payload, &cmds); err != nil { + return nil, false + } + results := h.server.tokenSale.runCommands(cmds, id, addr.IP.String()) + res, _ := rlp.EncodeToBytes(&results) + return res, true }