Stop reading webseed responses if chunks aren't wanted

diff --git a/peer-impl.go b/peer-impl.go
index 211097e84d9d41dda5a2b8a9d7f78f15ae8735b0..98d3aa7cd24b0d144cdc8afe5f25d45d84fb756d 100644 (file)
--- a/peer-impl.go
+++ b/peer-impl.go
@@ -14,7 +14,6 @@ import (
 type legacyPeerImpl interface {
        // Trigger the actual request state to get updated
        handleOnNeedUpdateRequests()
-       writeInterested(interested bool) bool
        // Actually go ahead and modify the pending requests.
        updateRequests()
 

diff --git a/peer.go b/peer.go
index 4ea6eb8789c0e4d40949604db801d3cf412a984d..26880bf361cdcd4b27a1764e5a306085acced07d 100644 (file)
--- a/peer.go
+++ b/peer.go
@@ -380,7 +380,7 @@ func (cn *Peer) totalExpectingTime() (ret time.Duration) {
        return
 }
 
-func (cn *Peer) setInterested(interested bool) bool {
+func (cn *PeerConn) setInterested(interested bool) bool {
        if cn.requestState.Interested == interested {
                return true
        }

diff --git a/requesting.go b/requesting.go
index 33232f14983e90d1eb6b12651875ac8beb9f97e8..f226e05aef5be93d6e791c50eb0f12326a4c17dc 100644 (file)
--- a/requesting.go
+++ b/requesting.go
@@ -76,7 +76,8 @@ func (p *peerId) GobDecode(b []byte) error {
 type (
        // A request index is a chunk indexed across the entire torrent. It's a single integer and can
        // be converted to a protocol request. TODO: This should be private.
-       RequestIndex   = requestStrategy.RequestIndex
+       RequestIndex = requestStrategy.RequestIndex
+       // This is request index but per-piece.
        chunkIndexType = requestStrategy.ChunkIndex
 )
 

diff --git a/torrent.go b/torrent.go
index 37f304dfcf8fb8040e39d42d10f03a6dc9015f4a..21b00514a1eaa91142abef3648e45ad386844379 100644 (file)
--- a/torrent.go
+++ b/torrent.go
@@ -1327,7 +1327,11 @@ func (t *Torrent) maybeDropMutuallyCompletePeer(
        p.drop()
 }
 
-func (t *Torrent) haveChunk(r Request) (ret bool) {
+func (t *Torrent) haveRequestIndexChunk(reqIndex RequestIndex) bool {
+       return t.haveChunk(t.requestIndexToRequest(reqIndex))
+}
+
+func (t *Torrent) haveChunk(r Request) bool {
        if !t.haveInfo() {
                return false
        }
@@ -3500,3 +3504,14 @@ func (t *Torrent) endRequestIndexForFileIndex(fileIndex int) RequestIndex {
        end := intCeilDiv(uint64(f.offset)+uint64(f.length), t.chunkSize.Uint64())
        return RequestIndex(end)
 }
+
+func (t *Torrent) wantReceiveChunk(reqIndex RequestIndex) bool {
+       pi := t.pieceIndexOfRequestIndex(reqIndex)
+       if !t.wantPieceIndex(pi) {
+               return false
+       }
+       if t.haveRequestIndexChunk(reqIndex) {
+               return false
+       }
+       return true
+}

diff --git a/webseed-peer.go b/webseed-peer.go
index 0f5a29447e5eb836693c67bde67d3639fef7fee3..a8f453c919dbde02da48e3369681917ce6c0a7ea 100644 (file)
--- a/webseed-peer.go
+++ b/webseed-peer.go
@@ -82,15 +82,8 @@ func (ws *webseedPeer) onGotInfo(info *metainfo.Info) {
        })
 }
 
-func (ws *webseedPeer) writeInterested(interested bool) bool {
-       return true
-}
-
-func (ws *webseedPeer) handleCancel(r RequestIndex) {
-       for wr := range ws.activeRequestsForIndex(r) {
-               wr.Cancel()
-       }
-}
+// Webseeds check the next request is wanted before reading it.
+func (ws *webseedPeer) handleCancel(RequestIndex) {}
 
 func (ws *webseedPeer) activeRequestsForIndex(r RequestIndex) iter.Seq[*webseedRequest] {
        return func(yield func(*webseedRequest) bool) {
@@ -268,6 +261,24 @@ func (ws *webseedPeer) onClose() {
        })
 }
 
+// Do we want a chunk, assuming it's valid etc.
+func (ws *webseedPeer) wantChunk(ri RequestIndex) bool {
+       return ws.peer.t.wantReceiveChunk(ri)
+}
+
+func (ws *webseedPeer) maxChunkDiscard() RequestIndex {
+       return RequestIndex(int(intCeilDiv(webseed.MaxDiscardBytes, ws.peer.t.chunkSize)))
+}
+
+func (ws *webseedPeer) keepReading(wr *webseedRequest) bool {
+       for ri := wr.next; ri < wr.end && ri < wr.next+ws.maxChunkDiscard(); ri++ {
+               if ws.wantChunk(ri) {
+                       return true
+               }
+       }
+       return false
+}
+
 func (ws *webseedPeer) readChunks(wr *webseedRequest) (err error) {
        t := ws.peer.t
        buf := t.getChunkBuffer()
@@ -275,7 +286,7 @@ func (ws *webseedPeer) readChunks(wr *webseedRequest) (err error) {
        msg := pp.Message{
                Type: pp.Piece,
        }
-       for wr.next < wr.end {
+       for ws.keepReading(wr) {
                reqSpec := t.requestIndexToRequest(wr.next)
                chunkLen := reqSpec.Length.Int()
                buf = buf[:chunkLen]

diff --git a/webseed-request.go b/webseed-request.go
index 7e601d6b00cce199c3ff15b3d3a2a13c9848ef99..51f76199b46997a03279980e8f570180c10dcfc5 100644 (file)
--- a/webseed-request.go
+++ b/webseed-request.go
@@ -1,6 +1,8 @@
 package torrent
 
 import (
+       "fmt"
+
        "github.com/anacrolix/torrent/webseed"
 )
 
@@ -23,6 +25,11 @@ func (me *webseedRequest) Close() {
 
 // Record that it was exceptionally cancelled.
 func (me *webseedRequest) Cancel() {
-       me.cancelled = true
        me.request.Cancel()
+       if !me.cancelled {
+               me.cancelled = true
+               if webseed.PrintDebug {
+                       fmt.Printf("cancelled webseed request\n")
+               }
+       }
 }

diff --git a/webseed/client.go b/webseed/client.go
index 52d4bb818a91260af1954ebe760e8fef703ffbf8..edf1ef6e9a2c140f7aa8c4b9147fa4a54ee6c217 100644 (file)
--- a/webseed/client.go
+++ b/webseed/client.go
@@ -12,11 +12,21 @@ import (
 
        "github.com/RoaringBitmap/roaring"
        "github.com/anacrolix/missinggo/v2/panicif"
+       "github.com/dustin/go-humanize"
 
        "github.com/anacrolix/torrent/metainfo"
        "github.com/anacrolix/torrent/segments"
 )
 
+// How many consecutive bytes to allow discarding from responses. This number is based on
+// https://archive.org/download/BloodyPitOfHorror/BloodyPitOfHorror.asr.srt. It seems that
+// archive.org might be using a webserver implementation that refuses to do partial responses to
+// small files.
+const MaxDiscardBytes = 48 << 10
+
+// Output debug information to stdout.
+const PrintDebug = false
+
 type RequestSpec = segments.Extent
 
 type requestPart struct {
@@ -45,12 +55,14 @@ type Client struct {
        // Max concurrent requests to a WebSeed for a given torrent.
        MaxRequests int
 
+       // TODO: Share this with Torrent.
        fileIndex segments.Index
        info      *metainfo.Info
        // The pieces we can request with the Url. We're more likely to ban/block at the file-level
        // given that's how requests are mapped to webseeds, but the torrent.Client works at the piece
        // level. We can map our file-level adjustments to the pieces here. This probably need to be
-       // private in the future, if Client ever starts removing pieces.
+       // private in the future, if Client ever starts removing pieces. TODO: This belongs in
+       // webseedPeer.
        Pieces roaring.Bitmap
        // This wraps http.Response bodies, for example to limit the download rate.
        ResponseBodyWrapper ResponseBodyWrapper
@@ -98,6 +110,14 @@ func (ws *Client) StartNewRequest(r RequestSpec) Request {
                        responseBodyWrapper: ws.ResponseBodyWrapper,
                }
                part.do = func() (*http.Response, error) {
+                       if PrintDebug {
+                               fmt.Printf(
+                                       "doing request for %q (file size %v), Range: %q\n",
+                                       req.URL,
+                                       humanize.Bytes(uint64(ws.fileIndex.Index(i).Length)),
+                                       req.Header.Get("Range"),
+                               )
+                       }
                        return ws.HttpClient.Do(req)
                }
                requestParts = append(requestParts, part)
@@ -171,12 +191,8 @@ func (me *Client) recvPartResult(ctx context.Context, w io.Writer, part requestP
        case http.StatusOK:
                // The response is from the beginning.
                me.checkContentLength(resp, part, part.e.End())
-               // This number is based on
-               // https://archive.org/download/BloodyPitOfHorror/BloodyPitOfHorror.asr.srt. It seems that
-               // archive.org might be using a webserver implementation that refuses to do partial
-               // responses to small files.
                discard := part.e.Start
-               if discard > 48<<10 {
+               if discard > MaxDiscardBytes {
                        return ErrBadResponse{"resp status ok but requested range", resp}
                }
                if discard != 0 {