webseed-peer.go

   1 package torrent
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "math/rand"
   8         "sync"
   9         "time"
  10
  11         "github.com/RoaringBitmap/roaring"
  12         "github.com/anacrolix/log"
  13
  14         "github.com/anacrolix/torrent/metainfo"
  15         pp "github.com/anacrolix/torrent/peer_protocol"
  16         "github.com/anacrolix/torrent/webseed"
  17 )
  18
  19 const (
  20         webseedPeerUnhandledErrorSleep   = 5 * time.Second
  21         webseedPeerCloseOnUnhandledError = false
  22 )
  23
  24 type webseedPeer struct {
  25         // First field for stats alignment.
  26         peer             Peer
  27         client           webseed.Client
  28         activeRequests   map[Request]webseed.Request
  29         requesterCond    sync.Cond
  30         lastUnhandledErr time.Time
  31 }
  32
  33 var _ peerImpl = (*webseedPeer)(nil)
  34
  35 func (me *webseedPeer) peerImplStatusLines() []string {
  36         return []string{
  37                 me.client.Url,
  38                 fmt.Sprintf("last unhandled error: %v", eventAgeString(me.lastUnhandledErr)),
  39         }
  40 }
  41
  42 func (ws *webseedPeer) String() string {
  43         return fmt.Sprintf("webseed peer for %q", ws.client.Url)
  44 }
  45
  46 func (ws *webseedPeer) onGotInfo(info *metainfo.Info) {
  47         ws.client.SetInfo(info)
  48         // There should be probably be a callback in Client instead, so it can remove pieces at its whim
  49         // too.
  50         ws.client.Pieces.Iterate(func(x uint32) bool {
  51                 ws.peer.t.incPieceAvailability(pieceIndex(x))
  52                 return true
  53         })
  54 }
  55
  56 func (ws *webseedPeer) writeInterested(interested bool) bool {
  57         return true
  58 }
  59
  60 func (ws *webseedPeer) _cancel(r RequestIndex) bool {
  61         if active, ok := ws.activeRequests[ws.peer.t.requestIndexToRequest(r)]; ok {
  62                 active.Cancel()
  63                 // The requester is running and will handle the result.
  64                 return true
  65         }
  66         // There should be no requester handling this, so no further events will occur.
  67         return false
  68 }
  69
  70 func (ws *webseedPeer) intoSpec(r Request) webseed.RequestSpec {
  71         return webseed.RequestSpec{ws.peer.t.requestOffset(r), int64(r.Length)}
  72 }
  73
  74 func (ws *webseedPeer) _request(r Request) bool {
  75         ws.requesterCond.Signal()
  76         return true
  77 }
  78
  79 func (ws *webseedPeer) doRequest(r Request) error {
  80         webseedRequest := ws.client.NewRequest(ws.intoSpec(r))
  81         ws.activeRequests[r] = webseedRequest
  82         err := func() error {
  83                 ws.requesterCond.L.Unlock()
  84                 defer ws.requesterCond.L.Lock()
  85                 return ws.requestResultHandler(r, webseedRequest)
  86         }()
  87         delete(ws.activeRequests, r)
  88         return err
  89 }
  90
  91 func (ws *webseedPeer) requester(i int) {
  92         ws.requesterCond.L.Lock()
  93         defer ws.requesterCond.L.Unlock()
  94 start:
  95         for !ws.peer.closed.IsSet() {
  96                 // Restart is set if we don't need to wait for the requestCond before trying again.
  97                 restart := false
  98                 ws.peer.requestState.Requests.Iterate(func(x RequestIndex) bool {
  99                         r := ws.peer.t.requestIndexToRequest(x)
 100                         if _, ok := ws.activeRequests[r]; ok {
 101                                 return true
 102                         }
 103                         err := ws.doRequest(r)
 104                         ws.requesterCond.L.Unlock()
 105                         if err != nil && !errors.Is(err, context.Canceled) {
 106                                 log.Printf("requester %v: error doing webseed request %v: %v", i, r, err)
 107                         }
 108                         restart = true
 109                         if errors.Is(err, webseed.ErrTooFast) {
 110                                 time.Sleep(time.Duration(rand.Int63n(int64(10 * time.Second))))
 111                         }
 112                         time.Sleep(time.Until(ws.lastUnhandledErr.Add(webseedPeerUnhandledErrorSleep)))
 113                         ws.requesterCond.L.Lock()
 114                         return false
 115                 })
 116                 if restart {
 117                         goto start
 118                 }
 119                 ws.requesterCond.Wait()
 120         }
 121 }
 122
 123 func (ws *webseedPeer) connectionFlags() string {
 124         return "WS"
 125 }
 126
 127 // Maybe this should drop all existing connections, or something like that.
 128 func (ws *webseedPeer) drop() {}
 129
 130 func (cn *webseedPeer) ban() {
 131         cn.peer.close()
 132 }
 133
 134 func (ws *webseedPeer) handleUpdateRequests() {
 135         // Because this is synchronous, webseed peers seem to get first dibs on newly prioritized
 136         // pieces.
 137         go func() {
 138                 ws.peer.t.cl.lock()
 139                 defer ws.peer.t.cl.unlock()
 140                 ws.peer.maybeUpdateActualRequestState()
 141         }()
 142 }
 143
 144 func (ws *webseedPeer) onClose() {
 145         ws.peer.logger.Levelf(log.Debug, "closing")
 146         // Just deleting them means we would have to manually cancel active requests.
 147         ws.peer.cancelAllRequests()
 148         ws.peer.t.iterPeers(func(p *Peer) {
 149                 if p.isLowOnRequests() {
 150                         p.updateRequests("webseedPeer.onClose")
 151                 }
 152         })
 153         ws.requesterCond.Broadcast()
 154 }
 155
 156 func (ws *webseedPeer) requestResultHandler(r Request, webseedRequest webseed.Request) error {
 157         result := <-webseedRequest.Result
 158         close(webseedRequest.Result) // one-shot
 159         // We do this here rather than inside receiveChunk, since we want to count errors too. I'm not
 160         // sure if we can divine which errors indicate cancellation on our end without hitting the
 161         // network though.
 162         if len(result.Bytes) != 0 || result.Err == nil {
 163                 // Increment ChunksRead and friends
 164                 ws.peer.doChunkReadStats(int64(len(result.Bytes)))
 165         }
 166         ws.peer.readBytes(int64(len(result.Bytes)))
 167         ws.peer.t.cl.lock()
 168         defer ws.peer.t.cl.unlock()
 169         if ws.peer.t.closed.IsSet() {
 170                 return nil
 171         }
 172         err := result.Err
 173         if err != nil {
 174                 switch {
 175                 case errors.Is(err, context.Canceled):
 176                 case errors.Is(err, webseed.ErrTooFast):
 177                 case ws.peer.closed.IsSet():
 178                 default:
 179                         ws.peer.logger.Printf("Request %v rejected: %v", r, result.Err)
 180                         // // Here lies my attempt to extract something concrete from Go's error system. RIP.
 181                         // cfg := spew.NewDefaultConfig()
 182                         // cfg.DisableMethods = true
 183                         // cfg.Dump(result.Err)
 184
 185                         if webseedPeerCloseOnUnhandledError {
 186                                 log.Printf("closing %v", ws)
 187                                 ws.peer.close()
 188                         } else {
 189                                 ws.lastUnhandledErr = time.Now()
 190                         }
 191                 }
 192                 if !ws.peer.remoteRejectedRequest(ws.peer.t.requestIndexFromRequest(r)) {
 193                         panic("invalid reject")
 194                 }
 195                 return err
 196         }
 197         err = ws.peer.receiveChunk(&pp.Message{
 198                 Type:  pp.Piece,
 199                 Index: r.Index,
 200                 Begin: r.Begin,
 201                 Piece: result.Bytes,
 202         })
 203         if err != nil {
 204                 panic(err)
 205         }
 206         return err
 207 }
 208
 209 func (me *webseedPeer) peerPieces() *roaring.Bitmap {
 210         return &me.client.Pieces
 211 }
 212
 213 func (cn *webseedPeer) peerHasAllPieces() (all, known bool) {
 214         if !cn.peer.t.haveInfo() {
 215                 return true, false
 216         }
 217         return cn.client.Pieces.GetCardinality() == uint64(cn.peer.t.numPieces()), true
 218 }