diff --git a/les/distributor.go b/les/distributor.go index c59b36146f..64f11f5b61 100644 --- a/les/distributor.go +++ b/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 +} diff --git a/les/odr.go b/les/odr.go index 06b44d3186..217ac8773b 100644 --- a/les/odr.go +++ b/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) @@ -73,10 +67,8 @@ 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 + valFunc validatorFunc + req *distReq } 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() - delivered, ok = req.sentTo[peer] - req.lock.Unlock() + ok = req.req.expectResponseFrom(peer) } 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.lock.Lock() - delete(req.sentTo, peer) - if req.answered != nil { - close(req.answered) - req.answered = nil - } - req.lock.Unlock() + req.req.delivered(peer, true) 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{}) +// 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) 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 !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) } }, } + 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 { - 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 - } + stopChn := self.reqDist.retrieve(rq, func(p distPeer, respTime time.Duration, srto, hrto bool) { + reqWg.Done() + pp := p.(*peer) + if self.serverPool != nil { + self.serverPool.adjustResponseTime(pp.poolEntry, respTime, srto) } + if hrto { + pp.Log().Debug("Request timed out hard") + self.removePeer(pp.id) + } + }) - select { - case <-ctx.Done(): - return ctx.Err() - case <-answered: - return nil - case <-timeout: - } + select { + case <-stopChn: + case <-ctx.Done(): + rq.stop(ctx.Err()) } -} -// 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 = rq.getError(); err == nil { // retrieved from network, store in db req.StoreResult(self.db) } else {