les: improved header fetcher and server statistics

les/handler.go

@@ -24,10 +24,8 @@ import (
 	"math/big"
 	"net"
 	"sync"
-	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -60,7 +58,7 @@ const (
 	MaxHeaderProofsFetch = 64  // Amount of merkle proofs to be fetched per retrieval request
 	MaxTxSend            = 64  // Amount of transactions to be send per request
 
-	disableClientRemovePeer = true
+	disableClientRemovePeer = false
 )
 
 // errIncompatibleConfig is returned if the requested protocols and configs are
@@ -157,44 +155,27 @@ func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, network
 			Length:  ProtocolLengths[i],
 			Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 				var entry *poolEntry
+				peer := manager.newPeer(int(version), networkId, p, rw)
 				if manager.serverPool != nil {
 					addr := p.RemoteAddr().(*net.TCPAddr)
-					entry = manager.serverPool.connect(p.ID(), addr.IP, uint16(addr.Port))
+					entry = manager.serverPool.connect(peer, addr.IP, uint16(addr.Port))
 					if entry == nil {
 						return fmt.Errorf("unwanted connection")
 					}
 				}
-				peer := manager.newPeer(int(version), networkId, p, rw)
 				peer.poolEntry = entry
 				select {
 				case manager.newPeerCh <- peer:
 					manager.wg.Add(1)
 					defer manager.wg.Done()
-					start := mclock.Now()
 					err := manager.handle(peer)
 					if entry != nil {
-						connTime := time.Duration(mclock.Now() - start)
-						stopped := false
-						select {
-						case <-manager.quitSync:
-							stopped = true
-						default:
-						}
-						//fmt.Println("connTime", peer.id, connTime, stopped, err)
-						quality := float64(1)
-						setQuality := true
-						if connTime < time.Minute*10 {
-							quality = 0
-							if stopped {
-								setQuality = false
-							}
-						}
-						manager.serverPool.disconnect(entry, quality, setQuality)
+						manager.serverPool.disconnect(entry)
 					}
 					return err
 				case <-manager.quitSync:
 					if entry != nil {
-						manager.serverPool.disconnect(entry, 0, false)
+						manager.serverPool.disconnect(entry)
 					}
 					return p2p.DiscQuitting
 				}
@@ -224,7 +205,6 @@ func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, network
 		manager.downloader = downloader.New(downloader.LightSync, chainDb, manager.eventMux, blockchain.HasHeader, nil, blockchain.GetHeaderByHash,
 			nil, blockchain.CurrentHeader, nil, nil, nil, blockchain.GetTdByHash,
 			blockchain.InsertHeaderChain, nil, nil, blockchain.Rollback, removePeer)
-		manager.fetcher = newLightFetcher(manager)
 	}
 
 	if odr != nil {
@@ -254,10 +234,12 @@ func (pm *ProtocolManager) removePeer(id string) {
 	glog.V(logger.Debug).Infof("LES: unregister peer %v", id)
 	if pm.lightSync {
 		pm.downloader.UnregisterPeer(id)
-		pm.odr.UnregisterPeer(peer)
 		if pm.txrelay != nil {
 			pm.txrelay.removePeer(id)
 		}
+		if pm.fetcher != nil {
+			pm.fetcher.removePeer(peer)
+		}
 	}
 	if err := pm.peers.Unregister(id); err != nil {
 		glog.V(logger.Error).Infoln("Removal failed:", err)
@@ -276,8 +258,10 @@ func (pm *ProtocolManager) Start(srvr *p2p.Server) {
 	lesTopic := discv5.Topic("LES@" + common.Bytes2Hex(pm.blockchain.Genesis().Hash().Bytes()[0:8]))
 	if pm.lightSync {
 		// start sync handler
-		if srvr != nil {
+		if srvr != nil { // srvr is nil during testing
 			pm.serverPool = newServerPool(pm.chainDb, []byte("serverPool/"), srvr, lesTopic, pm.quitSync, &pm.wg)
+			pm.odr.serverPool = pm.serverPool
+			pm.fetcher = newLightFetcher(pm)
 		}
 		go pm.syncer()
 	} else {
@@ -369,12 +353,17 @@ func (pm *ProtocolManager) handle(p *peer) error {
 			requestHeadersByHash, requestHeadersByNumber, nil, nil, nil); err != nil {
 			return err
 		}
-		pm.odr.RegisterPeer(p)
 		if pm.txrelay != nil {
 			pm.txrelay.addPeer(p)
 		}
 
-		pm.fetcher.notify(p, nil)
+		p.lock.Lock()
+		head := p.headInfo
+		p.lock.Unlock()
+		if pm.fetcher != nil {
+			pm.fetcher.addPeer(p)
+			pm.fetcher.announce(p, head)
+		}
 
 		if p.poolEntry != nil {
 			pm.serverPool.registered(p.poolEntry)
@@ -460,7 +449,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrDecode, "%v: %v", msg, err)
 		}
 		glog.V(logger.Detail).Infoln("AnnounceMsg:", req.Number, req.Hash, req.Td, req.ReorgDepth)
-		pm.fetcher.notify(p, &req)
+		if pm.fetcher != nil {
+			go pm.fetcher.announce(p, &req)
+		}
 
 	case GetBlockHeadersMsg:
 		glog.V(logger.Debug).Infof("<=== GetBlockHeadersMsg from peer %v", p.id)
@@ -558,7 +549,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
 		p.fcServer.GotReply(resp.ReqID, resp.BV)
-		if pm.fetcher.requestedID(resp.ReqID) {
+		if pm.fetcher != nil && pm.fetcher.requestedID(resp.ReqID) {
 			pm.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers)
 		} else {
 			err := pm.downloader.DeliverHeaders(p.id, resp.Headers)

les/helper_test.go

@@ -25,6 +25,7 @@ import (
 	"math/big"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
@@ -334,3 +335,13 @@ func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNu
 func (p *testPeer) close() {
 	p.app.Close()
 }
+
+type testServerPool peer
+
+func (p *testServerPool) selectPeer(func(*peer) (bool, uint64)) *peer {
+	return (*peer)(p)
+}
+
+func (p *testServerPool) adjustResponseTime(*poolEntry, time.Duration, bool) {
+
+}

les/odr.go

@@ -37,6 +37,11 @@ var (
 // peerDropFn is a callback type for dropping a peer detected as malicious.
 type peerDropFn func(id string)
 
+type odrPeerSelector interface {
+	selectPeer(func(*peer) (bool, uint64)) *peer
+	adjustResponseTime(*poolEntry, time.Duration, bool)
+}
+
 type LesOdr struct {
 	light.OdrBackend
 	db           ethdb.Database
@@ -44,7 +49,7 @@ type LesOdr struct {
 	removePeer   peerDropFn
 	mlock, clock sync.Mutex
 	sentReqs     map[uint64]*sentReq
-	peers        *odrPeerSet
+	serverPool   odrPeerSelector
 	lastReqID    uint64
 }
 
@@ -52,7 +57,6 @@ func NewLesOdr(db ethdb.Database) *LesOdr {
 	return &LesOdr{
 		db:       db,
 		stop:     make(chan struct{}),
-		peers:    newOdrPeerSet(),
 		sentReqs: make(map[uint64]*sentReq),
 	}
 }
@@ -77,16 +81,6 @@ type sentReq struct {
 	answered chan struct{} // closed and set to nil when any peer answers it
 }
 
-// RegisterPeer registers a new LES peer to the ODR capable peer set
-func (self *LesOdr) RegisterPeer(p *peer) error {
-	return self.peers.register(p)
-}
-
-// UnregisterPeer removes a peer from the ODR capable peer set
-func (self *LesOdr) UnregisterPeer(p *peer) {
-	self.peers.unregister(p)
-}
-
 const (
 	MsgBlockBodies = iota
 	MsgCode
@@ -142,29 +136,26 @@ func (self *LesOdr) requestPeer(req *sentReq, peer *peer, delivered, timeout cha
 
 	select {
 	case <-delivered:
-		servTime := uint64(mclock.Now() - stime)
-		self.peers.updateTimeout(peer, false)
-		self.peers.updateServTime(peer, servTime)
+		if self.serverPool != nil {
+			self.serverPool.adjustResponseTime(peer.poolEntry, time.Duration(mclock.Now()-stime), false)
+		}
 		return
 	case <-time.After(softRequestTimeout):
 		close(timeout)
-		if self.peers.updateTimeout(peer, true) {
-			self.removePeer(peer.id)
-		}
 	case <-self.stop:
 		return
 	}
 
 	select {
 	case <-delivered:
-		servTime := uint64(mclock.Now() - stime)
-		self.peers.updateServTime(peer, servTime)
-		return
 	case <-time.After(hardRequestTimeout):
-		self.removePeer(peer.id)
+		go self.removePeer(peer.id)
 	case <-self.stop:
 		return
 	}
+	if self.serverPool != nil {
+		self.serverPool.adjustResponseTime(peer.poolEntry, time.Duration(mclock.Now()-stime), true)
+	}
 }
 
 // networkRequest sends a request to known peers until an answer is received
@@ -193,7 +184,13 @@ func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) erro
 
 	exclude := make(map[*peer]struct{})
 	for {
-		if peer := self.peers.bestPeer(lreq, exclude); peer == nil {
+		var p *peer
+		if self.serverPool != nil {
+			p = self.serverPool.selectPeer(func(p *peer) (bool, uint64) {
+				return true, p.fcServer.CanSend(lreq.GetCost(p))
+			})
+		}
+		if p == nil {
 			select {
 			case <-ctx.Done():
 				return ctx.Err()
@@ -202,17 +199,17 @@ func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) erro
 			case <-time.After(retryPeers):
 			}
 		} else {
-			exclude[peer] = struct{}{}
+			exclude[p] = struct{}{}
 			delivered := make(chan struct{})
 			timeout := make(chan struct{})
 			req.lock.Lock()
-			req.sentTo[peer] = delivered
+			req.sentTo[p] = delivered
 			req.lock.Unlock()
 			reqWg.Add(1)
-			cost := lreq.GetCost(peer)
-			peer.fcServer.SendRequest(reqID, cost)
-			go self.requestPeer(req, peer, delivered, timeout, reqWg)
-			lreq.Request(reqID, peer)
+			cost := lreq.GetCost(p)
+			p.fcServer.SendRequest(reqID, cost)
+			go self.requestPeer(req, p, delivered, timeout, reqWg)
+			lreq.Request(reqID, p)
 
 			select {
 			case <-ctx.Done():

les/odr_peerset.go

-// 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 <http://www.gnu.org/licenses/>.
-
-package les
-
-import (
-	"sync"
-)
-
-const dropTimeoutRatio = 20
-
-type odrPeerInfo struct {
-	reqTimeSum, reqTimeCnt, reqCnt, timeoutCnt uint64
-}
-
-// odrPeerSet represents the collection of active peer participating in the block
-// download procedure.
-type odrPeerSet struct {
-	peers map[*peer]*odrPeerInfo
-	lock  sync.RWMutex
-}
-
-// newPeerSet creates a new peer set top track the active download sources.
-func newOdrPeerSet() *odrPeerSet {
-	return &odrPeerSet{
-		peers: make(map[*peer]*odrPeerInfo),
-	}
-}
-
-// Register injects a new peer into the working set, or returns an error if the
-// peer is already known.
-func (ps *odrPeerSet) register(p *peer) error {
-	ps.lock.Lock()
-	defer ps.lock.Unlock()
-
-	if _, ok := ps.peers[p]; ok {
-		return errAlreadyRegistered
-	}
-	ps.peers[p] = &odrPeerInfo{}
-	return nil
-}
-
-// Unregister removes a remote peer from the active set, disabling any further
-// actions to/from that particular entity.
-func (ps *odrPeerSet) unregister(p *peer) error {
-	ps.lock.Lock()
-	defer ps.lock.Unlock()
-
-	if _, ok := ps.peers[p]; !ok {
-		return errNotRegistered
-	}
-	delete(ps.peers, p)
-	return nil
-}
-
-func (ps *odrPeerSet) peerPriority(p *peer, info *odrPeerInfo, req LesOdrRequest) uint64 {
-	tm := p.fcServer.CanSend(req.GetCost(p))
-	if info.reqTimeCnt > 0 {
-		tm += info.reqTimeSum / info.reqTimeCnt
-	}
-	return tm
-}
-
-func (ps *odrPeerSet) bestPeer(req LesOdrRequest, exclude map[*peer]struct{}) *peer {
-	var best *peer
-	var bpv uint64
-	ps.lock.Lock()
-	defer ps.lock.Unlock()
-
-	for p, info := range ps.peers {
-		if _, ok := exclude[p]; !ok {
-			pv := ps.peerPriority(p, info, req)
-			if best == nil || pv < bpv {
-				best = p
-				bpv = pv
-			}
-		}
-	}
-	return best
-}
-
-func (ps *odrPeerSet) updateTimeout(p *peer, timeout bool) (drop bool) {
-	ps.lock.Lock()
-	defer ps.lock.Unlock()
-
-	if info, ok := ps.peers[p]; ok {
-		info.reqCnt++
-		if timeout {
-			// check ratio before increase to allow an extra timeout
-			if info.timeoutCnt*dropTimeoutRatio >= info.reqCnt {
-				return true
-			}
-			info.timeoutCnt++
-		}
-	}
-	return false
-}
-
-func (ps *odrPeerSet) updateServTime(p *peer, servTime uint64) {
-	ps.lock.Lock()
-	defer ps.lock.Unlock()
-
-	if info, ok := ps.peers[p]; ok {
-		info.reqTimeSum += servTime
-		info.reqTimeCnt++
-	}
-}

les/odr_test.go

@@ -160,6 +160,8 @@ func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) {
 	pm, db, odr := newTestProtocolManagerMust(t, false, 4, testChainGen)
 	lpm, ldb, odr := newTestProtocolManagerMust(t, true, 0, nil)
 	_, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm)
+	pool := (*testServerPool)(lpeer)
+	odr.serverPool = pool
 	select {
 	case <-time.After(time.Millisecond * 100):
 	case err := <-err1:
@@ -188,13 +190,13 @@ func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) {
 	}
 
 	// temporarily remove peer to test odr fails
-	odr.UnregisterPeer(lpeer)
+	odr.serverPool = nil
 	// expect retrievals to fail (except genesis block) without a les peer
 	test(expFail)
-	odr.RegisterPeer(lpeer)
+	odr.serverPool = pool
 	// expect all retrievals to pass
 	test(5)
-	odr.UnregisterPeer(lpeer)
+	odr.serverPool = nil
 	// still expect all retrievals to pass, now data should be cached locally
 	test(5)
 }

les/peer.go

@@ -51,9 +51,8 @@ type peer struct {
 
 	id string
 
-	firstHeadInfo, headInfo *announceData
-	headInfoLen             int
-	lock                    sync.RWMutex
+	headInfo *announceData
+	lock     sync.RWMutex
 
 	announceChn chan announceData
 
@@ -111,67 +110,6 @@ func (p *peer) headBlockInfo() blockInfo {
 	return blockInfo{Hash: p.headInfo.Hash, Number: p.headInfo.Number, Td: p.headInfo.Td}
 }
 
-func (p *peer) addNotify(announce *announceData) bool {
-	p.lock.Lock()
-	defer p.lock.Unlock()
-
-	if announce.Td.Cmp(p.headInfo.Td) < 1 {
-		return false
-	}
-	if p.headInfoLen >= maxHeadInfoLen {
-		//return false
-		p.firstHeadInfo = p.firstHeadInfo.next
-		p.headInfoLen--
-	}
-	if announce.haveHeaders == 0 {
-		hh := p.headInfo.Number - announce.ReorgDepth
-		if p.headInfo.haveHeaders < hh {
-			hh = p.headInfo.haveHeaders
-		}
-		announce.haveHeaders = hh
-	}
-	p.headInfo.next = announce
-	p.headInfo = announce
-	p.headInfoLen++
-	return true
-}
-
-func (p *peer) gotHeader(hash common.Hash, number uint64, td *big.Int) bool {
-	h := p.firstHeadInfo
-	ptr := 0
-	for h != nil {
-		if h.Hash == hash {
-			if h.Number != number || h.Td.Cmp(td) != 0 {
-				return false
-			}
-			h.headKnown = true
-			h.haveHeaders = h.Number
-			p.firstHeadInfo = h
-			p.headInfoLen -= ptr
-			last := h
-			h = h.next
-			// propagate haveHeaders through the chain
-			for h != nil {
-				hh := last.Number - h.ReorgDepth
-				if last.haveHeaders < hh {
-					hh = last.haveHeaders
-				}
-				if hh > h.haveHeaders {
-					h.haveHeaders = hh
-				} else {
-					return true
-				}
-				last = h
-				h = h.next
-			}
-			return true
-		}
-		h = h.next
-		ptr++
-	}
-	return true
-}
-
 // Td retrieves the current total difficulty of a peer.
 func (p *peer) Td() *big.Int {
 	p.lock.RLock()
@@ -455,9 +393,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 		p.fcCosts = MRC.decode()
 	}
 
-	p.firstHeadInfo = &announceData{Td: rTd, Hash: rHash, Number: rNum}
-	p.headInfo = p.firstHeadInfo
-	p.headInfoLen = 1
+	p.headInfo = &announceData{Td: rTd, Hash: rHash, Number: rNum}
 	return nil
 }
 

les/request_test.go

@@ -71,6 +71,8 @@ func testAccess(t *testing.T, protocol int, fn accessTestFn) {
 	pm, db, _ := newTestProtocolManagerMust(t, false, 4, testChainGen)
 	lpm, ldb, odr := newTestProtocolManagerMust(t, true, 0, nil)
 	_, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm)
+	pool := (*testServerPool)(lpeer)
+	odr.serverPool = pool
 	select {
 	case <-time.After(time.Millisecond * 100):
 	case err := <-err1:
@@ -100,11 +102,10 @@ func testAccess(t *testing.T, protocol int, fn accessTestFn) {
 	}
 
 	// temporarily remove peer to test odr fails
-	odr.UnregisterPeer(lpeer)
+	odr.serverPool = nil
 	// expect retrievals to fail (except genesis block) without a les peer
 	test(0)
-	odr.RegisterPeer(lpeer)
+	odr.serverPool = pool
 	// expect all retrievals to pass
 	test(5)
-	odr.UnregisterPeer(lpeer)
 }

light/lightchain.go

@@ -505,3 +505,14 @@ func (self *LightChain) SyncCht(ctx context.Context) bool {
 	}
 	return false
 }
+
+// LockChain locks the chain mutex for reading so that multiple canonical hashes can be
+// retrieved while it is guaranteed that they belong to the same version of the chain
+func (self *LightChain) LockChain() {
+	self.chainmu.RLock()
+}
+
+// UnlockChain unlocks the chain mutex
+func (self *LightChain) UnlockChain() {
+	self.chainmu.RUnlock()
+}