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