torrent.go

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