torrent.go

   1 package torrent
   2
   3 import (
   4         "container/heap"
   5         "crypto/sha1"
   6         "errors"
   7         "fmt"
   8         "io"
   9         "log"
  10         "math"
  11         "math/rand"
  12         "net"
  13         "os"
  14         "sync"
  15         "text/tabwriter"
  16         "time"
  17
  18         "github.com/anacrolix/dht"
  19         "github.com/anacrolix/missinggo"
  20         "github.com/anacrolix/missinggo/bitmap"
  21         "github.com/anacrolix/missinggo/perf"
  22         "github.com/anacrolix/missinggo/pubsub"
  23         "github.com/anacrolix/missinggo/slices"
  24         "github.com/bradfitz/iter"
  25
  26         "github.com/anacrolix/torrent/bencode"
  27         "github.com/anacrolix/torrent/metainfo"
  28         pp "github.com/anacrolix/torrent/peer_protocol"
  29         "github.com/anacrolix/torrent/storage"
  30         "github.com/anacrolix/torrent/tracker"
  31 )
  32
  33 func (t *Torrent) chunkIndexSpec(chunkIndex, piece int) chunkSpec {
  34         return chunkIndexSpec(chunkIndex, t.pieceLength(piece), t.chunkSize)
  35 }
  36
  37 type peersKey struct {
  38         IPBytes string
  39         Port    int
  40 }
  41
  42 // Maintains state of torrent within a Client.
  43 type Torrent struct {
  44         cl *Client
  45
  46         networkingEnabled bool
  47
  48         closed   missinggo.Event
  49         infoHash metainfo.Hash
  50         pieces   []piece
  51         // Values are the piece indices that changed.
  52         pieceStateChanges *pubsub.PubSub
  53         // The size of chunks to request from peers over the wire. This is
  54         // normally 16KiB by convention these days.
  55         chunkSize pp.Integer
  56         chunkPool *sync.Pool
  57         // Total length of the torrent in bytes. Stored because it's not O(1) to
  58         // get this from the info dict.
  59         length int64
  60
  61         // The storage to open when the info dict becomes available.
  62         storageOpener *storage.Client
  63         // Storage for torrent data.
  64         storage *storage.Torrent
  65
  66         metainfo metainfo.MetaInfo
  67
  68         // The info dict. nil if we don't have it (yet).
  69         info *metainfo.Info
  70         // Active peer connections, running message stream loops.
  71         conns               map[*connection]struct{}
  72         maxEstablishedConns int
  73         // Set of addrs to which we're attempting to connect. Connections are
  74         // half-open until all handshakes are completed.
  75         halfOpen map[string]struct{}
  76
  77         // Reserve of peers to connect to. A peer can be both here and in the
  78         // active connections if were told about the peer after connecting with
  79         // them. That encourages us to reconnect to peers that are well known in
  80         // the swarm.
  81         peers          map[peersKey]Peer
  82         wantPeersEvent missinggo.Event
  83         // An announcer for each tracker URL.
  84         trackerAnnouncers map[string]*trackerScraper
  85         // How many times we've initiated a DHT announce. TODO: Move into stats.
  86         numDHTAnnounces int
  87
  88         // Name used if the info name isn't available. Should be cleared when the
  89         // Info does become available.
  90         displayName string
  91         // The bencoded bytes of the info dict. This is actively manipulated if
  92         // the info bytes aren't initially available, and we try to fetch them
  93         // from peers.
  94         metadataBytes []byte
  95         // Each element corresponds to the 16KiB metadata pieces. If true, we have
  96         // received that piece.
  97         metadataCompletedChunks []bool
  98
  99         // Set when .Info is obtained.
 100         gotMetainfo missinggo.Event
 101
 102         readers               map[*Reader]struct{}
 103         readerNowPieces       bitmap.Bitmap
 104         readerReadaheadPieces bitmap.Bitmap
 105
 106         // The indexes of pieces we want with normal priority, that aren't
 107         // currently available.
 108         pendingPieces bitmap.Bitmap
 109         // A cache of completed piece indices.
 110         completedPieces bitmap.Bitmap
 111
 112         // A pool of piece priorities []int for assignment to new connections.
 113         // These "inclinations" are used to give connections preference for
 114         // different pieces.
 115         connPieceInclinationPool sync.Pool
 116         // Torrent-level statistics.
 117         stats TorrentStats
 118 }
 119
 120 // Returns a channel that is closed when the Torrent is closed.
 121 func (t *Torrent) Closed() <-chan struct{} {
 122         return t.closed.LockedChan(&t.cl.mu)
 123 }
 124
 125 func (t *Torrent) setChunkSize(size pp.Integer) {
 126         t.chunkSize = size
 127         t.chunkPool = &sync.Pool{
 128                 New: func() interface{} {
 129                         return make([]byte, size)
 130                 },
 131         }
 132 }
 133
 134 func (t *Torrent) setDisplayName(dn string) {
 135         if t.haveInfo() {
 136                 return
 137         }
 138         t.displayName = dn
 139 }
 140
 141 func (t *Torrent) pieceComplete(piece int) bool {
 142         return t.completedPieces.Get(piece)
 143 }
 144
 145 func (t *Torrent) pieceCompleteUncached(piece int) bool {
 146         return t.pieces[piece].Storage().GetIsComplete()
 147 }
 148
 149 // There's a connection to that address already.
 150 func (t *Torrent) addrActive(addr string) bool {
 151         if _, ok := t.halfOpen[addr]; ok {
 152                 return true
 153         }
 154         for c := range t.conns {
 155                 if c.remoteAddr().String() == addr {
 156                         return true
 157                 }
 158         }
 159         return false
 160 }
 161
 162 func (t *Torrent) worstUnclosedConns() (ret []*connection) {
 163         ret = make([]*connection, 0, len(t.conns))
 164         for c := range t.conns {
 165                 if !c.closed.IsSet() {
 166                         ret = append(ret, c)
 167                 }
 168         }
 169         return
 170 }
 171
 172 func (t *Torrent) addPeer(p Peer) {
 173         cl := t.cl
 174         cl.openNewConns(t)
 175         if len(t.peers) >= torrentPeersHighWater {
 176                 return
 177         }
 178         key := peersKey{string(p.IP), p.Port}
 179         if _, ok := t.peers[key]; ok {
 180                 return
 181         }
 182         t.peers[key] = p
 183         peersAddedBySource.Add(string(p.Source), 1)
 184         cl.openNewConns(t)
 185
 186 }
 187
 188 func (t *Torrent) invalidateMetadata() {
 189         for i := range t.metadataCompletedChunks {
 190                 t.metadataCompletedChunks[i] = false
 191         }
 192         t.info = nil
 193 }
 194
 195 func (t *Torrent) saveMetadataPiece(index int, data []byte) {
 196         if t.haveInfo() {
 197                 return
 198         }
 199         if index >= len(t.metadataCompletedChunks) {
 200                 log.Printf("%s: ignoring metadata piece %d", t, index)
 201                 return
 202         }
 203         copy(t.metadataBytes[(1<<14)*index:], data)
 204         t.metadataCompletedChunks[index] = true
 205 }
 206
 207 func (t *Torrent) metadataPieceCount() int {
 208         return (len(t.metadataBytes) + (1 << 14) - 1) / (1 << 14)
 209 }
 210
 211 func (t *Torrent) haveMetadataPiece(piece int) bool {
 212         if t.haveInfo() {
 213                 return (1<<14)*piece < len(t.metadataBytes)
 214         } else {
 215                 return piece < len(t.metadataCompletedChunks) && t.metadataCompletedChunks[piece]
 216         }
 217 }
 218
 219 func (t *Torrent) metadataSizeKnown() bool {
 220         return t.metadataBytes != nil
 221 }
 222
 223 func (t *Torrent) metadataSize() int {
 224         return len(t.metadataBytes)
 225 }
 226
 227 func infoPieceHashes(info *metainfo.Info) (ret []string) {
 228         for i := 0; i < len(info.Pieces); i += sha1.Size {
 229                 ret = append(ret, string(info.Pieces[i:i+sha1.Size]))
 230         }
 231         return
 232 }
 233
 234 func (t *Torrent) makePieces() {
 235         hashes := infoPieceHashes(t.info)
 236         t.pieces = make([]piece, len(hashes))
 237         for i, hash := range hashes {
 238                 piece := &t.pieces[i]
 239                 piece.t = t
 240                 piece.index = i
 241                 piece.noPendingWrites.L = &piece.pendingWritesMutex
 242                 missinggo.CopyExact(piece.Hash[:], hash)
 243         }
 244 }
 245
 246 // Called when metadata for a torrent becomes available.
 247 func (t *Torrent) setInfoBytes(b []byte) error {
 248         if t.haveInfo() {
 249                 return nil
 250         }
 251         if metainfo.HashBytes(b) != t.infoHash {
 252                 return errors.New("info bytes have wrong hash")
 253         }
 254         var info metainfo.Info
 255         err := bencode.Unmarshal(b, &info)
 256         if err != nil {
 257                 return fmt.Errorf("error unmarshalling info bytes: %s", err)
 258         }
 259         err = validateInfo(&info)
 260         if err != nil {
 261                 return fmt.Errorf("bad info: %s", err)
 262         }
 263         defer t.updateWantPeersEvent()
 264         t.info = &info
 265         t.displayName = "" // Save a few bytes lol.
 266         t.cl.event.Broadcast()
 267         t.gotMetainfo.Set()
 268         t.storage, err = t.storageOpener.OpenTorrent(t.info, t.infoHash)
 269         if err != nil {
 270                 return fmt.Errorf("error opening torrent storage: %s", err)
 271         }
 272         t.length = 0
 273         for _, f := range t.info.UpvertedFiles() {
 274                 t.length += f.Length
 275         }
 276         t.metadataBytes = b
 277         t.metadataCompletedChunks = nil
 278         t.makePieces()
 279         for conn := range t.conns {
 280                 if err := conn.setNumPieces(t.numPieces()); err != nil {
 281                         log.Printf("closing connection: %s", err)
 282                         conn.Close()
 283                 }
 284         }
 285         for i := range t.pieces {
 286                 t.updatePieceCompletion(i)
 287                 t.pieces[i].QueuedForHash = true
 288         }
 289         go func() {
 290                 for i := range t.pieces {
 291                         t.verifyPiece(i)
 292                 }
 293         }()
 294         return nil
 295 }
 296
 297 func (t *Torrent) haveAllMetadataPieces() bool {
 298         if t.haveInfo() {
 299                 return true
 300         }
 301         if t.metadataCompletedChunks == nil {
 302                 return false
 303         }
 304         for _, have := range t.metadataCompletedChunks {
 305                 if !have {
 306                         return false
 307                 }
 308         }
 309         return true
 310 }
 311
 312 // TODO: Propagate errors to disconnect peer.
 313 func (t *Torrent) setMetadataSize(bytes int64) (err error) {
 314         if t.haveInfo() {
 315                 // We already know the correct metadata size.
 316                 return
 317         }
 318         if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
 319                 return errors.New("bad size")
 320         }
 321         if t.metadataBytes != nil && len(t.metadataBytes) == int(bytes) {
 322                 return
 323         }
 324         t.metadataBytes = make([]byte, bytes)
 325         t.metadataCompletedChunks = make([]bool, (bytes+(1<<14)-1)/(1<<14))
 326         for c := range t.conns {
 327                 c.requestPendingMetadata()
 328         }
 329         return
 330 }
 331
 332 // The current working name for the torrent. Either the name in the info dict,
 333 // or a display name given such as by the dn value in a magnet link, or "".
 334 func (t *Torrent) name() string {
 335         if t.haveInfo() {
 336                 return t.info.Name
 337         }
 338         return t.displayName
 339 }
 340
 341 func (t *Torrent) pieceState(index int) (ret PieceState) {
 342         p := &t.pieces[index]
 343         ret.Priority = t.piecePriority(index)
 344         if t.pieceComplete(index) {
 345                 ret.Complete = true
 346         }
 347         if p.QueuedForHash || p.Hashing {
 348                 ret.Checking = true
 349         }
 350         if !ret.Complete && t.piecePartiallyDownloaded(index) {
 351                 ret.Partial = true
 352         }
 353         return
 354 }
 355
 356 func (t *Torrent) metadataPieceSize(piece int) int {
 357         return metadataPieceSize(len(t.metadataBytes), piece)
 358 }
 359
 360 func (t *Torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
 361         d := map[string]int{
 362                 "msg_type": msgType,
 363                 "piece":    piece,
 364         }
 365         if data != nil {
 366                 d["total_size"] = len(t.metadataBytes)
 367         }
 368         p, err := bencode.Marshal(d)
 369         if err != nil {
 370                 panic(err)
 371         }
 372         return pp.Message{
 373                 Type:            pp.Extended,
 374                 ExtendedID:      c.PeerExtensionIDs["ut_metadata"],
 375                 ExtendedPayload: append(p, data...),
 376         }
 377 }
 378
 379 func (t *Torrent) pieceStateRuns() (ret []PieceStateRun) {
 380         rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
 381                 ret = append(ret, PieceStateRun{
 382                         PieceState: el.(PieceState),
 383                         Length:     int(count),
 384                 })
 385         })
 386         for index := range t.pieces {
 387                 rle.Append(t.pieceState(index), 1)
 388         }
 389         rle.Flush()
 390         return
 391 }
 392
 393 // Produces a small string representing a PieceStateRun.
 394 func pieceStateRunStatusChars(psr PieceStateRun) (ret string) {
 395         ret = fmt.Sprintf("%d", psr.Length)
 396         ret += func() string {
 397                 switch psr.Priority {
 398                 case PiecePriorityNext:
 399                         return "N"
 400                 case PiecePriorityNormal:
 401                         return "."
 402                 case PiecePriorityReadahead:
 403                         return "R"
 404                 case PiecePriorityNow:
 405                         return "!"
 406                 default:
 407                         return ""
 408                 }
 409         }()
 410         if psr.Checking {
 411                 ret += "H"
 412         }
 413         if psr.Partial {
 414                 ret += "P"
 415         }
 416         if psr.Complete {
 417                 ret += "C"
 418         }
 419         return
 420 }
 421
 422 func (t *Torrent) writeStatus(w io.Writer) {
 423         fmt.Fprintf(w, "Infohash: %s\n", t.infoHash.HexString())
 424         fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
 425         if !t.haveInfo() {
 426                 fmt.Fprintf(w, "Metadata have: ")
 427                 for _, h := range t.metadataCompletedChunks {
 428                         fmt.Fprintf(w, "%c", func() rune {
 429                                 if h {
 430                                         return 'H'
 431                                 } else {
 432                                         return '.'
 433                                 }
 434                         }())
 435                 }
 436                 fmt.Fprintln(w)
 437         }
 438         fmt.Fprintf(w, "Piece length: %s\n", func() string {
 439                 if t.haveInfo() {
 440                         return fmt.Sprint(t.usualPieceSize())
 441                 } else {
 442                         return "?"
 443                 }
 444         }())
 445         if t.Info() != nil {
 446                 fmt.Fprintf(w, "Num Pieces: %d\n", t.numPieces())
 447                 fmt.Fprint(w, "Piece States:")
 448                 for _, psr := range t.pieceStateRuns() {
 449                         w.Write([]byte(" "))
 450                         w.Write([]byte(pieceStateRunStatusChars(psr)))
 451                 }
 452                 fmt.Fprintln(w)
 453         }
 454         fmt.Fprintf(w, "Reader Pieces:")
 455         t.forReaderOffsetPieces(func(begin, end int) (again bool) {
 456                 fmt.Fprintf(w, " %d:%d", begin, end)
 457                 return true
 458         })
 459         fmt.Fprintln(w)
 460
 461         fmt.Fprintf(w, "Trackers:\n")
 462         func() {
 463                 tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
 464                 fmt.Fprintf(tw, "    URL\tNext announce\tLast announce\n")
 465                 for _, ta := range slices.Sort(slices.FromMapElems(t.trackerAnnouncers), func(l, r *trackerScraper) bool {
 466                         return l.url < r.url
 467                 }).([]*trackerScraper) {
 468                         fmt.Fprintf(tw, "    %s\n", ta.statusLine())
 469                 }
 470                 tw.Flush()
 471         }()
 472
 473         fmt.Fprintf(w, "DHT Announces: %d\n", t.numDHTAnnounces)
 474
 475         fmt.Fprintf(w, "Pending peers: %d\n", len(t.peers))
 476         fmt.Fprintf(w, "Half open: %d\n", len(t.halfOpen))
 477         fmt.Fprintf(w, "Active peers: %d\n", len(t.conns))
 478         conns := t.connsAsSlice()
 479         slices.Sort(conns, worseConn)
 480         for i, c := range conns {
 481                 fmt.Fprintf(w, "%2d. ", i+1)
 482                 c.WriteStatus(w, t)
 483         }
 484 }
 485
 486 func (t *Torrent) haveInfo() bool {
 487         return t.info != nil
 488 }
 489
 490 // Returns a run-time generated MetaInfo that includes the info bytes and
 491 // announce-list as currently known to the client.
 492 func (t *Torrent) newMetaInfo() metainfo.MetaInfo {
 493         return metainfo.MetaInfo{
 494                 CreationDate: time.Now().Unix(),
 495                 Comment:      "dynamic metainfo from client",
 496                 CreatedBy:    "go.torrent",
 497                 AnnounceList: t.metainfo.UpvertedAnnounceList(),
 498                 InfoBytes:    t.metadataBytes,
 499         }
 500 }
 501
 502 func (t *Torrent) BytesMissing() int64 {
 503         t.mu().RLock()
 504         defer t.mu().RUnlock()
 505         return t.bytesMissingLocked()
 506 }
 507
 508 func (t *Torrent) bytesMissingLocked() int64 {
 509         return t.bytesLeft()
 510 }
 511
 512 func (t *Torrent) bytesLeft() (left int64) {
 513         for i := 0; i < t.numPieces(); i++ {
 514                 left += int64(t.pieces[i].bytesLeft())
 515         }
 516         return
 517 }
 518
 519 // Bytes left to give in tracker announces.
 520 func (t *Torrent) bytesLeftAnnounce() uint64 {
 521         if t.haveInfo() {
 522                 return uint64(t.bytesLeft())
 523         } else {
 524                 return math.MaxUint64
 525         }
 526 }
 527
 528 func (t *Torrent) piecePartiallyDownloaded(piece int) bool {
 529         if t.pieceComplete(piece) {
 530                 return false
 531         }
 532         if t.pieceAllDirty(piece) {
 533                 return false
 534         }
 535         return t.pieces[piece].hasDirtyChunks()
 536 }
 537
 538 func (t *Torrent) usualPieceSize() int {
 539         return int(t.info.PieceLength)
 540 }
 541
 542 func (t *Torrent) numPieces() int {
 543         return t.info.NumPieces()
 544 }
 545
 546 func (t *Torrent) numPiecesCompleted() (num int) {
 547         return t.completedPieces.Len()
 548 }
 549
 550 func (t *Torrent) close() (err error) {
 551         t.closed.Set()
 552         if t.storage != nil {
 553                 t.storage.Close()
 554         }
 555         for conn := range t.conns {
 556                 conn.Close()
 557         }
 558         t.cl.event.Broadcast()
 559         t.pieceStateChanges.Close()
 560         t.updateWantPeersEvent()
 561         return
 562 }
 563
 564 func (t *Torrent) requestOffset(r request) int64 {
 565         return torrentRequestOffset(t.length, int64(t.usualPieceSize()), r)
 566 }
 567
 568 // Return the request that would include the given offset into the torrent
 569 // data. Returns !ok if there is no such request.
 570 func (t *Torrent) offsetRequest(off int64) (req request, ok bool) {
 571         return torrentOffsetRequest(t.length, t.info.PieceLength, int64(t.chunkSize), off)
 572 }
 573
 574 func (t *Torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
 575         tr := perf.NewTimer()
 576
 577         n, err := t.pieces[piece].Storage().WriteAt(data, begin)
 578         if err == nil && n != len(data) {
 579                 err = io.ErrShortWrite
 580         }
 581         if err == nil {
 582                 tr.Mark("write chunk")
 583         }
 584         return
 585 }
 586
 587 func (t *Torrent) bitfield() (bf []bool) {
 588         bf = make([]bool, t.numPieces())
 589         t.completedPieces.IterTyped(func(piece int) (again bool) {
 590                 bf[piece] = true
 591                 return true
 592         })
 593         return
 594 }
 595
 596 func (t *Torrent) pieceNumChunks(piece int) int {
 597         return int((t.pieceLength(piece) + t.chunkSize - 1) / t.chunkSize)
 598 }
 599
 600 func (t *Torrent) pendAllChunkSpecs(pieceIndex int) {
 601         t.pieces[pieceIndex].DirtyChunks.Clear()
 602 }
 603
 604 type Peer struct {
 605         Id     [20]byte
 606         IP     net.IP
 607         Port   int
 608         Source peerSource
 609         // Peer is known to support encryption.
 610         SupportsEncryption bool
 611 }
 612
 613 func (t *Torrent) pieceLength(piece int) (len_ pp.Integer) {
 614         if piece < 0 || piece >= t.info.NumPieces() {
 615                 return
 616         }
 617         if piece == t.numPieces()-1 {
 618                 len_ = pp.Integer(t.length % t.info.PieceLength)
 619         }
 620         if len_ == 0 {
 621                 len_ = pp.Integer(t.info.PieceLength)
 622         }
 623         return
 624 }
 625
 626 func (t *Torrent) hashPiece(piece int) (ret metainfo.Hash) {
 627         hash := pieceHash.New()
 628         p := &t.pieces[piece]
 629         p.waitNoPendingWrites()
 630         ip := t.info.Piece(piece)
 631         pl := ip.Length()
 632         n, err := io.Copy(hash, io.NewSectionReader(t.pieces[piece].Storage(), 0, pl))
 633         if n == pl {
 634                 missinggo.CopyExact(&ret, hash.Sum(nil))
 635                 return
 636         }
 637         if err != io.ErrUnexpectedEOF && !os.IsNotExist(err) {
 638                 log.Printf("unexpected error hashing piece with %T: %s", t.storage.TorrentImpl, err)
 639         }
 640         return
 641 }
 642
 643 func (t *Torrent) haveAnyPieces() bool {
 644         for i := range t.pieces {
 645                 if t.pieceComplete(i) {
 646                         return true
 647                 }
 648         }
 649         return false
 650 }
 651
 652 func (t *Torrent) havePiece(index int) bool {
 653         return t.haveInfo() && t.pieceComplete(index)
 654 }
 655
 656 func (t *Torrent) haveChunk(r request) (ret bool) {
 657         // defer func() {
 658         //      log.Println("have chunk", r, ret)
 659         // }()
 660         if !t.haveInfo() {
 661                 return false
 662         }
 663         if t.pieceComplete(int(r.Index)) {
 664                 return true
 665         }
 666         p := &t.pieces[r.Index]
 667         return !p.pendingChunk(r.chunkSpec, t.chunkSize)
 668 }
 669
 670 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
 671         return int(cs.Begin / chunkSize)
 672 }
 673
 674 func (t *Torrent) wantPiece(r request) bool {
 675         if !t.wantPieceIndex(int(r.Index)) {
 676                 return false
 677         }
 678         if t.pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) {
 679                 return true
 680         }
 681         // TODO: What about pieces that were wanted, but aren't now, and aren't
 682         // completed either? That used to be done here.
 683         return false
 684 }
 685
 686 func (t *Torrent) wantPieceIndex(index int) bool {
 687         if !t.haveInfo() {
 688                 return false
 689         }
 690         if index < 0 || index >= t.numPieces() {
 691                 return false
 692         }
 693         p := &t.pieces[index]
 694         if p.QueuedForHash {
 695                 return false
 696         }
 697         if p.Hashing {
 698                 return false
 699         }
 700         if t.pieceComplete(index) {
 701                 return false
 702         }
 703         if t.pendingPieces.Contains(index) {
 704                 return true
 705         }
 706         return !t.forReaderOffsetPieces(func(begin, end int) bool {
 707                 return index < begin || index >= end
 708         })
 709 }
 710
 711 // The worst connection is one that hasn't been sent, or sent anything useful
 712 // for the longest. A bad connection is one that usually sends us unwanted
 713 // pieces, or has been in worser half of the established connections for more
 714 // than a minute.
 715 func (t *Torrent) worstBadConn() *connection {
 716         wcs := worseConnSlice{t.worstUnclosedConns()}
 717         for wcs.Len() != 0 {
 718                 c := heap.Pop(&wcs).(*connection)
 719                 if c.UnwantedChunksReceived >= 6 && c.UnwantedChunksReceived > c.UsefulChunksReceived {
 720                         return c
 721                 }
 722                 if wcs.Len() >= (t.maxEstablishedConns+1)/2 {
 723                         // Give connections 1 minute to prove themselves.
 724                         if time.Since(c.completedHandshake) > time.Minute {
 725                                 return c
 726                         }
 727                 }
 728         }
 729         return nil
 730 }
 731
 732 type PieceStateChange struct {
 733         Index int
 734         PieceState
 735 }
 736
 737 func (t *Torrent) publishPieceChange(piece int) {
 738         cur := t.pieceState(piece)
 739         p := &t.pieces[piece]
 740         if cur != p.PublicPieceState {
 741                 p.PublicPieceState = cur
 742                 t.pieceStateChanges.Publish(PieceStateChange{
 743                         piece,
 744                         cur,
 745                 })
 746         }
 747 }
 748
 749 func (t *Torrent) pieceNumPendingChunks(piece int) int {
 750         if t.pieceComplete(piece) {
 751                 return 0
 752         }
 753         return t.pieceNumChunks(piece) - t.pieces[piece].numDirtyChunks()
 754 }
 755
 756 func (t *Torrent) pieceAllDirty(piece int) bool {
 757         return t.pieces[piece].DirtyChunks.Len() == t.pieceNumChunks(piece)
 758 }
 759
 760 func (t *Torrent) readersChanged() {
 761         t.updateReaderPieces()
 762         t.updateAllPiecePriorities()
 763 }
 764
 765 func (t *Torrent) updateReaderPieces() {
 766         t.readerNowPieces, t.readerReadaheadPieces = t.readerPiecePriorities()
 767 }
 768
 769 func (t *Torrent) readerPosChanged(from, to pieceRange) {
 770         if from == to {
 771                 return
 772         }
 773         t.updateReaderPieces()
 774         // Order the ranges, high and low.
 775         l, h := from, to
 776         if l.begin > h.begin {
 777                 l, h = h, l
 778         }
 779         if l.end < h.begin {
 780                 // Two distinct ranges.
 781                 t.updatePiecePriorities(l.begin, l.end)
 782                 t.updatePiecePriorities(h.begin, h.end)
 783         } else {
 784                 // Ranges overlap.
 785                 end := l.end
 786                 if h.end > end {
 787                         end = h.end
 788                 }
 789                 t.updatePiecePriorities(l.begin, end)
 790         }
 791 }
 792
 793 func (t *Torrent) maybeNewConns() {
 794         // Tickle the accept routine.
 795         t.cl.event.Broadcast()
 796         t.openNewConns()
 797 }
 798
 799 func (t *Torrent) piecePriorityChanged(piece int) {
 800         for c := range t.conns {
 801                 c.updatePiecePriority(piece)
 802         }
 803         t.maybeNewConns()
 804         t.publishPieceChange(piece)
 805 }
 806
 807 func (t *Torrent) updatePiecePriority(piece int) {
 808         p := &t.pieces[piece]
 809         newPrio := t.piecePriorityUncached(piece)
 810         if newPrio == p.priority {
 811                 return
 812         }
 813         p.priority = newPrio
 814         t.piecePriorityChanged(piece)
 815 }
 816
 817 func (t *Torrent) updateAllPiecePriorities() {
 818         t.updatePiecePriorities(0, len(t.pieces))
 819 }
 820
 821 // Update all piece priorities in one hit. This function should have the same
 822 // output as updatePiecePriority, but across all pieces.
 823 func (t *Torrent) updatePiecePriorities(begin, end int) {
 824         for i := begin; i < end; i++ {
 825                 t.updatePiecePriority(i)
 826         }
 827 }
 828
 829 // Returns the range of pieces [begin, end) that contains the extent of bytes.
 830 func (t *Torrent) byteRegionPieces(off, size int64) (begin, end int) {
 831         if off >= t.length {
 832                 return
 833         }
 834         if off < 0 {
 835                 size += off
 836                 off = 0
 837         }
 838         if size <= 0 {
 839                 return
 840         }
 841         begin = int(off / t.info.PieceLength)
 842         end = int((off + size + t.info.PieceLength - 1) / t.info.PieceLength)
 843         if end > t.info.NumPieces() {
 844                 end = t.info.NumPieces()
 845         }
 846         return
 847 }
 848
 849 // Returns true if all iterations complete without breaking. Returns the read
 850 // regions for all readers. The reader regions should not be merged as some
 851 // callers depend on this method to enumerate readers.
 852 func (t *Torrent) forReaderOffsetPieces(f func(begin, end int) (more bool)) (all bool) {
 853         for r := range t.readers {
 854                 p := r.pieces
 855                 if p.begin >= p.end {
 856                         continue
 857                 }
 858                 if !f(p.begin, p.end) {
 859                         return false
 860                 }
 861         }
 862         return true
 863 }
 864
 865 func (t *Torrent) piecePriority(piece int) piecePriority {
 866         if !t.haveInfo() {
 867                 return PiecePriorityNone
 868         }
 869         return t.pieces[piece].priority
 870 }
 871
 872 func (t *Torrent) piecePriorityUncached(piece int) piecePriority {
 873         if t.pieceComplete(piece) {
 874                 return PiecePriorityNone
 875         }
 876         if t.readerNowPieces.Contains(piece) {
 877                 return PiecePriorityNow
 878         }
 879         // if t.readerNowPieces.Contains(piece - 1) {
 880         //      return PiecePriorityNext
 881         // }
 882         if t.readerReadaheadPieces.Contains(piece) {
 883                 return PiecePriorityReadahead
 884         }
 885         if t.pendingPieces.Contains(piece) {
 886                 return PiecePriorityNormal
 887         }
 888         return PiecePriorityNone
 889 }
 890
 891 func (t *Torrent) pendPiece(piece int) {
 892         if t.pendingPieces.Contains(piece) {
 893                 return
 894         }
 895         if t.havePiece(piece) {
 896                 return
 897         }
 898         t.pendingPieces.Add(piece)
 899         t.updatePiecePriority(piece)
 900 }
 901
 902 func (t *Torrent) unpendPieces(unpend *bitmap.Bitmap) {
 903         t.pendingPieces.Sub(unpend)
 904         unpend.IterTyped(func(piece int) (again bool) {
 905                 t.updatePiecePriority(piece)
 906                 return true
 907         })
 908 }
 909
 910 func (t *Torrent) pendPieceRange(begin, end int) {
 911         for i := begin; i < end; i++ {
 912                 t.pendPiece(i)
 913         }
 914 }
 915
 916 func (t *Torrent) unpendPieceRange(begin, end int) {
 917         var bm bitmap.Bitmap
 918         bm.AddRange(begin, end)
 919         t.unpendPieces(&bm)
 920 }
 921
 922 func (t *Torrent) pendRequest(req request) {
 923         ci := chunkIndex(req.chunkSpec, t.chunkSize)
 924         t.pieces[req.Index].pendChunkIndex(ci)
 925 }
 926
 927 func (t *Torrent) pieceCompletionChanged(piece int) {
 928         t.cl.event.Broadcast()
 929         if t.pieceComplete(piece) {
 930                 t.onPieceCompleted(piece)
 931         } else {
 932                 t.onIncompletePiece(piece)
 933         }
 934         t.updatePiecePriority(piece)
 935 }
 936
 937 func (t *Torrent) openNewConns() {
 938         t.cl.openNewConns(t)
 939 }
 940
 941 func (t *Torrent) getConnPieceInclination() []int {
 942         _ret := t.connPieceInclinationPool.Get()
 943         if _ret == nil {
 944                 pieceInclinationsNew.Add(1)
 945                 return rand.Perm(t.numPieces())
 946         }
 947         pieceInclinationsReused.Add(1)
 948         return _ret.([]int)
 949 }
 950
 951 func (t *Torrent) putPieceInclination(pi []int) {
 952         t.connPieceInclinationPool.Put(pi)
 953         pieceInclinationsPut.Add(1)
 954 }
 955
 956 func (t *Torrent) updatePieceCompletion(piece int) {
 957         pcu := t.pieceCompleteUncached(piece)
 958         changed := t.completedPieces.Get(piece) != pcu
 959         t.completedPieces.Set(piece, pcu)
 960         if changed {
 961                 t.pieceCompletionChanged(piece)
 962         }
 963 }
 964
 965 // Non-blocking read. Client lock is not required.
 966 func (t *Torrent) readAt(b []byte, off int64) (n int, err error) {
 967         p := &t.pieces[off/t.info.PieceLength]
 968         p.waitNoPendingWrites()
 969         return p.Storage().ReadAt(b, off-p.Info().Offset())
 970 }
 971
 972 func (t *Torrent) updateAllPieceCompletions() {
 973         for i := range iter.N(t.numPieces()) {
 974                 t.updatePieceCompletion(i)
 975         }
 976 }
 977
 978 // Returns an error if the metadata was completed, but couldn't be set for
 979 // some reason. Blame it on the last peer to contribute.
 980 func (t *Torrent) maybeCompleteMetadata() error {
 981         if t.haveInfo() {
 982                 // Nothing to do.
 983                 return nil
 984         }
 985         if !t.haveAllMetadataPieces() {
 986                 // Don't have enough metadata pieces.
 987                 return nil
 988         }
 989         err := t.setInfoBytes(t.metadataBytes)
 990         if err != nil {
 991                 t.invalidateMetadata()
 992                 return fmt.Errorf("error setting info bytes: %s", err)
 993         }
 994         if t.cl.config.Debug {
 995                 log.Printf("%s: got metadata from peers", t)
 996         }
 997         return nil
 998 }
 999
