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         "os"
  12         "sync"
  13         "text/tabwriter"
  14         "time"
  15         "unsafe"
  16
  17         "github.com/davecgh/go-spew/spew"
  18
  19         "github.com/anacrolix/dht/v2"
  20         "github.com/anacrolix/log"
  21         "github.com/anacrolix/missinggo"
  22         "github.com/anacrolix/missinggo/bitmap"
  23         "github.com/anacrolix/missinggo/perf"
  24         "github.com/anacrolix/missinggo/prioritybitmap"
  25         "github.com/anacrolix/missinggo/pubsub"
  26         "github.com/anacrolix/missinggo/slices"
  27
  28         "github.com/anacrolix/torrent/bencode"
  29         "github.com/anacrolix/torrent/metainfo"
  30         pp "github.com/anacrolix/torrent/peer_protocol"
  31         "github.com/anacrolix/torrent/storage"
  32         "github.com/anacrolix/torrent/tracker"
  33 )
  34
  35 func (t *Torrent) chunkIndexSpec(chunkIndex pp.Integer, piece pieceIndex) chunkSpec {
  36         return chunkIndexSpec(chunkIndex, t.pieceLength(piece), t.chunkSize)
  37 }
  38
  39 // Maintains state of torrent within a Client.
  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) {
 747         hash := pieceHash.New()
 748         p := t.piece(piece)
 749         p.waitNoPendingWrites()
 750         ip := t.info.Piece(int(piece))
 751         pl := ip.Length()
 752         n, err := io.Copy(hash, io.NewSectionReader(t.pieces[piece].Storage(), 0, pl))
 753         if n == pl {
 754                 missinggo.CopyExact(&ret, hash.Sum(nil))
 755                 return
 756         }
 757         if err != io.ErrUnexpectedEOF && !os.IsNotExist(err) {
 758                 t.logger.Printf("unexpected error hashing piece %d through %T: %s", piece, t.storage.TorrentImpl, err)
 759         }
 760         return
 761 }
 762
 763 func (t *Torrent) haveAnyPieces() bool {
 764         return t.completedPieces.Len() != 0
 765 }
 766
 767 func (t *Torrent) haveAllPieces() bool {
 768         if !t.haveInfo() {
 769                 return false
 770         }
 771         return t.completedPieces.Len() == bitmap.BitIndex(t.numPieces())
 772 }
 773
 774 func (t *Torrent) havePiece(index pieceIndex) bool {
 775         return t.haveInfo() && t.pieceComplete(index)
 776 }
 777
 778 func (t *Torrent) haveChunk(r request) (ret bool) {
 779         // defer func() {
 780         //      log.Println("have chunk", r, ret)
 781         // }()
 782         if !t.haveInfo() {
 783                 return false
 784         }
 785         if t.pieceComplete(pieceIndex(r.Index)) {
 786                 return true
 787         }
 788         p := &t.pieces[r.Index]
 789         return !p.pendingChunk(r.chunkSpec, t.chunkSize)
 790 }
 791
 792 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
 793         return int(cs.Begin / chunkSize)
 794 }
 795
 796 func (t *Torrent) wantPieceIndex(index pieceIndex) bool {
 797         if !t.haveInfo() {
 798                 return false
 799         }
 800         if index < 0 || index >= t.numPieces() {
 801                 return false
 802         }
 803         p := &t.pieces[index]
 804         if p.queuedForHash() {
 805                 return false
 806         }
 807         if p.hashing {
 808                 return false
 809         }
 810         if t.pieceComplete(index) {
 811                 return false
 812         }
 813         if t.pendingPieces.Contains(bitmap.BitIndex(index)) {
 814                 return true
 815         }
 816         // t.logger.Printf("piece %d not pending", index)
 817         return !t.forReaderOffsetPieces(func(begin, end pieceIndex) bool {
 818                 return index < begin || index >= end
 819         })
 820 }
 821
 822 // The worst connection is one that hasn't been sent, or sent anything useful
 823 // for the longest. A bad connection is one that usually sends us unwanted
 824 // pieces, or has been in worser half of the established connections for more
 825 // than a minute.
 826 func (t *Torrent) worstBadConn() *connection {
 827         wcs := worseConnSlice{t.unclosedConnsAsSlice()}
 828         heap.Init(&wcs)
 829         for wcs.Len() != 0 {
 830                 c := heap.Pop(&wcs).(*connection)
 831                 if c.stats.ChunksReadWasted.Int64() >= 6 && c.stats.ChunksReadWasted.Int64() > c.stats.ChunksReadUseful.Int64() {
 832                         return c
 833                 }
 834                 // If the connection is in the worst half of the established
 835                 // connection quota and is older than a minute.
 836                 if wcs.Len() >= (t.maxEstablishedConns+1)/2 {
 837                         // Give connections 1 minute to prove themselves.
 838                         if time.Since(c.completedHandshake) > time.Minute {
 839                                 return c
 840                         }
 841                 }
 842         }
 843         return nil
 844 }
 845
 846 type PieceStateChange struct {
 847         Index int
 848         PieceState
 849 }
 850
 851 func (t *Torrent) publishPieceChange(piece pieceIndex) {
 852         t.cl._mu.Defer(func() {
 853                 cur := t.pieceState(piece)
 854                 p := &t.pieces[piece]
 855                 if cur != p.publicPieceState {
 856                         p.publicPieceState = cur
 857                         t.pieceStateChanges.Publish(PieceStateChange{
 858                                 int(piece),
 859                                 cur,
 860                         })
 861                 }
 862         })
 863 }
 864
 865 func (t *Torrent) pieceNumPendingChunks(piece pieceIndex) pp.Integer {
 866         if t.pieceComplete(piece) {
 867                 return 0
 868         }
 869         return t.pieceNumChunks(piece) - t.pieces[piece].numDirtyChunks()
 870 }
 871
 872 func (t *Torrent) pieceAllDirty(piece pieceIndex) bool {
 873         return t.pieces[piece].dirtyChunks.Len() == int(t.pieceNumChunks(piece))
 874 }
 875
 876 func (t *Torrent) readersChanged() {
 877         t.updateReaderPieces()
 878         t.updateAllPiecePriorities()
 879 }
 880
 881 func (t *Torrent) updateReaderPieces() {
 882         t.readerNowPieces, t.readerReadaheadPieces = t.readerPiecePriorities()
 883 }
 884
 885 func (t *Torrent) readerPosChanged(from, to pieceRange) {
 886         if from == to {
 887                 return
 888         }
 889         t.updateReaderPieces()
 890         // Order the ranges, high and low.
 891         l, h := from, to
 892         if l.begin > h.begin {
 893                 l, h = h, l
 894         }
 895         if l.end < h.begin {
 896                 // Two distinct ranges.
 897                 t.updatePiecePriorities(l.begin, l.end)
 898                 t.updatePiecePriorities(h.begin, h.end)
 899         } else {
 900                 // Ranges overlap.
 901                 end := l.end
 902                 if h.end > end {
 903                         end = h.end
 904                 }
 905                 t.updatePiecePriorities(l.begin, end)
 906         }
 907 }
 908
 909 func (t *Torrent) maybeNewConns() {
 910         // Tickle the accept routine.
 911         t.cl.event.Broadcast()
 912         t.openNewConns()
 913 }
 914
 915 func (t *Torrent) piecePriorityChanged(piece pieceIndex) {
 916         // t.logger.Printf("piece %d priority changed", piece)
 917         for c := range t.conns {
 918                 if c.updatePiecePriority(piece) {
 919                         // log.Print("conn piece priority changed")
 920                         c.updateRequests()
 921                 }
 922         }
 923         t.maybeNewConns()
 924         t.publishPieceChange(piece)
 925 }
 926
 927 func (t *Torrent) updatePiecePriority(piece pieceIndex) {
 928         p := &t.pieces[piece]
 929         newPrio := p.uncachedPriority()
 930         // t.logger.Printf("torrent %p: piece %d: uncached priority: %v", t, piece, newPrio)
 931         if newPrio == PiecePriorityNone {
 932                 if !t.pendingPieces.Remove(bitmap.BitIndex(piece)) {
 933                         return
 934                 }
 935         } else {
 936                 if !t.pendingPieces.Set(bitmap.BitIndex(piece), newPrio.BitmapPriority()) {
 937                         return
 938                 }
 939         }
 940         t.piecePriorityChanged(piece)
 941 }
 942
 943 func (t *Torrent) updateAllPiecePriorities() {
 944         t.updatePiecePriorities(0, t.numPieces())
 945 }
 946
 947 // Update all piece priorities in one hit. This function should have the same
 948 // output as updatePiecePriority, but across all pieces.
 949 func (t *Torrent) updatePiecePriorities(begin, end pieceIndex) {
 950         for i := begin; i < end; i++ {
 951                 t.updatePiecePriority(i)
 952         }
 953 }
 954
 955 // Returns the range of pieces [begin, end) that contains the extent of bytes.
 956 func (t *Torrent) byteRegionPieces(off, size int64) (begin, end pieceIndex) {
 957         if off >= *t.length {
 958                 return
 959         }
 960         if off < 0 {
 961                 size += off
 962                 off = 0
 963         }
 964         if size <= 0 {
 965                 return
 966         }
 967         begin = pieceIndex(off / t.info.PieceLength)
 968         end = pieceIndex((off + size + t.info.PieceLength - 1) / t.info.PieceLength)
 969         if end > pieceIndex(t.info.NumPieces()) {
 970                 end = pieceIndex(t.info.NumPieces())
 971         }
 972         return
 973 }
 974
 975 // Returns true if all iterations complete without breaking. Returns the read
 976 // regions for all readers. The reader regions should not be merged as some
 977 // callers depend on this method to enumerate readers.
 978 func (t *Torrent) forReaderOffsetPieces(f func(begin, end pieceIndex) (more bool)) (all bool) {
 979         for r := range t.readers {
 980                 p := r.pieces
 981                 if p.begin >= p.end {
 982                         continue
 983                 }
 984                 if !f(p.begin, p.end) {
 985                         return false
 986                 }
 987         }
 988         return true
 989 }
 990
 991 func (t *Torrent) piecePriority(piece pieceIndex) piecePriority {
 992         prio, ok := t.pendingPieces.GetPriority(bitmap.BitIndex(piece))
 993         if !ok {
 994                 return PiecePriorityNone
 995         }
 996         if prio > 0 {
 997                 panic(prio)
 998         }
 999         ret := piecePriority(-prio)
1000         if ret == PiecePriorityNone {
1001                 panic(piece)
1002         }
1003         return ret
1004 }
1005
1006 func (t *Torrent) pendRequest(req request) {
1007         ci := chunkIndex(req.chunkSpec, t.chunkSize)
1008         t.pieces[req.Index].pendChunkIndex(ci)
1009 }
1010
1011 func (t *Torrent) pieceCompletionChanged(piece pieceIndex) {
1012         t.tickleReaders()
1013         t.cl.event.Broadcast()
1014         if t.pieceComplete(piece) {
1015                 t.onPieceCompleted(piece)
1016         } else {
1017                 t.onIncompletePiece(piece)
1018         }
1019         t.updatePiecePriority(piece)
1020 }
1021
1022 func (t *Torrent) numReceivedConns() (ret int) {
1023         for c := range t.conns {
1024                 if c.Discovery == peerSourceIncoming {
1025                         ret++
1026                 }
1027         }
1028         return
1029 }
1030
1031 func (t *Torrent) maxHalfOpen() int {
1032         // Note that if we somehow exceed the maximum established conns, we want
1033         // the negative value to have an effect.
1034         establishedHeadroom := int64(t.maxEstablishedConns - len(t.conns))
1035         extraIncoming := int64(t.numReceivedConns() - t.maxEstablishedConns/2)
1036         // We want to allow some experimentation with new peers, and to try to
1037         // upset an oversupply of received connections.
1038         return int(min(max(5, extraIncoming)+establishedHeadroom, int64(t.cl.config.HalfOpenConnsPerTorrent)))
1039 }
1040
1041 func (t *Torrent) openNewConns() {
1042         defer t.updateWantPeersEvent()
1043         for t.peers.Len() != 0 {
1044                 if !t.wantConns() {
1045                         return
1046                 }
1047                 if len(t.halfOpen) >= t.maxHalfOpen() {
1048                         return
1049                 }
1050                 p := t.peers.PopMax()
1051                 t.initiateConn(p)
1052         }
1053 }
1054
1055 func (t *Torrent) getConnPieceInclination() []int {
1056         _ret := t.connPieceInclinationPool.Get()
1057         if _ret == nil {
1058                 pieceInclinationsNew.Add(1)
1059                 return rand.Perm(int(t.numPieces()))
1060         }
1061         pieceInclinationsReused.Add(1)
1062         return *_ret.(*[]int)
1063 }
1064
1065 func (t *Torrent) putPieceInclination(pi []int) {
1066         t.connPieceInclinationPool.Put(&pi)
1067         pieceInclinationsPut.Add(1)
1068 }
1069
1070 func (t *Torrent) updatePieceCompletion(piece pieceIndex) bool {
1071         p := t.piece(piece)
1072         uncached := t.pieceCompleteUncached(piece)
1073         cached := p.completion()
1074         changed := cached != uncached
1075         p.storageCompletionOk = uncached.Ok
1076         t.completedPieces.Set(bitmap.BitIndex(piece), uncached.Complete)
1077         if changed {
1078                 log.Fstr("piece %d completion changed: %+v -> %+v", piece, cached, uncached).WithValues(debugLogValue).Log(t.logger)
1079                 t.pieceCompletionChanged(piece)
1080         }
1081         return changed
1082 }
1083
1084 // Non-blocking read. Client lock is not required.
1085 func (t *Torrent) readAt(b []byte, off int64) (n int, err error) {
1086         p := &t.pieces[off/t.info.PieceLength]
1087         p.waitNoPendingWrites()
1088         return p.Storage().ReadAt(b, off-p.Info().Offset())
1089 }
1090
1091 func (t *Torrent) updateAllPieceCompletions() {
1092         for i := pieceIndex(0); i < t.numPieces(); i++ {
1093                 t.updatePieceCompletion(i)
1094         }
1095 }
1096
1097 // Returns an error if the metadata was completed, but couldn't be set for
1098 // some reason. Blame it on the last peer to contribute.
1099 func (t *Torrent) maybeCompleteMetadata() error {
1100         if t.haveInfo() {
1101                 // Nothing to do.
1102                 return nil
1103         }
1104         if !t.haveAllMetadataPieces() {
1105                 // Don't have enough metadata pieces.
1106                 return nil
1107         }
1108         err := t.setInfoBytes(t.metadataBytes)
1109         if err != nil {
1110                 t.invalidateMetadata()
1111                 return fmt.Errorf("error setting info bytes: %s", err)
1112         }
1113         if t.cl.config.Debug {
1114                 t.logger.Printf("%s: got metadata from peers", t)
1115         }
1116         return nil
1117 }
1118
1119 func (t *Torrent) readerPiecePriorities() (now, readahead bitmap.Bitmap) {
1120         t.forReaderOffsetPieces(func(begin, end pieceIndex) bool {
1121                 if end > begin {
1122                         now.Add(bitmap.BitIndex(begin))
1123                         readahead.AddRange(bitmap.BitIndex(begin)+1, bitmap.BitIndex(end))
1124                 }
1125                 return true
1126         })
1127         return
1128 }
1129
1130 func (t *Torrent) needData() bool {
1131         if t.closed.IsSet() {
1132                 return false
1133         }
1134         if !t.haveInfo() {
1135                 return true
1136         }
1137         return t.pendingPieces.Len() != 0
1138 }
1139
1140 func appendMissingStrings(old, new []string) (ret []string) {
1141         ret = old
1142 new:
1143         for _, n := range new {
1144                 for _, o := range old {
1145                         if o == n {
1146                                 continue new
1147                         }
1148                 }
1149                 ret = append(ret, n)
1150         }
1151         return
1152 }
1153
1154 func appendMissingTrackerTiers(existing [][]string, minNumTiers int) (ret [][]string) {
1155         ret = existing
1156         for minNumTiers > len(ret) {
1157                 ret = append(ret, nil)
1158         }
1159         return
1160 }
1161
1162 func (t *Torrent) addTrackers(announceList [][]string) {
1163         fullAnnounceList := &t.metainfo.AnnounceList
1164         t.metainfo.AnnounceList = appendMissingTrackerTiers(*fullAnnounceList, len(announceList))
1165         for tierIndex, trackerURLs := range announceList {
1166                 (*fullAnnounceList)[tierIndex] = appendMissingStrings((*fullAnnounceList)[tierIndex], trackerURLs)
1167         }
1168         t.startMissingTrackerScrapers()
1169         t.updateWantPeersEvent()
1170 }
1171
1172 // Don't call this before the info is available.
1173 func (t *Torrent) bytesCompleted() int64 {
1174         if !t.haveInfo() {
1175                 return 0
1176         }
1177         return t.info.TotalLength() - t.bytesLeft()
1178 }
1179
1180 func (t *Torrent) SetInfoBytes(b []byte) (err error) {
1181         t.cl.lock()
1182         defer t.cl.unlock()
1183         return t.setInfoBytes(b)
1184 }
1185
1186 // Returns true if connection is removed from torrent.Conns.
1187 func (t *Torrent) deleteConnection(c *connection) (ret bool) {
1188         if !c.closed.IsSet() {
1189                 panic("connection is not closed")
1190                 // There are behaviours prevented by the closed state that will fail
1191                 // if the connection has been deleted.
1192         }
1193         _, ret = t.conns[c]
1194         delete(t.conns, c)
1195         torrent.Add("deleted connections", 1)
1196         c.deleteAllRequests()
1197         if len(t.conns) == 0 {
1198                 t.assertNoPendingRequests()
1199         }
1200         return
1201 }
1202
1203 func (t *Torrent) assertNoPendingRequests() {
1204         if len(t.pendingRequests) != 0 {
1205                 panic(t.pendingRequests)
1206         }
1207         if len(t.lastRequested) != 0 {
1208                 panic(t.lastRequested)
1209         }
1210 }
1211
1212 func (t *Torrent) dropConnection(c *connection) {
1213         t.cl.event.Broadcast()
1214         c.Close()
1215         if t.deleteConnection(c) {
1216                 t.openNewConns()
1217         }
1218 }
1219
1220 func (t *Torrent) wantPeers() bool {
1221         if t.closed.IsSet() {
1222                 return false
1223         }
1224         if t.peers.Len() > t.cl.config.TorrentPeersLowWater {
1225                 return false
1226         }
1227         return t.needData() || t.seeding()
1228 }
1229
1230 func (t *Torrent) updateWantPeersEvent() {
1231         if t.wantPeers() {
1232                 t.wantPeersEvent.Set()
1233         } else {
1234                 t.wantPeersEvent.Clear()
1235         }
1236 }
1237
1238 // Returns whether the client should make effort to seed the torrent.
1239 func (t *Torrent) seeding() bool {
1240         cl := t.cl
1241         if t.closed.IsSet() {
1242                 return false
1243         }
1244         if cl.config.NoUpload {
1245                 return false
1246         }
1247         if !cl.config.Seed {
1248                 return false
1249         }
1250         if cl.config.DisableAggressiveUpload && t.needData() {
1251                 return false
1252         }
1253         return true
1254 }
1255
1256 func (t *Torrent) startScrapingTracker(_url string) {
1257         if _url == "" {
1258                 return
1259         }
1260         u, err := url.Parse(_url)
1261         if err != nil {
1262                 // URLs with a leading '*' appear to be a uTorrent convention to
1263                 // disable trackers.
1264                 if _url[0] != '*' {
1265                         log.Str("error parsing tracker url").AddValues("url", _url).Log(t.logger)
1266                 }
1267                 return
1268         }
1269         if u.Scheme == "udp" {
1270                 u.Scheme = "udp4"
1271                 t.startScrapingTracker(u.String())
1272                 u.Scheme = "udp6"
1273                 t.startScrapingTracker(u.String())
1274                 return
1275         }
1276         if u.Scheme == "udp4" && (t.cl.config.DisableIPv4Peers || t.cl.config.DisableIPv4) {
1277                 return
1278         }
1279         if u.Scheme == "udp6" && t.cl.config.DisableIPv6 {
1280                 return
1281         }
1282         if _, ok := t.trackerAnnouncers[_url]; ok {
1283                 return
1284         }
1285         newAnnouncer := &trackerScraper{
1286                 u: *u,
1287                 t: t,
1288         }
1289         if t.trackerAnnouncers == nil {
1290                 t.trackerAnnouncers = make(map[string]*trackerScraper)
1291         }
1292         t.trackerAnnouncers[_url] = newAnnouncer
1293         go newAnnouncer.Run()
1294 }
1295
1296 // Adds and starts tracker scrapers for tracker URLs that aren't already
1297 // running.
1298 func (t *Torrent) startMissingTrackerScrapers() {
1299         if t.cl.config.DisableTrackers {
1300                 return
1301         }
1302         t.startScrapingTracker(t.metainfo.Announce)
1303         for _, tier := range t.metainfo.AnnounceList {
1304                 for _, url := range tier {
1305                         t.startScrapingTracker(url)
1306                 }
1307         }
1308 }
1309
1310 // Returns an AnnounceRequest with fields filled out to defaults and current
1311 // values.
1312 func (t *Torrent) announceRequest(event tracker.AnnounceEvent) tracker.AnnounceRequest {
1313         // Note that IPAddress is not set. It's set for UDP inside the tracker
1314         // code, since it's dependent on the network in use.
1315         return tracker.AnnounceRequest{
1316                 Event:    event,
1317                 NumWant:  -1,
1318                 Port:     uint16(t.cl.incomingPeerPort()),
1319                 PeerId:   t.cl.peerID,
1320                 InfoHash: t.infoHash,
1321                 Key:      t.cl.announceKey(),
1322
1323                 // The following are vaguely described in BEP 3.
1324
1325                 Left:     t.bytesLeftAnnounce(),
1326                 Uploaded: t.stats.BytesWrittenData.Int64(),
1327                 // There's no mention of wasted or unwanted download in the BEP.
1328                 Downloaded: t.stats.BytesReadUsefulData.Int64(),
1329         }
1330 }
1331
1332 // Adds peers revealed in an announce until the announce ends, or we have
1333 // enough peers.
1334 func (t *Torrent) consumeDhtAnnouncePeers(pvs <-chan dht.PeersValues) {
1335         cl := t.cl
1336         for v := range pvs {
1337                 cl.lock()
1338                 for _, cp := range v.Peers {
1339                         if cp.Port == 0 {
1340                                 // Can't do anything with this.
1341                                 continue
1342                         }
1343                         t.addPeer(Peer{
1344                                 IP:     cp.IP[:],
1345                                 Port:   cp.Port,
1346                                 Source: peerSourceDhtGetPeers,
1347                         })
1348                 }
1349                 cl.unlock()
1350         }
1351 }
1352
1353 func (t *Torrent) announceToDht(impliedPort bool, s *dht.Server) error {
1354         ps, err := s.Announce(t.infoHash, t.cl.incomingPeerPort(), impliedPort)
1355         if err != nil {
1356                 return err
1357         }
1358         go t.consumeDhtAnnouncePeers(ps.Peers)
1359         select {
1360         case <-t.closed.LockedChan(t.cl.locker()):
1361         case <-time.After(5 * time.Minute):
1362         }
1363         ps.Close()
1364         return nil
1365 }
1366
1367 func (t *Torrent) dhtAnnouncer(s *dht.Server) {
1368         cl := t.cl
1369         for {
1370                 select {
1371                 case <-t.closed.LockedChan(cl.locker()):
1372                         return
1373                 case <-t.wantPeersEvent.LockedChan(cl.locker()):
1374                 }
1375                 cl.lock()
1376                 t.numDHTAnnounces++
1377                 cl.unlock()
1378                 err := t.announceToDht(true, s)
1379                 if err != nil {
1380                         t.logger.Printf("error announcing %q to DHT: %s", t, err)
1381                 }
1382         }
1383 }
1384
1385 func (t *Torrent) addPeers(peers []Peer) {
1386         for _, p := range peers {
1387                 t.addPeer(p)
1388         }
1389 }
1390
1391 func (t *Torrent) Stats() TorrentStats {
1392         t.cl.rLock()
1393         defer t.cl.rUnlock()
1394         return t.statsLocked()
1395 }
1396
1397 func (t *Torrent) statsLocked() (ret TorrentStats) {
1398         ret.ActivePeers = len(t.conns)
1399         ret.HalfOpenPeers = len(t.halfOpen)
1400         ret.PendingPeers = t.peers.Len()
1401         ret.TotalPeers = t.numTotalPeers()
1402         ret.ConnectedSeeders = 0
1403         for c := range t.conns {
1404                 if all, ok := c.peerHasAllPieces(); all && ok {
1405                         ret.ConnectedSeeders++
1406                 }
1407         }
1408         ret.ConnStats = t.stats.Copy()
1409         return
1410 }
1411
1412 // The total number of peers in the torrent.
1413 func (t *Torrent) numTotalPeers() int {
1414         peers := make(map[string]struct{})
1415         for conn := range t.conns {
1416                 ra := conn.conn.RemoteAddr()
1417                 if ra == nil {
1418                         // It's been closed and doesn't support RemoteAddr.
1419                         continue
1420                 }
1421                 peers[ra.String()] = struct{}{}
1422         }
1423         for addr := range t.halfOpen {
1424                 peers[addr] = struct{}{}
1425         }
1426         t.peers.Each(func(peer Peer) {
1427                 peers[fmt.Sprintf("%s:%d", peer.IP, peer.Port)] = struct{}{}
1428         })
1429         return len(peers)
1430 }
1431
1432 // Reconcile bytes transferred before connection was associated with a
1433 // torrent.
1434 func (t *Torrent) reconcileHandshakeStats(c *connection) {
1435         if c.stats != (ConnStats{
1436                 // Handshakes should only increment these fields:
1437                 BytesWritten: c.stats.BytesWritten,
1438                 BytesRead:    c.stats.BytesRead,
1439         }) {
1440                 panic("bad stats")
1441         }
1442         c.postHandshakeStats(func(cs *ConnStats) {
1443                 cs.BytesRead.Add(c.stats.BytesRead.Int64())
1444                 cs.BytesWritten.Add(c.stats.BytesWritten.Int64())
1445         })
1446         c.reconciledHandshakeStats = true
1447 }
1448
1449 // Returns true if the connection is added.
1450 func (t *Torrent) addConnection(c *connection) (err error) {
1451         defer func() {
1452                 if err == nil {
1453                         torrent.Add("added connections", 1)
1454                 }
1455         }()
1456         if t.closed.IsSet() {
1457                 return errors.New("torrent closed")
1458         }
1459         for c0 := range t.conns {
1460                 if c.PeerID != c0.PeerID {
1461                         continue
1462                 }
1463                 if !t.cl.config.dropDuplicatePeerIds {
1464                         continue
1465                 }
1466                 if left, ok := c.hasPreferredNetworkOver(c0); ok && left {
1467                         c0.Close()
1468                         t.deleteConnection(c0)
1469                 } else {
1470                         return errors.New("existing connection preferred")
1471                 }
1472         }
1473         if len(t.conns) >= t.maxEstablishedConns {
1474                 c := t.worstBadConn()
1475                 if c == nil {
1476                         return errors.New("don't want conns")
1477                 }
1478                 c.Close()
1479                 t.deleteConnection(c)
1480         }
1481         if len(t.conns) >= t.maxEstablishedConns {
1482                 panic(len(t.conns))
1483         }
1484         t.conns[c] = struct{}{}
1485         return nil
1486 }
1487
1488 func (t *Torrent) wantConns() bool {
1489         if !t.networkingEnabled {
1490                 return false
1491         }
1492         if t.closed.IsSet() {
1493                 return false
1494         }
1495         if !t.seeding() && !t.needData() {
1496                 return false
1497         }
1498         if len(t.conns) < t.maxEstablishedConns {
1499                 return true
1500         }
1501         return t.worstBadConn() != nil
1502 }
1503
1504 func (t *Torrent) SetMaxEstablishedConns(max int) (oldMax int) {
1505         t.cl.lock()
1506         defer t.cl.unlock()
1507         oldMax = t.maxEstablishedConns
1508         t.maxEstablishedConns = max
1509         wcs := slices.HeapInterface(slices.FromMapKeys(t.conns), worseConn)
1510         for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 {
1511                 t.dropConnection(wcs.Pop().(*connection))
1512         }
1513         t.openNewConns()
1514         return oldMax
1515 }
1516
1517 func (t *Torrent) pieceHashed(piece pieceIndex, correct bool) {
1518         t.logger.Log(log.Fstr("hashed piece %d (passed=%t)", piece, correct).WithValues(debugLogValue))
1519         p := t.piece(piece)
1520         p.numVerifies++
1521         t.cl.event.Broadcast()
1522         if t.closed.IsSet() {
1523                 return
1524         }
1525         touchers := t.reapPieceTouchers(piece)
1526         if p.storageCompletionOk {
1527                 // Don't score the first time a piece is hashed, it could be an
1528                 // initial check.
1529                 if correct {
1530                         pieceHashedCorrect.Add(1)
1531                 } else {
1532                         log.Fmsg("piece %d failed hash: %d connections contributed", piece, len(touchers)).AddValues(t, p).Log(t.logger)
1533                         pieceHashedNotCorrect.Add(1)
1534                 }
1535         }
1536         if correct {
1537                 if len(touchers) != 0 {
1538                         // Don't increment stats above connection-level for every involved
1539                         // connection.
1540                         t.allStats((*ConnStats).incrementPiecesDirtiedGood)
1541                 }
1542                 for _, c := range touchers {
1543                         c.stats.incrementPiecesDirtiedGood()
1544                 }
1545                 err := p.Storage().MarkComplete()
1546                 if err != nil {
1547                         t.logger.Printf("%T: error marking piece complete %d: %s", t.storage, piece, err)
1548                 }
1549         } else {
1550                 if len(touchers) != 0 {
1551                         // Don't increment stats above connection-level for every involved
1552                         // connection.
1553                         t.allStats((*ConnStats).incrementPiecesDirtiedBad)
1554                         for _, c := range touchers {
1555                                 // Y u do dis peer?!
1556                                 c.stats.incrementPiecesDirtiedBad()
1557                         }
1558                         slices.Sort(touchers, connLessTrusted)
1559                         if t.cl.config.Debug {
1560                                 t.logger.Printf("dropping first corresponding conn from trust: %v", func() (ret []int64) {
1561                                         for _, c := range touchers {
1562                                                 ret = append(ret, c.netGoodPiecesDirtied())
1563                                         }
1564                                         return
1565                                 }())
1566                         }
1567                         c := touchers[0]
1568                         t.cl.banPeerIP(c.remoteAddr.IP)
1569                         c.Drop()
1570                 }
1571                 t.onIncompletePiece(piece)
1572                 p.Storage().MarkNotComplete()
1573         }
1574         t.updatePieceCompletion(piece)
1575 }
1576
1577 func (t *Torrent) cancelRequestsForPiece(piece pieceIndex) {
1578         // TODO: Make faster
1579         for cn := range t.conns {
1580                 cn.tickleWriter()
1581         }
1582 }
1583
1584 func (t *Torrent) onPieceCompleted(piece pieceIndex) {
1585         t.pendAllChunkSpecs(piece)
1586         t.cancelRequestsForPiece(piece)
1587         for conn := range t.conns {
1588                 conn.Have(piece)
1589         }
1590 }
1591
1592 // Called when a piece is found to be not complete.
1593 func (t *Torrent) onIncompletePiece(piece pieceIndex) {
1594         if t.pieceAllDirty(piece) {
1595                 t.pendAllChunkSpecs(piece)
1596         }
1597         if !t.wantPieceIndex(piece) {
1598                 // t.logger.Printf("piece %d incomplete and unwanted", piece)
1599                 return
1600         }
1601         // We could drop any connections that we told we have a piece that we
1602         // don't here. But there's a test failure, and it seems clients don't care
1603         // if you request pieces that you already claim to have. Pruning bad
1604         // connections might just remove any connections that aren't treating us
1605         // favourably anyway.
1606
1607         // for c := range t.conns {
1608         //      if c.sentHave(piece) {
1609         //              c.Drop()
1610         //      }
1611         // }
1612         for conn := range t.conns {
1613                 if conn.PeerHasPiece(piece) {
1614                         conn.updateRequests()
1615                 }
1616         }
1617 }
1618
1619 func (t *Torrent) tryCreateMorePieceHashers() {
1620         for t.activePieceHashes < 2 && t.tryCreatePieceHasher() {
1621         }
1622 }
1623
1624 func (t *Torrent) tryCreatePieceHasher() bool {
1625         if t.storage == nil {
1626                 return false
1627         }
1628         pi, ok := t.getPieceToHash()
1629         if !ok {
1630                 return false
1631         }
1632         p := t.piece(pi)
1633         t.piecesQueuedForHash.Remove(pi)
1634         p.hashing = true
1635         t.publishPieceChange(pi)
1636         t.updatePiecePriority(pi)
1637         t.storageLock.RLock()
1638         t.activePieceHashes++
1639         go t.pieceHasher(pi)
1640         return true
1641 }
1642
1643 func (t *Torrent) getPieceToHash() (ret pieceIndex, ok bool) {
1644         t.piecesQueuedForHash.IterTyped(func(i pieceIndex) bool {
1645                 if t.piece(i).hashing {
1646                         return true
1647                 }
1648                 ret = i
1649                 ok = true
1650                 return false
1651         })
1652         return
1653 }
1654
1655 func (t *Torrent) pieceHasher(index pieceIndex) {
1656         p := t.piece(index)
1657         sum := t.hashPiece(index)
1658         t.storageLock.RUnlock()
1659         t.cl.lock()
1660         defer t.cl.unlock()
1661         p.hashing = false
1662         t.updatePiecePriority(index)
1663         t.pieceHashed(index, sum == *p.hash)
1664         t.publishPieceChange(index)
1665         t.activePieceHashes--
1666         t.tryCreateMorePieceHashers()
1667 }
1668
1669 // Return the connections that touched a piece, and clear the entries while
1670 // doing it.
1671 func (t *Torrent) reapPieceTouchers(piece pieceIndex) (ret []*connection) {
1672         for c := range t.pieces[piece].dirtiers {
1673                 delete(c.peerTouchedPieces, piece)
1674                 ret = append(ret, c)
1675         }
1676         t.pieces[piece].dirtiers = nil
1677         return
1678 }
1679
1680 func (t *Torrent) connsAsSlice() (ret []*connection) {
1681         for c := range t.conns {
1682                 ret = append(ret, c)
1683         }
1684         return
1685 }
1686
1687 func (t *Torrent) queuePieceCheck(pieceIndex pieceIndex) {
1688         piece := t.piece(pieceIndex)
1689         if piece.queuedForHash() {
1690                 return
1691         }
1692         t.piecesQueuedForHash.Add(bitmap.BitIndex(pieceIndex))
1693         t.publishPieceChange(pieceIndex)
1694         t.updatePiecePriority(pieceIndex)
1695         t.tryCreateMorePieceHashers()
1696 }
1697
1698 // Forces all the pieces to be re-hashed. See also Piece.VerifyData.
1699 func (t *Torrent) VerifyData() {
1700         for i := pieceIndex(0); i < t.NumPieces(); i++ {
1701                 t.Piece(i).VerifyData()
1702         }
1703 }
1704
1705 // Start the process of connecting to the given peer for the given torrent if
1706 // appropriate.
1707 func (t *Torrent) initiateConn(peer Peer) {
1708         if peer.Id == t.cl.peerID {
1709                 return
1710         }
1711         if t.cl.badPeerIPPort(peer.IP, peer.Port) {
1712                 return
1713         }
1714         addr := IpPort{peer.IP, uint16(peer.Port)}
1715         if t.addrActive(addr.String()) {
1716                 return
1717         }
1718         t.halfOpen[addr.String()] = peer
1719         go t.cl.outgoingConnection(t, addr, peer.Source)
1720 }
1721
1722 func (t *Torrent) AddClientPeer(cl *Client) {
1723         t.AddPeers(func() (ps []Peer) {
1724                 for _, la := range cl.ListenAddrs() {
1725                         ps = append(ps, Peer{
1726                                 IP:   missinggo.AddrIP(la),
1727                                 Port: missinggo.AddrPort(la),
1728                         })
1729                 }
1730                 return
1731         }())
1732 }
1733
1734 // All stats that include this Torrent. Useful when we want to increment
1735 // ConnStats but not for every connection.
1736 func (t *Torrent) allStats(f func(*ConnStats)) {
1737         f(&t.stats)
1738         f(&t.cl.stats)
1739 }
1740
1741 func (t *Torrent) hashingPiece(i pieceIndex) bool {
1742         return t.pieces[i].hashing
1743 }
1744
1745 func (t *Torrent) pieceQueuedForHash(i pieceIndex) bool {
1746         return t.piecesQueuedForHash.Get(bitmap.BitIndex(i))
1747 }
1748
1749 func (t *Torrent) dialTimeout() time.Duration {
1750         return reducedDialTimeout(t.cl.config.MinDialTimeout, t.cl.config.NominalDialTimeout, t.cl.config.HalfOpenConnsPerTorrent, t.peers.Len())
1751 }
1752
1753 func (t *Torrent) piece(i int) *Piece {
1754         return &t.pieces[i]
1755 }