clientpool.go 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862
  1. // Copyright 2019 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. package les
  17. import (
  18. "bytes"
  19. "encoding/binary"
  20. "fmt"
  21. "io"
  22. "math"
  23. "sync"
  24. "time"
  25. "github.com/ethereum/go-ethereum/common"
  26. "github.com/ethereum/go-ethereum/common/mclock"
  27. "github.com/ethereum/go-ethereum/common/prque"
  28. "github.com/ethereum/go-ethereum/ethdb"
  29. "github.com/ethereum/go-ethereum/log"
  30. "github.com/ethereum/go-ethereum/p2p/enode"
  31. "github.com/ethereum/go-ethereum/rlp"
  32. lru "github.com/hashicorp/golang-lru"
  33. )
  34. const (
  35. negBalanceExpTC = time.Hour // time constant for exponentially reducing negative balance
  36. fixedPointMultiplier = 0x1000000 // constant to convert logarithms to fixed point format
  37. lazyQueueRefresh = time.Second * 10 // refresh period of the connected queue
  38. persistCumulativeTimeRefresh = time.Minute * 5 // refresh period of the cumulative running time persistence
  39. posBalanceCacheLimit = 8192 // the maximum number of cached items in positive balance queue
  40. negBalanceCacheLimit = 8192 // the maximum number of cached items in negative balance queue
  41. // connectedBias is applied to already connected clients So that
  42. // already connected client won't be kicked out very soon and we
  43. // can ensure all connected clients can have enough time to request
  44. // or sync some data.
  45. //
  46. // todo(rjl493456442) make it configurable. It can be the option of
  47. // free trial time!
  48. connectedBias = time.Minute * 3
  49. )
  50. // clientPool implements a client database that assigns a priority to each client
  51. // based on a positive and negative balance. Positive balance is externally assigned
  52. // to prioritized clients and is decreased with connection time and processed
  53. // requests (unless the price factors are zero). If the positive balance is zero
  54. // then negative balance is accumulated.
  55. //
  56. // Balance tracking and priority calculation for connected clients is done by
  57. // balanceTracker. connectedQueue ensures that clients with the lowest positive or
  58. // highest negative balance get evicted when the total capacity allowance is full
  59. // and new clients with a better balance want to connect.
  60. //
  61. // Already connected nodes receive a small bias in their favor in order to avoid
  62. // accepting and instantly kicking out clients. In theory, we try to ensure that
  63. // each client can have several minutes of connection time.
  64. //
  65. // Balances of disconnected clients are stored in nodeDB including positive balance
  66. // and negative banalce. Negative balance is transformed into a logarithmic form
  67. // with a constantly shifting linear offset in order to implement an exponential
  68. // decrease. Besides nodeDB will have a background thread to check the negative
  69. // balance of disconnected client. If the balance is low enough, then the record
  70. // will be dropped.
  71. type clientPool struct {
  72. ndb *nodeDB
  73. lock sync.Mutex
  74. clock mclock.Clock
  75. stopCh chan struct{}
  76. closed bool
  77. removePeer func(enode.ID)
  78. connectedMap map[enode.ID]*clientInfo
  79. connectedQueue *prque.LazyQueue
  80. defaultPosFactors, defaultNegFactors priceFactors
  81. connLimit int // The maximum number of connections that clientpool can support
  82. capLimit uint64 // The maximum cumulative capacity that clientpool can support
  83. connectedCap uint64 // The sum of the capacity of the current clientpool connected
  84. priorityConnected uint64 // The sum of the capacity of currently connected priority clients
  85. freeClientCap uint64 // The capacity value of each free client
  86. startTime mclock.AbsTime // The timestamp at which the clientpool started running
  87. cumulativeTime int64 // The cumulative running time of clientpool at the start point.
  88. disableBias bool // Disable connection bias(used in testing)
  89. }
  90. // clientPoolPeer represents a client peer in the pool.
  91. // Positive balances are assigned to node key while negative balances are assigned
  92. // to freeClientId. Currently network IP address without port is used because
  93. // clients have a limited access to IP addresses while new node keys can be easily
  94. // generated so it would be useless to assign a negative value to them.
  95. type clientPoolPeer interface {
  96. ID() enode.ID
  97. freeClientId() string
  98. updateCapacity(uint64)
  99. freezeClient()
  100. }
  101. // clientInfo represents a connected client
  102. type clientInfo struct {
  103. address string
  104. id enode.ID
  105. connectedAt mclock.AbsTime
  106. capacity uint64
  107. priority bool
  108. pool *clientPool
  109. peer clientPoolPeer
  110. queueIndex int // position in connectedQueue
  111. balanceTracker balanceTracker
  112. posFactors, negFactors priceFactors
  113. balanceMetaInfo string
  114. }
  115. // connSetIndex callback updates clientInfo item index in connectedQueue
  116. func connSetIndex(a interface{}, index int) {
  117. a.(*clientInfo).queueIndex = index
  118. }
  119. // connPriority callback returns actual priority of clientInfo item in connectedQueue
  120. func connPriority(a interface{}, now mclock.AbsTime) int64 {
  121. c := a.(*clientInfo)
  122. return c.balanceTracker.getPriority(now)
  123. }
  124. // connMaxPriority callback returns estimated maximum priority of clientInfo item in connectedQueue
  125. func connMaxPriority(a interface{}, until mclock.AbsTime) int64 {
  126. c := a.(*clientInfo)
  127. pri := c.balanceTracker.estimatedPriority(until, true)
  128. c.balanceTracker.addCallback(balanceCallbackQueue, pri+1, func() {
  129. c.pool.lock.Lock()
  130. if c.queueIndex != -1 {
  131. c.pool.connectedQueue.Update(c.queueIndex)
  132. }
  133. c.pool.lock.Unlock()
  134. })
  135. return pri
  136. }
  137. // priceFactors determine the pricing policy (may apply either to positive or
  138. // negative balances which may have different factors).
  139. // - timeFactor is cost unit per nanosecond of connection time
  140. // - capacityFactor is cost unit per nanosecond of connection time per 1000000 capacity
  141. // - requestFactor is cost unit per request "realCost" unit
  142. type priceFactors struct {
  143. timeFactor, capacityFactor, requestFactor float64
  144. }
  145. // newClientPool creates a new client pool
  146. func newClientPool(db ethdb.Database, freeClientCap uint64, clock mclock.Clock, removePeer func(enode.ID)) *clientPool {
  147. ndb := newNodeDB(db, clock)
  148. pool := &clientPool{
  149. ndb: ndb,
  150. clock: clock,
  151. connectedMap: make(map[enode.ID]*clientInfo),
  152. connectedQueue: prque.NewLazyQueue(connSetIndex, connPriority, connMaxPriority, clock, lazyQueueRefresh),
  153. freeClientCap: freeClientCap,
  154. removePeer: removePeer,
  155. startTime: clock.Now(),
  156. cumulativeTime: ndb.getCumulativeTime(),
  157. stopCh: make(chan struct{}),
  158. }
  159. // If the negative balance of free client is even lower than 1,
  160. // delete this entry.
  161. ndb.nbEvictCallBack = func(now mclock.AbsTime, b negBalance) bool {
  162. balance := math.Exp(float64(b.logValue-pool.logOffset(now)) / fixedPointMultiplier)
  163. return balance <= 1
  164. }
  165. go func() {
  166. for {
  167. select {
  168. case <-clock.After(lazyQueueRefresh):
  169. pool.lock.Lock()
  170. pool.connectedQueue.Refresh()
  171. pool.lock.Unlock()
  172. case <-clock.After(persistCumulativeTimeRefresh):
  173. pool.ndb.setCumulativeTime(pool.logOffset(clock.Now()))
  174. case <-pool.stopCh:
  175. return
  176. }
  177. }
  178. }()
  179. return pool
  180. }
  181. // stop shuts the client pool down
  182. func (f *clientPool) stop() {
  183. close(f.stopCh)
  184. f.lock.Lock()
  185. f.closed = true
  186. f.lock.Unlock()
  187. f.ndb.setCumulativeTime(f.logOffset(f.clock.Now()))
  188. f.ndb.close()
  189. }
  190. // connect should be called after a successful handshake. If the connection was
  191. // rejected, there is no need to call disconnect.
  192. func (f *clientPool) connect(peer clientPoolPeer, capacity uint64) bool {
  193. f.lock.Lock()
  194. defer f.lock.Unlock()
  195. // Short circuit if clientPool is already closed.
  196. if f.closed {
  197. return false
  198. }
  199. // Dedup connected peers.
  200. id, freeID := peer.ID(), peer.freeClientId()
  201. if _, ok := f.connectedMap[id]; ok {
  202. clientRejectedMeter.Mark(1)
  203. log.Debug("Client already connected", "address", freeID, "id", peerIdToString(id))
  204. return false
  205. }
  206. // Create a clientInfo but do not add it yet
  207. var (
  208. posBalance uint64
  209. negBalance uint64
  210. now = f.clock.Now()
  211. )
  212. pb := f.ndb.getOrNewPB(id)
  213. posBalance = pb.value
  214. nb := f.ndb.getOrNewNB(freeID)
  215. if nb.logValue != 0 {
  216. negBalance = uint64(math.Exp(float64(nb.logValue-f.logOffset(now))/fixedPointMultiplier) * float64(time.Second))
  217. }
  218. e := &clientInfo{
  219. pool: f,
  220. peer: peer,
  221. address: freeID,
  222. queueIndex: -1,
  223. id: id,
  224. connectedAt: now,
  225. priority: posBalance != 0,
  226. posFactors: f.defaultPosFactors,
  227. negFactors: f.defaultNegFactors,
  228. balanceMetaInfo: pb.meta,
  229. }
  230. // If the client is a free client, assign with a low free capacity,
  231. // Otherwise assign with the given value(priority client)
  232. if !e.priority || capacity == 0 {
  233. capacity = f.freeClientCap
  234. }
  235. e.capacity = capacity
  236. // Starts a balance tracker
  237. e.balanceTracker.init(f.clock, capacity)
  238. e.balanceTracker.setBalance(posBalance, negBalance)
  239. e.updatePriceFactors()
  240. // If the number of clients already connected in the clientpool exceeds its
  241. // capacity, evict some clients with lowest priority.
  242. //
  243. // If the priority of the newly added client is lower than the priority of
  244. // all connected clients, the client is rejected.
  245. newCapacity := f.connectedCap + capacity
  246. newCount := f.connectedQueue.Size() + 1
  247. if newCapacity > f.capLimit || newCount > f.connLimit {
  248. var (
  249. kickList []*clientInfo
  250. kickPriority int64
  251. )
  252. f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool {
  253. c := data.(*clientInfo)
  254. kickList = append(kickList, c)
  255. kickPriority = priority
  256. newCapacity -= c.capacity
  257. newCount--
  258. return newCapacity > f.capLimit || newCount > f.connLimit
  259. })
  260. bias := connectedBias
  261. if f.disableBias {
  262. bias = 0
  263. }
  264. if newCapacity > f.capLimit || newCount > f.connLimit || (e.balanceTracker.estimatedPriority(now+mclock.AbsTime(bias), false)-kickPriority) > 0 {
  265. for _, c := range kickList {
  266. f.connectedQueue.Push(c)
  267. }
  268. clientRejectedMeter.Mark(1)
  269. log.Debug("Client rejected", "address", freeID, "id", peerIdToString(id))
  270. return false
  271. }
  272. // accept new client, drop old ones
  273. for _, c := range kickList {
  274. f.dropClient(c, now, true)
  275. }
  276. }
  277. // Register new client to connection queue.
  278. f.connectedMap[id] = e
  279. f.connectedQueue.Push(e)
  280. f.connectedCap += e.capacity
  281. // If the current client is a paid client, monitor the status of client,
  282. // downgrade it to normal client if positive balance is used up.
  283. if e.priority {
  284. f.priorityConnected += capacity
  285. e.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) })
  286. }
  287. // If the capacity of client is not the default value(free capacity), notify
  288. // it to update capacity.
  289. if e.capacity != f.freeClientCap {
  290. e.peer.updateCapacity(e.capacity)
  291. }
  292. totalConnectedGauge.Update(int64(f.connectedCap))
  293. clientConnectedMeter.Mark(1)
  294. log.Debug("Client accepted", "address", freeID)
  295. return true
  296. }
  297. // disconnect should be called when a connection is terminated. If the disconnection
  298. // was initiated by the pool itself using disconnectFn then calling disconnect is
  299. // not necessary but permitted.
  300. func (f *clientPool) disconnect(p clientPoolPeer) {
  301. f.lock.Lock()
  302. defer f.lock.Unlock()
  303. // Short circuit if client pool is already closed.
  304. if f.closed {
  305. return
  306. }
  307. // Short circuit if the peer hasn't been registered.
  308. e := f.connectedMap[p.ID()]
  309. if e == nil {
  310. log.Debug("Client not connected", "address", p.freeClientId(), "id", peerIdToString(p.ID()))
  311. return
  312. }
  313. f.dropClient(e, f.clock.Now(), false)
  314. }
  315. // forClients iterates through a list of clients, calling the callback for each one.
  316. // If a client is not connected then clientInfo is nil. If the specified list is empty
  317. // then the callback is called for all connected clients.
  318. func (f *clientPool) forClients(ids []enode.ID, callback func(*clientInfo, enode.ID) error) error {
  319. f.lock.Lock()
  320. defer f.lock.Unlock()
  321. if len(ids) > 0 {
  322. for _, id := range ids {
  323. if err := callback(f.connectedMap[id], id); err != nil {
  324. return err
  325. }
  326. }
  327. } else {
  328. for _, c := range f.connectedMap {
  329. if err := callback(c, c.id); err != nil {
  330. return err
  331. }
  332. }
  333. }
  334. return nil
  335. }
  336. // setDefaultFactors sets the default price factors applied to subsequently connected clients
  337. func (f *clientPool) setDefaultFactors(posFactors, negFactors priceFactors) {
  338. f.lock.Lock()
  339. defer f.lock.Unlock()
  340. f.defaultPosFactors = posFactors
  341. f.defaultNegFactors = negFactors
  342. }
  343. // dropClient removes a client from the connected queue and finalizes its balance.
  344. // If kick is true then it also initiates the disconnection.
  345. func (f *clientPool) dropClient(e *clientInfo, now mclock.AbsTime, kick bool) {
  346. if _, ok := f.connectedMap[e.id]; !ok {
  347. return
  348. }
  349. f.finalizeBalance(e, now)
  350. f.connectedQueue.Remove(e.queueIndex)
  351. delete(f.connectedMap, e.id)
  352. f.connectedCap -= e.capacity
  353. if e.priority {
  354. f.priorityConnected -= e.capacity
  355. }
  356. totalConnectedGauge.Update(int64(f.connectedCap))
  357. if kick {
  358. clientKickedMeter.Mark(1)
  359. log.Debug("Client kicked out", "address", e.address)
  360. f.removePeer(e.id)
  361. } else {
  362. clientDisconnectedMeter.Mark(1)
  363. log.Debug("Client disconnected", "address", e.address)
  364. }
  365. }
  366. // capacityInfo returns the total capacity allowance, the total capacity of connected
  367. // clients and the total capacity of connected and prioritized clients
  368. func (f *clientPool) capacityInfo() (uint64, uint64, uint64) {
  369. f.lock.Lock()
  370. defer f.lock.Unlock()
  371. return f.capLimit, f.connectedCap, f.priorityConnected
  372. }
  373. // finalizeBalance stops the balance tracker, retrieves the final balances and
  374. // stores them in posBalanceQueue and negBalanceQueue
  375. func (f *clientPool) finalizeBalance(c *clientInfo, now mclock.AbsTime) {
  376. c.balanceTracker.stop(now)
  377. pos, neg := c.balanceTracker.getBalance(now)
  378. pb, nb := f.ndb.getOrNewPB(c.id), f.ndb.getOrNewNB(c.address)
  379. pb.value = pos
  380. f.ndb.setPB(c.id, pb)
  381. neg /= uint64(time.Second) // Convert the expanse to second level.
  382. if neg > 1 {
  383. nb.logValue = int64(math.Log(float64(neg))*fixedPointMultiplier) + f.logOffset(now)
  384. f.ndb.setNB(c.address, nb)
  385. } else {
  386. f.ndb.delNB(c.address) // Negative balance is small enough, drop it directly.
  387. }
  388. }
  389. // balanceExhausted callback is called by balanceTracker when positive balance is exhausted.
  390. // It revokes priority status and also reduces the client capacity if necessary.
  391. func (f *clientPool) balanceExhausted(id enode.ID) {
  392. f.lock.Lock()
  393. defer f.lock.Unlock()
  394. c := f.connectedMap[id]
  395. if c == nil || !c.priority {
  396. return
  397. }
  398. if c.priority {
  399. f.priorityConnected -= c.capacity
  400. }
  401. c.priority = false
  402. if c.capacity != f.freeClientCap {
  403. f.connectedCap += f.freeClientCap - c.capacity
  404. totalConnectedGauge.Update(int64(f.connectedCap))
  405. c.capacity = f.freeClientCap
  406. c.balanceTracker.setCapacity(c.capacity)
  407. c.peer.updateCapacity(c.capacity)
  408. }
  409. pb := f.ndb.getOrNewPB(id)
  410. pb.value = 0
  411. f.ndb.setPB(id, pb)
  412. }
  413. // setConnLimit sets the maximum number and total capacity of connected clients,
  414. // dropping some of them if necessary.
  415. func (f *clientPool) setLimits(totalConn int, totalCap uint64) {
  416. f.lock.Lock()
  417. defer f.lock.Unlock()
  418. f.connLimit = totalConn
  419. f.capLimit = totalCap
  420. if f.connectedCap > f.capLimit || f.connectedQueue.Size() > f.connLimit {
  421. f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool {
  422. f.dropClient(data.(*clientInfo), mclock.Now(), true)
  423. return f.connectedCap > f.capLimit || f.connectedQueue.Size() > f.connLimit
  424. })
  425. }
  426. }
  427. // setCapacity sets the assigned capacity of a connected client
  428. func (f *clientPool) setCapacity(c *clientInfo, capacity uint64) error {
  429. if f.connectedMap[c.id] != c {
  430. return fmt.Errorf("client %064x is not connected", c.id[:])
  431. }
  432. if c.capacity == capacity {
  433. return nil
  434. }
  435. if !c.priority {
  436. return errNoPriority
  437. }
  438. oldCapacity := c.capacity
  439. c.capacity = capacity
  440. f.connectedCap += capacity - oldCapacity
  441. c.balanceTracker.setCapacity(capacity)
  442. f.connectedQueue.Update(c.queueIndex)
  443. if f.connectedCap > f.capLimit {
  444. var kickList []*clientInfo
  445. kick := true
  446. f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool {
  447. client := data.(*clientInfo)
  448. kickList = append(kickList, client)
  449. f.connectedCap -= client.capacity
  450. if client == c {
  451. kick = false
  452. }
  453. return kick && (f.connectedCap > f.capLimit)
  454. })
  455. if kick {
  456. now := mclock.Now()
  457. for _, c := range kickList {
  458. f.dropClient(c, now, true)
  459. }
  460. } else {
  461. c.capacity = oldCapacity
  462. c.balanceTracker.setCapacity(oldCapacity)
  463. for _, c := range kickList {
  464. f.connectedCap += c.capacity
  465. f.connectedQueue.Push(c)
  466. }
  467. return errNoPriority
  468. }
  469. }
  470. totalConnectedGauge.Update(int64(f.connectedCap))
  471. f.priorityConnected += capacity - oldCapacity
  472. c.updatePriceFactors()
  473. c.peer.updateCapacity(c.capacity)
  474. return nil
  475. }
  476. // requestCost feeds request cost after serving a request from the given peer.
  477. func (f *clientPool) requestCost(p *clientPeer, cost uint64) {
  478. f.lock.Lock()
  479. defer f.lock.Unlock()
  480. info, exist := f.connectedMap[p.ID()]
  481. if !exist || f.closed {
  482. return
  483. }
  484. info.balanceTracker.requestCost(cost)
  485. }
  486. // logOffset calculates the time-dependent offset for the logarithmic
  487. // representation of negative balance
  488. //
  489. // From another point of view, the result returned by the function represents
  490. // the total time that the clientpool is cumulatively running(total_hours/multiplier).
  491. func (f *clientPool) logOffset(now mclock.AbsTime) int64 {
  492. // Note: fixedPointMultiplier acts as a multiplier here; the reason for dividing the divisor
  493. // is to avoid int64 overflow. We assume that int64(negBalanceExpTC) >> fixedPointMultiplier.
  494. cumulativeTime := int64((time.Duration(now - f.startTime)) / (negBalanceExpTC / fixedPointMultiplier))
  495. return f.cumulativeTime + cumulativeTime
  496. }
  497. // setClientPriceFactors sets the pricing factors for an individual connected client
  498. func (c *clientInfo) updatePriceFactors() {
  499. c.balanceTracker.setFactors(true, c.negFactors.timeFactor+float64(c.capacity)*c.negFactors.capacityFactor/1000000, c.negFactors.requestFactor)
  500. c.balanceTracker.setFactors(false, c.posFactors.timeFactor+float64(c.capacity)*c.posFactors.capacityFactor/1000000, c.posFactors.requestFactor)
  501. }
  502. // getPosBalance retrieves a single positive balance entry from cache or the database
  503. func (f *clientPool) getPosBalance(id enode.ID) posBalance {
  504. f.lock.Lock()
  505. defer f.lock.Unlock()
  506. return f.ndb.getOrNewPB(id)
  507. }
  508. // addBalance updates the balance of a client (either overwrites it or adds to it).
  509. // It also updates the balance meta info string.
  510. func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64, uint64, error) {
  511. f.lock.Lock()
  512. defer f.lock.Unlock()
  513. pb := f.ndb.getOrNewPB(id)
  514. var negBalance uint64
  515. c := f.connectedMap[id]
  516. if c != nil {
  517. pb.value, negBalance = c.balanceTracker.getBalance(f.clock.Now())
  518. }
  519. oldBalance := pb.value
  520. if amount > 0 {
  521. if amount > maxBalance || pb.value > maxBalance-uint64(amount) {
  522. return oldBalance, oldBalance, errBalanceOverflow
  523. }
  524. pb.value += uint64(amount)
  525. } else {
  526. if uint64(-amount) > pb.value {
  527. pb.value = 0
  528. } else {
  529. pb.value -= uint64(-amount)
  530. }
  531. }
  532. pb.meta = meta
  533. f.ndb.setPB(id, pb)
  534. if c != nil {
  535. c.balanceTracker.setBalance(pb.value, negBalance)
  536. if !c.priority && pb.value > 0 {
  537. // The capacity should be adjusted based on the requirement,
  538. // but we have no idea about the new capacity, need a second
  539. // call to udpate it.
  540. c.priority = true
  541. f.priorityConnected += c.capacity
  542. c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) })
  543. }
  544. // if balance is set to zero then reverting to non-priority status
  545. // is handled by the balanceExhausted callback
  546. c.balanceMetaInfo = meta
  547. }
  548. return oldBalance, pb.value, nil
  549. }
  550. // posBalance represents a recently accessed positive balance entry
  551. type posBalance struct {
  552. value uint64
  553. meta string
  554. }
  555. // EncodeRLP implements rlp.Encoder
  556. func (e *posBalance) EncodeRLP(w io.Writer) error {
  557. return rlp.Encode(w, []interface{}{e.value, e.meta})
  558. }
  559. // DecodeRLP implements rlp.Decoder
  560. func (e *posBalance) DecodeRLP(s *rlp.Stream) error {
  561. var entry struct {
  562. Value uint64
  563. Meta string
  564. }
  565. if err := s.Decode(&entry); err != nil {
  566. return err
  567. }
  568. e.value = entry.Value
  569. e.meta = entry.Meta
  570. return nil
  571. }
  572. // negBalance represents a negative balance entry of a disconnected client
  573. type negBalance struct{ logValue int64 }
  574. // EncodeRLP implements rlp.Encoder
  575. func (e *negBalance) EncodeRLP(w io.Writer) error {
  576. return rlp.Encode(w, []interface{}{uint64(e.logValue)})
  577. }
  578. // DecodeRLP implements rlp.Decoder
  579. func (e *negBalance) DecodeRLP(s *rlp.Stream) error {
  580. var entry struct {
  581. LogValue uint64
  582. }
  583. if err := s.Decode(&entry); err != nil {
  584. return err
  585. }
  586. e.logValue = int64(entry.LogValue)
  587. return nil
  588. }
  589. const (
  590. // nodeDBVersion is the version identifier of the node data in db
  591. //
  592. // Changelog:
  593. // * Replace `lastTotal` with `meta` in positive balance: version 0=>1
  594. nodeDBVersion = 1
  595. // dbCleanupCycle is the cycle of db for useless data cleanup
  596. dbCleanupCycle = time.Hour
  597. )
  598. var (
  599. positiveBalancePrefix = []byte("pb:") // dbVersion(uint16 big endian) + positiveBalancePrefix + id -> balance
  600. negativeBalancePrefix = []byte("nb:") // dbVersion(uint16 big endian) + negativeBalancePrefix + ip -> balance
  601. cumulativeRunningTimeKey = []byte("cumulativeTime:") // dbVersion(uint16 big endian) + cumulativeRunningTimeKey -> cumulativeTime
  602. )
  603. type nodeDB struct {
  604. db ethdb.Database
  605. pcache *lru.Cache
  606. ncache *lru.Cache
  607. auxbuf []byte // 37-byte auxiliary buffer for key encoding
  608. verbuf [2]byte // 2-byte auxiliary buffer for db version
  609. nbEvictCallBack func(mclock.AbsTime, negBalance) bool // Callback to determine whether the negative balance can be evicted.
  610. clock mclock.Clock
  611. closeCh chan struct{}
  612. cleanupHook func() // Test hook used for testing
  613. }
  614. func newNodeDB(db ethdb.Database, clock mclock.Clock) *nodeDB {
  615. pcache, _ := lru.New(posBalanceCacheLimit)
  616. ncache, _ := lru.New(negBalanceCacheLimit)
  617. ndb := &nodeDB{
  618. db: db,
  619. pcache: pcache,
  620. ncache: ncache,
  621. auxbuf: make([]byte, 37),
  622. clock: clock,
  623. closeCh: make(chan struct{}),
  624. }
  625. binary.BigEndian.PutUint16(ndb.verbuf[:], uint16(nodeDBVersion))
  626. go ndb.expirer()
  627. return ndb
  628. }
  629. func (db *nodeDB) close() {
  630. close(db.closeCh)
  631. }
  632. func (db *nodeDB) key(id []byte, neg bool) []byte {
  633. prefix := positiveBalancePrefix
  634. if neg {
  635. prefix = negativeBalancePrefix
  636. }
  637. if len(prefix)+len(db.verbuf)+len(id) > len(db.auxbuf) {
  638. db.auxbuf = append(db.auxbuf, make([]byte, len(prefix)+len(db.verbuf)+len(id)-len(db.auxbuf))...)
  639. }
  640. copy(db.auxbuf[:len(db.verbuf)], db.verbuf[:])
  641. copy(db.auxbuf[len(db.verbuf):len(db.verbuf)+len(prefix)], prefix)
  642. copy(db.auxbuf[len(prefix)+len(db.verbuf):len(prefix)+len(db.verbuf)+len(id)], id)
  643. return db.auxbuf[:len(prefix)+len(db.verbuf)+len(id)]
  644. }
  645. func (db *nodeDB) getCumulativeTime() int64 {
  646. blob, err := db.db.Get(append(cumulativeRunningTimeKey, db.verbuf[:]...))
  647. if err != nil || len(blob) == 0 {
  648. return 0
  649. }
  650. return int64(binary.BigEndian.Uint64(blob))
  651. }
  652. func (db *nodeDB) setCumulativeTime(v int64) {
  653. binary.BigEndian.PutUint64(db.auxbuf[:8], uint64(v))
  654. db.db.Put(append(cumulativeRunningTimeKey, db.verbuf[:]...), db.auxbuf[:8])
  655. }
  656. func (db *nodeDB) getOrNewPB(id enode.ID) posBalance {
  657. key := db.key(id.Bytes(), false)
  658. item, exist := db.pcache.Get(string(key))
  659. if exist {
  660. return item.(posBalance)
  661. }
  662. var balance posBalance
  663. if enc, err := db.db.Get(key); err == nil {
  664. if err := rlp.DecodeBytes(enc, &balance); err != nil {
  665. log.Error("Failed to decode positive balance", "err", err)
  666. }
  667. }
  668. db.pcache.Add(string(key), balance)
  669. return balance
  670. }
  671. func (db *nodeDB) setPB(id enode.ID, b posBalance) {
  672. if b.value == 0 && len(b.meta) == 0 {
  673. db.delPB(id)
  674. return
  675. }
  676. key := db.key(id.Bytes(), false)
  677. enc, err := rlp.EncodeToBytes(&(b))
  678. if err != nil {
  679. log.Error("Failed to encode positive balance", "err", err)
  680. return
  681. }
  682. db.db.Put(key, enc)
  683. db.pcache.Add(string(key), b)
  684. }
  685. func (db *nodeDB) delPB(id enode.ID) {
  686. key := db.key(id.Bytes(), false)
  687. db.db.Delete(key)
  688. db.pcache.Remove(string(key))
  689. }
  690. // getPosBalanceIDs returns a lexicographically ordered list of IDs of accounts
  691. // with a positive balance
  692. func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID) {
  693. if maxCount <= 0 {
  694. return
  695. }
  696. it := db.db.NewIteratorWithStart(db.key(start.Bytes(), false))
  697. defer it.Release()
  698. for i := len(stop[:]) - 1; i >= 0; i-- {
  699. stop[i]--
  700. if stop[i] != 255 {
  701. break
  702. }
  703. }
  704. stopKey := db.key(stop.Bytes(), false)
  705. keyLen := len(stopKey)
  706. for it.Next() {
  707. var id enode.ID
  708. if len(it.Key()) != keyLen || bytes.Compare(it.Key(), stopKey) == 1 {
  709. return
  710. }
  711. copy(id[:], it.Key()[keyLen-len(id):])
  712. result = append(result, id)
  713. if len(result) == maxCount {
  714. return
  715. }
  716. }
  717. return
  718. }
  719. func (db *nodeDB) getOrNewNB(id string) negBalance {
  720. key := db.key([]byte(id), true)
  721. item, exist := db.ncache.Get(string(key))
  722. if exist {
  723. return item.(negBalance)
  724. }
  725. var balance negBalance
  726. if enc, err := db.db.Get(key); err == nil {
  727. if err := rlp.DecodeBytes(enc, &balance); err != nil {
  728. log.Error("Failed to decode negative balance", "err", err)
  729. }
  730. }
  731. db.ncache.Add(string(key), balance)
  732. return balance
  733. }
  734. func (db *nodeDB) setNB(id string, b negBalance) {
  735. key := db.key([]byte(id), true)
  736. enc, err := rlp.EncodeToBytes(&(b))
  737. if err != nil {
  738. log.Error("Failed to encode negative balance", "err", err)
  739. return
  740. }
  741. db.db.Put(key, enc)
  742. db.ncache.Add(string(key), b)
  743. }
  744. func (db *nodeDB) delNB(id string) {
  745. key := db.key([]byte(id), true)
  746. db.db.Delete(key)
  747. db.ncache.Remove(string(key))
  748. }
  749. func (db *nodeDB) expirer() {
  750. for {
  751. select {
  752. case <-db.clock.After(dbCleanupCycle):
  753. db.expireNodes()
  754. case <-db.closeCh:
  755. return
  756. }
  757. }
  758. }
  759. // expireNodes iterates the whole node db and checks whether the negative balance
  760. // entry can deleted.
  761. //
  762. // The rationale behind this is: server doesn't need to keep the negative balance
  763. // records if they are low enough.
  764. func (db *nodeDB) expireNodes() {
  765. var (
  766. visited int
  767. deleted int
  768. start = time.Now()
  769. )
  770. iter := db.db.NewIteratorWithPrefix(append(db.verbuf[:], negativeBalancePrefix...))
  771. for iter.Next() {
  772. visited += 1
  773. var balance negBalance
  774. if err := rlp.DecodeBytes(iter.Value(), &balance); err != nil {
  775. log.Error("Failed to decode negative balance", "err", err)
  776. continue
  777. }
  778. if db.nbEvictCallBack != nil && db.nbEvictCallBack(db.clock.Now(), balance) {
  779. deleted += 1
  780. db.db.Delete(iter.Key())
  781. }
  782. }
  783. // Invoke testing hook if it's not nil.
  784. if db.cleanupHook != nil {
  785. db.cleanupHook()
  786. }
  787. log.Debug("Expire nodes", "visited", visited, "deleted", deleted, "elapsed", common.PrettyDuration(time.Since(start)))
  788. }