les: implement retrieval mechanism in request distributor

les/distributor.go

@@ -23,6 +23,8 @@ import (
 	"errors"
 	"sync"
 	"time"
+
+	"github.com/ethereum/go-ethereum/common/mclock"
 )
 
 // ErrNoPeers is returned if no peers capable of serving a queued request are available
@@ -52,6 +54,16 @@ type distPeer interface {
 	queueSend(f func())
 }
 
+var (
+	retryQueue         = time.Millisecond * 100
+	softRequestTimeout = time.Millisecond * 500
+	hardRequestTimeout = time.Second * 10
+)
+
+// 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)
+
 // distReq is the request abstraction used by the distributor. It is based on
 // three callback functions:
 // - getCost returns the upper estimate of the cost of sending the request to a given peer
@@ -60,6 +72,9 @@ type distPeer interface {
 // does the actual sending. Request order should be preserved but the callback itself should not
 // block until it is sent because other peers might still be able to receive requests while
 // one of them is blocking. Instead, the returned function is put in the peer's send queue.
+//
+// Requests can either be queued manually or by the retrieval management mechanism which
+// takes care of timeouts, resends and not sending to the same peer again.
 type distReq struct {
 	getCost func(distPeer) uint64
 	canSend func(distPeer) bool
@@ -68,6 +83,16 @@ type distReq struct {
 	reqOrder uint64
 	sentChn  chan distPeer
 	element  *list.Element
+
+	// retrieval management fields (optional)
+	dist                    *requestDistributor
+	peerCallback            reqPeerCallback
+	lock                    sync.RWMutex
+	stopChn                 chan struct{}
+	stopped                 bool
+	err                     error
+	sentTo                  map[distPeer]chan struct{} // channel signaling a reply from the given peer
+	sentCnt, softTimeoutCnt int
 }
 
 // newRequestDistributor creates a new request distributor
@@ -112,6 +137,13 @@ func (d *requestDistributor) loop() {
 					if send != nil {
 						peer.queueSend(send)
 					}
+					req.lock.Lock()
+					if req.sentTo != nil {
+						// if the retrieval manager is used, create deliver channel
+						req.sentTo[peer] = make(chan struct{})
+						req.sentCnt++
+					}
+					req.lock.Unlock()
 					chn <- peer
 					close(chn)
 				} else {
@@ -165,8 +197,10 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) {
 	for (len(peers) > 0 || elem == d.reqQueue.Front()) && elem != nil {
 		req := elem.Value.(*distReq)
 		canSend := false
+		req.lock.RLock()
 		for peer, _ := range peers {
-			if peer.canQueue() && req.canSend(peer) {
+			// if retrieve manager is not used and sentTo is nil, ok is always false
+			if _, ok := req.sentTo[peer]; !ok && peer.canQueue() && req.canSend(peer) {
 				canSend = true
 				cost := req.getCost(peer)
 				wait, bufRemain := peer.waitBefore(cost)
@@ -185,6 +219,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) {
 				delete(peers, peer)
 			}
 		}
+		req.lock.RUnlock()
 		next := elem.Next()
 		if !canSend && elem == d.reqQueue.Front() {
 			close(req.sentChn)
@@ -257,3 +292,156 @@ func (d *requestDistributor) remove(r *distReq) {
 		r.element = nil
 	}
 }
+
+// retrieve starts a process that keeps trying to retrieve a valid answer for a
+// request from any suitable peers until stopped or succeeded.
+func (d *requestDistributor) retrieve(r *distReq, cb reqPeerCallback) chan struct{} {
+	r.dist = d
+	r.sentTo = make(map[distPeer]chan struct{})
+	r.stopChn = make(chan struct{})
+	r.peerCallback = cb
+
+	go r.retrieve()
+	return r.stopChn
+}
+
+// retrieve 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 *distReq) retrieve() {
+	for {
+		sent := r.dist.queue(r)
+		var p distPeer
+		select {
+		case p = <-sent:
+		case <-r.stopChn:
+			if r.dist.cancel(r) {
+				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 {
+			respTime, srto, hrto := r.runTimer(p)
+			r.peerCallback(p, respTime, srto, hrto)
+		}
+		break
+	}
+}
+
+// getError returns any retrieval error (either internally generated or set by the
+// stop function) after stopChn has been closed
+func (r *distReq) 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 *distReq) 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 *distReq) 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 {
+			go r.retrieve()
+		}
+
+	}
+	return ok
+}
+
+// stop stops the retrieval process and sets an error code that will be returned
+// by getError
+func (r *distReq) 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 *distReq) 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 *distReq) 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 {
+			go r.retrieve()
+		}
+	}
+
+	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
+}

