diff --git a/les/distributor.go b/les/distributor.go index f90765b624..f5c5d944be 100644 --- a/les/distributor.go +++ b/les/distributor.go @@ -127,7 +127,15 @@ func (d *requestDistributor) loop() { loop: for { peer, req, wait := d.nextRequest() - if req != nil && wait == 0 { + // no request to send and nothing to wait for; the next + // queued request will wake up the loop + if req == nil { + break loop + } + // send non-wait or short-wait request immediately. + if wait == 0 || wait <= distMaxWait { + time.Sleep(wait) + chn := req.sentChn // save sentChn because remove sets it to nil d.remove(req) send := req.request(peer) @@ -136,23 +144,14 @@ func (d *requestDistributor) loop() { } chn <- peer close(chn) - } else { - if wait == 0 { - // no request to send and nothing to wait for; the next - // queued request will wake up the loop - break loop - } - d.loopNextSent = true // a "next" signal has been sent, do not send another one until this one has been received - if wait > distMaxWait { - // waiting times may be reduced by incoming request replies, if it is too long, recalculate it periodically - wait = distMaxWait - } - go func() { - time.Sleep(wait) - d.loopChn <- struct{}{} - }() - break loop + continue } + d.loopNextSent = true + go func() { + time.Sleep(distMaxWait) + d.loopChn <- struct{}{} + }() + break loop } d.lock.Unlock() } @@ -174,24 +173,29 @@ func (sp selectPeerItem) Weight() int64 { // nextRequest returns the next possible request from any peer, along with the // associated peer and necessary waiting time func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { - checkedPeers := make(map[distPeer]struct{}) - elem := d.reqQueue.Front() - var ( - bestPeer distPeer - bestReq *distReq - bestWait time.Duration - sel *weightedRandomSelect - ) - d.peerLock.RLock() defer d.peerLock.RUnlock() - for (len(d.peers) > 0 || elem == d.reqQueue.Front()) && elem != nil { + // Short circuit if there is no available peer or no pending request. + if len(d.peers) == 0 || d.reqQueue.Len() == 0 { + return nil, nil, 0 + } + + type queueReq struct { + req *distReq + peer distPeer + waitNeeded time.Duration + } + // Iterate request queue to find out the request with the least waiting time. + var bestReq *queueReq + for elem := d.reqQueue.Front(); elem != nil; { req := elem.Value.(*distReq) - canSend := false + var ( + sel *weightedRandomSelect + currentReq *queueReq + ) for peer := range d.peers { - if _, ok := checkedPeers[peer]; !ok && peer.canQueue() && req.canSend(peer) { - canSend = true + if peer.canQueue() && req.canSend(peer) { cost := req.getCost(peer) wait, bufRemain := peer.waitBefore(cost) if wait == 0 { @@ -200,28 +204,44 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { } sel.update(selectPeerItem{peer: peer, req: req, weight: int64(bufRemain*1000000) + 1}) } else { - if bestReq == nil || wait < bestWait { - bestPeer = peer - bestReq = req - bestWait = wait + if currentReq == nil || wait < currentReq.waitNeeded { + currentReq = &queueReq{ + req: req, + peer: peer, + waitNeeded: wait, + } } } - checkedPeers[peer] = struct{}{} } } + // Find out no-wait-request, pass back directly. + if sel != nil { + c := sel.choose().(selectPeerItem) + return c.peer, c.req, 0 + } next := elem.Next() - if !canSend && elem == d.reqQueue.Front() { + if currentReq != nil { + // Find out short-wait-request, abort further filtering + if currentReq.waitNeeded <= distMaxWait { + return currentReq.peer, currentReq.req, currentReq.waitNeeded + } else { + if bestReq == nil || currentReq.waitNeeded < bestReq.waitNeeded { + bestReq = currentReq + } + } + } else { + // Can't find any suitable peers, drop the request directly close(req.sentChn) d.remove(req) } elem = next - } - if sel != nil { - c := sel.choose().(selectPeerItem) - return c.peer, c.req, 0 } - return bestPeer, bestReq, bestWait + if bestReq != nil { + return bestReq.peer, bestReq.req, bestReq.waitNeeded + } + // Can't find anymore request that can be sent. + return nil, nil, 0 } // queue adds a request to the distribution queue, returns a channel where the diff --git a/les/distributor_test.go b/les/distributor_test.go index 2891bcab49..cfa2fe7111 100644 --- a/les/distributor_test.go +++ b/les/distributor_test.go @@ -57,8 +57,7 @@ func (p *testDistPeer) send(r *testDistReq) { p.sumCost += r.cost } -func (p *testDistPeer) worker(t *testing.T, checkOrder bool, stop chan struct{}) { - var last uint64 +func (p *testDistPeer) worker(t *testing.T, stop chan struct{}) { for { wait := time.Millisecond p.lock.Lock() @@ -66,12 +65,6 @@ func (p *testDistPeer) worker(t *testing.T, checkOrder bool, stop chan struct{}) rq := p.sent[0] wait = time.Duration(rq.procTime) p.sumCost -= rq.cost - if checkOrder { - if rq.order <= last { - t.Errorf("Requests processed in wrong order") - } - last = rq.order - } p.sent = p.sent[1:] } p.lock.Unlock() @@ -125,7 +118,7 @@ func testRequestDistributor(t *testing.T, resend bool) { var peers [testDistPeerCount]*testDistPeer for i := range peers { peers[i] = &testDistPeer{} - go peers[i].worker(t, !resend, stop) + go peers[i].worker(t, stop) dist.registerTestPeer(peers[i]) } @@ -183,3 +176,42 @@ func testRequestDistributor(t *testing.T, resend bool) { wg.Wait() } + +func TestProcessRequestOutOrder(t *testing.T) { + stop := make(chan struct{}) + defer close(stop) + + dist := newRequestDistributor(nil, stop) + var peers [2]*testDistPeer + for i := range peers { + peers[i] = &testDistPeer{} + go peers[i].worker(t, stop) + dist.registerTestPeer(peers[i]) + } + peers[0].sumCost = 10 * testDistMaxCost // distMaxWait + + peerChs := make([]chan distPeer, 2) + for i, cost := range []uint64{11 * testDistMaxCost, 100} { + rq := &testDistReq{ + cost: cost, + procTime: cost, + order: uint64(i + 1), + canSendTo: make(map[*testDistPeer]struct{}), + } + rq.canSendTo[peers[i]] = struct{}{} + req := &distReq{ + getCost: rq.getCost, + canSend: rq.canSend, + request: rq.request, + } + peerChs[i] = dist.queue(req) + } + + select { + case <-peerChs[0]: + t.Error("expect to be processed later") + case <-peerChs[1]: + case <-time.NewTicker(time.Second).C: + t.Error("waiting assigned peer timeout") + } +} diff --git a/les/flowcontrol/control.go b/les/flowcontrol/control.go index d50eb809cc..d62f44e4aa 100644 --- a/les/flowcontrol/control.go +++ b/les/flowcontrol/control.go @@ -155,6 +155,7 @@ func (peer *ServerNode) QueueRequest(reqID, maxCost uint64) { peer.lock.Lock() defer peer.lock.Unlock() + peer.recalcBLE(mclock.Now()) peer.bufEstimate -= maxCost peer.sumCost += maxCost peer.pending[reqID] = peer.sumCost