tracker_scraper.go

   1 package torrent
   2
   3 import (
   4         "bytes"
   5         "errors"
   6         "fmt"
   7         "net"
   8         "net/url"
   9         "time"
  10
  11         "github.com/anacrolix/dht/v2/krpc"
  12         "github.com/anacrolix/log"
  13
  14         "github.com/anacrolix/torrent/tracker"
  15 )
  16
  17 // Announces a torrent to a tracker at regular intervals, when peers are
  18 // required.
  19 type trackerScraper struct {
  20         u            url.URL
  21         t            *Torrent
  22         lastAnnounce trackerAnnounceResult
  23         allow        func()
  24         // The slowdown argument lets us indicate if we think there should be some backpressure on
  25         // access to the tracker. It doesn't necessarily have to be used.
  26         done func(slowdown bool)
  27 }
  28
  29 type torrentTrackerAnnouncer interface {
  30         statusLine() string
  31         URL() *url.URL
  32 }
  33
  34 func (me trackerScraper) URL() *url.URL {
  35         return &me.u
  36 }
  37
  38 func (ts *trackerScraper) statusLine() string {
  39         var w bytes.Buffer
  40         fmt.Fprintf(&w, "next ann: %v, last ann: %v",
  41                 func() string {
  42                         na := time.Until(ts.lastAnnounce.Completed.Add(ts.lastAnnounce.Interval))
  43                         if na > 0 {
  44                                 na /= time.Second
  45                                 na *= time.Second
  46                                 return na.String()
  47                         } else {
  48                                 return "anytime"
  49                         }
  50                 }(),
  51                 func() string {
  52                         if ts.lastAnnounce.Err != nil {
  53                                 return ts.lastAnnounce.Err.Error()
  54                         }
  55                         if ts.lastAnnounce.Completed.IsZero() {
  56                                 return "never"
  57                         }
  58                         return fmt.Sprintf("%d peers", ts.lastAnnounce.NumPeers)
  59                 }(),
  60         )
  61         return w.String()
  62 }
  63
  64 type trackerAnnounceResult struct {
  65         Err       error
  66         NumPeers  int
  67         Interval  time.Duration
  68         Completed time.Time
  69 }
  70
  71 func (me *trackerScraper) getIp() (ip net.IP, err error) {
  72         ips, err := net.LookupIP(me.u.Hostname())
  73         if err != nil {
  74                 return
  75         }
  76         if len(ips) == 0 {
  77                 err = errors.New("no ips")
  78                 return
  79         }
  80         for _, ip = range ips {
  81                 if me.t.cl.ipIsBlocked(ip) {
  82                         continue
  83                 }
  84                 switch me.u.Scheme {
  85                 case "udp4":
  86                         if ip.To4() == nil {
  87                                 continue
  88                         }
  89                 case "udp6":
  90                         if ip.To4() != nil {
  91                                 continue
  92                         }
  93                 }
  94                 return
  95         }
  96         err = errors.New("no acceptable ips")
  97         return
  98 }
  99
 100 func (me *trackerScraper) trackerUrl(ip net.IP) string {
 101         u := me.u
 102         if u.Port() != "" {
 103                 u.Host = net.JoinHostPort(ip.String(), u.Port())
 104         }
 105         return u.String()
 106 }
 107
 108 // Return how long to wait before trying again. For most errors, we return 5
 109 // minutes, a relatively quick turn around for DNS changes.
 110 func (me *trackerScraper) announce(event tracker.AnnounceEvent) (ret trackerAnnounceResult) {
 111         defer func() {
 112                 ret.Completed = time.Now()
 113         }()
 114         ret.Interval = time.Minute
 115         me.allow()
 116         // We might pass true if we got an error. Currently we don't because timing out with a
 117         // reasonably long timeout is its own form of backpressure (it remains to be seen if it's
 118         // enough).
 119         defer me.done(false)
 120         ip, err := me.getIp()
 121         if err != nil {
 122                 ret.Err = fmt.Errorf("error getting ip: %s", err)
 123                 return
 124         }
 125         me.t.cl.rLock()
 126         req := me.t.announceRequest(event)
 127         me.t.cl.rUnlock()
 128         // The default timeout is currently 15s, and that works well as backpressure on concurrent
 129         // access to the tracker.
 130         //ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 131         //defer cancel()
 132         me.t.logger.WithDefaultLevel(log.Debug).Printf("announcing to %q: %#v", me.u.String(), req)
 133         res, err := tracker.Announce{
 134                 //Context:    ctx,
 135                 HTTPProxy:  me.t.cl.config.HTTPProxy,
 136                 UserAgent:  me.t.cl.config.HTTPUserAgent,
 137                 TrackerUrl: me.trackerUrl(ip),
 138                 Request:    req,
 139                 HostHeader: me.u.Host,
 140                 ServerName: me.u.Hostname(),
 141                 UdpNetwork: me.u.Scheme,
 142                 ClientIp4:  krpc.NodeAddr{IP: me.t.cl.config.PublicIp4},
 143                 ClientIp6:  krpc.NodeAddr{IP: me.t.cl.config.PublicIp6},
 144         }.Do()
 145         me.t.logger.WithDefaultLevel(log.Debug).Printf("announce to %q returned %#v: %v", me.u.String(), res, err)
 146         if err != nil {
 147                 ret.Err = fmt.Errorf("announcing: %w", err)
 148                 return
 149         }
 150         me.t.AddPeers(peerInfos(nil).AppendFromTracker(res.Peers))
 151         ret.NumPeers = len(res.Peers)
 152         ret.Interval = time.Duration(res.Interval) * time.Second
 153         return
 154 }
 155
 156 // Returns whether we can shorten the interval, and sets notify to a channel that receives when we
 157 // might change our mind, or leaves it if we won't.
 158 func (me *trackerScraper) canIgnoreInterval(notify *<-chan struct{}) bool {
 159         gotInfo := me.t.GotInfo()
 160         select {
 161         case <-gotInfo:
 162                 // Private trackers really don't like us announcing more than they specify. They're also
 163                 // tracking us very carefully, so it's best to comply.
 164                 private := me.t.info.Private
 165                 return private == nil || !*private
 166         default:
 167                 *notify = gotInfo
 168                 return false
 169         }
 170 }
 171
 172 func (me *trackerScraper) Run() {
 173         defer me.announceStopped()
 174         // make sure first announce is a "started"
 175         e := tracker.Started
 176         for {
 177                 ar := me.announce(e)
 178                 // after first announce, get back to regular "none"
 179                 e = tracker.None
 180                 me.t.cl.lock()
 181                 me.lastAnnounce = ar
 182                 me.t.cl.unlock()
 183
 184         recalculate:
 185                 // Make sure we don't announce for at least a minute since the last one.
 186                 interval := ar.Interval
 187                 if interval < time.Minute {
 188                         interval = time.Minute
 189                 }
 190
 191                 me.t.cl.lock()
 192                 wantPeers := me.t.wantPeersEvent.C()
 193                 closed := me.t.closed.C()
 194                 me.t.cl.unlock()
 195
 196                 // If we want peers, reduce the interval to the minimum if it's appropriate.
 197
 198                 // A channel that receives when we should reconsider our interval. Starts as nil since that
 199                 // never receives.
 200                 var reconsider <-chan struct{}
 201                 select {
 202                 case <-wantPeers:
 203                         if interval > time.Minute && me.canIgnoreInterval(&reconsider) {
 204                                 interval = time.Minute
 205                         }
 206                 default:
 207                         reconsider = wantPeers
 208                 }
 209
 210                 select {
 211                 case <-closed:
 212                         return
 213                 case <-reconsider:
 214                         // Recalculate the interval.
 215                         goto recalculate
 216                 case <-time.After(time.Until(ar.Completed.Add(interval))):
 217                 }
 218         }
 219 }
 220
 221 func (me *trackerScraper) announceStopped() {
 222         me.announce(tracker.Stopped)
 223 }