diff --git a/les/fetcher.go b/les/fetcher.go
index cc539c42bf..67685145c0 100644
--- a/les/fetcher.go
+++ b/les/fetcher.go
@@ -242,7 +242,7 @@ func (f *lightFetcher) unregisterPeer(p *peer) {
// announce processes a new announcement message received from a peer, adding new
// nodes to the peer's block tree and removing old nodes if necessary
-func (f *lightFetcher) announce(p *peer, head *announceData) {
+func (f *lightFetcher) announce(p *peer, head *announceData, acceptNonMonotonic bool) {
f.lock.Lock()
defer f.lock.Unlock()
p.Log().Debug("Received new announcement", "number", head.Number, "hash", head.Hash, "reorg", head.ReorgDepth)
@@ -254,6 +254,9 @@ func (f *lightFetcher) announce(p *peer, head *announceData) {
}
if fp.lastAnnounced != nil && head.Td.Cmp(fp.lastAnnounced.td) <= 0 {
+ if acceptNonMonotonic {
+ return
+ }
// announced tds should be strictly monotonic
p.Log().Debug("Received non-monotonic td", "current", head.Td, "previous", fp.lastAnnounced.td)
go f.pm.removePeer(p.id)
diff --git a/les/freeclient.go b/les/freeclient.go
new file mode 100644
index 0000000000..8f3c7bb61a
--- /dev/null
+++ b/les/freeclient.go
@@ -0,0 +1,279 @@
+// Copyright 2016 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 implements the Light Ethereum Subprotocol.
+package les
+
+import (
+ "io"
+ "math"
+ "sync"
+ "time"
+
+ "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/rlp"
+)
+
+// freeClientPool implements a client database that limits the connection time
+// of each client and manages accepting/rejecting incoming connections and even
+// kicking out some connected clients. The pool calculates recent usage time
+// for each known client (a value that increases linearly when the client is
+// connected and decreases exponentially when not connected). Clients with lower
+// recent usage are preferred, unknown nodes have the highest priority. Already
+// connected nodes receive a small bias in their favor in order to avoid accepting
+// and instantly kicking out clients.
+//
+// Note: the pool can use any string for client identification. Using signature
+// keys for that purpose would not make sense when being known has a negative
+// value for the client. Currently the LES protocol manager uses IP addresses
+// (without port address) to identify clients.
+type freeClientPool struct {
+ db ethdb.Database
+ lock sync.Mutex
+ clock mclock.Clock
+ closed bool
+
+ connectedLimit, totalLimit int
+
+ addressMap map[string]*freeClientPoolEntry
+ connPool, disconnPool *prque.Prque
+ startupTime mclock.AbsTime
+ logOffsetAtStartup int64
+}
+
+const (
+ recentUsageExpTC = time.Hour // time constant of the exponential weighting window for "recent" server usage
+ fixedPointMultiplier = 0x1000000 // constant to convert logarithms to fixed point format
+ connectedBias = time.Minute // this bias is applied in favor of already connected clients in order to avoid kicking them out very soon
+)
+
+// newFreeClientPool creates a new free client pool
+func newFreeClientPool(db ethdb.Database, connectedLimit, totalLimit int, quit chan struct{}, wg *sync.WaitGroup, clock mclock.Clock) *freeClientPool {
+ pool := &freeClientPool{
+ db: db,
+ clock: clock,
+ addressMap: make(map[string]*freeClientPoolEntry),
+ connPool: prque.New(poolSetIndex),
+ disconnPool: prque.New(poolSetIndex),
+ connectedLimit: connectedLimit,
+ totalLimit: totalLimit,
+ }
+ pool.loadFromDb()
+ wg.Add(1)
+ go func() {
+ <-quit
+ pool.lock.Lock()
+ pool.closed = true
+ pool.saveToDb()
+ pool.lock.Unlock()
+ wg.Done()
+ }()
+
+ return pool
+}
+
+// connect should be called after a successful handshake. If the connection was
+// rejected, there is no need to call disconnect.
+func (f *freeClientPool) connect(address string, disconnectFn func()) bool {
+ f.lock.Lock()
+ defer f.lock.Unlock()
+
+ if f.closed {
+ return false
+ }
+ e := f.addressMap[address]
+ now := f.clock.Now()
+ var recentUsage int64
+ if e == nil {
+ e = &freeClientPoolEntry{address: address, index: -1}
+ f.addressMap[address] = e
+ } else {
+ if e.connected {
+ log.Debug("Client already connected", "address", address)
+ return false
+ }
+ recentUsage = int64(math.Exp(float64(e.logUsage-f.logOffset(now)) / fixedPointMultiplier))
+ }
+ e.linUsage = recentUsage - int64(now)
+ // check whether (linUsage+connectedBias) is smaller than the highest entry in the connected pool
+ if f.connPool.Size() == f.connectedLimit {
+ i := f.connPool.PopItem().(*freeClientPoolEntry)
+ if e.linUsage+int64(connectedBias)-i.linUsage < 0 {
+ // kick it out and accept the new client
+ f.connPool.Remove(i.index)
+ f.calcLogUsage(i, now)
+ i.connected = false
+ f.disconnPool.Push(i, -i.logUsage)
+ log.Debug("Client kicked out", "address", i.address)
+ go i.disconnectFn()
+ } else {
+ // keep the old client and reject the new one
+ f.connPool.Push(i, i.linUsage)
+ log.Debug("Client rejected", "address", address)
+ return false
+ }
+ }
+ f.disconnPool.Remove(e.index)
+ e.connected = true
+ e.disconnectFn = disconnectFn
+ f.connPool.Push(e, e.linUsage)
+ if f.connPool.Size()+f.disconnPool.Size() > f.totalLimit {
+ f.disconnPool.Pop()
+ }
+ log.Debug("Client accepted", "address", address)
+ return true
+}
+
+// disconnect should be called when a connection is terminated. If the disconnection
+// was initiated by the pool itself using disconnectFn then calling disconnect is
+// not necessary but permitted.
+func (f *freeClientPool) disconnect(address string) {
+ f.lock.Lock()
+ defer f.lock.Unlock()
+
+ if f.closed {
+ return
+ }
+ e := f.addressMap[address]
+ now := f.clock.Now()
+ if !e.connected {
+ log.Debug("Client already disconnected", "address", address)
+ return
+ }
+
+ f.connPool.Remove(e.index)
+ f.calcLogUsage(e, now)
+ e.connected = false
+ f.disconnPool.Push(e, -e.logUsage)
+ log.Debug("Client disconnected", "address", address)
+}
+
+// logOffset calculates the time-dependent offset for the logarithmic
+// representation of recent usage
+func (f *freeClientPool) logOffset(now mclock.AbsTime) int64 {
+ // Note: fixedPointMultiplier acts as a multiplier here; the reason for dividing the divisor
+ // is to avoid int64 overflow. We assume that int64(recentUsageExpTC) >> fixedPointMultiplier.
+ logDecay := int64((time.Duration(now - f.startupTime)) / (recentUsageExpTC / fixedPointMultiplier))
+ return f.logOffsetAtStartup + logDecay
+}
+
+// calcLogUsage converts recent usage from linear to logarithmic representation
+// when disconnecting a peer or closing the client pool
+func (f *freeClientPool) calcLogUsage(e *freeClientPoolEntry, now mclock.AbsTime) {
+ dt := e.linUsage + int64(now)
+ if dt < 1 {
+ dt = 1
+ }
+ e.logUsage = int64(math.Log(float64(dt))*fixedPointMultiplier) + f.logOffset(now)
+}
+
+// freeClientPoolStorage is the RLP representation of the pool's database storage
+type freeClientPoolStorage struct {
+ LogOffset uint64
+ List []*freeClientPoolEntry
+}
+
+// loadFromDb restores pool status from the database storage
+// (automatically called at initialization)
+func (f *freeClientPool) loadFromDb() {
+ enc, err := f.db.Get([]byte("freeClientPool"))
+ if err != nil {
+ return
+ }
+ var storage freeClientPoolStorage
+ err = rlp.DecodeBytes(enc, &storage)
+ if err != nil {
+ log.Error("Failed to decode client list", "err", err)
+ return
+ }
+ f.logOffsetAtStartup = int64(storage.LogOffset)
+ f.startupTime = f.clock.Now()
+ for _, e := range storage.List {
+ log.Debug("Loaded free client record", "address", e.address, "logUsage", e.logUsage)
+ f.addressMap[e.address] = e
+ f.disconnPool.Push(e, -e.logUsage)
+ }
+}
+
+// saveToDb saves pool status to the database storage
+// (automatically called during shutdown)
+func (f *freeClientPool) saveToDb() {
+ now := f.clock.Now()
+ storage := freeClientPoolStorage{
+ LogOffset: uint64(f.logOffset(now)),
+ List: make([]*freeClientPoolEntry, len(f.addressMap)),
+ }
+ i := 0
+ for _, e := range f.addressMap {
+ if e.connected {
+ f.calcLogUsage(e, now)
+ }
+ storage.List[i] = e
+ i++
+ }
+ enc, err := rlp.EncodeToBytes(storage)
+ if err != nil {
+ log.Error("Failed to encode client list", "err", err)
+ } else {
+ f.db.Put([]byte("freeClientPool"), enc)
+ }
+}
+
+// freeClientPoolEntry represents a client address known by the pool.
+// When connected, recent usage is calculated as linUsage + int64(clock.Now())
+// When disconnected, it is calculated as exp(logUsage - logOffset) where logOffset
+// also grows linearly with time while the server is running.
+// Conversion between linear and logarithmic representation happens when connecting
+// or disconnecting the node.
+//
+// Note: linUsage and logUsage are values used with constantly growing offsets so
+// even though they are close to each other at any time they may wrap around int64
+// limits over time. Comparison should be performed accordingly.
+type freeClientPoolEntry struct {
+ address string
+ connected bool
+ disconnectFn func()
+ linUsage, logUsage int64
+ index int
+}
+
+func (e *freeClientPoolEntry) EncodeRLP(w io.Writer) error {
+ return rlp.Encode(w, []interface{}{e.address, uint64(e.logUsage)})
+}
+
+func (e *freeClientPoolEntry) DecodeRLP(s *rlp.Stream) error {
+ var entry struct {
+ Address string
+ LogUsage uint64
+ }
+ if err := s.Decode(&entry); err != nil {
+ return err
+ }
+ e.address = entry.Address
+ e.logUsage = int64(entry.LogUsage)
+ e.connected = false
+ e.index = -1
+ return nil
+}
+
+// poolSetIndex callback is used by both priority queues to set/update the index of
+// the element in the queue. Index is needed to remove elements other than the top one.
+func poolSetIndex(a interface{}, i int) {
+ a.(*freeClientPoolEntry).index = i
+}
diff --git a/les/freeclient_test.go b/les/freeclient_test.go
new file mode 100644
index 0000000000..f863cc0f90
--- /dev/null
+++ b/les/freeclient_test.go
@@ -0,0 +1,150 @@
+// Copyright 2017 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 light implements on-demand retrieval capable state and chain objects
+// for the Ethereum Light Client.
+package les
+
+import (
+ "fmt"
+ "math/rand"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/common/mclock"
+ "github.com/ethereum/go-ethereum/ethdb"
+)
+
+func TestFreeClientPoolL10C100(t *testing.T) {
+ testFreeClientPool(t, 10, 100)
+}
+
+func TestFreeClientPoolL40C200(t *testing.T) {
+ testFreeClientPool(t, 40, 200)
+}
+
+func TestFreeClientPoolL100C300(t *testing.T) {
+ testFreeClientPool(t, 100, 300)
+}
+
+const testFreeClientPoolTicks = 500000
+
+func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
+ quit := make(chan struct{})
+ clock := mclock.NewSimulatedClock(time.Second, 10)
+ defer clock.Stop()
+ var wg sync.WaitGroup
+ db := ethdb.NewMemDatabase()
+
+ pool := newFreeClientPool(db, connLimit, 10000, quit, &wg, clock)
+
+ connected := make([]bool, clientCount)
+ connTicks := make([]int, clientCount)
+ disconnCh := make(chan int)
+ tickCh := clock.After(time.Second)
+ tickCounter := 0
+
+ peerId := func(i int) string {
+ return fmt.Sprintf("test peer #%d", i)
+ }
+
+ disconnFn := func(i int) func() {
+ return func() {
+ disconnCh <- i
+ }
+ }
+
+ // pool should accept new peers up to its connected limit
+ for i := 0; i < connLimit; i++ {
+ if pool.connect(peerId(i), disconnFn(i)) {
+ connected[i] = true
+ } else {
+ t.Errorf("Test peer #%d rejected", i)
+ }
+ }
+ // since all accepted peers are new and should not be kicked out, the next one should be rejected
+ if pool.connect(peerId(connLimit), disconnFn(connLimit)) {
+ connected[connLimit] = true
+ t.Errorf("Peer accepted over connected limit")
+ }
+
+ // randomly connect and disconnect peers, expect to have a similar total connection time at the end
+loop:
+ for {
+ select {
+ case <-tickCh:
+ i := rand.Intn(clientCount)
+ if connected[i] {
+ pool.disconnect(peerId(i))
+ connected[i] = false
+ connTicks[i] += tickCounter
+ } else {
+ if pool.connect(peerId(i), disconnFn(i)) {
+ connected[i] = true
+ connTicks[i] -= tickCounter
+ }
+ }
+ tickCounter++
+ if tickCounter >= testFreeClientPoolTicks {
+ break loop
+ }
+ tickCh = clock.After(time.Second)
+ case i := <-disconnCh:
+ pool.disconnect(peerId(i))
+ if connected[i] {
+ connTicks[i] += tickCounter
+ connected[i] = false
+ }
+ }
+ }
+
+ expTicks := testFreeClientPoolTicks * connLimit / clientCount
+ expMin := expTicks - expTicks/10
+ expMax := expTicks + expTicks/10
+
+ // check if the total connected time of peers are all in the expected range
+ for i, c := range connected {
+ if c {
+ connTicks[i] += tickCounter
+ }
+ if connTicks[i] < expMin || connTicks[i] > expMax {
+ t.Errorf("Total connected time of test node #%d (%d) outside expected range (%d to %d)", i, connTicks[i], expMin, expMax)
+ }
+ }
+
+ // a previously unknown peer should be accepted now
+ if !pool.connect("newPeer", func() {}) {
+ t.Errorf("Previously unknown peer rejected")
+ }
+
+ // close and restart pool
+ close(quit)
+ wg.Wait()
+ quit2 := make(chan struct{})
+ var wg2 sync.WaitGroup
+ pool = newFreeClientPool(db, connLimit, 10000, quit2, &wg2, clock)
+
+ // try connecting all known peers (connLimit should be filled up)
+ for i := 0; i < clientCount; i++ {
+ pool.connect(peerId(i), func() {})
+ }
+ // expect pool to remember known nodes and kick out one of them to accept a new one
+ if !pool.connect("newPeer2", func() {}) {
+ t.Errorf("Previously unknown peer rejected after restarting pool")
+ }
+ close(quit2)
+}
diff --git a/les/handler.go b/les/handler.go
index a1c16cb875..716a3651cd 100644
--- a/les/handler.go
+++ b/les/handler.go
@@ -28,6 +28,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/consensus"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/rawdb"
@@ -104,6 +105,7 @@ type ProtocolManager struct {
odr *LesOdr
server *LesServer
serverPool *serverPool
+ clientPool *freeClientPool
lesTopic discv5.Topic
reqDist *requestDistributor
retriever *retrieveManager
@@ -226,6 +228,7 @@ func (pm *ProtocolManager) Start(maxPeers int) {
if pm.lightSync {
go pm.syncer()
} else {
+ pm.clientPool = newFreeClientPool(pm.chainDb, maxPeers, 10000, pm.quitSync, pm.wg, mclock.MonotonicClock{})
go func() {
for range pm.newPeerCh {
}
@@ -264,7 +267,8 @@ func (pm *ProtocolManager) newPeer(pv int, nv uint64, p *p2p.Peer, rw p2p.MsgRea
// this function terminates, the peer is disconnected.
func (pm *ProtocolManager) handle(p *peer) error {
// Ignore maxPeers if this is a trusted peer
- if pm.peers.Len() >= pm.maxPeers && !p.Peer.Info().Network.Trusted {
+ // In server mode we try to check into the client pool after handshake
+ if pm.lightSync && pm.peers.Len() >= pm.maxPeers && !p.Peer.Info().Network.Trusted {
return p2p.DiscTooManyPeers
}
@@ -282,6 +286,19 @@ func (pm *ProtocolManager) handle(p *peer) error {
p.Log().Debug("Light Ethereum handshake failed", "err", err)
return err
}
+
+ if !pm.lightSync && !p.Peer.Info().Network.Trusted {
+ addr, ok := p.RemoteAddr().(*net.TCPAddr)
+ // test peer address is not a tcp address, don't use client pool if can not typecast
+ if ok {
+ id := addr.IP.String()
+ if !pm.clientPool.connect(id, func() { pm.removePeer(p.id) }) {
+ return p2p.DiscTooManyPeers
+ }
+ defer pm.clientPool.disconnect(id)
+ }
+ }
+
if rw, ok := p.rw.(*meteredMsgReadWriter); ok {
rw.Init(p.version)
}
@@ -302,7 +319,10 @@ func (pm *ProtocolManager) handle(p *peer) error {
head := p.headInfo
p.lock.Unlock()
if pm.fetcher != nil {
- pm.fetcher.announce(p, head)
+ go func() {
+ time.Sleep(time.Second * 2)
+ pm.fetcher.announce(p, head, true)
+ }()
}
if p.poolEntry != nil {
@@ -399,7 +419,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
p.Log().Trace("Announce message content", "number", req.Number, "hash", req.Hash, "td", req.Td, "reorg", req.ReorgDepth)
if pm.fetcher != nil {
- pm.fetcher.announce(p, &req)
+ pm.fetcher.announce(p, &req, false)
}
case GetBlockHeadersMsg: