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) maybeDropMutuallyCompletePeer(
 830         // I'm not sure about taking peer here, not all peer implementations actually drop. Maybe that's okay?
 831         p *peer,
 832 ) {
 833         if !t.cl.config.DropMutuallyCompletePeers {
 834                 return
 835         }
 836         if !t.haveAllPieces() {
 837                 return
 838         }
 839         if all, known := p.peerHasAllPieces(); !(known && all) {
 840                 return
 841         }
 842         if p.useful() {
 843                 return
 844         }
 845         t.logger.WithDefaultLevel(log.Debug).Printf("dropping %v, which is mutually complete", p)
 846         p.drop()
 847 }
 848
 849 func (t *Torrent) haveChunk(r request) (ret bool) {
 850         // defer func() {
 851         //      log.Println("have chunk", r, ret)
 852         // }()
 853         if !t.haveInfo() {
 854                 return false
 855         }
 856         if t.pieceComplete(pieceIndex(r.Index)) {
 857                 return true
 858         }
 859         p := &t.pieces[r.Index]
 860         return !p.pendingChunk(r.chunkSpec, t.chunkSize)
 861 }
 862
 863 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
 864         return int(cs.Begin / chunkSize)
 865 }
 866
 867 func (t *Torrent) wantPieceIndex(index pieceIndex) bool {
 868         if !t.haveInfo() {
 869                 return false
 870         }
 871         if index < 0 || index >= t.numPieces() {
 872                 return false
 873         }
 874         p := &t.pieces[index]
 875         if p.queuedForHash() {
 876                 return false
 877         }
 878         if p.hashing {
 879                 return false
 880         }
 881         if t.pieceComplete(index) {
 882                 return false
 883         }
 884         if t._pendingPieces.Contains(bitmap.BitIndex(index)) {
 885                 return true
 886         }
 887         // t.logger.Printf("piece %d not pending", index)
 888         return !t.forReaderOffsetPieces(func(begin, end pieceIndex) bool {
 889                 return index < begin || index >= end
 890         })
 891 }
 892
 893 // The worst connection is one that hasn't been sent, or sent anything useful for the longest. A bad
 894 // connection is one that usually sends us unwanted pieces, or has been in worser half of the
 895 // established connections for more than a minute.
 896 func (t *Torrent) worstBadConn() *PeerConn {
 897         wcs := worseConnSlice{t.unclosedConnsAsSlice()}
 898         heap.Init(&wcs)
 899         for wcs.Len() != 0 {
 900                 c := heap.Pop(&wcs).(*PeerConn)
 901                 if c._stats.ChunksReadWasted.Int64() >= 6 && c._stats.ChunksReadWasted.Int64() > c._stats.ChunksReadUseful.Int64() {
 902                         return c
 903                 }
 904                 // If the connection is in the worst half of the established
 905                 // connection quota and is older than a minute.
 906                 if wcs.Len() >= (t.maxEstablishedConns+1)/2 {
 907                         // Give connections 1 minute to prove themselves.
 908                         if time.Since(c.completedHandshake) > time.Minute {
 909                                 return c
 910                         }
 911                 }
 912         }
 913         return nil
 914 }
 915
 916 type PieceStateChange struct {
 917         Index int
 918         PieceState
 919 }
 920
 921 func (t *Torrent) publishPieceChange(piece pieceIndex) {
 922         t.cl._mu.Defer(func() {
 923                 cur := t.pieceState(piece)
 924                 p := &t.pieces[piece]
 925                 if cur != p.publicPieceState {
 926                         p.publicPieceState = cur
 927                         t.pieceStateChanges.Publish(PieceStateChange{
 928                                 int(piece),
 929                                 cur,
 930                         })
 931                 }
 932         })
 933 }
 934
 935 func (t *Torrent) pieceNumPendingChunks(piece pieceIndex) pp.Integer {
 936         if t.pieceComplete(piece) {
 937                 return 0
 938         }
 939         return t.pieceNumChunks(piece) - t.pieces[piece].numDirtyChunks()
 940 }
 941
 942 func (t *Torrent) pieceAllDirty(piece pieceIndex) bool {
 943         return t.pieces[piece]._dirtyChunks.Len() == int(t.pieceNumChunks(piece))
 944 }
 945
 946 func (t *Torrent) readersChanged() {
 947         t.updateReaderPieces()
 948         t.updateAllPiecePriorities()
 949 }
 950
 951 func (t *Torrent) updateReaderPieces() {
 952         t._readerNowPieces, t._readerReadaheadPieces = t.readerPiecePriorities()
 953 }
 954
 955 func (t *Torrent) readerPosChanged(from, to pieceRange) {
 956         if from == to {
 957                 return
 958         }
 959         t.updateReaderPieces()
 960         // Order the ranges, high and low.
 961         l, h := from, to
 962         if l.begin > h.begin {
 963                 l, h = h, l
 964         }
 965         if l.end < h.begin {
 966                 // Two distinct ranges.
 967                 t.updatePiecePriorities(l.begin, l.end)
 968                 t.updatePiecePriorities(h.begin, h.end)
 969         } else {
 970                 // Ranges overlap.
 971                 end := l.end
 972                 if h.end > end {
 973                         end = h.end
 974                 }
 975                 t.updatePiecePriorities(l.begin, end)
 976         }
 977 }
 978
 979 func (t *Torrent) maybeNewConns() {
 980         // Tickle the accept routine.
 981         t.cl.event.Broadcast()
 982         t.openNewConns()
 983 }
 984
 985 func (t *Torrent) piecePriorityChanged(piece pieceIndex) {
 986         // t.logger.Printf("piece %d priority changed", piece)
 987         t.iterPeers(func(c *peer) {
 988                 if c.updatePiecePriority(piece) {
 989                         // log.Print("conn piece priority changed")
 990                         c.updateRequests()
 991                 }
 992         })
 993         t.maybeNewConns()
 994         t.publishPieceChange(piece)
 995 }
 996
 997 func (t *Torrent) updatePiecePriority(piece pieceIndex) {
 998         p := &t.pieces[piece]
 999         newPrio := p.uncachedPriority()
1000         // t.logger.Printf("torrent %p: piece %d: uncached priority: %v", t, piece, newPrio)
1001         if newPrio == PiecePriorityNone {
1002                 if !t._pendingPieces.Remove(bitmap.BitIndex(piece)) {
1003                         return
1004                 }
1005         } else {
1006                 if !t._pendingPieces.Set(bitmap.BitIndex(piece), newPrio.BitmapPriority()) {
1007                         return
1008                 }
1009         }
1010         t.piecePriorityChanged(piece)
1011 }
1012
1013 func (t *Torrent) updateAllPiecePriorities() {
1014         t.updatePiecePriorities(0, t.numPieces())
1015 }
1016
1017 // Update all piece priorities in one hit. This function should have the same
1018 // output as updatePiecePriority, but across all pieces.
1019 func (t *Torrent) updatePiecePriorities(begin, end pieceIndex) {
1020         for i := begin; i < end; i++ {
1021                 t.updatePiecePriority(i)
1022         }
1023 }
1024
1025 // Returns the range of pieces [begin, end) that contains the extent of bytes.
1026 func (t *Torrent) byteRegionPieces(off, size int64) (begin, end pieceIndex) {
1027         if off >= *t.length {
1028                 return
1029         }
1030         if off < 0 {
1031                 size += off
1032                 off = 0
1033         }
1034         if size <= 0 {
1035                 return
1036         }
1037         begin = pieceIndex(off / t.info.PieceLength)
1038         end = pieceIndex((off + size + t.info.PieceLength - 1) / t.info.PieceLength)
1039         if end > pieceIndex(t.info.NumPieces()) {
1040                 end = pieceIndex(t.info.NumPieces())
1041         }
1042         return
1043 }
1044
1045 // Returns true if all iterations complete without breaking. Returns the read
1046 // regions for all readers. The reader regions should not be merged as some
1047 // callers depend on this method to enumerate readers.
1048 func (t *Torrent) forReaderOffsetPieces(f func(begin, end pieceIndex) (more bool)) (all bool) {
1049         for r := range t.readers {
1050                 p := r.pieces
1051                 if p.begin >= p.end {
1052                         continue
1053                 }
1054                 if !f(p.begin, p.end) {
1055                         return false
1056                 }
1057         }
1058         return true
1059 }
1060
1061 func (t *Torrent) piecePriority(piece pieceIndex) piecePriority {
1062         prio, ok := t._pendingPieces.GetPriority(bitmap.BitIndex(piece))
1063         if !ok {
1064                 return PiecePriorityNone
1065         }
1066         if prio > 0 {
1067                 panic(prio)
1068         }
1069         ret := piecePriority(-prio)
1070         if ret == PiecePriorityNone {
1071                 panic(piece)
1072         }
1073         return ret
1074 }
1075
1076 func (t *Torrent) pendRequest(req request) {
1077         ci := chunkIndex(req.chunkSpec, t.chunkSize)
1078         t.pieces[req.Index].pendChunkIndex(ci)
1079 }
1080
1081 func (t *Torrent) pieceCompletionChanged(piece pieceIndex) {
1082         t.tickleReaders()
1083         t.cl.event.Broadcast()
1084         if t.pieceComplete(piece) {
1085                 t.onPieceCompleted(piece)
1086         } else {
1087                 t.onIncompletePiece(piece)
1088         }
1089         t.updatePiecePriority(piece)
1090 }
1091
1092 func (t *Torrent) numReceivedConns() (ret int) {
1093         for c := range t.conns {
1094                 if c.Discovery == PeerSourceIncoming {
1095                         ret++
1096                 }
1097         }
1098         return
1099 }
1100
1101 func (t *Torrent) maxHalfOpen() int {
1102         // Note that if we somehow exceed the maximum established conns, we want
1103         // the negative value to have an effect.
1104         establishedHeadroom := int64(t.maxEstablishedConns - len(t.conns))
1105         extraIncoming := int64(t.numReceivedConns() - t.maxEstablishedConns/2)
1106         // We want to allow some experimentation with new peers, and to try to
1107         // upset an oversupply of received connections.
1108         return int(min(max(5, extraIncoming)+establishedHeadroom, int64(t.cl.config.HalfOpenConnsPerTorrent)))
1109 }
1110
1111 func (t *Torrent) openNewConns() (initiated int) {
1112         defer t.updateWantPeersEvent()
1113         for t.peers.Len() != 0 {
1114                 if !t.wantConns() {
1115                         return
1116                 }
1117                 if len(t.halfOpen) >= t.maxHalfOpen() {
1118                         return
1119                 }
1120                 if len(t.cl.dialers) == 0 {
1121                         return
1122                 }
1123                 if t.cl.numHalfOpen >= t.cl.config.TotalHalfOpenConns {
1124                         return
1125                 }
1126                 p := t.peers.PopMax()
1127                 t.initiateConn(p)
1128                 initiated++
1129         }
1130         return
1131 }
1132
1133 func (t *Torrent) getConnPieceInclination() []int {
1134         _ret := t.connPieceInclinationPool.Get()
1135         if _ret == nil {
1136                 pieceInclinationsNew.Add(1)
1137                 return rand.Perm(int(t.numPieces()))
1138         }
1139         pieceInclinationsReused.Add(1)
1140         return *_ret.(*[]int)
1141 }
1142
1143 func (t *Torrent) putPieceInclination(pi []int) {
1144         t.connPieceInclinationPool.Put(&pi)
1145         pieceInclinationsPut.Add(1)
1146 }
1147
1148 func (t *Torrent) updatePieceCompletion(piece pieceIndex) bool {
1149         p := t.piece(piece)
1150         uncached := t.pieceCompleteUncached(piece)
1151         cached := p.completion()
1152         changed := cached != uncached
1153         complete := uncached.Complete
1154         p.storageCompletionOk = uncached.Ok
1155         t._completedPieces.Set(bitmap.BitIndex(piece), complete)
1156         if complete && len(p.dirtiers) != 0 {
1157                 t.logger.Printf("marked piece %v complete but still has dirtiers", piece)
1158         }
1159         if changed {
1160                 log.Fstr("piece %d completion changed: %+v -> %+v", piece, cached, uncached).SetLevel(log.Debug).Log(t.logger)
1161                 t.pieceCompletionChanged(piece)
1162         }
1163         return changed
1164 }
1165
1166 // Non-blocking read. Client lock is not required.
1167 func (t *Torrent) readAt(b []byte, off int64) (n int, err error) {
1168         for len(b) != 0 {
1169                 p := &t.pieces[off/t.info.PieceLength]
1170                 p.waitNoPendingWrites()
1171                 var n1 int
1172                 n1, err = p.Storage().ReadAt(b, off-p.Info().Offset())
1173                 if n1 == 0 {
1174                         break
1175                 }
1176                 off += int64(n1)
1177                 n += n1
1178                 b = b[n1:]
1179         }
1180         return
1181 }
1182
1183 // Returns an error if the metadata was completed, but couldn't be set for
1184 // some reason. Blame it on the last peer to contribute.
1185 func (t *Torrent) maybeCompleteMetadata() error {
1186         if t.haveInfo() {
1187                 // Nothing to do.
1188                 return nil
1189         }
1190         if !t.haveAllMetadataPieces() {
1191                 // Don't have enough metadata pieces.
1192                 return nil
1193         }
1194         err := t.setInfoBytes(t.metadataBytes)
1195         if err != nil {
1196                 t.invalidateMetadata()
1197                 return fmt.Errorf("error setting info bytes: %s", err)
1198         }
1199         if t.cl.config.Debug {
1200                 t.logger.Printf("%s: got metadata from peers", t)
1201         }
1202         return nil
1203 }
1204
1205 func (t *Torrent) readerPiecePriorities() (now, readahead bitmap.Bitmap) {
1206         t.forReaderOffsetPieces(func(begin, end pieceIndex) bool {
1207                 if end > begin {
1208                         now.Add(bitmap.BitIndex(begin))
1209                         readahead.AddRange(bitmap.BitIndex(begin)+1, bitmap.BitIndex(end))
1210                 }
1211                 return true
1212         })
1213         return
1214 }
1215
1216 func (t *Torrent) needData() bool {
1217         if t.closed.IsSet() {
1218                 return false
1219         }
1220         if !t.haveInfo() {
1221                 return true
1222         }
1223         return t._pendingPieces.Len() != 0
1224 }
1225
1226 func appendMissingStrings(old, new []string) (ret []string) {
1227         ret = old
1228 new:
1229         for _, n := range new {
1230                 for _, o := range old {
1231                         if o == n {
1232                                 continue new
1233                         }
1234                 }
1235                 ret = append(ret, n)
1236         }
1237         return
1238 }
1239
1240 func appendMissingTrackerTiers(existing [][]string, minNumTiers int) (ret [][]string) {
1241         ret = existing
1242         for minNumTiers > len(ret) {
1243                 ret = append(ret, nil)
1244         }
1245         return
1246 }
1247
1248 func (t *Torrent) addTrackers(announceList [][]string) {
1249         fullAnnounceList := &t.metainfo.AnnounceList
1250         t.metainfo.AnnounceList = appendMissingTrackerTiers(*fullAnnounceList, len(announceList))
1251         for tierIndex, trackerURLs := range announceList {
1252                 (*fullAnnounceList)[tierIndex] = appendMissingStrings((*fullAnnounceList)[tierIndex], trackerURLs)
1253         }
1254         t.startMissingTrackerScrapers()
1255         t.updateWantPeersEvent()
1256 }
1257
1258 // Don't call this before the info is available.
1259 func (t *Torrent) bytesCompleted() int64 {
1260         if !t.haveInfo() {
1261                 return 0
1262         }
1263         return t.info.TotalLength() - t.bytesLeft()
1264 }
1265
1266 func (t *Torrent) SetInfoBytes(b []byte) (err error) {
1267         t.cl.lock()
1268         defer t.cl.unlock()
1269         return t.setInfoBytes(b)
1270 }
1271
1272 // Returns true if connection is removed from torrent.Conns.
1273 func (t *Torrent) deleteConnection(c *PeerConn) (ret bool) {
1274         if !c.closed.IsSet() {
1275                 panic("connection is not closed")
1276                 // There are behaviours prevented by the closed state that will fail
1277                 // if the connection has been deleted.
1278         }
1279         _, ret = t.conns[c]
1280         delete(t.conns, c)
1281         // Avoid adding a drop event more than once. Probably we should track whether we've generated
1282         // the drop event against the PexConnState instead.
1283         if ret {
1284                 if !t.cl.config.DisablePEX {
1285                         t.pex.Drop(c)
1286                 }
1287         }
1288         torrent.Add("deleted connections", 1)
1289         c.deleteAllRequests()
1290         if t.numActivePeers() == 0 {
1291                 t.assertNoPendingRequests()
1292         }
1293         return
1294 }
1295
1296 func (t *Torrent) numActivePeers() (num int) {
1297         t.iterPeers(func(*peer) {
1298                 num++
1299         })
1300         return
1301 }
1302
1303 func (t *Torrent) assertNoPendingRequests() {
1304         if len(t.pendingRequests) != 0 {
1305                 panic(t.pendingRequests)
1306         }
1307         //if len(t.lastRequested) != 0 {
1308         //      panic(t.lastRequested)
1309         //}
1310 }
1311
1312 func (t *Torrent) dropConnection(c *PeerConn) {
1313         t.cl.event.Broadcast()
1314         c.close()
1315         if t.deleteConnection(c) {
1316                 t.openNewConns()
1317         }
1318 }
1319
1320 func (t *Torrent) wantPeers() bool {
1321         if t.closed.IsSet() {
1322                 return false
1323         }
1324         if t.peers.Len() > t.cl.config.TorrentPeersLowWater {
1325                 return false
1326         }
1327         return t.needData() || t.seeding()
1328 }
1329
1330 func (t *Torrent) updateWantPeersEvent() {
1331         if t.wantPeers() {
1332                 t.wantPeersEvent.Set()
1333         } else {
1334                 t.wantPeersEvent.Clear()
1335         }
1336 }
1337
1338 // Returns whether the client should make effort to seed the torrent.
1339 func (t *Torrent) seeding() bool {
1340         cl := t.cl
1341         if t.closed.IsSet() {
1342                 return false
1343         }
1344         if t.dataUploadDisallowed {
1345                 return false
1346         }
1347         if cl.config.NoUpload {
1348                 return false
1349         }
1350         if !cl.config.Seed {
1351                 return false
1352         }
1353         if cl.config.DisableAggressiveUpload && t.needData() {
1354                 return false
1355         }
1356         return true
1357 }
1358
1359 func (t *Torrent) onWebRtcConn(
1360         c datachannel.ReadWriteCloser,
1361         dcc webtorrent.DataChannelContext,
1362 ) {
1363         defer c.Close()
1364         pc, err := t.cl.initiateProtocolHandshakes(
1365                 context.Background(),
1366                 webrtcNetConn{c, dcc},
1367                 t,
1368                 dcc.LocalOffered,
1369                 false,
1370                 webrtcNetAddr{dcc.Remote},
1371                 webrtcNetwork,
1372                 fmt.Sprintf("webrtc offer_id %x", dcc.OfferId),
1373         )
1374         if err != nil {
1375                 t.logger.WithDefaultLevel(log.Error).Printf("error in handshaking webrtc connection: %v", err)
1376                 return
1377         }
1378         if dcc.LocalOffered {
1379                 pc.Discovery = PeerSourceTracker
1380         } else {
1381                 pc.Discovery = PeerSourceIncoming
1382         }
1383         t.cl.lock()
1384         defer t.cl.unlock()
1385         err = t.cl.runHandshookConn(pc, t)
1386         if err != nil {
1387                 t.logger.WithDefaultLevel(log.Critical).Printf("error running handshook webrtc conn: %v", err)
1388         }
1389 }
1390
1391 func (t *Torrent) logRunHandshookConn(pc *PeerConn, logAll bool, level log.Level) {
1392         err := t.cl.runHandshookConn(pc, t)
1393         if err != nil || logAll {
1394                 t.logger.WithDefaultLevel(level).Printf("error running handshook conn: %v", err)
1395         }
1396 }
1397
1398 func (t *Torrent) runHandshookConnLoggingErr(pc *PeerConn) {
1399         t.logRunHandshookConn(pc, false, log.Debug)
1400 }
1401
1402 func (t *Torrent) startWebsocketAnnouncer(u url.URL) torrentTrackerAnnouncer {
1403         wtc, release := t.cl.websocketTrackers.Get(u.String())
1404         go func() {
1405                 <-t.closed.LockedChan(t.cl.locker())
1406                 release()
1407         }()
1408         wst := websocketTrackerStatus{u, wtc}
1409         go func() {
1410                 err := wtc.Announce(tracker.Started, t.infoHash)
1411                 if err != nil {
1412                         t.logger.WithDefaultLevel(log.Warning).Printf(
1413                                 "error in initial announce to %q: %v",
1414                                 u.String(), err,
1415                         )
1416                 }
1417         }()
1418         return wst
1419
1420 }
1421
1422 func (t *Torrent) startScrapingTracker(_url string) {
1423         if _url == "" {
1424                 return
1425         }
1426         u, err := url.Parse(_url)
1427         if err != nil {
1428                 // URLs with a leading '*' appear to be a uTorrent convention to
1429                 // disable trackers.
1430                 if _url[0] != '*' {
1431                         log.Str("error parsing tracker url").AddValues("url", _url).Log(t.logger)
1432                 }
1433                 return
1434         }
1435         if u.Scheme == "udp" {
1436                 u.Scheme = "udp4"
1437                 t.startScrapingTracker(u.String())
1438                 u.Scheme = "udp6"
1439                 t.startScrapingTracker(u.String())
1440                 return
1441         }
1442         if _, ok := t.trackerAnnouncers[_url]; ok {
1443                 return
1444         }
1445         sl := func() torrentTrackerAnnouncer {
1446                 switch u.Scheme {
1447                 case "ws", "wss":
1448                         if t.cl.config.DisableWebtorrent {
1449                                 return nil
1450                         }
1451                         return t.startWebsocketAnnouncer(*u)
1452                 case "udp4":
1453                         if t.cl.config.DisableIPv4Peers || t.cl.config.DisableIPv4 {
1454                                 return nil
1455                         }
1456                 case "udp6":
1457                         if t.cl.config.DisableIPv6 {
1458                                 return nil
1459                         }
1460                 }
1461                 newAnnouncer := &trackerScraper{
1462                         u: *u,
1463                         t: t,
1464                 }
1465                 go newAnnouncer.Run()
1466                 return newAnnouncer
1467         }()
1468         if sl == nil {
1469                 return
1470         }
1471         if t.trackerAnnouncers == nil {
1472                 t.trackerAnnouncers = make(map[string]torrentTrackerAnnouncer)
1473         }
1474         t.trackerAnnouncers[_url] = sl
1475 }
1476
1477 // Adds and starts tracker scrapers for tracker URLs that aren't already
1478 // running.
1479 func (t *Torrent) startMissingTrackerScrapers() {
1480         if t.cl.config.DisableTrackers {
1481                 return
1482         }
1483         t.startScrapingTracker(t.metainfo.Announce)
1484         for _, tier := range t.metainfo.AnnounceList {
1485                 for _, url := range tier {
1486                         t.startScrapingTracker(url)
1487                 }
1488         }
1489 }
1490
1491 // Returns an AnnounceRequest with fields filled out to defaults and current
1492 // values.
1493 func (t *Torrent) announceRequest(event tracker.AnnounceEvent) tracker.AnnounceRequest {
1494         // Note that IPAddress is not set. It's set for UDP inside the tracker code, since it's
1495         // dependent on the network in use.
1496         return tracker.AnnounceRequest{
1497                 Event: event,
1498                 NumWant: func() int32 {
1499                         if t.wantPeers() && len(t.cl.dialers) > 0 {
1500                                 return -1
1501                         } else {
1502                                 return 0
1503                         }
1504                 }(),
1505                 Port:     uint16(t.cl.incomingPeerPort()),
1506                 PeerId:   t.cl.peerID,
1507                 InfoHash: t.infoHash,
1508                 Key:      t.cl.announceKey(),
1509
1510                 // The following are vaguely described in BEP 3.
1511
1512                 Left:     t.bytesLeftAnnounce(),
1513                 Uploaded: t.stats.BytesWrittenData.Int64(),
1514                 // There's no mention of wasted or unwanted download in the BEP.
1515                 Downloaded: t.stats.BytesReadUsefulData.Int64(),
1516         }
1517 }
1518
1519 // Adds peers revealed in an announce until the announce ends, or we have
1520 // enough peers.
1521 func (t *Torrent) consumeDhtAnnouncePeers(pvs <-chan dht.PeersValues) {
1522         cl := t.cl
1523         for v := range pvs {
1524                 cl.lock()
1525                 for _, cp := range v.Peers {
1526                         if cp.Port == 0 {
1527                                 // Can't do anything with this.
1528                                 continue
1529                         }
1530                         t.addPeer(PeerInfo{
1531                                 Addr:   ipPortAddr{cp.IP, cp.Port},
1532                                 Source: PeerSourceDhtGetPeers,
1533                         })
1534                 }
1535                 cl.unlock()
1536         }
1537 }
1538
1539 func (t *Torrent) announceToDht(impliedPort bool, s DhtServer) error {
1540         ps, err := s.Announce(t.infoHash, t.cl.incomingPeerPort(), impliedPort)
1541         if err != nil {
1542                 return err
1543         }
1544         go t.consumeDhtAnnouncePeers(ps.Peers())
1545         select {
1546         case <-t.closed.LockedChan(t.cl.locker()):
1547         case <-time.After(5 * time.Minute):
1548         }
1549         ps.Close()
1550         return nil
1551 }
1552
1553 func (t *Torrent) dhtAnnouncer(s DhtServer) {
1554         cl := t.cl
1555         cl.lock()
1556         defer cl.unlock()
1557         for {
1558                 for {
1559                         if t.closed.IsSet() {
1560                                 return
1561                         }
1562                         if !t.wantPeers() {
1563                                 goto wait
1564                         }
1565                         // TODO: Determine if there's a listener on the port we're announcing.
1566                         if len(cl.dialers) == 0 && len(cl.listeners) == 0 {
1567                                 goto wait
1568                         }
1569                         break
1570                 wait:
1571                         cl.event.Wait()
1572                 }
1573                 func() {
1574                         t.numDHTAnnounces++
1575                         cl.unlock()
1576                         defer cl.lock()
1577                         err := t.announceToDht(true, s)
1578                         if err != nil {
1579                                 t.logger.WithDefaultLevel(log.Warning).Printf("error announcing %q to DHT: %s", t, err)
1580                         }
1581                 }()
1582         }
1583 }
1584
1585 func (t *Torrent) addPeers(peers []PeerInfo) (added int) {
1586         for _, p := range peers {
1587                 if t.addPeer(p) {
1588                         added++
1589                 }
1590         }
1591         return
1592 }
1593
1594 // The returned TorrentStats may require alignment in memory. See
1595 // https://github.com/anacrolix/torrent/issues/383.
1596 func (t *Torrent) Stats() TorrentStats {
1597         t.cl.rLock()
1598         defer t.cl.rUnlock()
1599         return t.statsLocked()
1600 }
1601
1602 func (t *Torrent) statsLocked() (ret TorrentStats) {
1603         ret.ActivePeers = len(t.conns)
1604         ret.HalfOpenPeers = len(t.halfOpen)
1605         ret.PendingPeers = t.peers.Len()
1606         ret.TotalPeers = t.numTotalPeers()
1607         ret.ConnectedSeeders = 0
1608         for c := range t.conns {
1609                 if all, ok := c.peerHasAllPieces(); all && ok {
1610                         ret.ConnectedSeeders++
1611                 }
1612         }
1613         ret.ConnStats = t.stats.Copy()
1614         return
1615 }
1616
1617 // The total number of peers in the torrent.
1618 func (t *Torrent) numTotalPeers() int {
1619         peers := make(map[string]struct{})
1620         for conn := range t.conns {
1621                 ra := conn.conn.RemoteAddr()
1622                 if ra == nil {
1623                         // It's been closed and doesn't support RemoteAddr.
1624                         continue
1625                 }
1626                 peers[ra.String()] = struct{}{}
1627         }
1628         for addr := range t.halfOpen {
1629                 peers[addr] = struct{}{}
1630         }
1631         t.peers.Each(func(peer PeerInfo) {
1632                 peers[peer.Addr.String()] = struct{}{}
1633         })
1634         return len(peers)
1635 }
1636
1637 // Reconcile bytes transferred before connection was associated with a
1638 // torrent.
1639 func (t *Torrent) reconcileHandshakeStats(c *PeerConn) {
1640         if c._stats != (ConnStats{
1641                 // Handshakes should only increment these fields:
1642                 BytesWritten: c._stats.BytesWritten,
1643                 BytesRead:    c._stats.BytesRead,
1644         }) {
1645                 panic("bad stats")
1646         }
1647         c.postHandshakeStats(func(cs *ConnStats) {
1648                 cs.BytesRead.Add(c._stats.BytesRead.Int64())
1649                 cs.BytesWritten.Add(c._stats.BytesWritten.Int64())
1650         })
1651         c.reconciledHandshakeStats = true
1652 }
1653
1654 // Returns true if the connection is added.
1655 func (t *Torrent) addConnection(c *PeerConn) (err error) {
1656         defer func() {
1657                 if err == nil {
1658                         torrent.Add("added connections", 1)
1659                 }
1660         }()
1661         if t.closed.IsSet() {
1662                 return errors.New("torrent closed")
1663         }
1664         for c0 := range t.conns {
1665                 if c.PeerID != c0.PeerID {
1666                         continue
1667                 }
1668                 if !t.cl.config.DropDuplicatePeerIds {
1669                         continue
1670                 }
1671                 if left, ok := c.hasPreferredNetworkOver(c0); ok && left {
1672                         c0.close()
1673                         t.deleteConnection(c0)
1674                 } else {
1675                         return errors.New("existing connection preferred")
1676                 }
1677         }
1678         if len(t.conns) >= t.maxEstablishedConns {
1679                 c := t.worstBadConn()
1680                 if c == nil {
1681                         return errors.New("don't want conns")
1682                 }
1683                 c.close()
1684                 t.deleteConnection(c)
1685         }
1686         if len(t.conns) >= t.maxEstablishedConns {
1687                 panic(len(t.conns))
1688         }
1689         t.conns[c] = struct{}{}
1690         if !t.cl.config.DisablePEX && !c.PeerExtensionBytes.SupportsExtended() {
1691                 t.pex.Add(c) // as no further extended handshake expected
1692         }
1693         return nil
1694 }
1695
1696 func (t *Torrent) wantConns() bool {
1697         if !t.networkingEnabled {
1698                 return false
1699         }
1700         if t.closed.IsSet() {
1701                 return false
1702         }
1703         if !t.seeding() && !t.needData() {
1704                 return false
1705         }
1706         if len(t.conns) < t.maxEstablishedConns {
1707                 return true
1708         }
1709         return t.worstBadConn() != nil
1710 }
1711
1712 func (t *Torrent) SetMaxEstablishedConns(max int) (oldMax int) {
1713         t.cl.lock()
1714         defer t.cl.unlock()
1715         oldMax = t.maxEstablishedConns
1716         t.maxEstablishedConns = max
1717         wcs := slices.HeapInterface(slices.FromMapKeys(t.conns), func(l, r *PeerConn) bool {
1718                 return worseConn(&l.peer, &r.peer)
1719         })
1720         for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 {
1721                 t.dropConnection(wcs.Pop().(*PeerConn))
1722         }
1723         t.openNewConns()
1724         return oldMax
1725 }
1726
1727 func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
1728         t.logger.Log(log.Fstr("hashed piece %d (passed=%t)", piece, passed).SetLevel(log.Debug))
1729         p := t.piece(piece)
1730         p.numVerifies++
1731         t.cl.event.Broadcast()
1732         if t.closed.IsSet() {
1733                 return
1734         }
1735
1736         // Don't score the first time a piece is hashed, it could be an initial check.
1737         if p.storageCompletionOk {
1738                 if passed {
1739                         pieceHashedCorrect.Add(1)
1740                 } else {
1741                         log.Fmsg("piece %d failed hash: %d connections contributed", piece, len(p.dirtiers)).AddValues(t, p).Log(t.logger)
1742                         pieceHashedNotCorrect.Add(1)
1743                 }
1744         }
1745
1746         p.marking = true
1747         t.publishPieceChange(piece)
1748         defer func() {
1749                 p.marking = false
1750                 t.publishPieceChange(piece)
1751         }()
1752
1753         if passed {
1754                 if len(p.dirtiers) != 0 {
1755                         // Don't increment stats above connection-level for every involved connection.
1756                         t.allStats((*ConnStats).incrementPiecesDirtiedGood)
1757                 }
1758                 for c := range p.dirtiers {
1759                         c._stats.incrementPiecesDirtiedGood()
1760                 }
1761                 t.clearPieceTouchers(piece)
1762                 t.cl.unlock()
1763                 err := p.Storage().MarkComplete()
1764                 if err != nil {
1765                         t.logger.Printf("%T: error marking piece complete %d: %s", t.storage, piece, err)
1766                 }
1767                 t.cl.lock()
1768
1769                 if t.closed.IsSet() {
1770                         return
1771                 }
1772                 t.pendAllChunkSpecs(piece)
1773         } else {
1774                 if len(p.dirtiers) != 0 && p.allChunksDirty() && hashIoErr == nil {
1775                         // Peers contributed to all the data for this piece hash failure, and the failure was
1776                         // not due to errors in the storage (such as data being dropped in a cache).
1777
1778                         // Increment Torrent and above stats, and then specific connections.
1779                         t.allStats((*ConnStats).incrementPiecesDirtiedBad)
1780                         for c := range p.dirtiers {
1781                                 // Y u do dis peer?!
1782                                 c.stats().incrementPiecesDirtiedBad()
1783                         }
1784
1785                         bannableTouchers := make([]*peer, 0, len(p.dirtiers))
1786                         for c := range p.dirtiers {
1787                                 if !c.trusted {
1788                                         bannableTouchers = append(bannableTouchers, c)
1789                                 }
1790                         }
1791                         t.clearPieceTouchers(piece)
1792                         slices.Sort(bannableTouchers, connLessTrusted)
1793
1794                         if t.cl.config.Debug {
1795                                 t.logger.Printf(
1796                                         "bannable conns by trust for piece %d: %v",
1797                                         piece,
1798                                         func() (ret []connectionTrust) {
1799                                                 for _, c := range bannableTouchers {
1800                                                         ret = append(ret, c.trust())
1801                                                 }
1802                                                 return
1803                                         }(),
1804                                 )
1805                         }
1806
1807                         if len(bannableTouchers) >= 1 {
1808                                 c := bannableTouchers[0]
1809                                 t.cl.banPeerIP(c.remoteIp())
1810                                 c.drop()
1811                         }
1812                 }
1813                 t.onIncompletePiece(piece)
1814                 p.Storage().MarkNotComplete()
1815         }
1816         t.updatePieceCompletion(piece)
1817 }
1818
1819 func (t *Torrent) cancelRequestsForPiece(piece pieceIndex) {
1820         // TODO: Make faster
1821         for cn := range t.conns {
1822                 cn.tickleWriter()
1823         }
1824 }
1825
1826 func (t *Torrent) onPieceCompleted(piece pieceIndex) {
1827         t.pendAllChunkSpecs(piece)
1828         t.cancelRequestsForPiece(piece)
1829         for conn := range t.conns {
1830                 conn.have(piece)
1831                 t.maybeDropMutuallyCompletePeer(&conn.peer)
1832         }
1833 }
1834
1835 // Called when a piece is found to be not complete.
1836 func (t *Torrent) onIncompletePiece(piece pieceIndex) {
1837         if t.pieceAllDirty(piece) {
1838                 t.pendAllChunkSpecs(piece)
1839         }
1840         if !t.wantPieceIndex(piece) {
1841                 // t.logger.Printf("piece %d incomplete and unwanted", piece)
1842                 return
1843         }
1844         // We could drop any connections that we told we have a piece that we
1845         // don't here. But there's a test failure, and it seems clients don't care
1846         // if you request pieces that you already claim to have. Pruning bad
1847         // connections might just remove any connections that aren't treating us
1848         // favourably anyway.
1849
1850         // for c := range t.conns {
1851         //      if c.sentHave(piece) {
1852         //              c.drop()
1853         //      }
1854         // }
1855         t.iterPeers(func(conn *peer) {
1856                 if conn.peerHasPiece(piece) {
1857                         conn.updateRequests()
1858                 }
1859         })
1860 }
1861
1862 func (t *Torrent) tryCreateMorePieceHashers() {
1863         for !t.closed.IsSet() && t.activePieceHashes < 2 && t.tryCreatePieceHasher() {
1864         }
1865 }
1866
1867 func (t *Torrent) tryCreatePieceHasher() bool {
1868         if t.storage == nil {
1869                 return false
1870         }
1871         pi, ok := t.getPieceToHash()
1872         if !ok {
1873                 return false
1874         }
1875         p := t.piece(pi)
1876         t.piecesQueuedForHash.Remove(pi)
1877         p.hashing = true
1878         t.publishPieceChange(pi)
1879         t.updatePiecePriority(pi)
1880         t.storageLock.RLock()
1881         t.activePieceHashes++
1882         go t.pieceHasher(pi)
1883         return true
1884 }
1885
1886 func (t *Torrent) getPieceToHash() (ret pieceIndex, ok bool) {
1887         t.piecesQueuedForHash.IterTyped(func(i pieceIndex) bool {
1888                 if t.piece(i).hashing {
1889                         return true
1890                 }
1891                 ret = i
1892                 ok = true
1893                 return false
1894         })
1895         return
1896 }
1897
1898 func (t *Torrent) pieceHasher(index pieceIndex) {
1899         p := t.piece(index)
1900         sum, copyErr := t.hashPiece(index)
1901         correct := sum == *p.hash
1902         switch copyErr {
1903         case nil, io.EOF:
1904         default:
1905                 log.Fmsg("piece %v (%s) hash failure copy error: %v", p, p.hash.HexString(), copyErr).Log(t.logger)
1906         }
1907         t.storageLock.RUnlock()
1908         t.cl.lock()
1909         defer t.cl.unlock()
1910         p.hashing = false
1911         t.updatePiecePriority(index)
1912         t.pieceHashed(index, correct, copyErr)
1913         t.publishPieceChange(index)
1914         t.activePieceHashes--
1915         t.tryCreateMorePieceHashers()
1916 }
1917
1918 // Return the connections that touched a piece, and clear the entries while doing it.
1919 func (t *Torrent) clearPieceTouchers(pi pieceIndex) {
1920         p := t.piece(pi)
1921         for c := range p.dirtiers {
1922                 delete(c.peerTouchedPieces, pi)
1923                 delete(p.dirtiers, c)
1924         }
1925 }
1926
1927 func (t *Torrent) peersAsSlice() (ret []*peer) {
1928         t.iterPeers(func(p *peer) {
1929                 ret = append(ret, p)
1930         })
1931         return
1932 }
1933
1934 func (t *Torrent) queuePieceCheck(pieceIndex pieceIndex) {
1935         piece := t.piece(pieceIndex)
1936         if piece.queuedForHash() {
1937                 return
1938         }
1939         t.piecesQueuedForHash.Add(bitmap.BitIndex(pieceIndex))
1940         t.publishPieceChange(pieceIndex)
1941         t.updatePiecePriority(pieceIndex)
1942         t.tryCreateMorePieceHashers()
1943 }
1944
1945 // Forces all the pieces to be re-hashed. See also Piece.VerifyData. This should not be called
1946 // before the Info is available.
1947 func (t *Torrent) VerifyData() {
1948         for i := pieceIndex(0); i < t.NumPieces(); i++ {
1949                 t.Piece(i).VerifyData()
1950         }
1951 }
1952
1953 // Start the process of connecting to the given peer for the given torrent if appropriate.
1954 func (t *Torrent) initiateConn(peer PeerInfo) {
1955         if peer.Id == t.cl.peerID {
1956                 return
1957         }
1958         if t.cl.badPeerAddr(peer.Addr) && !peer.Trusted {
1959                 return
1960         }
1961         addr := peer.Addr
1962         if t.addrActive(addr.String()) {
1963                 return
1964         }
1965         t.cl.numHalfOpen++
1966         t.halfOpen[addr.String()] = peer
1967         go t.cl.outgoingConnection(t, addr, peer.Source, peer.Trusted)
1968 }
1969
1970 // Adds a trusted, pending peer for each of the given Client's addresses. Typically used in tests to
1971 // quickly make one Client visible to the Torrent of another Client.
1972 func (t *Torrent) AddClientPeer(cl *Client) int {
1973         return t.AddPeers(func() (ps []PeerInfo) {
1974                 for _, la := range cl.ListenAddrs() {
1975                         ps = append(ps, PeerInfo{
1976                                 Addr:    la,
1977                                 Trusted: true,
1978                         })
1979                 }
1980                 return
1981         }())
1982 }
1983
1984 // All stats that include this Torrent. Useful when we want to increment ConnStats but not for every
1985 // connection.
1986 func (t *Torrent) allStats(f func(*ConnStats)) {
1987         f(&t.stats)
1988         f(&t.cl.stats)
1989 }
1990
1991 func (t *Torrent) hashingPiece(i pieceIndex) bool {
1992         return t.pieces[i].hashing
1993 }
1994
1995 func (t *Torrent) pieceQueuedForHash(i pieceIndex) bool {
1996         return t.piecesQueuedForHash.Get(bitmap.BitIndex(i))
1997 }
1998
1999 func (t *Torrent) dialTimeout() time.Duration {
2000         return reducedDialTimeout(t.cl.config.MinDialTimeout, t.cl.config.NominalDialTimeout, t.cl.config.HalfOpenConnsPerTorrent, t.peers.Len())
2001 }
2002
2003 func (t *Torrent) piece(i int) *Piece {
2004         return &t.pieces[i]
2005 }
2006
2007 func (t *Torrent) requestStrategyTorrent() requestStrategyTorrent {
2008         return t
2009 }
2010
2011 type torrentRequestStrategyCallbacks struct {
2012         t *Torrent
2013 }
2014
2015 func (cb torrentRequestStrategyCallbacks) requestTimedOut(r request) {
2016         torrent.Add("request timeouts", 1)
2017         cb.t.cl.lock()
2018         defer cb.t.cl.unlock()
2019         cb.t.iterPeers(func(cn *peer) {
2020                 if cn.peerHasPiece(pieceIndex(r.Index)) {
2021                         cn.updateRequests()
2022                 }
2023         })
2024
2025 }
2026
2027 func (t *Torrent) requestStrategyCallbacks() requestStrategyCallbacks {
2028         return torrentRequestStrategyCallbacks{t}
2029 }
2030
2031 func (t *Torrent) onWriteChunkErr(err error) {
2032         if t.userOnWriteChunkErr != nil {
2033                 go t.userOnWriteChunkErr(err)
2034                 return
2035         }
2036         t.logger.WithDefaultLevel(log.Critical).Printf("default chunk write error handler: disabling data download")
2037         t.disallowDataDownloadLocked()
2038 }
2039
2040 func (t *Torrent) DisallowDataDownload() {
2041         t.cl.lock()
2042         defer t.cl.unlock()
2043         t.disallowDataDownloadLocked()
2044 }
2045
2046 func (t *Torrent) disallowDataDownloadLocked() {
2047         t.dataDownloadDisallowed = true
2048         t.iterPeers(func(c *peer) {
2049                 c.updateRequests()
2050         })
2051         t.tickleReaders()
2052 }
2053
2054 func (t *Torrent) AllowDataDownload() {
2055         t.cl.lock()
2056         defer t.cl.unlock()
2057         t.dataDownloadDisallowed = false
2058         t.tickleReaders()
2059         t.iterPeers(func(c *peer) {
2060                 c.updateRequests()
2061         })
2062 }
2063
2064 // Enables uploading data, if it was disabled.
2065 func (t *Torrent) AllowDataUpload() {
2066         t.cl.lock()
2067         defer t.cl.unlock()
2068         t.dataUploadDisallowed = false
2069         for c := range t.conns {
2070                 c.updateRequests()
2071         }
2072 }
2073
2074 // Disables uploading data, if it was enabled.
2075 func (t *Torrent) DisallowDataUpload() {
2076         t.cl.lock()
2077         defer t.cl.unlock()
2078         t.dataUploadDisallowed = true
2079         for c := range t.conns {
2080                 c.updateRequests()
2081         }
2082 }
2083
2084 // Sets a handler that is called if there's an error writing a chunk to local storage. By default,
2085 // or if nil, a critical message is logged, and data download is disabled.
2086 func (t *Torrent) SetOnWriteChunkError(f func(error)) {
2087         t.cl.lock()
2088         defer t.cl.unlock()
2089         t.userOnWriteChunkErr = f
2090 }
2091
2092 func (t *Torrent) iterPeers(f func(*peer)) {
2093         for pc := range t.conns {
2094                 f(&pc.peer)
2095         }
2096         for _, ws := range t.webSeeds {
2097                 f(ws)
2098         }
2099 }
2100
2101 func (t *Torrent) addWebSeed(url string) {
2102         if t.cl.config.DisableWebseeds {
2103                 return
2104         }
2105         if _, ok := t.webSeeds[url]; ok {
2106                 return
2107         }
2108         const maxRequests = 10
2109         ws := webseedPeer{
2110                 peer: peer{
2111                         t:                        t,
2112                         outgoing:                 true,
2113                         network:                  "http",
2114                         reconciledHandshakeStats: true,
2115                         peerSentHaveAll:          true,
2116                         PeerMaxRequests:          maxRequests,
2117                 },
2118                 client: webseed.Client{
2119                         HttpClient: http.DefaultClient,
2120                         Url:        url,
2121                 },
2122                 requests: make(map[request]webseed.Request, maxRequests),
2123         }
2124         ws.peer.logger = t.logger.WithContextValue(&ws)
2125         ws.peer.peerImpl = &ws
2126         if t.haveInfo() {
2127                 ws.onGotInfo(t.info)
2128         }
2129         t.webSeeds[url] = &ws.peer
2130 }
2131
2132 func (t *Torrent) peerIsActive(p *peer) (active bool) {
2133         t.iterPeers(func(p1 *peer) {
2134                 if p1 == p {
2135                         active = true
2136                 }
2137         })
2138         return
2139 }