webseed-peer.go

   1 package torrent
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "net/http"
   8         "strings"
   9         "sync"
  10
  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"
  16 )
  17
  18 type webseedPeer struct {
  19         client         webseed.Client
  20         activeRequests map[Request]webseed.Request
  21         requesterCond  sync.Cond
  22         peer           Peer
  23         // Number of requester routines.
  24         maxRequests int
  25 }
  26
  27 var _ peerImpl = (*webseedPeer)(nil)
  28
  29 func (me *webseedPeer) connStatusString() string {
  30         return me.client.Url
  31 }
  32
  33 func (ws *webseedPeer) String() string {
  34         return fmt.Sprintf("webseed peer for %q", ws.client.Url)
  35 }
  36
  37 func (ws *webseedPeer) onGotInfo(info *metainfo.Info) {
  38         ws.client.SetInfo(info)
  39         // There should be probably be a callback in Client instead, so it can remove pieces at its whim
  40         // too.
  41         ws.client.Pieces.Iterate(func(x uint32) bool {
  42                 ws.peer.t.incPieceAvailability(pieceIndex(x))
  43                 return true
  44         })
  45 }
  46
  47 func (ws *webseedPeer) writeInterested(interested bool) bool {
  48         return true
  49 }
  50
  51 func (ws *webseedPeer) _cancel(r RequestIndex) bool {
  52         active, ok := ws.activeRequests[ws.peer.t.requestIndexToRequest(r)]
  53         if ok {
  54                 active.Cancel()
  55                 if !ws.peer.deleteRequest(r) {
  56                         panic("cancelled webseed request should exist")
  57                 }
  58                 if ws.peer.isLowOnRequests() {
  59                         ws.peer.updateRequests("webseedPeer._cancel")
  60                 }
  61         }
  62         return true
  63 }
  64
  65 func (ws *webseedPeer) intoSpec(r Request) webseed.RequestSpec {
  66         return webseed.RequestSpec{ws.peer.t.requestOffset(r), int64(r.Length)}
  67 }
  68
  69 func (ws *webseedPeer) _request(r Request) bool {
  70         ws.requesterCond.Signal()
  71         return true
  72 }
  73
  74 func (ws *webseedPeer) doRequest(r Request) {
  75         webseedRequest := ws.client.NewRequest(ws.intoSpec(r))
  76         ws.activeRequests[r] = webseedRequest
  77         func() {
  78                 ws.requesterCond.L.Unlock()
  79                 defer ws.requesterCond.L.Lock()
  80                 ws.requestResultHandler(r, webseedRequest)
  81         }()
  82         delete(ws.activeRequests, r)
  83 }
  84
  85 func (ws *webseedPeer) requester() {
  86         ws.requesterCond.L.Lock()
  87         defer ws.requesterCond.L.Unlock()
  88 start:
  89         for !ws.peer.closed.IsSet() {
  90                 restart := false
  91                 ws.peer.actualRequestState.Requests.Iterate(func(x uint32) bool {
  92                         r := ws.peer.t.requestIndexToRequest(x)
  93                         if _, ok := ws.activeRequests[r]; ok {
  94                                 return true
  95                         }
  96                         ws.doRequest(r)
  97                         restart = true
  98                         return false
  99                 })
 100                 if restart {
 101                         goto start
 102                 }
 103                 ws.requesterCond.Wait()
 104         }
 105 }
 106
 107 func (ws *webseedPeer) connectionFlags() string {
 108         return "WS"
 109 }
 110
 111 // TODO: This is called when banning peers. Perhaps we want to be able to ban webseeds too. We could
 112 // return bool if this is even possible, and if it isn't, skip to the next drop candidate.
 113 func (ws *webseedPeer) drop() {}
 114
 115 func (ws *webseedPeer) handleUpdateRequests() {
 116         ws.peer.maybeUpdateActualRequestState()
 117 }
 118
 119 func (ws *webseedPeer) onClose() {
 120         ws.peer.logger.WithLevel(log.Debug).Print("closing")
 121         for _, r := range ws.activeRequests {
 122                 r.Cancel()
 123         }
 124         ws.requesterCond.Broadcast()
 125 }
 126
 127 func (ws *webseedPeer) requestResultHandler(r Request, webseedRequest webseed.Request) {
 128         result := <-webseedRequest.Result
 129         // We do this here rather than inside receiveChunk, since we want to count errors too. I'm not
 130         // sure if we can divine which errors indicate cancellation on our end without hitting the
 131         // network though.
 132         ws.peer.doChunkReadStats(int64(len(result.Bytes)))
 133         ws.peer.readBytes(int64(len(result.Bytes)))
 134         ws.peer.t.cl.lock()
 135         defer ws.peer.t.cl.unlock()
 136         if result.Err != nil {
 137                 if !errors.Is(result.Err, context.Canceled) {
 138                         ws.peer.logger.Printf("Request %v rejected: %v", r, result.Err)
 139                 }
 140                 // We need to filter out temporary errors, but this is a nightmare in Go. Currently a bad
 141                 // webseed URL can starve out the good ones due to the chunk selection algorithm.
 142                 const closeOnAllErrors = false
 143                 if closeOnAllErrors ||
 144                         strings.Contains(result.Err.Error(), "unsupported protocol scheme") ||
 145                         func() bool {
 146                                 var err webseed.ErrBadResponse
 147                                 if !errors.As(result.Err, &err) {
 148                                         return false
 149                                 }
 150                                 return err.Response.StatusCode == http.StatusNotFound
 151                         }() {
 152                         ws.peer.close()
 153                 } else {
 154                         ws.peer.remoteRejectedRequest(ws.peer.t.requestIndexFromRequest(r))
 155                 }
 156         } else {
 157                 err := ws.peer.receiveChunk(&pp.Message{
 158                         Type:  pp.Piece,
 159                         Index: r.Index,
 160                         Begin: r.Begin,
 161                         Piece: result.Bytes,
 162                 })
 163                 if err != nil {
 164                         panic(err)
 165                 }
 166         }
 167 }
 168
 169 func (me *webseedPeer) isLowOnRequests() bool {
 170         return me.peer.actualRequestState.Requests.GetCardinality() < uint64(me.maxRequests)
 171 }
 172
 173 func (me *webseedPeer) peerPieces() *roaring.Bitmap {
 174         return &me.client.Pieces
 175 }
 176
 177 func (cn *webseedPeer) peerHasAllPieces() (all, known bool) {
 178         if !cn.peer.t.haveInfo() {
 179                 return true, false
 180         }
 181         return cn.client.Pieces.GetCardinality() == uint64(cn.peer.t.numPieces()), true
 182 }