torrent.go

   1 package torrent
   2
   3 import (
   4         "bytes"
   5         "container/heap"
   6         "context"
   7         "crypto/sha1"
   8         "errors"
   9         "fmt"
  10         "io"
  11         "math/rand"
  12         "net/http"
  13         "net/url"
  14         "sort"
  15         "sync"
  16         "text/tabwriter"
  17         "time"
  18         "unsafe"
  19
  20         "github.com/anacrolix/torrent/common"
  21         "github.com/anacrolix/torrent/segments"
  22         "github.com/anacrolix/torrent/webseed"
  23         "github.com/davecgh/go-spew/spew"
  24         "github.com/pion/datachannel"
  25
  26         "github.com/anacrolix/dht/v2"
  27         "github.com/anacrolix/log"
  28         "github.com/anacrolix/missinggo"
  29         "github.com/anacrolix/missinggo/perf"
  30         "github.com/anacrolix/missinggo/pubsub"
  31         "github.com/anacrolix/missinggo/slices"
  32         "github.com/anacrolix/missinggo/v2/bitmap"
  33         "github.com/anacrolix/missinggo/v2/prioritybitmap"
  34
  35         "github.com/anacrolix/torrent/bencode"
  36         "github.com/anacrolix/torrent/metainfo"
  37         pp "github.com/anacrolix/torrent/peer_protocol"
  38         "github.com/anacrolix/torrent/storage"
  39         "github.com/anacrolix/torrent/tracker"
  40         "github.com/anacrolix/torrent/webtorrent"
  41 )
  42
  43 // Maintains state of torrent within a Client. Many methods should not be called before the info is
  44 // available, see .Info and .GotInfo.
  45 type Torrent struct {
  46         // Torrent-level aggregate statistics. First in struct to ensure 64-bit
  47         // alignment. See #262.
  48         stats  ConnStats
  49         cl     *Client
  50         logger log.Logger
  51
  52         networkingEnabled      bool
  53         dataDownloadDisallowed bool
  54         dataUploadDisallowed   bool
  55         userOnWriteChunkErr    func(error)
  56
  57         // Determines what chunks to request from peers.
  58         requestStrategy requestStrategy
  59
  60         closed   missinggo.Event
  61         infoHash metainfo.Hash
  62         pieces   []Piece
  63         // Values are the piece indices that changed.
  64         pieceStateChanges *pubsub.PubSub
  65         // The size of chunks to request from peers over the wire. This is
  66         // normally 16KiB by convention these days.
  67         chunkSize pp.Integer
  68         chunkPool *sync.Pool
  69         // Total length of the torrent in bytes. Stored because it's not O(1) to
  70         // get this from the info dict.
  71         length *int64
  72
  73         // The storage to open when the info dict becomes available.
  74         storageOpener *storage.Client
  75         // Storage for torrent data.
  76         storage *storage.Torrent
  77         // Read-locked for using storage, and write-locked for Closing.
  78         storageLock sync.RWMutex
  79
  80         // TODO: Only announce stuff is used?
  81         metainfo metainfo.MetaInfo
  82
  83         // The info dict. nil if we don't have it (yet).
  84         info      *metainfo.Info
  85         fileIndex segments.Index
  86         files     *[]*File
  87
  88         webSeeds map[string]*peer
  89
  90         // Active peer connections, running message stream loops. TODO: Make this
  91         // open (not-closed) connections only.
  92         conns               map[*PeerConn]struct{}
  93         maxEstablishedConns int
  94         // Set of addrs to which we're attempting to connect. Connections are
  95         // half-open until all handshakes are completed.
  96         halfOpen    map[string]PeerInfo
  97         fastestPeer *peer
  98
  99         // Reserve of peers to connect to. A peer can be both here and in the
 100         // active connections if were told about the peer after connecting with
 101         // them. That encourages us to reconnect to peers that are well known in
 102         // the swarm.
 103         peers prioritizedPeers
 104         // Whether we want to know to know more peers.
 105         wantPeersEvent missinggo.Event
 106         // An announcer for each tracker URL.
 107         trackerAnnouncers map[string]torrentTrackerAnnouncer
 108         // How many times we've initiated a DHT announce. TODO: Move into stats.
 109         numDHTAnnounces int
 110
 111         // Name used if the info name isn't available. Should be cleared when the
 112         // Info does become available.
 113         nameMu      sync.RWMutex
 114         displayName string
 115
 116         // The bencoded bytes of the info dict. This is actively manipulated if
 117         // the info bytes aren't initially available, and we try to fetch them
 118         // from peers.
 119         metadataBytes []byte
 120         // Each element corresponds to the 16KiB metadata pieces. If true, we have
 121         // received that piece.
 122         metadataCompletedChunks []bool
 123         metadataChanged         sync.Cond
 124
 125         // Set when .Info is obtained.
 126         gotMetainfo missinggo.Event
 127
 128         readers                map[*reader]struct{}
 129         _readerNowPieces       bitmap.Bitmap
 130         _readerReadaheadPieces bitmap.Bitmap
 131
 132         // A cache of pieces we need to get. Calculated from various piece and
 133         // file priorities and completion states elsewhere.
 134         _pendingPieces prioritybitmap.PriorityBitmap
 135         // A cache of completed piece indices.
 136         _completedPieces bitmap.Bitmap
 137         // Pieces that need to be hashed.
 138         piecesQueuedForHash bitmap.Bitmap
 139         activePieceHashes   int
 140
 141         // A pool of piece priorities []int for assignment to new connections.
 142         // These "inclinations" are used to give connections preference for
 143         // different pieces.
 144         connPieceInclinationPool sync.Pool
 145
 146         // Count of each request across active connections.
 147         pendingRequests map[request]int
 148
 149         pex pexState
 150 }
 151
 152 func (t *Torrent) numConns() int {
 153         return len(t.conns)
 154 }
 155
 156 func (t *Torrent) numReaders() int {
 157         return len(t.readers)
 158 }
 159
 160 func (t *Torrent) readerNowPieces() bitmap.Bitmap {
 161         return t._readerNowPieces
 162 }
 163
 164 func (t *Torrent) readerReadaheadPieces() bitmap.Bitmap {
 165         return t._readerReadaheadPieces
 166 }
 167
 168 func (t *Torrent) ignorePieces() bitmap.Bitmap {
 169         ret := t._completedPieces.Copy()
 170         ret.Union(t.piecesQueuedForHash)
 171         for i := 0; i < t.numPieces(); i++ {
 172                 if t.piece(i).hashing {
 173                         ret.Set(i, true)
 174                 }
 175         }
 176         return ret
 177 }
 178
 179 func (t *Torrent) pendingPieces() *prioritybitmap.PriorityBitmap {
 180         return &t._pendingPieces
 181 }
 182
 183 func (t *Torrent) tickleReaders() {
 184         t.cl.event.Broadcast()
 185 }
 186
 187 // Returns a channel that is closed when the Torrent is closed.
 188 func (t *Torrent) Closed() <-chan struct{} {
 189         return t.closed.LockedChan(t.cl.locker())
 190 }
 191
 192 // KnownSwarm returns the known subset of the peers in the Torrent's swarm, including active,
 193 // pending, and half-open peers.
 194 func (t *Torrent) KnownSwarm() (ks []PeerInfo) {
 195         // Add pending peers to the list
 196         t.peers.Each(func(peer PeerInfo) {
 197                 ks = append(ks, peer)
 198         })
 199
 200         // Add half-open peers to the list
 201         for _, peer := range t.halfOpen {
 202                 ks = append(ks, peer)
 203         }
 204
 205         // Add active peers to the list
 206         for conn := range t.conns {
 207
 208                 ks = append(ks, PeerInfo{
 209                         Id:     conn.PeerID,
 210                         Addr:   conn.RemoteAddr,
 211                         Source: conn.Discovery,
 212                         // > If the connection is encrypted, that's certainly enough to set SupportsEncryption.
 213                         // > But if we're not connected to them with an encrypted connection, I couldn't say
 214                         // > what's appropriate. We can carry forward the SupportsEncryption value as we
 215                         // > received it from trackers/DHT/PEX, or just use the encryption state for the
 216                         // > connection. It's probably easiest to do the latter for now.
 217                         // https://github.com/anacrolix/torrent/pull/188
 218                         SupportsEncryption: conn.headerEncrypted,
 219                 })
 220         }
 221
 222         return
 223 }
 224
 225 func (t *Torrent) setChunkSize(size pp.Integer) {
 226         t.chunkSize = size
 227         t.chunkPool = &sync.Pool{
 228                 New: func() interface{} {
 229                         b := make([]byte, size)
 230                         return &b
 231                 },
 232         }
 233 }
 234
 235 func (t *Torrent) pieceComplete(piece pieceIndex) bool {
 236         return t._completedPieces.Get(bitmap.BitIndex(piece))
 237 }
 238
 239 func (t *Torrent) pieceCompleteUncached(piece pieceIndex) storage.Completion {
 240         return t.pieces[piece].Storage().Completion()
 241 }
 242
 243 // There's a connection to that address already.
 244 func (t *Torrent) addrActive(addr string) bool {
 245         if _, ok := t.halfOpen[addr]; ok {
 246                 return true
 247         }
 248         for c := range t.conns {
 249                 ra := c.RemoteAddr
 250                 if ra.String() == addr {
 251                         return true
 252                 }
 253         }
 254         return false
 255 }
 256
 257 func (t *Torrent) unclosedConnsAsSlice() (ret []*PeerConn) {
 258         ret = make([]*PeerConn, 0, len(t.conns))
 259         for c := range t.conns {
 260                 if !c.closed.IsSet() {
 261                         ret = append(ret, c)
 262                 }
 263         }
 264         return
 265 }
 266
 267 func (t *Torrent) addPeer(p PeerInfo) (added bool) {
 268         cl := t.cl
 269         torrent.Add(fmt.Sprintf("peers added by source %q", p.Source), 1)
 270         if t.closed.IsSet() {
 271                 return false
 272         }
 273         if ipAddr, ok := tryIpPortFromNetAddr(p.Addr); ok {
 274                 if cl.badPeerIPPort(ipAddr.IP, ipAddr.Port) {
 275                         torrent.Add("peers not added because of bad addr", 1)
 276                         // cl.logger.Printf("peers not added because of bad addr: %v", p)
 277                         return false
 278                 }
 279         }
 280         if replaced, ok := t.peers.AddReturningReplacedPeer(p); ok {
 281                 torrent.Add("peers replaced", 1)
 282                 if !replaced.equal(p) {
 283                         t.logger.WithDefaultLevel(log.Debug).Printf("added %v replacing %v", p, replaced)
 284                         added = true
 285                 }
 286         } else {
 287                 added = true
 288         }
 289         t.openNewConns()
 290         for t.peers.Len() > cl.config.TorrentPeersHighWater {
 291                 _, ok := t.peers.DeleteMin()
 292                 if ok {
 293                         torrent.Add("excess reserve peers discarded", 1)
 294                 }
 295         }
 296         return
 297 }
 298
 299 func (t *Torrent) invalidateMetadata() {
 300         for i := range t.metadataCompletedChunks {
 301                 t.metadataCompletedChunks[i] = false
 302         }
 303         t.nameMu.Lock()
 304         t.info = nil
 305         t.nameMu.Unlock()
 306 }
 307
 308 func (t *Torrent) saveMetadataPiece(index int, data []byte) {
 309         if t.haveInfo() {
 310                 return
 311         }
 312         if index >= len(t.metadataCompletedChunks) {
 313                 t.logger.Printf("%s: ignoring metadata piece %d", t, index)
 314                 return
 315         }
 316         copy(t.metadataBytes[(1<<14)*index:], data)
 317         t.metadataCompletedChunks[index] = true
 318 }
 319
 320 func (t *Torrent) metadataPieceCount() int {
 321         return (len(t.metadataBytes) + (1 << 14) - 1) / (1 << 14)
 322 }
 323
 324 func (t *Torrent) haveMetadataPiece(piece int) bool {
 325         if t.haveInfo() {
 326                 return (1<<14)*piece < len(t.metadataBytes)
 327         } else {
 328                 return piece < len(t.metadataCompletedChunks) && t.metadataCompletedChunks[piece]
 329         }
 330 }
 331
 332 func (t *Torrent) metadataSize() int {
 333         return len(t.metadataBytes)
 334 }
 335
 336 func infoPieceHashes(info *metainfo.Info) (ret [][]byte) {
 337         for i := 0; i < len(info.Pieces); i += sha1.Size {
 338                 ret = append(ret, info.Pieces[i:i+sha1.Size])
 339         }
 340         return
 341 }
 342
 343 func (t *Torrent) makePieces() {
 344         hashes := infoPieceHashes(t.info)
 345         t.pieces = make([]Piece, len(hashes))
 346         for i, hash := range hashes {
 347                 piece := &t.pieces[i]
 348                 piece.t = t
 349                 piece.index = pieceIndex(i)
 350                 piece.noPendingWrites.L = &piece.pendingWritesMutex
 351                 piece.hash = (*metainfo.Hash)(unsafe.Pointer(&hash[0]))
 352                 files := *t.files
 353                 beginFile := pieceFirstFileIndex(piece.torrentBeginOffset(), files)
 354                 endFile := pieceEndFileIndex(piece.torrentEndOffset(), files)
 355                 piece.files = files[beginFile:endFile]
 356         }
 357 }
 358
 359 // Returns the index of the first file containing the piece. files must be
 360 // ordered by offset.
 361 func pieceFirstFileIndex(pieceOffset int64, files []*File) int {
 362         for i, f := range files {
 363                 if f.offset+f.length > pieceOffset {
 364                         return i
 365                 }
 366         }
 367         return 0
 368 }
 369
 370 // Returns the index after the last file containing the piece. files must be
 371 // ordered by offset.
 372 func pieceEndFileIndex(pieceEndOffset int64, files []*File) int {
 373         for i, f := range files {
 374                 if f.offset+f.length >= pieceEndOffset {
 375                         return i + 1
 376                 }
 377         }
 378         return 0
 379 }
 380
 381 func (t *Torrent) cacheLength() {
 382         var l int64
 383         for _, f := range t.info.UpvertedFiles() {
 384                 l += f.Length
 385         }
 386         t.length = &l
 387 }
 388
 389 func (t *Torrent) setInfo(info *metainfo.Info) error {
 390         if err := validateInfo(info); err != nil {
 391                 return fmt.Errorf("bad info: %s", err)
 392         }
 393         if t.storageOpener != nil {
 394                 var err error
 395                 t.storage, err = t.storageOpener.OpenTorrent(info, t.infoHash)
 396                 if err != nil {
 397                         return fmt.Errorf("error opening torrent storage: %s", err)
 398                 }
 399         }
 400         t.nameMu.Lock()
 401         t.info = info
 402         t.nameMu.Unlock()
 403         t.fileIndex = segments.NewIndex(common.LengthIterFromUpvertedFiles(info.UpvertedFiles()))
 404         t.displayName = "" // Save a few bytes lol.
 405         t.initFiles()
 406         t.cacheLength()
 407         t.makePieces()
 408         return nil
 409 }
 410
 411 // This seems to be all the follow-up tasks after info is set, that can't fail.
 412 func (t *Torrent) onSetInfo() {
 413         t.iterPeers(func(p *peer) {
 414                 p.onGotInfo(t.info)
 415         })
 416         for i := range t.pieces {
 417                 t.updatePieceCompletion(pieceIndex(i))
 418                 p := &t.pieces[i]
 419                 if !p.storageCompletionOk {
 420                         // t.logger.Printf("piece %s completion unknown, queueing check", p)
 421                         t.queuePieceCheck(pieceIndex(i))
 422                 }
 423         }
 424         t.cl.event.Broadcast()
 425         t.gotMetainfo.Set()
 426         t.updateWantPeersEvent()
 427         t.pendingRequests = make(map[request]int)
 428         t.tryCreateMorePieceHashers()
 429 }
 430
 431 // Called when metadata for a torrent becomes available.
 432 func (t *Torrent) setInfoBytes(b []byte) error {
 433         if metainfo.HashBytes(b) != t.infoHash {
 434                 return errors.New("info bytes have wrong hash")
 435         }
 436         var info metainfo.Info
 437         if err := bencode.Unmarshal(b, &info); err != nil {
 438                 return fmt.Errorf("error unmarshalling info bytes: %s", err)
 439         }
 440         t.metadataBytes = b
 441         t.metadataCompletedChunks = nil
 442         if t.info != nil {
 443                 return nil
 444         }
 445         if err := t.setInfo(&info); err != nil {
 446                 return err
 447         }
 448         t.onSetInfo()
 449         return nil
 450 }
 451
 452 func (t *Torrent) haveAllMetadataPieces() bool {
 453         if t.haveInfo() {
 454                 return true
 455         }
 456         if t.metadataCompletedChunks == nil {
 457                 return false
 458         }
 459         for _, have := range t.metadataCompletedChunks {
 460                 if !have {
 461                         return false
 462                 }
 463         }
 464         return true
 465 }
 466
 467 // TODO: Propagate errors to disconnect peer.
 468 func (t *Torrent) setMetadataSize(bytes int) (err error) {
 469         if t.haveInfo() {
 470                 // We already know the correct metadata size.
 471                 return
 472         }
 473         if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
 474                 return errors.New("bad size")
 475         }
 476         if t.metadataBytes != nil && len(t.metadataBytes) == int(bytes) {
 477                 return
 478         }
 479         t.metadataBytes = make([]byte, bytes)
 480         t.metadataCompletedChunks = make([]bool, (bytes+(1<<14)-1)/(1<<14))
 481         t.metadataChanged.Broadcast()
 482         for c := range t.conns {
 483                 c.requestPendingMetadata()
 484         }
 485         return
 486 }
 487
 488 // The current working name for the torrent. Either the name in the info dict,
 489 // or a display name given such as by the dn value in a magnet link, or "".
 490 func (t *Torrent) name() string {
 491         t.nameMu.RLock()
 492         defer t.nameMu.RUnlock()
 493         if t.haveInfo() {
 494                 return t.info.Name
 495         }
 496         return t.displayName
 497 }
 498
 499 func (t *Torrent) pieceState(index pieceIndex) (ret PieceState) {
 500         p := &t.pieces[index]
 501         ret.Priority = t.piecePriority(index)
 502         ret.Completion = p.completion()
 503         ret.QueuedForHash = p.queuedForHash()
 504         ret.Hashing = p.hashing
 505         ret.Checking = ret.QueuedForHash || ret.Hashing
 506         ret.Marking = p.marking
 507         if !ret.Complete && t.piecePartiallyDownloaded(index) {
 508                 ret.Partial = true
 509         }
 510         return
 511 }
 512
 513 func (t *Torrent) metadataPieceSize(piece int) int {
 514         return metadataPieceSize(len(t.metadataBytes), piece)
 515 }
 516
 517 func (t *Torrent) newMetadataExtensionMessage(c *PeerConn, msgType int, piece int, data []byte) pp.Message {
 518         d := map[string]int{
 519                 "msg_type": msgType,
 520                 "piece":    piece,
 521         }
 522         if data != nil {
 523                 d["total_size"] = len(t.metadataBytes)
 524         }
 525         p := bencode.MustMarshal(d)
 526         return pp.Message{
 527                 Type:            pp.Extended,
 528                 ExtendedID:      c.PeerExtensionIDs[pp.ExtensionNameMetadata],
 529                 ExtendedPayload: append(p, data...),
 530         }
 531 }
 532
 533 func (t *Torrent) pieceStateRuns() (ret PieceStateRuns) {
 534         rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
 535                 ret = append(ret, PieceStateRun{
 536                         PieceState: el.(PieceState),
 537                         Length:     int(count),
 538                 })
 539         })
 540         for index := range t.pieces {
 541                 rle.Append(t.pieceState(pieceIndex(index)), 1)
 542         }
 543         rle.Flush()
 544         return
 545 }
 546
 547 // Produces a small string representing a PieceStateRun.
 548 func (psr PieceStateRun) String() (ret string) {
 549         ret = fmt.Sprintf("%d", psr.Length)
 550         ret += func() string {
 551                 switch psr.Priority {
 552                 case PiecePriorityNext:
 553                         return "N"
 554                 case PiecePriorityNormal:
 555                         return "."
 556                 case PiecePriorityReadahead:
 557                         return "R"
 558                 case PiecePriorityNow:
 559                         return "!"
 560                 case PiecePriorityHigh:
 561                         return "H"
 562                 default:
 563                         return ""
 564                 }
 565         }()
 566         if psr.Hashing {
 567                 ret += "H"
 568         }
 569         if psr.QueuedForHash {
 570                 ret += "Q"
 571         }
 572         if psr.Marking {
 573                 ret += "M"
 574         }
 575         if psr.Partial {
 576                 ret += "P"
 577         }
 578         if psr.Complete {
 579                 ret += "C"
 580         }
 581         if !psr.Ok {
 582                 ret += "?"
 583         }
 584         return
 585 }
 586
 587 func (t *Torrent) writeStatus(w io.Writer) {
 588         fmt.Fprintf(w, "Infohash: %s\n", t.infoHash.HexString())
 589         fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
 590         if !t.haveInfo() {
 591                 fmt.Fprintf(w, "Metadata have: ")
 592                 for _, h := range t.metadataCompletedChunks {
 593                         fmt.Fprintf(w, "%c", func() rune {
 594                                 if h {
 595                                         return 'H'
 596                                 } else {
 597                                         return '.'
 598                                 }
 599                         }())
 600                 }
 601                 fmt.Fprintln(w)
 602         }
 603         fmt.Fprintf(w, "Piece length: %s\n", func() string {
 604                 if t.haveInfo() {
 605                         return fmt.Sprint(t.usualPieceSize())
 606                 } else {
 607                         return "?"
 608                 }
 609         }())
 610         if t.info != nil {
 611                 fmt.Fprintf(w, "Num Pieces: %d (%d completed)\n", t.numPieces(), t.numPiecesCompleted())
 612                 fmt.Fprintf(w, "Piece States: %s", t.pieceStateRuns())
 613                 fmt.Fprintln(w)
 614         }
 615         fmt.Fprintf(w, "Reader Pieces:")
 616         t.forReaderOffsetPieces(func(begin, end pieceIndex) (again bool) {
 617                 fmt.Fprintf(w, " %d:%d", begin, end)
 618                 return true
 619         })
 620         fmt.Fprintln(w)
 621
 622         fmt.Fprintf(w, "Enabled trackers:\n")
 623         func() {
 624                 tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
 625                 fmt.Fprintf(tw, "    URL\tExtra\n")
 626                 for _, ta := range slices.Sort(slices.FromMapElems(t.trackerAnnouncers), func(l, r torrentTrackerAnnouncer) bool {
 627                         lu := l.URL()
 628                         ru := r.URL()
 629                         var luns, runs url.URL = *lu, *ru
 630                         luns.Scheme = ""
 631                         runs.Scheme = ""
 632                         var ml missinggo.MultiLess
 633                         ml.StrictNext(luns.String() == runs.String(), luns.String() < runs.String())
 634                         ml.StrictNext(lu.String() == ru.String(), lu.String() < ru.String())
 635                         return ml.Less()
 636                 }).([]torrentTrackerAnnouncer) {
 637                         fmt.Fprintf(tw, "    %q\t%v\n", ta.URL(), ta.statusLine())
 638                 }
 639                 tw.Flush()
 640         }()
 641
 642         fmt.Fprintf(w, "DHT Announces: %d\n", t.numDHTAnnounces)
 643
 644         spew.NewDefaultConfig()
 645         spew.Fdump(w, t.statsLocked())
 646
 647         peers := t.peersAsSlice()
 648         sort.Slice(peers, func(i, j int) bool {
 649                 return worseConn(peers[i], peers[j])
 650         })
 651         for i, c := range peers {
 652                 fmt.Fprintf(w, "%2d. ", i+1)
 653                 c.writeStatus(w, t)
 654         }
 655 }
 656
 657 func (t *Torrent) haveInfo() bool {
 658         return t.info != nil
 659 }
 660
 661 // Returns a run-time generated MetaInfo that includes the info bytes and
 662 // announce-list as currently known to the client.
 663 func (t *Torrent) newMetaInfo() metainfo.MetaInfo {
 664         return metainfo.MetaInfo{
 665                 CreationDate: time.Now().Unix(),
 666                 Comment:      "dynamic metainfo from client",
 667                 CreatedBy:    "go.torrent",
 668                 AnnounceList: t.metainfo.UpvertedAnnounceList().Clone(),
 669                 InfoBytes: func() []byte {
 670                         if t.haveInfo() {
 671                                 return t.metadataBytes
 672                         } else {
 673                                 return nil
 674                         }
 675                 }(),
 676         }
 677 }
 678
 679 func (t *Torrent) BytesMissing() int64 {
 680         t.cl.rLock()
 681         defer t.cl.rUnlock()
 682         return t.bytesMissingLocked()
 683 }
 684
 685 func (t *Torrent) bytesMissingLocked() int64 {
 686         return t.bytesLeft()
 687 }
 688
 689 func (t *Torrent) bytesLeft() (left int64) {
 690         bitmap.Flip(t._completedPieces, 0, bitmap.BitIndex(t.numPieces())).IterTyped(func(piece int) bool {
 691                 p := &t.pieces[piece]
 692                 left += int64(p.length() - p.numDirtyBytes())
 693                 return true
 694         })
 695         return
 696 }
 697
 698 // Bytes left to give in tracker announces.
 699 func (t *Torrent) bytesLeftAnnounce() int64 {
 700         if t.haveInfo() {
 701                 return t.bytesLeft()
 702         } else {
 703                 return -1
 704         }
 705 }
 706
 707 func (t *Torrent) piecePartiallyDownloaded(piece pieceIndex) bool {
 708         if t.pieceComplete(piece) {
 709                 return false
 710         }
 711         if t.pieceAllDirty(piece) {
 712                 return false
 713         }
 714         return t.pieces[piece].hasDirtyChunks()
 715 }
 716
 717 func (t *Torrent) usualPieceSize() int {
 718         return int(t.info.PieceLength)
 719 }
 720
 721 func (t *Torrent) numPieces() pieceIndex {
 722         return pieceIndex(t.info.NumPieces())
 723 }
 724
 725 func (t *Torrent) numPiecesCompleted() (num int) {
 726         return t._completedPieces.Len()
 727 }
 728
 729 func (t *Torrent) close() (err error) {
 730         t.closed.Set()
 731         t.tickleReaders()
 732         if t.storage != nil {
 733                 t.storageLock.Lock()
 734                 t.storage.Close()
 735                 t.storageLock.Unlock()
 736         }
 737         for conn := range t.conns {
 738                 conn.close()
 739         }
 740         t.pex.Reset()
 741         t.cl.event.Broadcast()
 742         t.pieceStateChanges.Close()
 743         t.updateWantPeersEvent()
 744         return
 745 }
 746
 747 func (t *Torrent) requestOffset(r request) int64 {
 748         return torrentRequestOffset(*t.length, int64(t.usualPieceSize()), r)
 749 }
 750
 751 // Return the request that would include the given offset into the torrent data. Returns !ok if
 752 // there is no such request.
 753 func (t *Torrent) offsetRequest(off int64) (req request, ok bool) {
 754         return torrentOffsetRequest(*t.length, t.info.PieceLength, int64(t.chunkSize), off)
 755 }
 756
 757 func (t *Torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
 758         defer perf.ScopeTimerErr(&err)()
 759         n, err := t.pieces[piece].Storage().WriteAt(data, begin)
 760         if err == nil && n != len(data) {
 761                 err = io.ErrShortWrite
 762         }
 763         return err
 764 }
 765
 766 func (t *Torrent) bitfield() (bf []bool) {
 767         bf = make([]bool, t.numPieces())
 768         t._completedPieces.IterTyped(func(piece int) (again bool) {
 769                 bf[piece] = true
 770                 return true
 771         })
 772         return
 773 }
 774
 775 func (t *Torrent) pieceNumChunks(piece pieceIndex) pp.Integer {
 776         return (t.pieceLength(piece) + t.chunkSize - 1) / t.chunkSize
 777 }
 778
 779 func (t *Torrent) pendAllChunkSpecs(pieceIndex pieceIndex) {
 780         t.pieces[pieceIndex]._dirtyChunks.Clear()
 781 }
 782
 783 func (t *Torrent) pieceLength(piece pieceIndex) pp.Integer {
 784         if t.info.PieceLength == 0 {
 785                 // There will be no variance amongst pieces. Only pain.
 786                 return 0
 787         }
 788         if piece == t.numPieces()-1 {
 789                 ret := pp.Integer(*t.length % t.info.PieceLength)
 790                 if ret != 0 {
 791                         return ret
 792                 }
 793         }
 794         return pp.Integer(t.info.PieceLength)
 795 }
 796
 797 func (t *Torrent) hashPiece(piece pieceIndex) (ret metainfo.Hash, err error) {
 798         hash := pieceHash.New()
 799         p := t.piece(piece)
 800         p.waitNoPendingWrites()
 801         storagePiece := t.pieces[piece].Storage()
 802         const logPieceContents = false
 803         if logPieceContents {
 804                 var examineBuf bytes.Buffer
 805                 _, err = storagePiece.WriteTo(io.MultiWriter(hash, &examineBuf))
 806                 log.Printf("hashed %q with copy err %v", examineBuf.Bytes(), err)
 807         } else {
 808                 _, err = storagePiece.WriteTo(hash)
 809         }
 810         missinggo.CopyExact(&ret, hash.Sum(nil))
 811         return
 812 }
 813
 814 func (t *Torrent) haveAnyPieces() bool {
 815         return t._completedPieces.Len() != 0
 816 }
 817
 818 func (t *Torrent) haveAllPieces() bool {
 819         if !t.haveInfo() {
 820                 return false
 821         }
 822         return t._completedPieces.Len() == bitmap.BitIndex(t.numPieces())
 823 }
 824
 825 func (t *Torrent) havePiece(index pieceIndex) bool {
 826         return t.haveInfo() && t.pieceComplete(index)
 827 }
 828
 829 func (t *Torrent) haveChunk(r request) (ret bool) {
 830         // defer func() {
 831         //      log.Println("have chunk", r, ret)
 832         // }()
 833         if !t.haveInfo() {
 834                 return false
 835         }
 836         if t.pieceComplete(pieceIndex(r.Index)) {
 837                 return true
 838         }
 839         p := &t.pieces[r.Index]
 840         return !p.pendingChunk(r.chunkSpec, t.chunkSize)
 841 }
 842
 843 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
 844         return int(cs.Begin / chunkSize)
 845 }
 846
 847 func (t *Torrent) wantPieceIndex(index pieceIndex) bool {
 848         if !t.haveInfo() {
 849                 return false
 850         }
 851         if index < 0 || index >= t.numPieces() {
 852                 return false
 853         }
 854         p := &t.pieces[index]
 855         if p.queuedForHash() {
 856                 return false
 857         }
 858         if p.hashing {
 859                 return false
 860         }
 861         if t.pieceComplete(index) {
 862                 return false
 863         }
 864         if t._pendingPieces.Contains(bitmap.BitIndex(index)) {
 865                 return true
 866         }
 867         // t.logger.Printf("piece %d not pending", index)
 868         return !t.forReaderOffsetPieces(func(begin, end pieceIndex) bool {
 869                 return index < begin || index >= end
 870         })
 871 }
 872
 873 // The worst connection is one that hasn't been sent, or sent anything useful for the longest. A bad
 874 // connection is one that usually sends us unwanted pieces, or has been in worser half of the
 875 // established connections for more than a minute.
 876 func (t *Torrent) worstBadConn() *PeerConn {
 877         wcs := worseConnSlice{t.unclosedConnsAsSlice()}
 878         heap.Init(&wcs)
 879         for wcs.Len() != 0 {
 880                 c := heap.Pop(&wcs).(*PeerConn)
 881                 if c._stats.ChunksReadWasted.Int64() >= 6 && c._stats.ChunksReadWasted.Int64() > c._stats.ChunksReadUseful.Int64() {
 882                         return c
 883                 }
 884                 // If the connection is in the worst half of the established
 885                 // connection quota and is older than a minute.
 886                 if wcs.Len() >= (t.maxEstablishedConns+1)/2 {
 887                         // Give connections 1 minute to prove themselves.
 888                         if time.Since(c.completedHandshake) > time.Minute {
 889                                 return c
 890                         }
 891                 }
 892         }
 893         return nil
 894 }
 895
 896 type PieceStateChange struct {
 897         Index int
 898         PieceState
 899 }
 900
 901 func (t *Torrent) publishPieceChange(piece pieceIndex) {
 902         t.cl._mu.Defer(func() {
 903                 cur := t.pieceState(piece)
 904                 p := &t.pieces[piece]
 905                 if cur != p.publicPieceState {
 906                         p.publicPieceState = cur
 907                         t.pieceStateChanges.Publish(PieceStateChange{
 908                                 int(piece),
 909                                 cur,
 910                         })
 911                 }
 912         })
 913 }
 914
 915 func (t *Torrent) pieceNumPendingChunks(piece pieceIndex) pp.Integer {
 916         if t.pieceComplete(piece) {
 917                 return 0
 918         }
 919         return t.pieceNumChunks(piece) - t.pieces[piece].numDirtyChunks()
 920 }
 921
 922 func (t *Torrent) pieceAllDirty(piece pieceIndex) bool {
 923         return t.pieces[piece]._dirtyChunks.Len() == int(t.pieceNumChunks(piece))
 924 }
 925
 926 func (t *Torrent) readersChanged() {
 927         t.updateReaderPieces()
 928         t.updateAllPiecePriorities()
 929 }
 930
 931 func (t *Torrent) updateReaderPieces() {
 932         t._readerNowPieces, t._readerReadaheadPieces = t.readerPiecePriorities()
 933 }
 934
 935 func (t *Torrent) readerPosChanged(from, to pieceRange) {
 936         if from == to {
 937                 return
 938         }
 939         t.updateReaderPieces()
 940         // Order the ranges, high and low.
 941         l, h := from, to
 942         if l.begin > h.begin {
 943                 l, h = h, l
 944         }
 945         if l.end < h.begin {
 946                 // Two distinct ranges.
 947                 t.updatePiecePriorities(l.begin, l.end)
 948                 t.updatePiecePriorities(h.begin, h.end)
 949         } else {
 950                 // Ranges overlap.
 951                 end := l.end
 952                 if h.end > end {
 953                         end = h.end
 954                 }
 955                 t.updatePiecePriorities(l.begin, end)
 956         }
 957 }
 958
 959 func (t *Torrent) maybeNewConns() {
 960         // Tickle the accept routine.
 961         t.cl.event.Broadcast()
 962         t.openNewConns()
 963 }
 964
 965 func (t *Torrent) piecePriorityChanged(piece pieceIndex) {
 966         // t.logger.Printf("piece %d priority changed", piece)
 967         t.iterPeers(func(c *peer) {
 968                 if c.updatePiecePriority(piece) {
 969                         // log.Print("conn piece priority changed")
 970                         c.updateRequests()
 971                 }
 972         })
 973         t.maybeNewConns()
 974         t.publishPieceChange(piece)
 975 }
 976
 977 func (t *Torrent) updatePiecePriority(piece pieceIndex) {
 978         p := &t.pieces[piece]
 979         newPrio := p.uncachedPriority()
 980         // t.logger.Printf("torrent %p: piece %d: uncached priority: %v", t, piece, newPrio)
 981         if newPrio == PiecePriorityNone {
 982                 if !t._pendingPieces.Remove(bitmap.BitIndex(piece)) {
 983                         return
 984                 }
 985         } else {
 986                 if !t._pendingPieces.Set(bitmap.BitIndex(piece), newPrio.BitmapPriority()) {
 987                         return
 988                 }
 989         }
 990         t.piecePriorityChanged(piece)
 991 }
 992
 993 func (t *Torrent) updateAllPiecePriorities() {
 994         t.updatePiecePriorities(0, t.numPieces())
 995 }
 996
 997 // Update all piece priorities in one hit. This function should have the same
 998 // output as updatePiecePriority, but across all pieces.
 999 func (t *Torrent) updatePiecePriorities(begin, end pieceIndex) {
1000         for i := begin; i < end; i++ {
1001                 t.updatePiecePriority(i)
1002         }
1003 }
1004
1005 // Returns the range of pieces [begin, end) that contains the extent of bytes.
1006 func (t *Torrent) byteRegionPieces(off, size int64) (begin, end pieceIndex) {
1007         if off >= *t.length {
1008                 return
1009         }
1010         if off < 0 {
1011                 size += off
1012                 off = 0
1013         }
1014         if size <= 0 {
1015                 return
1016         }
1017         begin = pieceIndex(off / t.info.PieceLength)
1018         end = pieceIndex((off + size + t.info.PieceLength - 1) / t.info.PieceLength)
1019         if end > pieceIndex(t.info.NumPieces()) {
1020                 end = pieceIndex(t.info.NumPieces())
1021         }
1022         return
1023 }
1024
1025 // Returns true if all iterations complete without breaking. Returns the read
1026 // regions for all readers. The reader regions should not be merged as some
1027 // callers depend on this method to enumerate readers.
1028 func (t *Torrent) forReaderOffsetPieces(f func(begin, end pieceIndex) (more bool)) (all bool) {
1029         for r := range t.readers {
1030                 p := r.pieces
1031                 if p.begin >= p.end {
1032                         continue
1033                 }
1034                 if !f(p.begin, p.end) {
1035                         return false
1036                 }
1037         }
1038         return true
1039 }
1040
1041 func (t *Torrent) piecePriority(piece pieceIndex) piecePriority {
1042         prio, ok := t._pendingPieces.GetPriority(bitmap.BitIndex(piece))
1043         if !ok {
1044                 return PiecePriorityNone
1045         }
1046         if prio > 0 {
1047                 panic(prio)
1048         }
1049         ret := piecePriority(-prio)
1050         if ret == PiecePriorityNone {
1051                 panic(piece)
1052         }
1053         return ret
1054 }
1055
1056 func (t *Torrent) pendRequest(req request) {
1057         ci := chunkIndex(req.chunkSpec, t.chunkSize)
1058         t.pieces[req.Index].pendChunkIndex(ci)
1059 }
1060
1061 func (t *Torrent) pieceCompletionChanged(piece pieceIndex) {
1062         t.tickleReaders()
1063         t.cl.event.Broadcast()
1064         if t.pieceComplete(piece) {
1065                 t.onPieceCompleted(piece)
1066         } else {
1067                 t.onIncompletePiece(piece)
1068         }
1069         t.updatePiecePriority(piece)
1070 }
1071
1072 func (t *Torrent) numReceivedConns() (ret int) {
1073         for c := range t.conns {
1074                 if c.Discovery == PeerSourceIncoming {
1075                         ret++
1076                 }
1077         }
1078         return
1079 }
1080
1081 func (t *Torrent) maxHalfOpen() int {
1082         // Note that if we somehow exceed the maximum established conns, we want
1083         // the negative value to have an effect.
1084         establishedHeadroom := int64(t.maxEstablishedConns - len(t.conns))
1085         extraIncoming := int64(t.numReceivedConns() - t.maxEstablishedConns/2)
1086         // We want to allow some experimentation with new peers, and to try to
1087         // upset an oversupply of received connections.
1088         return int(min(max(5, extraIncoming)+establishedHeadroom, int64(t.cl.config.HalfOpenConnsPerTorrent)))
1089 }
1090
1091 func (t *Torrent) openNewConns() (initiated int) {
1092         defer t.updateWantPeersEvent()
1093         for t.peers.Len() != 0 {
1094                 if !t.wantConns() {
1095                         return
1096                 }
1097                 if len(t.halfOpen) >= t.maxHalfOpen() {
1098                         return
1099                 }
1100                 if len(t.cl.dialers) == 0 {
1101                         return
1102                 }
1103                 if t.cl.numHalfOpen >= t.cl.config.TotalHalfOpenConns {
1104                         return
1105                 }
1106                 p := t.peers.PopMax()
1107                 t.initiateConn(p)
1108                 initiated++
1109         }
1110         return
1111 }
1112
1113 func (t *Torrent) getConnPieceInclination() []int {
1114         _ret := t.connPieceInclinationPool.Get()
1115         if _ret == nil {
1116                 pieceInclinationsNew.Add(1)
1117                 return rand.Perm(int(t.numPieces()))
1118         }
1119         pieceInclinationsReused.Add(1)
1120         return *_ret.(*[]int)
1121 }
1122
1123 func (t *Torrent) putPieceInclination(pi []int) {
1124         t.connPieceInclinationPool.Put(&pi)
1125         pieceInclinationsPut.Add(1)
1126 }
1127
1128 func (t *Torrent) updatePieceCompletion(piece pieceIndex) bool {
1129         p := t.piece(piece)
1130         uncached := t.pieceCompleteUncached(piece)
1131         cached := p.completion()
1132         changed := cached != uncached
1133         complete := uncached.Complete
1134         p.storageCompletionOk = uncached.Ok
1135         t._completedPieces.Set(bitmap.BitIndex(piece), complete)
1136         if complete && len(p.dirtiers) != 0 {
1137                 t.logger.Printf("marked piece %v complete but still has dirtiers", piece)
1138         }
1139         if changed {
1140                 log.Fstr("piece %d completion changed: %+v -> %+v", piece, cached, uncached).SetLevel(log.Debug).Log(t.logger)
1141                 t.pieceCompletionChanged(piece)
1142         }
1143         return changed
1144 }
1145
1146 // Non-blocking read. Client lock is not required.
1147 func (t *Torrent) readAt(b []byte, off int64) (n int, err error) {
1148         for len(b) != 0 {
1149                 p := &t.pieces[off/t.info.PieceLength]
1150                 p.waitNoPendingWrites()
1151                 var n1 int
1152                 n1, err = p.Storage().ReadAt(b, off-p.Info().Offset())
1153                 if n1 == 0 {
1154                         break
1155                 }
1156                 off += int64(n1)
1157                 n += n1
1158                 b = b[n1:]
1159         }
1160         return
1161 }
1162
1163 // Returns an error if the metadata was completed, but couldn't be set for
1164 // some reason. Blame it on the last peer to contribute.
1165 func (t *Torrent) maybeCompleteMetadata() error {
1166         if t.haveInfo() {
1167                 // Nothing to do.
1168                 return nil
1169         }
1170         if !t.haveAllMetadataPieces() {
1171                 // Don't have enough metadata pieces.
1172                 return nil
1173         }
1174         err := t.setInfoBytes(t.metadataBytes)
1175         if err != nil {
1176                 t.invalidateMetadata()
1177                 return fmt.Errorf("error setting info bytes: %s", err)
1178         }
1179         if t.cl.config.Debug {
1180                 t.logger.Printf("%s: got metadata from peers", t)
1181         }
1182         return nil
1183 }
1184
1185 func (t *Torrent) readerPiecePriorities() (now, readahead bitmap.Bitmap) {
1186         t.forReaderOffsetPieces(func(begin, end pieceIndex) bool {
1187                 if end > begin {
1188                         now.Add(bitmap.BitIndex(begin))
1189                         readahead.AddRange(bitmap.BitIndex(begin)+1, bitmap.BitIndex(end))
1190                 }
1191                 return true
1192         })
1193         return
1194 }
1195
1196 func (t *Torrent) needData() bool {
1197         if t.closed.IsSet() {
1198                 return false
1199         }
1200         if !t.haveInfo() {
1201                 return true
1202         }
1203         return t._pendingPieces.Len() != 0
1204 }
1205
1206 func appendMissingStrings(old, new []string) (ret []string) {
1207         ret = old
1208 new:
1209         for _, n := range new {
1210                 for _, o := range old {
1211                         if o == n {
1212                                 continue new
1213                         }
1214                 }
1215                 ret = append(ret, n)
1216         }
1217         return
1218 }
1219
1220 func appendMissingTrackerTiers(existing [][]string, minNumTiers int) (ret [][]string) {
1221         ret = existing
1222         for minNumTiers > len(ret) {
1223                 ret = append(ret, nil)
1224         }
1225         return
1226 }
1227
1228 func (t *Torrent) addTrackers(announceList [][]string) {
1229         fullAnnounceList := &t.metainfo.AnnounceList
1230         t.metainfo.AnnounceList = appendMissingTrackerTiers(*fullAnnounceList, len(announceList))
1231         for tierIndex, trackerURLs := range announceList {
1232                 (*fullAnnounceList)[tierIndex] = appendMissingStrings((*fullAnnounceList)[tierIndex], trackerURLs)
1233         }
1234         t.startMissingTrackerScrapers()
1235         t.updateWantPeersEvent()
1236 }
1237
1238 // Don't call this before the info is available.
1239 func (t *Torrent) bytesCompleted() int64 {
1240         if !t.haveInfo() {
1241                 return 0
1242         }
1243         return t.info.TotalLength() - t.bytesLeft()
1244 }
1245
1246 func (t *Torrent) SetInfoBytes(b []byte) (err error) {
1247         t.cl.lock()
1248         defer t.cl.unlock()
1249         return t.setInfoBytes(b)
1250 }
1251
1252 // Returns true if connection is removed from torrent.Conns.
1253 func (t *Torrent) deleteConnection(c *PeerConn) (ret bool) {
1254         if !c.closed.IsSet() {
1255                 panic("connection is not closed")
1256                 // There are behaviours prevented by the closed state that will fail
1257                 // if the connection has been deleted.
1258         }
1259         _, ret = t.conns[c]
1260         delete(t.conns, c)
1261         // Avoid adding a drop event more than once. Probably we should track whether we've generated
1262         // the drop event against the PexConnState instead.
1263         if ret {
1264                 if !t.cl.config.DisablePEX {
1265                         t.pex.Drop(c)
1266                 }
1267         }
1268         torrent.Add("deleted connections", 1)
1269         c.deleteAllRequests()
1270         if t.numActivePeers() == 0 {
1271                 t.assertNoPendingRequests()
1272         }
1273         return
1274 }
1275
1276 func (t *Torrent) numActivePeers() (num int) {
1277         t.iterPeers(func(*peer) {
1278                 num++
1279         })
1280         return
1281 }
1282
1283 func (t *Torrent) assertNoPendingRequests() {
1284         if len(t.pendingRequests) != 0 {
1285                 panic(t.pendingRequests)
1286         }
1287         //if len(t.lastRequested) != 0 {
1288         //      panic(t.lastRequested)
1289         //}
1290 }
1291
1292 func (t *Torrent) dropConnection(c *PeerConn) {
1293         t.cl.event.Broadcast()
1294         c.close()
1295         if t.deleteConnection(c) {
1296                 t.openNewConns()
1297         }
1298 }
1299
1300 func (t *Torrent) wantPeers() bool {
1301         if t.closed.IsSet() {
1302                 return false
1303         }
1304         if t.peers.Len() > t.cl.config.TorrentPeersLowWater {
1305                 return false
1306         }
1307         return t.needData() || t.seeding()
1308 }
1309
1310 func (t *Torrent) updateWantPeersEvent() {
1311         if t.wantPeers() {
1312                 t.wantPeersEvent.Set()
1313         } else {
1314                 t.wantPeersEvent.Clear()
1315         }
1316 }
1317
1318 // Returns whether the client should make effort to seed the torrent.
1319 func (t *Torrent) seeding() bool {
1320         cl := t.cl
1321         if t.closed.IsSet() {
1322                 return false
1323         }
1324         if t.dataUploadDisallowed {
1325                 return false
1326         }
1327         if cl.config.NoUpload {
1328                 return false
1329         }
1330         if !cl.config.Seed {
1331                 return false
1332         }
1333         if cl.config.DisableAggressiveUpload && t.needData() {
1334                 return false
1335         }
1336         return true
1337 }
1338
1339 func (t *Torrent) onWebRtcConn(
1340         c datachannel.ReadWriteCloser,
1341         dcc webtorrent.DataChannelContext,
1342 ) {
1343         defer c.Close()
1344         pc, err := t.cl.initiateProtocolHandshakes(
1345                 context.Background(),
1346                 webrtcNetConn{c, dcc},
1347                 t,
1348                 dcc.LocalOffered,
1349                 false,
1350                 webrtcNetAddr{dcc.Remote},
1351                 webrtcNetwork,
1352                 fmt.Sprintf("webrtc offer_id %x", dcc.OfferId),
1353         )
1354         if err != nil {
1355                 t.logger.WithDefaultLevel(log.Error).Printf("error in handshaking webrtc connection: %v", err)
1356                 return
1357         }
1358         if dcc.LocalOffered {
1359                 pc.Discovery = PeerSourceTracker
1360         } else {
1361                 pc.Discovery = PeerSourceIncoming
1362         }
1363         t.cl.lock()
1364         defer t.cl.unlock()
1365         err = t.cl.runHandshookConn(pc, t)
1366         if err != nil {
1367                 t.logger.WithDefaultLevel(log.Critical).Printf("error running handshook webrtc conn: %v", err)
1368         }
1369 }
1370
1371 func (t *Torrent) logRunHandshookConn(pc *PeerConn, logAll bool, level log.Level) {
1372         err := t.cl.runHandshookConn(pc, t)
1373         if err != nil || logAll {
1374                 t.logger.WithDefaultLevel(level).Printf("error running handshook conn: %v", err)
1375         }
1376 }
1377
1378 func (t *Torrent) runHandshookConnLoggingErr(pc *PeerConn) {
1379         t.logRunHandshookConn(pc, false, log.Debug)
1380 }
1381
1382 func (t *Torrent) startWebsocketAnnouncer(u url.URL) torrentTrackerAnnouncer {
1383         wtc, release := t.cl.websocketTrackers.Get(u.String())
1384         go func() {
1385                 <-t.closed.LockedChan(t.cl.locker())
1386                 release()
1387         }()
1388         wst := websocketTrackerStatus{u, wtc}
1389         go func() {
1390                 err := wtc.Announce(tracker.Started, t.infoHash)
1391                 if err != nil {
1392                         t.logger.WithDefaultLevel(log.Warning).Printf(
1393                                 "error in initial announce to %q: %v",
1394                                 u.String(), err,
1395                         )
1396                 }
1397         }()
1398         return wst
1399
1400 }
1401
1402 func (t *Torrent) startScrapingTracker(_url string) {
1403         if _url == "" {
1404                 return
1405         }
1406         u, err := url.Parse(_url)
1407         if err != nil {
1408                 // URLs with a leading '*' appear to be a uTorrent convention to
1409                 // disable trackers.
1410                 if _url[0] != '*' {
1411                         log.Str("error parsing tracker url").AddValues("url", _url).Log(t.logger)
1412                 }
1413                 return
1414         }
1415         if u.Scheme == "udp" {
1416                 u.Scheme = "udp4"
1417                 t.startScrapingTracker(u.String())
1418                 u.Scheme = "udp6"
1419                 t.startScrapingTracker(u.String())
1420                 return
1421         }
1422         if _, ok := t.trackerAnnouncers[_url]; ok {
1423                 return
1424         }
1425         sl := func() torrentTrackerAnnouncer {
1426                 switch u.Scheme {
1427                 case "ws", "wss":
1428                         if t.cl.config.DisableWebtorrent {
1429                                 return nil
1430                         }
1431                         return t.startWebsocketAnnouncer(*u)
1432                 case "udp4":
1433                         if t.cl.config.DisableIPv4Peers || t.cl.config.DisableIPv4 {
1434                                 return nil
1435                         }
1436                 case "udp6":
1437                         if t.cl.config.DisableIPv6 {
1438                                 return nil
1439                         }
1440                 }
1441                 urlString := (*u).String()
1442                 cl := t.cl
1443                 newAnnouncer := &trackerScraper{
1444                         u: *u,
1445                         t: t,
1446                         allow: func() {
1447                                 cl.lock()
1448                                 defer cl.unlock()
1449                                 if cl.activeAnnounces == nil {
1450                                         cl.activeAnnounces = make(map[string]struct{})
1451                                 }
1452                                 for {
1453                                         if _, ok := cl.activeAnnounces[urlString]; ok {
1454                                                 cl.event.Wait()
1455                                         } else {
1456                                                 break
1457                                         }
1458                                 }
1459                                 cl.activeAnnounces[urlString] = struct{}{}
1460                         },
1461                         done: func(slowdown bool) {
1462                                 cl.lock()
1463                                 defer cl.unlock()
1464                                 delete(cl.activeAnnounces, urlString)
1465                                 cl.event.Broadcast()
1466                         },
1467                 }
1468                 go newAnnouncer.Run()
1469                 return newAnnouncer
1470         }()
1471         if sl == nil {
1472                 return
1473         }
1474         if t.trackerAnnouncers == nil {
1475                 t.trackerAnnouncers = make(map[string]torrentTrackerAnnouncer)
1476         }
1477         t.trackerAnnouncers[_url] = sl
1478 }
1479
1480 // Adds and starts tracker scrapers for tracker URLs that aren't already
1481 // running.
1482 func (t *Torrent) startMissingTrackerScrapers() {
1483         if t.cl.config.DisableTrackers {
1484                 return
1485         }
1486         t.startScrapingTracker(t.metainfo.Announce)
1487         for _, tier := range t.metainfo.AnnounceList {
1488                 for _, url := range tier {
1489                         t.startScrapingTracker(url)
1490                 }
1491         }
1492 }
1493
1494 // Returns an AnnounceRequest with fields filled out to defaults and current
1495 // values.
1496 func (t *Torrent) announceRequest(event tracker.AnnounceEvent) tracker.AnnounceRequest {
1497         // Note that IPAddress is not set. It's set for UDP inside the tracker code, since it's
1498         // dependent on the network in use.
1499         return tracker.AnnounceRequest{
1500                 Event: event,
1501                 NumWant: func() int32 {
1502                         if t.wantPeers() && len(t.cl.dialers) > 0 {
1503                                 return -1
1504                         } else {
1505                                 return 0
1506                         }
1507                 }(),
1508                 Port:     uint16(t.cl.incomingPeerPort()),
1509                 PeerId:   t.cl.peerID,
1510                 InfoHash: t.infoHash,
1511                 Key:      t.cl.announceKey(),
1512
1513                 // The following are vaguely described in BEP 3.
1514
1515                 Left:     t.bytesLeftAnnounce(),
1516                 Uploaded: t.stats.BytesWrittenData.Int64(),
1517                 // There's no mention of wasted or unwanted download in the BEP.
1518                 Downloaded: t.stats.BytesReadUsefulData.Int64(),
1519         }
1520 }
1521
1522 // Adds peers revealed in an announce until the announce ends, or we have
1523 // enough peers.
1524 func (t *Torrent) consumeDhtAnnouncePeers(pvs <-chan dht.PeersValues) {
1525         cl := t.cl
1526         for v := range pvs {
1527                 cl.lock()
1528                 for _, cp := range v.Peers {
1529                         if cp.Port == 0 {
1530                                 // Can't do anything with this.
1531                                 continue
1532                         }
1533                         t.addPeer(PeerInfo{
1534                                 Addr:   ipPortAddr{cp.IP, cp.Port},
1535                                 Source: PeerSourceDhtGetPeers,
1536                         })
1537                 }
1538                 cl.unlock()
1539         }
1540 }
1541
1542 func (t *Torrent) announceToDht(impliedPort bool, s DhtServer) error {
1543         ps, err := s.Announce(t.infoHash, t.cl.incomingPeerPort(), impliedPort)
1544         if err != nil {
1545                 return err
1546         }
1547         go t.consumeDhtAnnouncePeers(ps.Peers())
1548         select {
1549         case <-t.closed.LockedChan(t.cl.locker()):
1550         case <-time.After(5 * time.Minute):
1551         }
1552         ps.Close()
1553         return nil
1554 }
1555
1556 func (t *Torrent) dhtAnnouncer(s DhtServer) {
1557         cl := t.cl
1558         cl.lock()
1559         defer cl.unlock()
1560         for {
1561                 for {
1562                         if t.closed.IsSet() {
1563                                 return
1564                         }
1565                         if !t.wantPeers() {
1566                                 goto wait
1567                         }
1568                         // TODO: Determine if there's a listener on the port we're announcing.
1569                         if len(cl.dialers) == 0 && len(cl.listeners) == 0 {
1570                                 goto wait
1571                         }
1572                         break
1573                 wait:
1574                         cl.event.Wait()
1575                 }
1576                 func() {
1577                         t.numDHTAnnounces++
1578                         cl.unlock()
1579                         defer cl.lock()
1580                         err := t.announceToDht(true, s)
1581                         if err != nil {
1582                                 t.logger.WithDefaultLevel(log.Warning).Printf("error announcing %q to DHT: %s", t, err)
1583                         }
1584                 }()
1585         }
1586 }
1587
1588 func (t *Torrent) addPeers(peers []PeerInfo) (added int) {
1589         for _, p := range peers {
1590                 if t.addPeer(p) {
1591                         added++
1592                 }
1593         }
1594         return
1595 }
1596
1597 // The returned TorrentStats may require alignment in memory. See
1598 // https://github.com/anacrolix/torrent/issues/383.
1599 func (t *Torrent) Stats() TorrentStats {
1600         t.cl.rLock()
1601         defer t.cl.rUnlock()
1602         return t.statsLocked()
1603 }
1604
1605 func (t *Torrent) statsLocked() (ret TorrentStats) {
1606         ret.ActivePeers = len(t.conns)
1607         ret.HalfOpenPeers = len(t.halfOpen)
1608         ret.PendingPeers = t.peers.Len()
1609         ret.TotalPeers = t.numTotalPeers()
1610         ret.ConnectedSeeders = 0
1611         for c := range t.conns {
1612                 if all, ok := c.peerHasAllPieces(); all && ok {
1613                         ret.ConnectedSeeders++
1614                 }
1615         }
1616         ret.ConnStats = t.stats.Copy()
1617         return
1618 }
1619
1620 // The total number of peers in the torrent.
1621 func (t *Torrent) numTotalPeers() int {
1622         peers := make(map[string]struct{})
1623         for conn := range t.conns {
1624                 ra := conn.conn.RemoteAddr()
1625                 if ra == nil {
1626                         // It's been closed and doesn't support RemoteAddr.
1627                         continue
1628                 }
1629                 peers[ra.String()] = struct{}{}
1630         }
1631         for addr := range t.halfOpen {
1632                 peers[addr] = struct{}{}
1633         }
1634         t.peers.Each(func(peer PeerInfo) {
1635                 peers[peer.Addr.String()] = struct{}{}
1636         })
1637         return len(peers)
1638 }
1639
1640 // Reconcile bytes transferred before connection was associated with a
1641 // torrent.
1642 func (t *Torrent) reconcileHandshakeStats(c *PeerConn) {
1643         if c._stats != (ConnStats{
1644                 // Handshakes should only increment these fields:
1645                 BytesWritten: c._stats.BytesWritten,
1646                 BytesRead:    c._stats.BytesRead,
1647         }) {
1648                 panic("bad stats")
1649         }
1650         c.postHandshakeStats(func(cs *ConnStats) {
1651                 cs.BytesRead.Add(c._stats.BytesRead.Int64())
1652                 cs.BytesWritten.Add(c._stats.BytesWritten.Int64())
1653         })
1654         c.reconciledHandshakeStats = true
1655 }
1656
1657 // Returns true if the connection is added.
1658 func (t *Torrent) addConnection(c *PeerConn) (err error) {
1659         defer func() {
1660                 if err == nil {
1661                         torrent.Add("added connections", 1)
1662                 }
1663         }()
1664         if t.closed.IsSet() {
1665                 return errors.New("torrent closed")
1666         }
1667         for c0 := range t.conns {
1668                 if c.PeerID != c0.PeerID {
1669                         continue
1670                 }
1671                 if !t.cl.config.DropDuplicatePeerIds {
1672                         continue
1673                 }
1674                 if left, ok := c.hasPreferredNetworkOver(c0); ok && left {
1675                         c0.close()
1676                         t.deleteConnection(c0)
1677                 } else {
1678                         return errors.New("existing connection preferred")
1679                 }
1680         }
1681         if len(t.conns) >= t.maxEstablishedConns {
1682                 c := t.worstBadConn()
1683                 if c == nil {
1684                         return errors.New("don't want conns")
1685                 }
1686                 c.close()
1687                 t.deleteConnection(c)
1688         }
1689         if len(t.conns) >= t.maxEstablishedConns {
1690                 panic(len(t.conns))
1691         }
1692         t.conns[c] = struct{}{}
1693         if !t.cl.config.DisablePEX && !c.PeerExtensionBytes.SupportsExtended() {
1694                 t.pex.Add(c) // as no further extended handshake expected
1695         }
1696         return nil
1697 }
1698
1699 func (t *Torrent) wantConns() bool {
1700         if !t.networkingEnabled {
1701                 return false
1702         }
1703         if t.closed.IsSet() {
1704                 return false
1705         }
1706         if !t.seeding() && !t.needData() {
1707                 return false
1708         }
1709         if len(t.conns) < t.maxEstablishedConns {
1710                 return true
1711         }
1712         return t.worstBadConn() != nil
1713 }
1714
1715 func (t *Torrent) SetMaxEstablishedConns(max int) (oldMax int) {
1716         t.cl.lock()
1717         defer t.cl.unlock()
1718         oldMax = t.maxEstablishedConns
1719         t.maxEstablishedConns = max
1720         wcs := slices.HeapInterface(slices.FromMapKeys(t.conns), func(l, r *PeerConn) bool {
1721                 return worseConn(&l.peer, &r.peer)
1722         })
1723         for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 {
1724                 t.dropConnection(wcs.Pop().(*PeerConn))
1725         }
1726         t.openNewConns()
1727         return oldMax
1728 }
1729
1730 func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
1731         t.logger.Log(log.Fstr("hashed piece %d (passed=%t)", piece, passed).SetLevel(log.Debug))
1732         p := t.piece(piece)
1733         p.numVerifies++
1734         t.cl.event.Broadcast()
1735         if t.closed.IsSet() {
1736                 return
1737         }
1738
1739         // Don't score the first time a piece is hashed, it could be an initial check.
1740         if p.storageCompletionOk {
1741                 if passed {
1742                         pieceHashedCorrect.Add(1)
1743                 } else {
1744                         log.Fmsg("piece %d failed hash: %d connections contributed", piece, len(p.dirtiers)).AddValues(t, p).Log(t.logger)
1745                         pieceHashedNotCorrect.Add(1)
1746                 }
1747         }
1748
1749         p.marking = true
1750         t.publishPieceChange(piece)
1751         defer func() {
1752                 p.marking = false
1753                 t.publishPieceChange(piece)
1754         }()
1755
1756         if passed {
1757                 if len(p.dirtiers) != 0 {
1758                         // Don't increment stats above connection-level for every involved connection.
1759                         t.allStats((*ConnStats).incrementPiecesDirtiedGood)
1760                 }
1761                 for c := range p.dirtiers {
1762                         c._stats.incrementPiecesDirtiedGood()
1763                 }
1764                 t.clearPieceTouchers(piece)
1765                 t.cl.unlock()
1766                 err := p.Storage().MarkComplete()
1767                 if err != nil {
1768                         t.logger.Printf("%T: error marking piece complete %d: %s", t.storage, piece, err)
1769                 }
1770                 t.cl.lock()
1771
1772                 if t.closed.IsSet() {
1773                         return
1774                 }
1775                 t.pendAllChunkSpecs(piece)
1776         } else {
1777                 if len(p.dirtiers) != 0 && p.allChunksDirty() && hashIoErr == nil {
1778                         // Peers contributed to all the data for this piece hash failure, and the failure was
1779                         // not due to errors in the storage (such as data being dropped in a cache).
1780
1781                         // Increment Torrent and above stats, and then specific connections.
1782                         t.allStats((*ConnStats).incrementPiecesDirtiedBad)
1783                         for c := range p.dirtiers {
1784                                 // Y u do dis peer?!
1785                                 c.stats().incrementPiecesDirtiedBad()
1786                         }
1787
1788                         bannableTouchers := make([]*peer, 0, len(p.dirtiers))
1789                         for c := range p.dirtiers {
1790                                 if !c.trusted {
1791                                         bannableTouchers = append(bannableTouchers, c)
1792                                 }
1793                         }
1794                         t.clearPieceTouchers(piece)
1795                         slices.Sort(bannableTouchers, connLessTrusted)
1796
1797                         if t.cl.config.Debug {
1798                                 t.logger.Printf(
1799                                         "bannable conns by trust for piece %d: %v",
1800                                         piece,
1801                                         func() (ret []connectionTrust) {
1802                                                 for _, c := range bannableTouchers {
1803                                                         ret = append(ret, c.trust())
1804                                                 }
1805                                                 return
1806                                         }(),
1807                                 )
1808                         }
1809
1810                         if len(bannableTouchers) >= 1 {
1811                                 c := bannableTouchers[0]
1812                                 t.cl.banPeerIP(c.remoteIp())
1813                                 c.drop()
1814                         }
1815                 }
1816                 t.onIncompletePiece(piece)
1817                 p.Storage().MarkNotComplete()
1818         }
1819         t.updatePieceCompletion(piece)
1820 }
1821
1822 func (t *Torrent) cancelRequestsForPiece(piece pieceIndex) {
1823         // TODO: Make faster
1824         for cn := range t.conns {
1825                 cn.tickleWriter()
1826         }
1827 }
1828
1829 func (t *Torrent) onPieceCompleted(piece pieceIndex) {
1830         t.pendAllChunkSpecs(piece)
1831         t.cancelRequestsForPiece(piece)
1832         for conn := range t.conns {
1833                 conn.have(piece)
1834         }
1835 }
1836
1837 // Called when a piece is found to be not complete.
1838 func (t *Torrent) onIncompletePiece(piece pieceIndex) {
1839         if t.pieceAllDirty(piece) {
1840                 t.pendAllChunkSpecs(piece)
1841         }
1842         if !t.wantPieceIndex(piece) {
1843                 // t.logger.Printf("piece %d incomplete and unwanted", piece)
1844                 return
1845         }
1846         // We could drop any connections that we told we have a piece that we
1847         // don't here. But there's a test failure, and it seems clients don't care
1848         // if you request pieces that you already claim to have. Pruning bad
1849         // connections might just remove any connections that aren't treating us
1850         // favourably anyway.
1851
1852         // for c := range t.conns {
1853         //      if c.sentHave(piece) {
1854         //              c.drop()
1855         //      }
1856         // }
1857         t.iterPeers(func(conn *peer) {
1858                 if conn.peerHasPiece(piece) {
1859                         conn.updateRequests()
1860                 }
1861         })
1862 }
1863
1864 func (t *Torrent) tryCreateMorePieceHashers() {
1865         for !t.closed.IsSet() && t.activePieceHashes < 2 && t.tryCreatePieceHasher() {
1866         }
1867 }
1868
1869 func (t *Torrent) tryCreatePieceHasher() bool {
1870         if t.storage == nil {
1871                 return false
1872         }
1873         pi, ok := t.getPieceToHash()
1874         if !ok {
1875                 return false
1876         }
1877         p := t.piece(pi)
1878         t.piecesQueuedForHash.Remove(pi)
1879         p.hashing = true
1880         t.publishPieceChange(pi)
1881         t.updatePiecePriority(pi)
1882         t.storageLock.RLock()
1883         t.activePieceHashes++
1884         go t.pieceHasher(pi)
1885         return true
1886 }
1887
1888 func (t *Torrent) getPieceToHash() (ret pieceIndex, ok bool) {
1889         t.piecesQueuedForHash.IterTyped(func(i pieceIndex) bool {
1890                 if t.piece(i).hashing {
1891                         return true
1892                 }
1893                 ret = i
1894                 ok = true
1895                 return false
1896         })
1897         return
1898 }
1899
1900 func (t *Torrent) pieceHasher(index pieceIndex) {
1901         p := t.piece(index)
1902         sum, copyErr := t.hashPiece(index)
1903         correct := sum == *p.hash
1904         switch copyErr {
1905         case nil, io.EOF:
1906         default:
1907                 log.Fmsg("piece %v (%s) hash failure copy error: %v", p, p.hash.HexString(), copyErr).Log(t.logger)
1908         }
1909         t.storageLock.RUnlock()
1910         t.cl.lock()
1911         defer t.cl.unlock()
1912         p.hashing = false
1913         t.updatePiecePriority(index)
1914         t.pieceHashed(index, correct, copyErr)
1915         t.publishPieceChange(index)
1916         t.activePieceHashes--
1917         t.tryCreateMorePieceHashers()
1918 }
1919
1920 // Return the connections that touched a piece, and clear the entries while doing it.
1921 func (t *Torrent) clearPieceTouchers(pi pieceIndex) {
1922         p := t.piece(pi)
1923         for c := range p.dirtiers {
1924                 delete(c.peerTouchedPieces, pi)
1925                 delete(p.dirtiers, c)
1926         }
1927 }
1928
1929 func (t *Torrent) peersAsSlice() (ret []*peer) {
1930         t.iterPeers(func(p *peer) {
1931                 ret = append(ret, p)
1932         })
1933         return
1934 }
1935
1936 func (t *Torrent) queuePieceCheck(pieceIndex pieceIndex) {
1937         piece := t.piece(pieceIndex)
1938         if piece.queuedForHash() {
1939                 return
1940         }
1941         t.piecesQueuedForHash.Add(bitmap.BitIndex(pieceIndex))
1942         t.publishPieceChange(pieceIndex)
1943         t.updatePiecePriority(pieceIndex)
1944         t.tryCreateMorePieceHashers()
1945 }
1946
1947 // Forces all the pieces to be re-hashed. See also Piece.VerifyData. This should not be called
1948 // before the Info is available.
1949 func (t *Torrent) VerifyData() {
1950         for i := pieceIndex(0); i < t.NumPieces(); i++ {
1951                 t.Piece(i).VerifyData()
1952         }
1953 }
1954
1955 // Start the process of connecting to the given peer for the given torrent if appropriate.
1956 func (t *Torrent) initiateConn(peer PeerInfo) {
1957         if peer.Id == t.cl.peerID {
1958                 return
1959         }
1960         if t.cl.badPeerAddr(peer.Addr) && !peer.Trusted {
1961                 return
1962         }
1963         addr := peer.Addr
1964         if t.addrActive(addr.String()) {
1965                 return
1966         }
1967         t.cl.numHalfOpen++
1968         t.halfOpen[addr.String()] = peer
1969         go t.cl.outgoingConnection(t, addr, peer.Source, peer.Trusted)
1970 }
1971
1972 // Adds a trusted, pending peer for each of the given Client's addresses. Typically used in tests to
1973 // quickly make one Client visible to the Torrent of another Client.
1974 func (t *Torrent) AddClientPeer(cl *Client) int {
1975         return t.AddPeers(func() (ps []PeerInfo) {
1976                 for _, la := range cl.ListenAddrs() {
1977                         ps = append(ps, PeerInfo{
1978                                 Addr:    la,
1979                                 Trusted: true,
1980                         })
1981                 }
1982                 return
1983         }())
1984 }
1985
1986 // All stats that include this Torrent. Useful when we want to increment ConnStats but not for every
1987 // connection.
1988 func (t *Torrent) allStats(f func(*ConnStats)) {
1989         f(&t.stats)
1990         f(&t.cl.stats)
1991 }
1992
1993 func (t *Torrent) hashingPiece(i pieceIndex) bool {
1994         return t.pieces[i].hashing
1995 }
1996
1997 func (t *Torrent) pieceQueuedForHash(i pieceIndex) bool {
1998         return t.piecesQueuedForHash.Get(bitmap.BitIndex(i))
1999 }
2000
2001 func (t *Torrent) dialTimeout() time.Duration {
2002         return reducedDialTimeout(t.cl.config.MinDialTimeout, t.cl.config.NominalDialTimeout, t.cl.config.HalfOpenConnsPerTorrent, t.peers.Len())
2003 }
2004
2005 func (t *Torrent) piece(i int) *Piece {
2006         return &t.pieces[i]
2007 }
2008
2009 func (t *Torrent) requestStrategyTorrent() requestStrategyTorrent {
2010         return t
2011 }
2012
2013 type torrentRequestStrategyCallbacks struct {
2014         t *Torrent
2015 }
2016
2017 func (cb torrentRequestStrategyCallbacks) requestTimedOut(r request) {
2018         torrent.Add("request timeouts", 1)
2019         cb.t.cl.lock()
2020         defer cb.t.cl.unlock()
2021         cb.t.iterPeers(func(cn *peer) {
2022                 if cn.peerHasPiece(pieceIndex(r.Index)) {
2023                         cn.updateRequests()
2024                 }
2025         })
2026
2027 }
2028
2029 func (t *Torrent) requestStrategyCallbacks() requestStrategyCallbacks {
2030         return torrentRequestStrategyCallbacks{t}
2031 }
2032
2033 func (t *Torrent) onWriteChunkErr(err error) {
2034         if t.userOnWriteChunkErr != nil {
2035                 go t.userOnWriteChunkErr(err)
2036                 return
2037         }
2038         t.logger.WithDefaultLevel(log.Critical).Printf("default chunk write error handler: disabling data download")
2039         t.disallowDataDownloadLocked()
2040 }
2041
2042 func (t *Torrent) DisallowDataDownload() {
2043         t.cl.lock()
2044         defer t.cl.unlock()
2045         t.disallowDataDownloadLocked()
2046 }
2047
2048 func (t *Torrent) disallowDataDownloadLocked() {
2049         t.dataDownloadDisallowed = true
2050         t.iterPeers(func(c *peer) {
2051                 c.updateRequests()
2052         })
2053         t.tickleReaders()
2054 }
2055
2056 func (t *Torrent) AllowDataDownload() {
2057         t.cl.lock()
2058         defer t.cl.unlock()
2059         t.dataDownloadDisallowed = false
2060         t.tickleReaders()
2061         t.iterPeers(func(c *peer) {
2062                 c.updateRequests()
2063         })
2064 }
2065
2066 // Enables uploading data, if it was disabled.
2067 func (t *Torrent) AllowDataUpload() {
2068         t.cl.lock()
2069         defer t.cl.unlock()
2070         t.dataUploadDisallowed = false
2071         for c := range t.conns {
2072                 c.updateRequests()
2073         }
2074 }
2075
2076 // Disables uploading data, if it was enabled.
2077 func (t *Torrent) DisallowDataUpload() {
2078         t.cl.lock()
2079         defer t.cl.unlock()
2080         t.dataUploadDisallowed = true
2081         for c := range t.conns {
2082                 c.updateRequests()
2083         }
2084 }
2085
2086 // Sets a handler that is called if there's an error writing a chunk to local storage. By default,
2087 // or if nil, a critical message is logged, and data download is disabled.
2088 func (t *Torrent) SetOnWriteChunkError(f func(error)) {
2089         t.cl.lock()
2090         defer t.cl.unlock()
2091         t.userOnWriteChunkErr = f
2092 }
2093
2094 func (t *Torrent) iterPeers(f func(*peer)) {
2095         for pc := range t.conns {
2096                 f(&pc.peer)
2097         }
2098         for _, ws := range t.webSeeds {
2099                 f(ws)
2100         }
2101 }
2102
2103 func (t *Torrent) addWebSeed(url string) {
2104         if t.cl.config.DisableWebseeds {
2105                 return
2106         }
2107         if _, ok := t.webSeeds[url]; ok {
2108                 return
2109         }
2110         const maxRequests = 10
2111         ws := webseedPeer{
2112                 peer: peer{
2113                         t:                        t,
2114                         outgoing:                 true,
2115                         network:                  "http",
2116                         reconciledHandshakeStats: true,
2117                         peerSentHaveAll:          true,
2118                         PeerMaxRequests:          maxRequests,
2119                 },
2120                 client: webseed.Client{
2121                         HttpClient: http.DefaultClient,
2122                         Url:        url,
2123                 },
2124                 requests: make(map[request]webseed.Request, maxRequests),
2125         }
2126         ws.peer.logger = t.logger.WithContextValue(&ws)
2127         ws.peer.peerImpl = &ws
2128         if t.haveInfo() {
2129                 ws.onGotInfo(t.info)
2130         }
2131         t.webSeeds[url] = &ws.peer
2132 }
2133
2134 func (t *Torrent) peerIsActive(p *peer) (active bool) {
2135         t.iterPeers(func(p1 *peer) {
2136                 if p1 == p {
2137                         active = true
2138                 }
2139         })
2140         return
2141 }