tracker_scraper.go

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