Add Torrent-level request cancel for consistency

peerconn.go

@@ -1550,8 +1550,10 @@ func (c *Peer) deleteRequest(r RequestIndex) bool {
 		f(PeerRequestEvent{c, c.t.requestIndexToRequest(r)})
 	}
 	c.updateExpectingChunks()
-	delete(c.t.pendingRequests, r)
-	delete(c.t.lastRequested, r)
+	if c.t.requestingPeer(r) == c {
+		delete(c.t.pendingRequests, r)
+		delete(c.t.lastRequested, r)
+	}
 	return true
 }
 
@@ -1694,6 +1696,10 @@ func (pc *PeerConn) isLowOnRequests() bool {
 	return pc.actualRequestState.Requests.IsEmpty()
 }
 
+func (p *Peer) uncancelledRequests() uint64 {
+	return p.actualRequestState.Requests.GetCardinality() - p.cancelledRequests.GetCardinality()
+}
+
 func (pc *PeerConn) remoteIsTransmission() bool {
 	return bytes.HasPrefix(pc.PeerID[:], []byte("-TR")) && pc.PeerID[7] == '-'
 }

requesting.go

@@ -211,6 +211,7 @@ func (p *Peer) applyRequestState(next desiredRequestState) bool {
 	}
 	more := true
 	requestHeap := &next.Requests
+	t := p.t
 	heap.Init(requestHeap)
 	for requestHeap.Len() != 0 && maxRequests(current.Requests.GetCardinality()) < p.nominalMaxRequests() {
 		req := heap.Pop(requestHeap).(RequestIndex)
@@ -219,9 +220,9 @@ func (p *Peer) applyRequestState(next desiredRequestState) bool {
 			// requests, so we can skip this one with no additional consideration.
 			continue
 		}
-		existing := p.t.pendingRequests[req]
-		if existing != nil && existing != p && existing.actualRequestState.Requests.GetCardinality()-existing.cancelledRequests.GetCardinality() > current.Requests.GetCardinality() {
-			existing.cancel(req)
+		existing := t.requestingPeer(req)
+		if existing != nil && existing != p && existing.uncancelledRequests() > current.Requests.GetCardinality() {
+			t.cancelRequest(req)
 		}
 		more = p.mustRequest(req)
 		if !more {

torrent.go

@@ -2310,3 +2310,16 @@ func (t *Torrent) pieceRequestIndexOffset(piece pieceIndex) RequestIndex {
 func (t *Torrent) updateComplete() {
 	t.Complete.SetBool(t.haveAllPieces())
 }
+
+func (t *Torrent) cancelRequest(r RequestIndex) *Peer {
+	p := t.pendingRequests[r]
+	if p != nil {
+		p.cancel(r)
+	}
+	delete(t.pendingRequests, r)
+	return p
+}
+
+func (t *Torrent) requestingPeer(r RequestIndex) *Peer {
+	return t.pendingRequests[r]
+}