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                         // Demeter is throwing a tantrum on Mount Olympus for this
 113                         ws.peer.t.cl.locker().RLock()
 114                         duration := time.Until(ws.lastUnhandledErr.Add(webseedPeerUnhandledErrorSleep))
 115                         ws.peer.t.cl.locker().RUnlock()
 116                         time.Sleep(duration)
 117                         ws.requesterCond.L.Lock()
 118                         return false
 119                 })
 120                 if restart {
 121                         goto start
 122                 }
 123                 ws.requesterCond.Wait()
 124         }
 125 }
 126
 127 func (ws *webseedPeer) connectionFlags() string {
 128         return "WS"
 129 }
 130
 131 // Maybe this should drop all existing connections, or something like that.
 132 func (ws *webseedPeer) drop() {}
 133
 134 func (cn *webseedPeer) ban() {
 135         cn.peer.close()
 136 }
 137
 138 func (ws *webseedPeer) handleUpdateRequests() {
 139         // Because this is synchronous, webseed peers seem to get first dibs on newly prioritized
 140         // pieces.
 141         go func() {
 142                 ws.peer.t.cl.lock()
 143                 defer ws.peer.t.cl.unlock()
 144                 ws.peer.maybeUpdateActualRequestState()
 145         }()
 146 }
 147
 148 func (ws *webseedPeer) onClose() {
 149         ws.peer.logger.Levelf(log.Debug, "closing")
 150         // Just deleting them means we would have to manually cancel active requests.
 151         ws.peer.cancelAllRequests()
 152         ws.peer.t.iterPeers(func(p *Peer) {
 153                 if p.isLowOnRequests() {
 154                         p.updateRequests("webseedPeer.onClose")
 155                 }
 156         })
 157         ws.requesterCond.Broadcast()
 158 }
 159
 160 func (ws *webseedPeer) requestResultHandler(r Request, webseedRequest webseed.Request) error {
 161         result := <-webseedRequest.Result
 162         close(webseedRequest.Result) // one-shot
 163         // We do this here rather than inside receiveChunk, since we want to count errors too. I'm not
 164         // sure if we can divine which errors indicate cancellation on our end without hitting the
 165         // network though.
 166         if len(result.Bytes) != 0 || result.Err == nil {
 167                 // Increment ChunksRead and friends
 168                 ws.peer.doChunkReadStats(int64(len(result.Bytes)))
 169         }
 170         ws.peer.readBytes(int64(len(result.Bytes)))
 171         ws.peer.t.cl.lock()
 172         defer ws.peer.t.cl.unlock()
 173         if ws.peer.t.closed.IsSet() {
 174                 return nil
 175         }
 176         err := result.Err
 177         if err != nil {
 178                 switch {
 179                 case errors.Is(err, context.Canceled):
 180                 case errors.Is(err, webseed.ErrTooFast):
 181                 case ws.peer.closed.IsSet():
 182                 default:
 183                         ws.peer.logger.Printf("Request %v rejected: %v", r, result.Err)
 184                         // // Here lies my attempt to extract something concrete from Go's error system. RIP.
 185                         // cfg := spew.NewDefaultConfig()
 186                         // cfg.DisableMethods = true
 187                         // cfg.Dump(result.Err)
 188
 189                         if webseedPeerCloseOnUnhandledError {
 190                                 log.Printf("closing %v", ws)
 191                                 ws.peer.close()
 192                         } else {
 193                                 ws.lastUnhandledErr = time.Now()
 194                         }
 195                 }
 196                 if !ws.peer.remoteRejectedRequest(ws.peer.t.requestIndexFromRequest(r)) {
 197                         panic("invalid reject")
 198                 }
 199                 return err
 200         }
 201         err = ws.peer.receiveChunk(&pp.Message{
 202                 Type:  pp.Piece,
 203                 Index: r.Index,
 204                 Begin: r.Begin,
 205                 Piece: result.Bytes,
 206         })
 207         if err != nil {
 208                 panic(err)
 209         }
 210         return err
 211 }
 212
 213 func (me *webseedPeer) peerPieces() *roaring.Bitmap {
 214         return &me.client.Pieces
 215 }
 216
 217 func (cn *webseedPeer) peerHasAllPieces() (all, known bool) {
 218         if !cn.peer.t.haveInfo() {
 219                 return true, false
 220         }
 221         return cn.client.Pieces.GetCardinality() == uint64(cn.peer.t.numPieces()), true
 222 }