les: implement flow control parameter update on client side

les/flowcontrol/control.go

@@ -31,7 +31,7 @@ type ServerParams struct {
 }
 
 type ClientNode struct {
-	params   *ServerParams
+	params   ServerParams
 	bufValue uint64
 	lastTime mclock.AbsTime
 	lock     sync.Mutex
@@ -39,7 +39,7 @@ type ClientNode struct {
 	cmNode   *cmNode
 }
 
-func NewClientNode(cm *ClientManager, params *ServerParams) *ClientNode {
+func NewClientNode(cm *ClientManager, params ServerParams) *ClientNode {
 	node := &ClientNode{
 		cm:       cm,
 		params:   params,
@@ -96,13 +96,13 @@ func (peer *ClientNode) RequestProcessed(cost uint64) (bv, realCost uint64) {
 type ServerNode struct {
 	bufEstimate uint64
 	lastTime    mclock.AbsTime
-	params      *ServerParams
+	params      ServerParams
 	sumCost     uint64            // sum of req costs sent to this server
 	pending     map[uint64]uint64 // value = sumCost after sending the given req
 	lock        sync.RWMutex
 }
 
-func NewServerNode(params *ServerParams) *ServerNode {
+func NewServerNode(params ServerParams) *ServerNode {
 	return &ServerNode{
 		bufEstimate: params.BufLimit,
 		lastTime:    mclock.Now(),
@@ -111,6 +111,22 @@ func NewServerNode(params *ServerParams) *ServerNode {
 	}
 }
 
+// UpdateParams updates flow control parameters
+func (peer *ServerNode) UpdateParams(params ServerParams) {
+	peer.lock.Lock()
+	defer peer.lock.Unlock()
+
+	peer.recalcBLE(mclock.Now())
+	if params.BufLimit > peer.params.BufLimit {
+		peer.bufEstimate += params.BufLimit - peer.params.BufLimit
+	} else {
+		if peer.bufEstimate > params.BufLimit {
+			peer.bufEstimate = params.BufLimit
+		}
+	}
+	peer.params = params
+}
+
 func (peer *ServerNode) recalcBLE(time mclock.AbsTime) {
 	dt := uint64(time - peer.lastTime)
 	if time < peer.lastTime {

les/handler.go

@@ -380,17 +380,25 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrDecode, "%v: %v", msg, err)
 		}
 
-		if p.requestAnnounceType == announceTypeSigned {
+		update, size := req.Update.decode()
-			if err := req.checkSignature(p.ID()); err != nil {
+		if p.rejectUpdate(size) {
-				p.Log().Trace("Invalid announcement signature", "err", err)
+			return errResp(ErrRequestRejected, "")
-				return err
-			}
-			p.Log().Trace("Valid announcement signature")
 		}
+		p.updateFlowControl(update)
 
-		p.Log().Trace("Announce message content", "number", req.Number, "hash", req.Hash, "td", req.Td, "reorg", req.ReorgDepth)
+		if req.Hash != (common.Hash{}) {
-		if pm.fetcher != nil {
+			if p.requestAnnounceType == announceTypeSigned {
-			pm.fetcher.announce(p, &req)
+				if err := req.checkSignature(p.ID(), update); err != nil {
+					p.Log().Trace("Invalid announcement signature", "err", err)
+					return err
+				}
+				p.Log().Trace("Valid announcement signature")
+			}
+
+			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)
+			}
 		}
 
 	case GetBlockHeadersMsg:

les/helper_test.go

@@ -184,7 +184,7 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
 		srv := &LesServer{lesCommons: lesCommons{protocolManager: pm}}
 		pm.server = srv
 
-		srv.defParams = &flowcontrol.ServerParams{
+		srv.defParams = flowcontrol.ServerParams{
 			BufLimit:    testBufLimit,
 			MinRecharge: 1,
 		}
@@ -313,7 +313,7 @@ func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNu
 		t.Fatalf("status send: %v", err)
 	}
 
-	p.fcServerParams = &flowcontrol.ServerParams{
+	p.fcServerParams = flowcontrol.ServerParams{
 		BufLimit:    testBufLimit,
 		MinRecharge: 1,
 	}

les/peer.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/les/flowcontrol"
@@ -42,6 +43,12 @@ var (
 
 const maxResponseErrors = 50 // number of invalid responses tolerated (makes the protocol less brittle but still avoids spam)
 
+// bandwidth limitation for parameter updates
+const (
+	allowedUpdateBytes = 100000                // initial/maximum allowed update size
+	allowedUpdateRate  = time.Millisecond * 10 // time constant for recharging one byte of allowance
+)
+
 const (
 	announceTypeNone = iota
 	announceTypeSimple
@@ -69,10 +76,12 @@ type peer struct {
 	poolEntry      *poolEntry
 	hasBlock       func(common.Hash, uint64, bool) bool
 	responseErrors int
+	updateCounter  uint64
+	updateTime     mclock.AbsTime
 
 	fcClient       *flowcontrol.ClientNode // nil if the peer is server only
 	fcServer       *flowcontrol.ServerNode // nil if the peer is client only
-	fcServerParams *flowcontrol.ServerParams
+	fcServerParams flowcontrol.ServerParams
 	fcCosts        requestCostTable
 }
 
@@ -89,6 +98,27 @@ func newPeer(version int, network uint64, p *p2p.Peer, rw p2p.MsgReadWriter) *pe
 	}
 }
 
+// rejectUpdate returns true if a parameter update has to be rejected because
+// the size and/or rate of updates exceed the bandwidth limitation
+func (p *peer) rejectUpdate(size uint64) bool {
+	now := mclock.Now()
+	if p.updateCounter == 0 {
+		p.updateTime = now
+	} else {
+		dt := now - p.updateTime
+		r := uint64(dt / mclock.AbsTime(allowedUpdateRate))
+		if p.updateCounter > r {
+			p.updateCounter -= r
+			p.updateTime += mclock.AbsTime(allowedUpdateRate * time.Duration(r))
+		} else {
+			p.updateCounter = 0
+			p.updateTime = now
+		}
+	}
+	p.updateCounter += size
+	return p.updateCounter > allowedUpdateBytes
+}
+
 func (p *peer) canQueue() bool {
 	return p.sendQueue.canQueue()
 }
@@ -340,12 +370,14 @@ func (l keyValueList) add(key string, val interface{}) keyValueList {
 	return append(l, entry)
 }
 
-func (l keyValueList) decode() keyValueMap {
+func (l keyValueList) decode() (keyValueMap, uint64) {
 	m := make(keyValueMap)
+	var size uint64
 	for _, entry := range l {
 		m[entry.Key] = entry.Value
+		size += uint64(len(entry.Key)) + uint64(len(entry.Value)) + 8
 	}
-	return m
+	return m, size
 }
 
 func (m keyValueMap) get(key string, val interface{}) error {
@@ -418,7 +450,10 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 	if err != nil {
 		return err
 	}
-	recv := recvList.decode()
+	recv, size := recvList.decode()
+	if p.rejectUpdate(size) {
+		return errResp(ErrRequestRejected, "")
+	}
 
 	var rGenesis, rHash common.Hash
 	var rVersion, rNetwork, rNum uint64
@@ -471,7 +506,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 		if recv.get("txRelay", nil) != nil {
 			return errResp(ErrUselessPeer, "peer cannot relay transactions")
 		}
-		params := &flowcontrol.ServerParams{}
+		var params flowcontrol.ServerParams
 		if err := recv.get("flowControl/BL", &params.BufLimit); err != nil {
 			return err
 		}
@@ -491,6 +526,30 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 	return nil
 }
 
+// updateFlowControl updates the flow control parameters belonging to the server
+// node if the announced key/value set contains relevant fields
+func (p *peer) updateFlowControl(update keyValueMap) {
+	if p.fcServer == nil {
+		return
+	}
+	params := p.fcServerParams
+	updateParams := false
+	if update.get("flowControl/BL", &params.BufLimit) == nil {
+		updateParams = true
+	}
+	if update.get("flowControl/MRR", &params.MinRecharge) == nil {
+		updateParams = true
+	}
+	if updateParams {
+		p.fcServerParams = params
+		p.fcServer.UpdateParams(params)
+	}
+	var MRC RequestCostList
+	if update.get("flowControl/MRC", &MRC) == nil {
+		p.fcCosts = MRC.decode()
+	}
+}
+
 // String implements fmt.Stringer.
 func (p *peer) String() string {
 	return fmt.Sprintf("Peer %s [%s]", p.id,

les/protocol.go

@@ -146,9 +146,9 @@ func (a *announceData) sign(privKey *ecdsa.PrivateKey) {
 }
 
 // checkSignature verifies if the block announcement has a valid signature by the given pubKey
-func (a *announceData) checkSignature(id enode.ID) error {
+func (a *announceData) checkSignature(id enode.ID, update keyValueMap) error {
 	var sig []byte
-	if err := a.Update.decode().get("sign", &sig); err != nil {
+	if err := update.get("sign", &sig); err != nil {
 		return err
 	}
 	rlp, _ := rlp.EncodeToBytes(announceBlock{a.Hash, a.Number, a.Td})

les/server.go

@@ -43,7 +43,7 @@ type LesServer struct {
 
 	fcManager   *flowcontrol.ClientManager // nil if our node is client only
 	fcCostStats *requestCostStats
-	defParams   *flowcontrol.ServerParams
+	defParams   flowcontrol.ServerParams
 	lesTopics   []discv5.Topic
 	privateKey  *ecdsa.PrivateKey
 	quitSync    chan struct{}
@@ -98,7 +98,7 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 	srv.chtIndexer.Start(eth.BlockChain())
 	pm.server = srv
 
-	srv.defParams = &flowcontrol.ServerParams{
+	srv.defParams = flowcontrol.ServerParams{
 		BufLimit:    300000000,
 		MinRecharge: 50000,
 	}