diff --git a/les/handler.go b/les/handler.go index 64023af0f5..ec70777264 100644 --- a/les/handler.go +++ b/les/handler.go @@ -102,7 +102,9 @@ type ProtocolManager struct { odr *LesOdr server *LesServer serverPool *serverPool + lesTopic discv5.Topic reqDist *requestDistributor + retriever *retrieveManager downloader *downloader.Downloader fetcher *lightFetcher @@ -202,29 +204,25 @@ 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.reqDist = newRequestDistributor(func() map[distPeer]struct{} { - m := make(map[distPeer]struct{}) - peers := manager.peers.AllPeers() - for _, peer := range peers { - m[peer] = struct{}{} + manager.reqDist = newRequestDistributor(func() map[distPeer]struct{} { + m := make(map[distPeer]struct{}) + peers := manager.peers.AllPeers() + for _, peer := range peers { + m[peer] = struct{}{} + } + return m + }, manager.quitSync) + + manager.lesTopic = discv5.Topic("LES@" + common.Bytes2Hex(manager.blockchain.Genesis().Hash().Bytes()[0:8])) + manager.serverPool = newServerPool(manager.chainDb, []byte("serverPool/"), manager.lesTopic, manager.quitSync, &manager.wg) + manager.retriever = newRetrieveManager(manager.reqDist, manager.serverPool, removePeer) + if odr != nil { + odr.retriever = manager.retriever } - return m - }, manager.quitSync) - if odr != nil { - odr.removePeer = removePeer - odr.reqDist = manager.reqDist + manager.fetcher = newLightFetcher(manager) } - /*validator := func(block *types.Block, parent *types.Block) error { - return core.ValidateHeader(pow, block.Header(), parent.Header(), true, false) - } - heighter := func() uint64 { - return chainman.LastBlockNumberU64() - } - manager.fetcher = fetcher.New(chainman.GetBlockNoOdr, validator, nil, heighter, chainman.InsertChain, manager.removePeer) - */ return manager, nil } @@ -261,23 +259,20 @@ func (pm *ProtocolManager) Start(srvr *p2p.Server) { if srvr != nil { topicDisc = srvr.DiscV5 } - lesTopic := discv5.Topic("LES@" + common.Bytes2Hex(pm.blockchain.Genesis().Hash().Bytes()[0:8])) if pm.lightSync { // start sync handler 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) + pm.serverPool.setServer(srvr) } go pm.syncer() } else { if topicDisc != nil { go func() { - logger := log.New("topic", lesTopic) + logger := log.New("topic", pm.lesTopic) logger.Info("Starting topic registration") defer logger.Info("Terminated topic registration") - topicDisc.RegisterTopic(lesTopic, pm.quitSync) + topicDisc.RegisterTopic(pm.lesTopic, pm.quitSync) }() } go func() { @@ -926,7 +921,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } if deliverMsg != nil { - err := pm.odr.Deliver(p, deliverMsg) + err := pm.retriever.deliver(p, deliverMsg) if err != nil { p.responseErrors++ if p.responseErrors > maxResponseErrors { diff --git a/les/odr.go b/les/odr.go index 684f36c761..ae6d73254f 100644 --- a/les/odr.go +++ b/les/odr.go @@ -18,45 +18,23 @@ package les import ( "context" - "crypto/rand" - "encoding/binary" - "sync" - "time" - "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" ) -var ( - softRequestTimeout = time.Millisecond * 500 - hardRequestTimeout = time.Second * 10 -) - -// peerDropFn is a callback type for dropping a peer detected as malicious. -type peerDropFn func(id string) - -type odrPeerSelector interface { - adjustResponseTime(*poolEntry, time.Duration, bool) -} - +// LesOdr implements light.OdrBackend type LesOdr struct { - light.OdrBackend - db ethdb.Database - stop chan struct{} - removePeer peerDropFn - mlock, clock sync.Mutex - sentReqs map[uint64]*sentReq - serverPool odrPeerSelector - reqDist *requestDistributor + db ethdb.Database + stop chan struct{} + retriever *retrieveManager } func NewLesOdr(db ethdb.Database) *LesOdr { return &LesOdr{ - db: db, - stop: make(chan struct{}), - sentReqs: make(map[uint64]*sentReq), + db: db, + stop: make(chan struct{}), } } @@ -68,17 +46,6 @@ func (odr *LesOdr) Database() ethdb.Database { return odr.db } -// validatorFunc is a function that processes a message. -type validatorFunc func(ethdb.Database, *Msg) error - -// sentReq is a request waiting for an answer that satisfies its valFunc -type sentReq struct { - valFunc validatorFunc - sentTo map[*peer]chan struct{} - lock sync.RWMutex // protects acces to sentTo - answered chan struct{} // closed and set to nil when any peer answers it -} - const ( MsgBlockBodies = iota MsgCode @@ -94,88 +61,11 @@ type Msg struct { Obj interface{} } -// Deliver is called by the LES protocol manager to deliver ODR reply messages to waiting requests -func (self *LesOdr) Deliver(peer *peer, msg *Msg) error { - var delivered chan struct{} - self.mlock.Lock() - req, ok := self.sentReqs[msg.ReqID] - self.mlock.Unlock() - if ok { - req.lock.Lock() - delivered, ok = req.sentTo[peer] - req.lock.Unlock() - } +// Retrieve tries to fetch an object from the LES network. +// If the network retrieval was successful, it stores the object in local db. +func (self *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err error) { + lreq := LesRequest(req) - if !ok { - return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID) - } - - if err := req.valFunc(self.db, msg); err != nil { - peer.Log().Warn("Invalid odr response", "err", err) - return errResp(ErrInvalidResponse, "reqID = %v", msg.ReqID) - } - close(delivered) - req.lock.Lock() - delete(req.sentTo, peer) - if req.answered != nil { - close(req.answered) - req.answered = nil - } - req.lock.Unlock() - return nil -} - -func (self *LesOdr) requestPeer(req *sentReq, peer *peer, delivered, timeout chan struct{}, reqWg *sync.WaitGroup) { - stime := mclock.Now() - defer func() { - req.lock.Lock() - delete(req.sentTo, peer) - req.lock.Unlock() - reqWg.Done() - }() - - select { - case <-delivered: - if self.serverPool != nil { - self.serverPool.adjustResponseTime(peer.poolEntry, time.Duration(mclock.Now()-stime), false) - } - return - case <-time.After(softRequestTimeout): - close(timeout) - case <-self.stop: - return - } - - select { - case <-delivered: - case <-time.After(hardRequestTimeout): - peer.Log().Debug("Request timed out hard") - 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 -// or the context is cancelled -func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) error { - answered := make(chan struct{}) - req := &sentReq{ - valFunc: lreq.Validate, - sentTo: make(map[*peer]chan struct{}), - answered: answered, // reply delivered by any peer - } - - exclude := make(map[*peer]struct{}) - - reqWg := new(sync.WaitGroup) - reqWg.Add(1) - defer reqWg.Done() - - var timeout chan struct{} reqID := getNextReqID() rq := &distReq{ getCost: func(dp distPeer) uint64 { @@ -183,67 +73,17 @@ func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) erro }, canSend: func(dp distPeer) bool { p := dp.(*peer) - _, ok := exclude[p] - return !ok && lreq.CanSend(p) + return lreq.CanSend(p) }, request: func(dp distPeer) func() { p := dp.(*peer) - exclude[p] = struct{}{} - delivered := make(chan struct{}) - timeout = make(chan struct{}) - req.lock.Lock() - req.sentTo[p] = delivered - req.lock.Unlock() - reqWg.Add(1) cost := lreq.GetCost(p) p.fcServer.QueueRequest(reqID, cost) - go self.requestPeer(req, p, delivered, timeout, reqWg) return func() { lreq.Request(reqID, p) } }, } - self.mlock.Lock() - self.sentReqs[reqID] = req - self.mlock.Unlock() - - go func() { - reqWg.Wait() - self.mlock.Lock() - delete(self.sentReqs, reqID) - self.mlock.Unlock() - }() - - for { - peerChn := self.reqDist.queue(rq) - select { - case <-ctx.Done(): - self.reqDist.cancel(rq) - return ctx.Err() - case <-answered: - self.reqDist.cancel(rq) - return nil - case _, ok := <-peerChn: - if !ok { - return ErrNoPeers - } - } - - select { - case <-ctx.Done(): - return ctx.Err() - case <-answered: - return nil - case <-timeout: - } - } -} - -// Retrieve tries to fetch an object from the LES network. -// If the network retrieval was successful, it stores the object in local db. -func (self *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err error) { - lreq := LesRequest(req) - err = self.networkRequest(ctx, lreq) - if err == nil { + if err = self.retriever.retrieve(ctx, reqID, rq, func(p distPeer, msg *Msg) error { return lreq.Validate(self.db, msg) }); err == nil { // retrieved from network, store in db req.StoreResult(self.db) } else { @@ -251,9 +91,3 @@ func (self *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err err } return } - -func getNextReqID() uint64 { - var rnd [8]byte - rand.Read(rnd[:]) - return binary.BigEndian.Uint64(rnd[:]) -} diff --git a/les/retrieve.go b/les/retrieve.go new file mode 100644 index 0000000000..8b3056da86 --- /dev/null +++ b/les/retrieve.go @@ -0,0 +1,323 @@ +// 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 light implements on-demand retrieval capable state and chain objects +// for the Ethereum Light Client. +package les + +import ( + "context" + "crypto/rand" + "encoding/binary" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common/mclock" +) + +var ( + retryQueue = time.Millisecond * 100 + softRequestTimeout = time.Millisecond * 500 + hardRequestTimeout = time.Second * 10 +) + +// retrieveManager is a layer on top of requestDistributor which takes care of +// matching replies by request ID and handles timeouts and resends if necessary. +type retrieveManager struct { + dist *requestDistributor + serverPool peerSelector + removePeer peerDropFn + + lock sync.RWMutex + sentReqs map[uint64]*sentReq +} + +// validatorFunc is a function that processes a reply message +type validatorFunc func(distPeer, *Msg) error + +// peerDropFn is a callback type for dropping a peer detected as malicious. +type peerDropFn func(id string) + +// peerSelector receives feedback info about response times and timeouts +type peerSelector interface { + adjustResponseTime(*poolEntry, time.Duration, bool) +} + +type sentReq struct { + rm *retrieveManager + req *distReq + validate validatorFunc + stopChn chan struct{} + stopped bool + err error + lock sync.RWMutex // protect access to sentTo map + sentTo map[distPeer]chan struct{} // channel signaling a reply from the given peer + reqWg sync.WaitGroup // wait until every peer replied or reached hard timeout and we can forget about the request + sentCnt, softTimeoutCnt int +} + +// reqPeerCallback is called after a request sent to a certain peer has either been +// answered or timed out hard +type reqPeerCallback func(p distPeer, respTime time.Duration, srto, hrto bool) + +// newRetrieveManager creates the retrieve manager +func newRetrieveManager(dist *requestDistributor, serverPool peerSelector, removePeer peerDropFn) *retrieveManager { + return &retrieveManager{ + dist: dist, + serverPool: serverPool, + removePeer: removePeer, + sentReqs: make(map[uint64]*sentReq), + } +} + +// retrieve sends a request (to multiple peers if necessary) and waits for an answer +// that is delivered through the deliver function and successfully validated by the +// validator callback. It returns when a valid answer is delivered or the context is +// cancelled. +func (rm *retrieveManager) retrieve(ctx context.Context, reqID uint64, req *distReq, val validatorFunc) error { + sentReq := rm.sendReq(reqID, req, val) + select { + case <-sentReq.stopChn: + case <-ctx.Done(): + sentReq.stop(ctx.Err()) + } + return sentReq.getError() +} + +// sendReq starts a process that keeps trying to retrieve a valid answer for a +// request from any suitable peers until stopped or succeeded. +func (rm *retrieveManager) sendReq(reqID uint64, req *distReq, val validatorFunc) *sentReq { + r := &sentReq{ + rm: rm, + req: req, + sentTo: make(map[distPeer]chan struct{}), + stopChn: make(chan struct{}), + validate: val, + } + + canSend := req.canSend + req.canSend = func(p distPeer) bool { + r.lock.RLock() + _, sent := r.sentTo[p] + r.lock.RUnlock() + return !sent && canSend(p) + } + + rm.lock.Lock() + rm.sentReqs[reqID] = r + rm.lock.Unlock() + + r.reqWg.Add(1) + r.tryRetrieve() + go func() { + r.reqWg.Wait() + rm.lock.Lock() + delete(rm.sentReqs, reqID) + rm.lock.Unlock() + }() + r.reqWg.Done() + + return r +} + +// deliver is called by the LES protocol manager to deliver reply messages to waiting requests +func (rm *retrieveManager) deliver(peer distPeer, msg *Msg) error { + rm.lock.RLock() + req, ok := rm.sentReqs[msg.ReqID] + rm.lock.RUnlock() + if ok { + ok = req.expectResponseFrom(peer) + } + + if !ok { + return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID) + } + + if err := req.validate(peer, msg); err != nil { + req.delivered(peer, false) + return errResp(ErrInvalidResponse, "reqID = %v", msg.ReqID) + } + req.delivered(peer, true) + return nil +} + +// tryRetrieve starts a retrieval process for a single peer. If no request has been +// sent yet and no suitable peer found, it sets an ErrNoPeers error code and returns. +// Otherwise it keeps trying until it sends the request to a peer, then waits for an +// answer or a timeout. +func (r *sentReq) tryRetrieve() { + r.reqWg.Add(1) + go func() { + defer r.reqWg.Done() + + for { + sent := r.rm.dist.queue(r.req) + var p distPeer + select { + case p = <-sent: + case <-r.stopChn: + if r.rm.dist.cancel(r.req) { + p = nil + } else { + p = <-sent + } + } + if p == nil { + if r.waiting() { + time.Sleep(retryQueue) + if r.waiting() { + continue + } + } else { + r.stop(ErrNoPeers) // no effect if already stopped for another reason + } + } else { + r.lock.Lock() + if r.sentTo[p] == nil { + r.sentTo[p] = make(chan struct{}) + r.sentCnt++ + } + r.lock.Unlock() + respTime, srto, hrto := r.runTimer(p) + // send feedback to server pool and remove peer if hard timeout happened + pp, ok := p.(*peer) + if ok && r.rm.serverPool != nil { + r.rm.serverPool.adjustResponseTime(pp.poolEntry, respTime, srto) + } + if hrto { + pp.Log().Debug("Request timed out hard") + if r.rm.removePeer != nil { + r.rm.removePeer(pp.id) + } + } + } + break + } + }() +} + +// getError returns any retrieval error (either internally generated or set by the +// stop function) after stopChn has been closed +func (r *sentReq) getError() error { + return r.err +} + +// expectResponseFrom tells if we are expecting a response from a given peer. +// A response to a sent request is expected even after another peer have delivered +// a valid response. If a hard timeout occurs, no response is expected any more. +func (r *sentReq) expectResponseFrom(peer distPeer) bool { + r.lock.Lock() + defer r.lock.Unlock() + + return r.sentTo[peer] != nil +} + +// delivered notifies the retrieval mechanism that a reply (either valid +// or invalid) has been received from a certain peer +func (r *sentReq) delivered(peer distPeer, valid bool) bool { + r.lock.Lock() + defer r.lock.Unlock() + + delivered, ok := r.sentTo[peer] + if ok { + close(delivered) + delete(r.sentTo, peer) + if valid && !r.stopped { + r.stopped = true + close(r.stopChn) + } + if !r.stopped && r.sentCnt == 0 { + r.tryRetrieve() + } + + } + return ok +} + +// stop stops the retrieval process and sets an error code that will be returned +// by getError +func (r *sentReq) stop(err error) { + r.lock.Lock() + if !r.stopped { + r.stopped = true + r.err = err + close(r.stopChn) + } + r.lock.Unlock() +} + +// waiting returns true if the retrieval mechanism is waiting for an answer from +// any peer +func (r *sentReq) waiting() bool { + r.lock.RLock() + defer r.lock.RUnlock() + + return !r.stopped && r.sentCnt+r.softTimeoutCnt != 0 +} + +// runTimer starts a request timeout for a single peer. If a certain (short) period +// of time passes with no reply received, it switches from "sent" to "soft timeout" +// state and the retrieval mechanism will start trying to send another request to a +// new peer. If another (longer) timeout period passes, it switches to "hard timeout" +// state, after which a reply is no longer expected and the peer should be dropped. +func (r *sentReq) runTimer(peer distPeer) (respTime time.Duration, srto, hrto bool) { + start := mclock.Now() + + r.lock.RLock() + delivered := r.sentTo[peer] + r.lock.RUnlock() + if delivered == nil { + panic(nil) + } + + select { + case <-delivered: + r.lock.Lock() + r.sentCnt-- + r.lock.Unlock() + return time.Duration(mclock.Now() - start), false, false + case <-time.After(softRequestTimeout): + r.lock.Lock() + r.sentCnt-- + resend := !r.stopped && r.sentCnt == 0 + r.softTimeoutCnt++ + r.lock.Unlock() + if resend { + r.tryRetrieve() + } + } + + hrto = false + select { + case <-delivered: + case <-time.After(hardRequestTimeout): + hrto = true + } + + r.lock.Lock() + r.softTimeoutCnt-- + // do not delete sentTo entry, nil means that we are not expecting an answer + // but we also do not want to send to that peer again + r.sentTo[peer] = nil + r.lock.Unlock() + return time.Duration(mclock.Now() - start), true, hrto +} + +func getNextReqID() uint64 { + var rnd [8]byte + rand.Read(rnd[:]) + return binary.BigEndian.Uint64(rnd[:]) +} diff --git a/les/serverpool.go b/les/serverpool.go index 64fe991c63..56ccfe465a 100644 --- a/les/serverpool.go +++ b/les/serverpool.go @@ -102,6 +102,8 @@ type serverPool struct { wg *sync.WaitGroup connWg sync.WaitGroup + topic discv5.Topic + discSetPeriod chan time.Duration discNodes chan *discv5.Node discLookups chan bool @@ -118,13 +120,13 @@ type serverPool struct { } // newServerPool creates a new serverPool instance -func newServerPool(db ethdb.Database, dbPrefix []byte, server *p2p.Server, topic discv5.Topic, quit chan struct{}, wg *sync.WaitGroup) *serverPool { +func newServerPool(db ethdb.Database, dbPrefix []byte, topic discv5.Topic, quit chan struct{}, wg *sync.WaitGroup) *serverPool { pool := &serverPool{ db: db, dbKey: append(dbPrefix, []byte(topic)...), - server: server, quit: quit, wg: wg, + topic: topic, entries: make(map[discover.NodeID]*poolEntry), timeout: make(chan *poolEntry, 1), adjustStats: make(chan poolStatAdjust, 100), @@ -137,17 +139,20 @@ func newServerPool(db ethdb.Database, dbPrefix []byte, server *p2p.Server, topic pool.newQueue = newPoolEntryQueue(maxNewEntries, pool.removeEntry) wg.Add(1) pool.loadNodes() - pool.checkDial() + go pool.eventLoop() + return pool +} + +func (pool *serverPool) setServer(server *p2p.Server) { + pool.server = server + pool.checkDial() if pool.server.DiscV5 != nil { pool.discSetPeriod = make(chan time.Duration, 1) pool.discNodes = make(chan *discv5.Node, 100) pool.discLookups = make(chan bool, 100) - go pool.server.DiscV5.SearchTopic(topic, pool.discSetPeriod, pool.discNodes, pool.discLookups) + go pool.server.DiscV5.SearchTopic(pool.topic, pool.discSetPeriod, pool.discNodes, pool.discLookups) } - - go pool.eventLoop() - return pool } // connect should be called upon any incoming connection. If the connection has been @@ -485,7 +490,7 @@ func (pool *serverPool) checkDial() { // dial initiates a new connection func (pool *serverPool) dial(entry *poolEntry, knownSelected bool) { - if entry.state != psNotConnected { + if pool.server == nil || entry.state != psNotConnected { return } entry.state = psDialed