les/odr.go

@@ -22,18 +22,12 @@ import (
 	"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"
 	"golang.org/x/net/context"
 )
 
-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)
 
@@ -74,9 +68,7 @@ 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{}
+	req     *distReq
-	lock     sync.RWMutex  // protects acces to sentTo
-	answered chan struct{} // closed and set to nil when any peer answers it
 }
 
 const (
@@ -96,14 +88,11 @@ type Msg struct {
 
 // 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()
+		ok = req.req.expectResponseFrom(peer)
-		delivered, ok = req.sentTo[peer]
-		req.lock.Unlock()
 	}
 
 	if !ok {
@@ -112,70 +101,22 @@ func (self *LesOdr) Deliver(peer *peer, msg *Msg) error {
 
 	if err := req.valFunc(self.db, msg); err != nil {
 		peer.Log().Warn("Invalid odr response", "err", err)
+		req.req.delivered(peer, false)
 		return errResp(ErrInvalidResponse, "reqID = %v", msg.ReqID)
 	}
-	close(delivered)
+	req.req.delivered(peer, true)
-	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) {
+// Retrieve tries to fetch an object from the LES network.
-	stime := mclock.Now()
+// If the network retrieval was successful, it stores the object in local db.
-	defer func() {
+func (self *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err error) {
-		req.lock.Lock()
+	lreq := LesRequest(req)
-		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,27 +124,24 @@ func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) erro
 		},
 		canSend: func(dp distPeer) bool {
 			p := dp.(*peer)
-			_, ok := exclude[p]
+			return lreq.CanSend(p)
-			return !ok && 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) }
 		},
 	}
 
+	sreq := &sentReq{
+		valFunc: lreq.Validate,
+		req:     rq,
+	}
+
 	self.mlock.Lock()
-	self.sentReqs[reqID] = req
+	self.sentReqs[reqID] = sreq
 	self.mlock.Unlock()
 
 	go func() {
@@ -213,37 +151,25 @@ func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) erro
 		self.mlock.Unlock()
 	}()
 
-	for {
+	stopChn := self.reqDist.retrieve(rq, func(p distPeer, respTime time.Duration, srto, hrto bool) {
-		peerChn := self.reqDist.queue(rq)
+		reqWg.Done()
-		select {
+		pp := p.(*peer)
-		case <-ctx.Done():
+		if self.serverPool != nil {
-			self.reqDist.cancel(rq)
+			self.serverPool.adjustResponseTime(pp.poolEntry, respTime, srto)
-			return ctx.Err()
-		case <-answered:
-			self.reqDist.cancel(rq)
-			return nil
-		case _, ok := <-peerChn:
-			if !ok {
-				return ErrNoPeers
 		}
+		if hrto {
+			pp.Log().Debug("Request timed out hard")
+			self.removePeer(pp.id)
 		}
+	})
 
 	select {
+	case <-stopChn:
 	case <-ctx.Done():
-			return ctx.Err()
+		rq.stop(ctx.Err())
-		case <-answered:
-			return nil
-		case <-timeout:
 	}
-	}
-}
 
-// Retrieve tries to fetch an object from the LES network.
+	if err = rq.getError(); err == nil {
-// 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 {
 		// retrieved from network, store in db
 		req.StoreResult(self.db)
 	} else {