|
|
@ -39,6 +39,7 @@ type stateReq struct { |
|
|
|
timeout time.Duration // Maximum round trip time for this to complete
|
|
|
|
timeout time.Duration // Maximum round trip time for this to complete
|
|
|
|
timer *time.Timer // Timer to fire when the RTT timeout expires
|
|
|
|
timer *time.Timer // Timer to fire when the RTT timeout expires
|
|
|
|
peer *peerConnection // Peer that we're requesting from
|
|
|
|
peer *peerConnection // Peer that we're requesting from
|
|
|
|
|
|
|
|
delivered time.Time // Time when the packet was delivered (independent when we process it)
|
|
|
|
response [][]byte // Response data of the peer (nil for timeouts)
|
|
|
|
response [][]byte // Response data of the peer (nil for timeouts)
|
|
|
|
dropped bool // Flag whether the peer dropped off early
|
|
|
|
dropped bool // Flag whether the peer dropped off early
|
|
|
|
} |
|
|
|
} |
|
|
@ -148,6 +149,7 @@ func (d *Downloader) runStateSync(s *stateSync) *stateSync { |
|
|
|
// Finalize the request and queue up for processing
|
|
|
|
// Finalize the request and queue up for processing
|
|
|
|
req.timer.Stop() |
|
|
|
req.timer.Stop() |
|
|
|
req.response = pack.(*statePack).states |
|
|
|
req.response = pack.(*statePack).states |
|
|
|
|
|
|
|
req.delivered = time.Now() |
|
|
|
|
|
|
|
|
|
|
|
finished = append(finished, req) |
|
|
|
finished = append(finished, req) |
|
|
|
delete(active, pack.PeerId()) |
|
|
|
delete(active, pack.PeerId()) |
|
|
@ -349,7 +351,6 @@ func (s *stateSync) loop() (err error) { |
|
|
|
return errCanceled |
|
|
|
return errCanceled |
|
|
|
|
|
|
|
|
|
|
|
case req := <-s.deliver: |
|
|
|
case req := <-s.deliver: |
|
|
|
deliveryTime := time.Now() |
|
|
|
|
|
|
|
// Response, disconnect or timeout triggered, drop the peer if stalling
|
|
|
|
// Response, disconnect or timeout triggered, drop the peer if stalling
|
|
|
|
log.Trace("Received node data response", "peer", req.peer.id, "count", len(req.response), "dropped", req.dropped, "timeout", !req.dropped && req.timedOut()) |
|
|
|
log.Trace("Received node data response", "peer", req.peer.id, "count", len(req.response), "dropped", req.dropped, "timeout", !req.dropped && req.timedOut()) |
|
|
|
if req.nItems <= 2 && !req.dropped && req.timedOut() { |
|
|
|
if req.nItems <= 2 && !req.dropped && req.timedOut() { |
|
|
@ -376,7 +377,7 @@ func (s *stateSync) loop() (err error) { |
|
|
|
} |
|
|
|
} |
|
|
|
// Process all the received blobs and check for stale delivery
|
|
|
|
// Process all the received blobs and check for stale delivery
|
|
|
|
delivered, err := s.process(req) |
|
|
|
delivered, err := s.process(req) |
|
|
|
req.peer.SetNodeDataIdle(delivered, deliveryTime) |
|
|
|
req.peer.SetNodeDataIdle(delivered, req.delivered) |
|
|
|
if err != nil { |
|
|
|
if err != nil { |
|
|
|
log.Warn("Node data write error", "err", err) |
|
|
|
log.Warn("Node data write error", "err", err) |
|
|
|
return err |
|
|
|
return err |
|
|
|