1000 func (t *Torrent) readerPieces() (ret bitmap.Bitmap) {
1001         t.forReaderOffsetPieces(func(begin, end int) bool {
1002                 ret.AddRange(begin, end)
1003                 return true
1004         })
1005         return
1006 }
1007
1008 func (t *Torrent) readerPiecePriorities() (now, readahead bitmap.Bitmap) {
1009         t.forReaderOffsetPieces(func(begin, end int) bool {
1010                 if end > begin {
1011                         now.Add(begin)
1012                         readahead.AddRange(begin+1, end)
1013                 }
1014                 return true
1015         })
1016         return
1017 }
1018
1019 func (t *Torrent) needData() bool {
1020         if t.closed.IsSet() {
1021                 return false
1022         }
1023         if !t.haveInfo() {
1024                 return true
1025         }
1026         if t.pendingPieces.Len() != 0 {
1027                 return true
1028         }
1029         // Read as "not all complete".
1030         return !t.readerPieces().IterTyped(func(piece int) bool {
1031                 return t.pieceComplete(piece)
1032         })
1033 }
1034
1035 func appendMissingStrings(old, new []string) (ret []string) {
1036         ret = old
1037 new:
1038         for _, n := range new {
1039                 for _, o := range old {
1040                         if o == n {
1041                                 continue new
1042                         }
1043                 }
1044                 ret = append(ret, n)
1045         }
1046         return
1047 }
1048
1049 func appendMissingTrackerTiers(existing [][]string, minNumTiers int) (ret [][]string) {
1050         ret = existing
1051         for minNumTiers > len(ret) {
1052                 ret = append(ret, nil)
1053         }
1054         return
1055 }
1056
1057 func (t *Torrent) addTrackers(announceList [][]string) {
1058         fullAnnounceList := &t.metainfo.AnnounceList
1059         t.metainfo.AnnounceList = appendMissingTrackerTiers(*fullAnnounceList, len(announceList))
1060         for tierIndex, trackerURLs := range announceList {
1061                 (*fullAnnounceList)[tierIndex] = appendMissingStrings((*fullAnnounceList)[tierIndex], trackerURLs)
1062         }
1063         t.startMissingTrackerScrapers()
1064         t.updateWantPeersEvent()
1065 }
1066
1067 // Don't call this before the info is available.
1068 func (t *Torrent) bytesCompleted() int64 {
1069         if !t.haveInfo() {
1070                 return 0
1071         }
1072         return t.info.TotalLength() - t.bytesLeft()
1073 }
1074
1075 func (t *Torrent) SetInfoBytes(b []byte) (err error) {
1076         t.cl.mu.Lock()
1077         defer t.cl.mu.Unlock()
1078         return t.setInfoBytes(b)
1079 }
1080
1081 // Returns true if connection is removed from torrent.Conns.
1082 func (t *Torrent) deleteConnection(c *connection) (ret bool) {
1083         _, ret = t.conns[c]
1084         delete(t.conns, c)
1085         return
1086 }
1087
1088 func (t *Torrent) dropConnection(c *connection) {
1089         t.cl.event.Broadcast()
1090         c.Close()
1091         if t.deleteConnection(c) {
1092                 t.openNewConns()
1093         }
1094 }
1095
1096 func (t *Torrent) wantPeers() bool {
1097         if t.closed.IsSet() {
1098                 return false
1099         }
1100         if len(t.peers) > torrentPeersLowWater {
1101                 return false
1102         }
1103         return t.needData() || t.seeding()
1104 }
1105
1106 func (t *Torrent) updateWantPeersEvent() {
1107         if t.wantPeers() {
1108                 t.wantPeersEvent.Set()
1109         } else {
1110                 t.wantPeersEvent.Clear()
1111         }
1112 }
1113
1114 // Returns whether the client should make effort to seed the torrent.
1115 func (t *Torrent) seeding() bool {
1116         cl := t.cl
1117         if t.closed.IsSet() {
1118                 return false
1119         }
1120         if cl.config.NoUpload {
1121                 return false
1122         }
1123         if !cl.config.Seed {
1124                 return false
1125         }
1126         if t.needData() {
1127                 return false
1128         }
1129         return true
1130 }
1131
1132 func (t *Torrent) startScrapingTracker(url string) {
1133         if url == "" {
1134                 return
1135         }
1136         if _, ok := t.trackerAnnouncers[url]; ok {
1137                 return
1138         }
1139         newAnnouncer := &trackerScraper{
1140                 url: url,
1141                 t:   t,
1142         }
1143         if t.trackerAnnouncers == nil {
1144                 t.trackerAnnouncers = make(map[string]*trackerScraper)
1145         }
1146         t.trackerAnnouncers[url] = newAnnouncer
1147         go newAnnouncer.Run()
1148 }
1149
1150 // Adds and starts tracker scrapers for tracker URLs that aren't already
1151 // running.
1152 func (t *Torrent) startMissingTrackerScrapers() {
1153         if t.cl.config.DisableTrackers {
1154                 return
1155         }
1156         t.startScrapingTracker(t.metainfo.Announce)
1157         for _, tier := range t.metainfo.AnnounceList {
1158                 for _, url := range tier {
1159                         t.startScrapingTracker(url)
1160                 }
1161         }
1162 }
1163
1164 // Returns an AnnounceRequest with fields filled out to defaults and current
1165 // values.
1166 func (t *Torrent) announceRequest() tracker.AnnounceRequest {
1167         return tracker.AnnounceRequest{
1168                 Event:    tracker.None,
1169                 NumWant:  -1,
1170                 Port:     uint16(t.cl.incomingPeerPort()),
1171                 PeerId:   t.cl.peerID,
1172                 InfoHash: t.infoHash,
1173                 Left:     t.bytesLeftAnnounce(),
1174         }
1175 }
1176
1177 // Adds peers revealed in an announce until the announce ends, or we have
1178 // enough peers.
1179 func (t *Torrent) consumeDHTAnnounce(pvs <-chan dht.PeersValues) {
1180         cl := t.cl
1181         // Count all the unique addresses we got during this announce.
1182         allAddrs := make(map[string]struct{})
1183         for {
1184                 select {
1185                 case v, ok := <-pvs:
1186                         if !ok {
1187                                 return
1188                         }
1189                         addPeers := make([]Peer, 0, len(v.Peers))
1190                         for _, cp := range v.Peers {
1191                                 if cp.Port == 0 {
1192                                         // Can't do anything with this.
1193                                         continue
1194                                 }
1195                                 addPeers = append(addPeers, Peer{
1196                                         IP:     cp.IP[:],
1197                                         Port:   cp.Port,
1198                                         Source: peerSourceDHTGetPeers,
1199                                 })
1200                                 key := (&net.UDPAddr{
1201                                         IP:   cp.IP[:],
1202                                         Port: cp.Port,
1203                                 }).String()
1204                                 allAddrs[key] = struct{}{}
1205                         }
1206                         cl.mu.Lock()
1207                         t.addPeers(addPeers)
1208                         numPeers := len(t.peers)
1209                         cl.mu.Unlock()
1210                         if numPeers >= torrentPeersHighWater {
1211                                 return
1212                         }
1213                 case <-t.closed.LockedChan(&cl.mu):
1214                         return
1215                 }
1216         }
1217 }
1218
1219 func (t *Torrent) announceDHT(impliedPort bool) (err error) {
1220         cl := t.cl
1221         ps, err := cl.dHT.Announce(t.infoHash, cl.incomingPeerPort(), impliedPort)
1222         if err != nil {
1223                 return
1224         }
1225         t.consumeDHTAnnounce(ps.Peers)
1226         ps.Close()
1227         return
1228 }
1229
1230 func (t *Torrent) dhtAnnouncer() {
1231         cl := t.cl
1232         for {
1233                 select {
1234                 case <-t.wantPeersEvent.LockedChan(&cl.mu):
1235                 case <-t.closed.LockedChan(&cl.mu):
1236                         return
1237                 }
1238                 err := t.announceDHT(true)
1239                 func() {
1240                         cl.mu.Lock()
1241                         defer cl.mu.Unlock()
1242                         if err == nil {
1243                                 t.numDHTAnnounces++
1244                         } else {
1245                                 log.Printf("error announcing %q to DHT: %s", t, err)
1246                         }
1247                 }()
1248                 select {
1249                 case <-t.closed.LockedChan(&cl.mu):
1250                         return
1251                 case <-time.After(5 * time.Minute):
1252                 }
1253         }
1254 }
1255
1256 func (t *Torrent) addPeers(peers []Peer) {
1257         for _, p := range peers {
1258                 if t.cl.badPeerIPPort(p.IP, p.Port) {
1259                         continue
1260                 }
1261                 t.addPeer(p)
1262         }
1263 }
1264
1265 func (t *Torrent) Stats() TorrentStats {
1266         t.cl.mu.Lock()
1267         defer t.cl.mu.Unlock()
1268
1269         t.stats.ActivePeers = len(t.conns)
1270         t.stats.HalfOpenPeers = len(t.halfOpen)
1271         t.stats.PendingPeers = len(t.peers)
1272         t.stats.TotalPeers = t.numTotalPeers()
1273
1274         return t.stats
1275 }
1276
1277 // The total number of peers in the torrent.
1278 func (t *Torrent) numTotalPeers() int {
1279         peers := make(map[string]struct{})
1280         for conn := range t.conns {
1281                 peers[conn.conn.RemoteAddr().String()] = struct{}{}
1282         }
1283         for addr := range t.halfOpen {
1284                 peers[addr] = struct{}{}
1285         }
1286         for _, peer := range t.peers {
1287                 peers[fmt.Sprintf("%s:%d", peer.IP, peer.Port)] = struct{}{}
1288         }
1289         return len(peers)
1290 }
1291
1292 // Returns true if the connection is added.
1293 func (t *Torrent) addConnection(c *connection, outgoing bool) bool {
1294         if t.cl.closed.IsSet() {
1295                 return false
1296         }
1297         if !t.wantConns() {
1298                 return false
1299         }
1300         for c0 := range t.conns {
1301                 if c.PeerID == c0.PeerID {
1302                         // Already connected to a client with that ID.
1303                         duplicateClientConns.Add(1)
1304                         lower := string(t.cl.peerID[:]) < string(c.PeerID[:])
1305                         // Retain the connection from initiated from lower peer ID to
1306                         // higher.
1307                         if outgoing == lower {
1308                                 // Close the other one.
1309                                 c0.Close()
1310                                 // Is it safe to delete from the map while we're iterating
1311                                 // over it?
1312                                 t.deleteConnection(c0)
1313                         } else {
1314                                 // Abandon this one.
1315                                 return false
1316                         }
1317                 }
1318         }
1319         if len(t.conns) >= t.maxEstablishedConns {
1320                 c := t.worstBadConn()
1321                 if c == nil {
1322                         return false
1323                 }
1324                 if t.cl.config.Debug && missinggo.CryHeard() {
1325                         log.Printf("%s: dropping connection to make room for new one:\n    %s", t, c)
1326                 }
1327                 c.Close()
1328                 t.deleteConnection(c)
1329         }
1330         if len(t.conns) >= t.maxEstablishedConns {
1331                 panic(len(t.conns))
1332         }
1333         if c.t != nil {
1334                 panic("connection already associated with a torrent")
1335         }
1336         // Reconcile bytes transferred before connection was associated with a
1337         // torrent.
1338         t.stats.wroteBytes(c.stats.BytesWritten)
1339         t.stats.readBytes(c.stats.BytesRead)
1340         c.t = t
1341         t.conns[c] = struct{}{}
1342         return true
1343 }
1344
1345 func (t *Torrent) wantConns() bool {
1346         if !t.networkingEnabled {
1347                 return false
1348         }
1349         if t.closed.IsSet() {
1350                 return false
1351         }
1352         if !t.seeding() && !t.needData() {
1353                 return false
1354         }
1355         if len(t.conns) < t.maxEstablishedConns {
1356                 return true
1357         }
1358         return t.worstBadConn() != nil
1359 }
1360
1361 func (t *Torrent) SetMaxEstablishedConns(max int) (oldMax int) {
1362         t.cl.mu.Lock()
1363         defer t.cl.mu.Unlock()
1364         oldMax = t.maxEstablishedConns
1365         t.maxEstablishedConns = max
1366         wcs := slices.HeapInterface(slices.FromMapKeys(t.conns), worseConn)
1367         for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 {
1368                 t.dropConnection(wcs.Pop().(*connection))
1369         }
1370         t.openNewConns()
1371         return oldMax
1372 }
1373
1374 func (t *Torrent) mu() missinggo.RWLocker {
1375         return &t.cl.mu
1376 }
1377
1378 func (t *Torrent) pieceHashed(piece int, correct bool) {
1379         if t.closed.IsSet() {
1380                 return
1381         }
1382         p := &t.pieces[piece]
1383         touchers := t.reapPieceTouchers(piece)
1384         if p.EverHashed {
1385                 // Don't score the first time a piece is hashed, it could be an
1386                 // initial check.
1387                 if correct {
1388                         pieceHashedCorrect.Add(1)
1389                 } else {
1390                         log.Printf("%s: piece %d (%s) failed hash: %d connections contributed", t, piece, p.Hash, len(touchers))
1391                         pieceHashedNotCorrect.Add(1)
1392                 }
1393         }
1394         p.EverHashed = true
1395         if correct {
1396                 for _, c := range touchers {
1397                         c.goodPiecesDirtied++
1398                 }
1399                 err := p.Storage().MarkComplete()
1400                 if err != nil {
1401                         log.Printf("%T: error completing piece %d: %s", t.storage, piece, err)
1402                 }
1403                 t.updatePieceCompletion(piece)
1404         } else {
1405                 if len(touchers) != 0 {
1406                         for _, c := range touchers {
1407                                 // Y u do dis peer?!
1408                                 c.badPiecesDirtied++
1409                         }
1410                         slices.Sort(touchers, connLessTrusted)
1411                         log.Printf("dropping first corresponding conn from trust: %v", func() (ret []int) {
1412                                 for _, c := range touchers {
1413                                         ret = append(ret, c.netGoodPiecesDirtied())
1414                                 }
1415                                 return
1416                         }())
1417                         c := touchers[0]
1418                         t.cl.banPeerIP(missinggo.AddrIP(c.remoteAddr()))
1419                         c.Drop()
1420                 }
1421                 t.onIncompletePiece(piece)
1422         }
1423 }
1424
1425 func (t *Torrent) onPieceCompleted(piece int) {
1426         t.pendingPieces.Remove(piece)
1427         t.pendAllChunkSpecs(piece)
1428         for conn := range t.conns {
1429                 conn.Have(piece)
1430                 for r := range conn.Requests {
1431                         if int(r.Index) == piece {
1432                                 conn.Cancel(r)
1433                         }
1434                 }
1435                 // Could check here if peer doesn't have piece, but due to caching
1436                 // some peers may have said they have a piece but they don't.
1437                 conn.upload()
1438         }
1439 }
1440
1441 func (t *Torrent) onIncompletePiece(piece int) {
1442         if t.pieceAllDirty(piece) {
1443                 t.pendAllChunkSpecs(piece)
1444         }
1445         if !t.wantPieceIndex(piece) {
1446                 return
1447         }
1448         // We could drop any connections that we told we have a piece that we
1449         // don't here. But there's a test failure, and it seems clients don't care
1450         // if you request pieces that you already claim to have. Pruning bad
1451         // connections might just remove any connections that aren't treating us
1452         // favourably anyway.
1453
1454         // for c := range t.conns {
1455         //      if c.sentHave(piece) {
1456         //              c.Drop()
1457         //      }
1458         // }
1459         for conn := range t.conns {
1460                 if conn.PeerHasPiece(piece) {
1461                         conn.updateRequests()
1462                 }
1463         }
1464 }
1465
1466 func (t *Torrent) verifyPiece(piece int) {
1467         cl := t.cl
1468         cl.mu.Lock()
1469         defer cl.mu.Unlock()
1470         p := &t.pieces[piece]
1471         for p.Hashing || t.storage == nil {
1472                 cl.event.Wait()
1473         }
1474         p.QueuedForHash = false
1475         if t.closed.IsSet() || t.pieceComplete(piece) {
1476                 t.updatePiecePriority(piece)
1477                 return
1478         }
1479         p.Hashing = true
1480         t.publishPieceChange(piece)
1481         cl.mu.Unlock()
1482         sum := t.hashPiece(piece)
1483         cl.mu.Lock()
1484         p.Hashing = false
1485         t.pieceHashed(piece, sum == p.Hash)
1486 }
1487
1488 // Return the connections that touched a piece, and clear the entry while
1489 // doing it.
1490 func (t *Torrent) reapPieceTouchers(piece int) (ret []*connection) {
1491         for c := range t.conns {
1492                 if _, ok := c.peerTouchedPieces[piece]; ok {
1493                         ret = append(ret, c)
1494                         delete(c.peerTouchedPieces, piece)
1495                 }
1496         }
1497         return
1498 }
1499
1500 func (t *Torrent) connsAsSlice() (ret []*connection) {
1501         for c := range t.conns {
1502                 ret = append(ret, c)
1503         }
1504         return
1505 }
1506
1507 // Currently doesn't really queue, but should in the future.
1508 func (t *Torrent) queuePieceCheck(pieceIndex int) {
1509         piece := &t.pieces[pieceIndex]
1510         if piece.QueuedForHash {
1511                 return
1512         }
1513         piece.QueuedForHash = true
1514         t.publishPieceChange(pieceIndex)
1515         go t.verifyPiece(pieceIndex)
1516 }