11 "github.com/RoaringBitmap/roaring"
12 "github.com/anacrolix/log"
13 "github.com/anacrolix/torrent/metainfo"
14 pp "github.com/anacrolix/torrent/peer_protocol"
15 "github.com/anacrolix/torrent/webseed"
18 type webseedPeer struct {
20 activeRequests map[Request]webseed.Request
21 requesterCond sync.Cond
23 // Number of requester routines.
27 var _ peerImpl = (*webseedPeer)(nil)
29 func (me *webseedPeer) connStatusString() string {
33 func (ws *webseedPeer) String() string {
34 return fmt.Sprintf("webseed peer for %q", ws.client.Url)
37 func (ws *webseedPeer) onGotInfo(info *metainfo.Info) {
38 ws.client.SetInfo(info)
41 func (ws *webseedPeer) writeInterested(interested bool) bool {
45 func (ws *webseedPeer) _cancel(r RequestIndex) bool {
46 active, ok := ws.activeRequests[ws.peer.t.requestIndexToRequest(r)]
49 if !ws.peer.deleteRequest(r) {
50 panic("cancelled webseed request should exist")
52 if ws.peer.isLowOnRequests() {
53 ws.peer.updateRequests("webseedPeer._cancel")
59 func (ws *webseedPeer) intoSpec(r Request) webseed.RequestSpec {
60 return webseed.RequestSpec{ws.peer.t.requestOffset(r), int64(r.Length)}
63 func (ws *webseedPeer) _request(r Request) bool {
64 ws.requesterCond.Signal()
68 func (ws *webseedPeer) doRequest(r Request) {
69 webseedRequest := ws.client.NewRequest(ws.intoSpec(r))
70 ws.activeRequests[r] = webseedRequest
72 ws.requesterCond.L.Unlock()
73 defer ws.requesterCond.L.Lock()
74 ws.requestResultHandler(r, webseedRequest)
76 delete(ws.activeRequests, r)
79 func (ws *webseedPeer) requester() {
80 ws.requesterCond.L.Lock()
81 defer ws.requesterCond.L.Unlock()
83 for !ws.peer.closed.IsSet() {
85 ws.peer.actualRequestState.Requests.Iterate(func(x uint32) bool {
86 r := ws.peer.t.requestIndexToRequest(x)
87 if _, ok := ws.activeRequests[r]; ok {
97 ws.requesterCond.Wait()
101 func (ws *webseedPeer) connectionFlags() string {
105 // TODO: This is called when banning peers. Perhaps we want to be able to ban webseeds too. We could
106 // return bool if this is even possible, and if it isn't, skip to the next drop candidate.
107 func (ws *webseedPeer) drop() {}
109 func (ws *webseedPeer) handleUpdateRequests() {
110 ws.peer.maybeUpdateActualRequestState()
113 func (ws *webseedPeer) onClose() {
114 ws.peer.logger.WithLevel(log.Debug).Print("closing")
115 for _, r := range ws.activeRequests {
118 ws.requesterCond.Broadcast()
121 func (ws *webseedPeer) requestResultHandler(r Request, webseedRequest webseed.Request) {
122 result := <-webseedRequest.Result
123 // We do this here rather than inside receiveChunk, since we want to count errors too. I'm not
124 // sure if we can divine which errors indicate cancellation on our end without hitting the
126 ws.peer.doChunkReadStats(int64(len(result.Bytes)))
127 ws.peer.readBytes(int64(len(result.Bytes)))
129 defer ws.peer.t.cl.unlock()
130 if result.Err != nil {
131 if !errors.Is(result.Err, context.Canceled) {
132 ws.peer.logger.Printf("Request %v rejected: %v", r, result.Err)
134 // We need to filter out temporary errors, but this is a nightmare in Go. Currently a bad
135 // webseed URL can starve out the good ones due to the chunk selection algorithm.
136 const closeOnAllErrors = false
137 if closeOnAllErrors ||
138 strings.Contains(result.Err.Error(), "unsupported protocol scheme") ||
140 var err webseed.ErrBadResponse
141 if !errors.As(result.Err, &err) {
144 return err.Response.StatusCode == http.StatusNotFound
148 ws.peer.remoteRejectedRequest(ws.peer.t.requestIndexFromRequest(r))
151 err := ws.peer.receiveChunk(&pp.Message{
163 func (me *webseedPeer) isLowOnRequests() bool {
164 return me.peer.actualRequestState.Requests.GetCardinality() < uint64(me.maxRequests)
167 func (me *webseedPeer) peerPieces() *roaring.Bitmap {
168 return &me.client.Pieces
171 func (cn *webseedPeer) peerHasAllPieces() (all, known bool) {
172 if !cn.peer.t.haveInfo() {
175 return cn.client.Pieces.GetCardinality() == uint64(cn.peer.t.numPieces()), true