]> Sergey Matveev's repositories - btrtrc.git/blob - torrent.go
Implement smart ban using generics
[btrtrc.git] / torrent.go
1 package torrent
2
3 import (
4         "bytes"
5         "container/heap"
6         "context"
7         "crypto/sha1"
8         "errors"
9         "fmt"
10         "io"
11         "net/url"
12         "sort"
13         "strings"
14         "text/tabwriter"
15         "time"
16         "unsafe"
17
18         "github.com/RoaringBitmap/roaring"
19         "github.com/anacrolix/chansync"
20         "github.com/anacrolix/chansync/events"
21         "github.com/anacrolix/dht/v2"
22         "github.com/anacrolix/log"
23         "github.com/anacrolix/missinggo/perf"
24         "github.com/anacrolix/missinggo/pubsub"
25         "github.com/anacrolix/missinggo/slices"
26         "github.com/anacrolix/missinggo/v2"
27         "github.com/anacrolix/missinggo/v2/bitmap"
28         "github.com/anacrolix/multiless"
29         "github.com/anacrolix/sync"
30         request_strategy "github.com/anacrolix/torrent/request-strategy"
31         "github.com/davecgh/go-spew/spew"
32         "github.com/pion/datachannel"
33
34         "github.com/anacrolix/torrent/bencode"
35         "github.com/anacrolix/torrent/common"
36         "github.com/anacrolix/torrent/metainfo"
37         pp "github.com/anacrolix/torrent/peer_protocol"
38         "github.com/anacrolix/torrent/segments"
39         "github.com/anacrolix/torrent/storage"
40         "github.com/anacrolix/torrent/tracker"
41         "github.com/anacrolix/torrent/webseed"
42         "github.com/anacrolix/torrent/webtorrent"
43 )
44
45 // Maintains state of torrent within a Client. Many methods should not be called before the info is
46 // available, see .Info and .GotInfo.
47 type Torrent struct {
48         // Torrent-level aggregate statistics. First in struct to ensure 64-bit
49         // alignment. See #262.
50         stats  ConnStats
51         cl     *Client
52         logger log.Logger
53
54         networkingEnabled      chansync.Flag
55         dataDownloadDisallowed chansync.Flag
56         dataUploadDisallowed   bool
57         userOnWriteChunkErr    func(error)
58
59         closed   chansync.SetOnce
60         infoHash metainfo.Hash
61         pieces   []Piece
62         // Values are the piece indices that changed.
63         pieceStateChanges *pubsub.PubSub
64         // The size of chunks to request from peers over the wire. This is
65         // normally 16KiB by convention these days.
66         chunkSize pp.Integer
67         chunkPool sync.Pool
68         // Total length of the torrent in bytes. Stored because it's not O(1) to
69         // get this from the info dict.
70         length *int64
71
72         // The storage to open when the info dict becomes available.
73         storageOpener *storage.Client
74         // Storage for torrent data.
75         storage *storage.Torrent
76         // Read-locked for using storage, and write-locked for Closing.
77         storageLock sync.RWMutex
78
79         // TODO: Only announce stuff is used?
80         metainfo metainfo.MetaInfo
81
82         // The info dict. nil if we don't have it (yet).
83         info      *metainfo.Info
84         fileIndex segments.Index
85         files     *[]*File
86
87         webSeeds map[string]*Peer
88         // Active peer connections, running message stream loops. TODO: Make this
89         // open (not-closed) connections only.
90         conns               map[*PeerConn]struct{}
91         maxEstablishedConns int
92         // Set of addrs to which we're attempting to connect. Connections are
93         // half-open until all handshakes are completed.
94         halfOpen map[string]PeerInfo
95
96         // Reserve of peers to connect to. A peer can be both here and in the
97         // active connections if were told about the peer after connecting with
98         // them. That encourages us to reconnect to peers that are well known in
99         // the swarm.
100         peers prioritizedPeers
101         // Whether we want to know to know more peers.
102         wantPeersEvent missinggo.Event
103         // An announcer for each tracker URL.
104         trackerAnnouncers map[string]torrentTrackerAnnouncer
105         // How many times we've initiated a DHT announce. TODO: Move into stats.
106         numDHTAnnounces int
107
108         // Name used if the info name isn't available. Should be cleared when the
109         // Info does become available.
110         nameMu      sync.RWMutex
111         displayName string
112
113         // The bencoded bytes of the info dict. This is actively manipulated if
114         // the info bytes aren't initially available, and we try to fetch them
115         // from peers.
116         metadataBytes []byte
117         // Each element corresponds to the 16KiB metadata pieces. If true, we have
118         // received that piece.
119         metadataCompletedChunks []bool
120         metadataChanged         sync.Cond
121
122         // Closed when .Info is obtained.
123         gotMetainfoC chan struct{}
124
125         readers                map[*reader]struct{}
126         _readerNowPieces       bitmap.Bitmap
127         _readerReadaheadPieces bitmap.Bitmap
128
129         // A cache of pieces we need to get. Calculated from various piece and
130         // file priorities and completion states elsewhere.
131         _pendingPieces roaring.Bitmap
132         // A cache of completed piece indices.
133         _completedPieces roaring.Bitmap
134         // Pieces that need to be hashed.
135         piecesQueuedForHash       bitmap.Bitmap
136         activePieceHashes         int
137         initialPieceCheckDisabled bool
138
139         connsWithAllPieces map[*Peer]struct{}
140         // Count of each request across active connections.
141         pendingRequests map[RequestIndex]*Peer
142         lastRequested   map[RequestIndex]time.Time
143         // Chunks we've written to since the corresponding piece was last checked.
144         dirtyChunks roaring.Bitmap
145
146         pex pexState
147
148         // Is On when all pieces are complete.
149         Complete chansync.Flag
150
151         smartBanCache smartBanCache
152 }
153
154 func (t *Torrent) selectivePieceAvailabilityFromPeers(i pieceIndex) (count int) {
155         // This could be done with roaring.BitSliceIndexing.
156         t.iterPeers(func(peer *Peer) {
157                 if _, ok := t.connsWithAllPieces[peer]; ok {
158                         return
159                 }
160                 if peer.peerHasPiece(i) {
161                         count++
162                 }
163         })
164         return
165 }
166
167 func (t *Torrent) decPieceAvailability(i pieceIndex) {
168         if !t.haveInfo() {
169                 return
170         }
171         p := t.piece(i)
172         if p.relativeAvailability <= 0 {
173                 panic(p.relativeAvailability)
174         }
175         p.relativeAvailability--
176         t.updatePieceRequestOrder(i)
177 }
178
179 func (t *Torrent) incPieceAvailability(i pieceIndex) {
180         // If we don't the info, this should be reconciled when we do.
181         if t.haveInfo() {
182                 p := t.piece(i)
183                 p.relativeAvailability++
184                 t.updatePieceRequestOrder(i)
185         }
186 }
187
188 func (t *Torrent) readerNowPieces() bitmap.Bitmap {
189         return t._readerNowPieces
190 }
191
192 func (t *Torrent) readerReadaheadPieces() bitmap.Bitmap {
193         return t._readerReadaheadPieces
194 }
195
196 func (t *Torrent) ignorePieceForRequests(i pieceIndex) bool {
197         return !t.wantPieceIndex(i)
198 }
199
200 // Returns a channel that is closed when the Torrent is closed.
201 func (t *Torrent) Closed() events.Done {
202         return t.closed.Done()
203 }
204
205 // KnownSwarm returns the known subset of the peers in the Torrent's swarm, including active,
206 // pending, and half-open peers.
207 func (t *Torrent) KnownSwarm() (ks []PeerInfo) {
208         // Add pending peers to the list
209         t.peers.Each(func(peer PeerInfo) {
210                 ks = append(ks, peer)
211         })
212
213         // Add half-open peers to the list
214         for _, peer := range t.halfOpen {
215                 ks = append(ks, peer)
216         }
217
218         // Add active peers to the list
219         for conn := range t.conns {
220                 ks = append(ks, PeerInfo{
221                         Id:     conn.PeerID,
222                         Addr:   conn.RemoteAddr,
223                         Source: conn.Discovery,
224                         // > If the connection is encrypted, that's certainly enough to set SupportsEncryption.
225                         // > But if we're not connected to them with an encrypted connection, I couldn't say
226                         // > what's appropriate. We can carry forward the SupportsEncryption value as we
227                         // > received it from trackers/DHT/PEX, or just use the encryption state for the
228                         // > connection. It's probably easiest to do the latter for now.
229                         // https://github.com/anacrolix/torrent/pull/188
230                         SupportsEncryption: conn.headerEncrypted,
231                 })
232         }
233
234         return
235 }
236
237 func (t *Torrent) setChunkSize(size pp.Integer) {
238         t.chunkSize = size
239         t.chunkPool = sync.Pool{
240                 New: func() interface{} {
241                         b := make([]byte, size)
242                         return &b
243                 },
244         }
245 }
246
247 func (t *Torrent) pieceComplete(piece pieceIndex) bool {
248         return t._completedPieces.Contains(bitmap.BitIndex(piece))
249 }
250
251 func (t *Torrent) pieceCompleteUncached(piece pieceIndex) storage.Completion {
252         if t.storage == nil {
253                 return storage.Completion{Complete: false, Ok: true}
254         }
255         return t.pieces[piece].Storage().Completion()
256 }
257
258 // There's a connection to that address already.
259 func (t *Torrent) addrActive(addr string) bool {
260         if _, ok := t.halfOpen[addr]; ok {
261                 return true
262         }
263         for c := range t.conns {
264                 ra := c.RemoteAddr
265                 if ra.String() == addr {
266                         return true
267                 }
268         }
269         return false
270 }
271
272 func (t *Torrent) appendUnclosedConns(ret []*PeerConn) []*PeerConn {
273         return t.appendConns(ret, func(conn *PeerConn) bool {
274                 return !conn.closed.IsSet()
275         })
276 }
277
278 func (t *Torrent) appendConns(ret []*PeerConn, f func(*PeerConn) bool) []*PeerConn {
279         for c := range t.conns {
280                 if f(c) {
281                         ret = append(ret, c)
282                 }
283         }
284         return ret
285 }
286
287 func (t *Torrent) addPeer(p PeerInfo) (added bool) {
288         cl := t.cl
289         torrent.Add(fmt.Sprintf("peers added by source %q", p.Source), 1)
290         if t.closed.IsSet() {
291                 return false
292         }
293         if ipAddr, ok := tryIpPortFromNetAddr(p.Addr); ok {
294                 if cl.badPeerIPPort(ipAddr.IP, ipAddr.Port) {
295                         torrent.Add("peers not added because of bad addr", 1)
296                         // cl.logger.Printf("peers not added because of bad addr: %v", p)
297                         return false
298                 }
299         }
300         if replaced, ok := t.peers.AddReturningReplacedPeer(p); ok {
301                 torrent.Add("peers replaced", 1)
302                 if !replaced.equal(p) {
303                         t.logger.WithDefaultLevel(log.Debug).Printf("added %v replacing %v", p, replaced)
304                         added = true
305                 }
306         } else {
307                 added = true
308         }
309         t.openNewConns()
310         for t.peers.Len() > cl.config.TorrentPeersHighWater {
311                 _, ok := t.peers.DeleteMin()
312                 if ok {
313                         torrent.Add("excess reserve peers discarded", 1)
314                 }
315         }
316         return
317 }
318
319 func (t *Torrent) invalidateMetadata() {
320         for i := 0; i < len(t.metadataCompletedChunks); i++ {
321                 t.metadataCompletedChunks[i] = false
322         }
323         t.nameMu.Lock()
324         t.gotMetainfoC = make(chan struct{})
325         t.info = nil
326         t.nameMu.Unlock()
327 }
328
329 func (t *Torrent) saveMetadataPiece(index int, data []byte) {
330         if t.haveInfo() {
331                 return
332         }
333         if index >= len(t.metadataCompletedChunks) {
334                 t.logger.Printf("%s: ignoring metadata piece %d", t, index)
335                 return
336         }
337         copy(t.metadataBytes[(1<<14)*index:], data)
338         t.metadataCompletedChunks[index] = true
339 }
340
341 func (t *Torrent) metadataPieceCount() int {
342         return (len(t.metadataBytes) + (1 << 14) - 1) / (1 << 14)
343 }
344
345 func (t *Torrent) haveMetadataPiece(piece int) bool {
346         if t.haveInfo() {
347                 return (1<<14)*piece < len(t.metadataBytes)
348         } else {
349                 return piece < len(t.metadataCompletedChunks) && t.metadataCompletedChunks[piece]
350         }
351 }
352
353 func (t *Torrent) metadataSize() int {
354         return len(t.metadataBytes)
355 }
356
357 func infoPieceHashes(info *metainfo.Info) (ret [][]byte) {
358         for i := 0; i < len(info.Pieces); i += sha1.Size {
359                 ret = append(ret, info.Pieces[i:i+sha1.Size])
360         }
361         return
362 }
363
364 func (t *Torrent) makePieces() {
365         hashes := infoPieceHashes(t.info)
366         t.pieces = make([]Piece, len(hashes))
367         for i, hash := range hashes {
368                 piece := &t.pieces[i]
369                 piece.t = t
370                 piece.index = pieceIndex(i)
371                 piece.noPendingWrites.L = &piece.pendingWritesMutex
372                 piece.hash = (*metainfo.Hash)(unsafe.Pointer(&hash[0]))
373                 files := *t.files
374                 beginFile := pieceFirstFileIndex(piece.torrentBeginOffset(), files)
375                 endFile := pieceEndFileIndex(piece.torrentEndOffset(), files)
376                 piece.files = files[beginFile:endFile]
377                 piece.undirtiedChunksIter = undirtiedChunksIter{
378                         TorrentDirtyChunks: &t.dirtyChunks,
379                         StartRequestIndex:  piece.requestIndexOffset(),
380                         EndRequestIndex:    piece.requestIndexOffset() + piece.numChunks(),
381                 }
382         }
383 }
384
385 // Returns the index of the first file containing the piece. files must be
386 // ordered by offset.
387 func pieceFirstFileIndex(pieceOffset int64, files []*File) int {
388         for i, f := range files {
389                 if f.offset+f.length > pieceOffset {
390                         return i
391                 }
392         }
393         return 0
394 }
395
396 // Returns the index after the last file containing the piece. files must be
397 // ordered by offset.
398 func pieceEndFileIndex(pieceEndOffset int64, files []*File) int {
399         for i, f := range files {
400                 if f.offset+f.length >= pieceEndOffset {
401                         return i + 1
402                 }
403         }
404         return 0
405 }
406
407 func (t *Torrent) cacheLength() {
408         var l int64
409         for _, f := range t.info.UpvertedFiles() {
410                 l += f.Length
411         }
412         t.length = &l
413 }
414
415 // TODO: This shouldn't fail for storage reasons. Instead we should handle storage failure
416 // separately.
417 func (t *Torrent) setInfo(info *metainfo.Info) error {
418         if err := validateInfo(info); err != nil {
419                 return fmt.Errorf("bad info: %s", err)
420         }
421         if t.storageOpener != nil {
422                 var err error
423                 t.storage, err = t.storageOpener.OpenTorrent(info, t.infoHash)
424                 if err != nil {
425                         return fmt.Errorf("error opening torrent storage: %s", err)
426                 }
427         }
428         t.nameMu.Lock()
429         t.info = info
430         t.nameMu.Unlock()
431         t.updateComplete()
432         t.fileIndex = segments.NewIndex(common.LengthIterFromUpvertedFiles(info.UpvertedFiles()))
433         t.displayName = "" // Save a few bytes lol.
434         t.initFiles()
435         t.cacheLength()
436         t.makePieces()
437         return nil
438 }
439
440 func (t *Torrent) pieceRequestOrderKey(i int) request_strategy.PieceRequestOrderKey {
441         return request_strategy.PieceRequestOrderKey{
442                 InfoHash: t.infoHash,
443                 Index:    i,
444         }
445 }
446
447 // This seems to be all the follow-up tasks after info is set, that can't fail.
448 func (t *Torrent) onSetInfo() {
449         t.initPieceRequestOrder()
450         for i := range t.pieces {
451                 p := &t.pieces[i]
452                 // Need to add relativeAvailability before updating piece completion, as that may result in conns
453                 // being dropped.
454                 if p.relativeAvailability != 0 {
455                         panic(p.relativeAvailability)
456                 }
457                 p.relativeAvailability = t.selectivePieceAvailabilityFromPeers(i)
458                 t.addRequestOrderPiece(i)
459                 t.updatePieceCompletion(pieceIndex(i))
460                 if !t.initialPieceCheckDisabled && !p.storageCompletionOk {
461                         // t.logger.Printf("piece %s completion unknown, queueing check", p)
462                         t.queuePieceCheck(pieceIndex(i))
463                 }
464         }
465         t.cl.event.Broadcast()
466         close(t.gotMetainfoC)
467         t.updateWantPeersEvent()
468         t.pendingRequests = make(map[RequestIndex]*Peer)
469         t.lastRequested = make(map[RequestIndex]time.Time)
470         t.tryCreateMorePieceHashers()
471         t.iterPeers(func(p *Peer) {
472                 p.onGotInfo(t.info)
473                 p.updateRequests("onSetInfo")
474         })
475 }
476
477 // Called when metadata for a torrent becomes available.
478 func (t *Torrent) setInfoBytesLocked(b []byte) error {
479         if metainfo.HashBytes(b) != t.infoHash {
480                 return errors.New("info bytes have wrong hash")
481         }
482         var info metainfo.Info
483         if err := bencode.Unmarshal(b, &info); err != nil {
484                 return fmt.Errorf("error unmarshalling info bytes: %s", err)
485         }
486         t.metadataBytes = b
487         t.metadataCompletedChunks = nil
488         if t.info != nil {
489                 return nil
490         }
491         if err := t.setInfo(&info); err != nil {
492                 return err
493         }
494         t.onSetInfo()
495         return nil
496 }
497
498 func (t *Torrent) haveAllMetadataPieces() bool {
499         if t.haveInfo() {
500                 return true
501         }
502         if t.metadataCompletedChunks == nil {
503                 return false
504         }
505         for _, have := range t.metadataCompletedChunks {
506                 if !have {
507                         return false
508                 }
509         }
510         return true
511 }
512
513 // TODO: Propagate errors to disconnect peer.
514 func (t *Torrent) setMetadataSize(size int) (err error) {
515         if t.haveInfo() {
516                 // We already know the correct metadata size.
517                 return
518         }
519         if uint32(size) > maxMetadataSize {
520                 return errors.New("bad size")
521         }
522         if len(t.metadataBytes) == size {
523                 return
524         }
525         t.metadataBytes = make([]byte, size)
526         t.metadataCompletedChunks = make([]bool, (size+(1<<14)-1)/(1<<14))
527         t.metadataChanged.Broadcast()
528         for c := range t.conns {
529                 c.requestPendingMetadata()
530         }
531         return
532 }
533
534 // The current working name for the torrent. Either the name in the info dict,
535 // or a display name given such as by the dn value in a magnet link, or "".
536 func (t *Torrent) name() string {
537         t.nameMu.RLock()
538         defer t.nameMu.RUnlock()
539         if t.haveInfo() {
540                 return t.info.Name
541         }
542         if t.displayName != "" {
543                 return t.displayName
544         }
545         return "infohash:" + t.infoHash.HexString()
546 }
547
548 func (t *Torrent) pieceState(index pieceIndex) (ret PieceState) {
549         p := &t.pieces[index]
550         ret.Priority = t.piecePriority(index)
551         ret.Completion = p.completion()
552         ret.QueuedForHash = p.queuedForHash()
553         ret.Hashing = p.hashing
554         ret.Checking = ret.QueuedForHash || ret.Hashing
555         ret.Marking = p.marking
556         if !ret.Complete && t.piecePartiallyDownloaded(index) {
557                 ret.Partial = true
558         }
559         return
560 }
561
562 func (t *Torrent) metadataPieceSize(piece int) int {
563         return metadataPieceSize(len(t.metadataBytes), piece)
564 }
565
566 func (t *Torrent) newMetadataExtensionMessage(c *PeerConn, msgType pp.ExtendedMetadataRequestMsgType, piece int, data []byte) pp.Message {
567         return pp.Message{
568                 Type:       pp.Extended,
569                 ExtendedID: c.PeerExtensionIDs[pp.ExtensionNameMetadata],
570                 ExtendedPayload: append(bencode.MustMarshal(pp.ExtendedMetadataRequestMsg{
571                         Piece:     piece,
572                         TotalSize: len(t.metadataBytes),
573                         Type:      msgType,
574                 }), data...),
575         }
576 }
577
578 type pieceAvailabilityRun struct {
579         Count        pieceIndex
580         Availability int
581 }
582
583 func (me pieceAvailabilityRun) String() string {
584         return fmt.Sprintf("%v(%v)", me.Count, me.Availability)
585 }
586
587 func (t *Torrent) pieceAvailabilityRuns() (ret []pieceAvailabilityRun) {
588         rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
589                 ret = append(ret, pieceAvailabilityRun{Availability: el.(int), Count: int(count)})
590         })
591         for i := range t.pieces {
592                 rle.Append(t.pieces[i].availability(), 1)
593         }
594         rle.Flush()
595         return
596 }
597
598 func (t *Torrent) pieceAvailabilityFrequencies() (freqs []int) {
599         freqs = make([]int, t.numActivePeers()+1)
600         for i := range t.pieces {
601                 freqs[t.piece(i).availability()]++
602         }
603         return
604 }
605
606 func (t *Torrent) pieceStateRuns() (ret PieceStateRuns) {
607         rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
608                 ret = append(ret, PieceStateRun{
609                         PieceState: el.(PieceState),
610                         Length:     int(count),
611                 })
612         })
613         for index := range t.pieces {
614                 rle.Append(t.pieceState(pieceIndex(index)), 1)
615         }
616         rle.Flush()
617         return
618 }
619
620 // Produces a small string representing a PieceStateRun.
621 func (psr PieceStateRun) String() (ret string) {
622         ret = fmt.Sprintf("%d", psr.Length)
623         ret += func() string {
624                 switch psr.Priority {
625                 case PiecePriorityNext:
626                         return "N"
627                 case PiecePriorityNormal:
628                         return "."
629                 case PiecePriorityReadahead:
630                         return "R"
631                 case PiecePriorityNow:
632                         return "!"
633                 case PiecePriorityHigh:
634                         return "H"
635                 default:
636                         return ""
637                 }
638         }()
639         if psr.Hashing {
640                 ret += "H"
641         }
642         if psr.QueuedForHash {
643                 ret += "Q"
644         }
645         if psr.Marking {
646                 ret += "M"
647         }
648         if psr.Partial {
649                 ret += "P"
650         }
651         if psr.Complete {
652                 ret += "C"
653         }
654         if !psr.Ok {
655                 ret += "?"
656         }
657         return
658 }
659
660 func (t *Torrent) writeStatus(w io.Writer) {
661         fmt.Fprintf(w, "Infohash: %s\n", t.infoHash.HexString())
662         fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
663         if !t.haveInfo() {
664                 fmt.Fprintf(w, "Metadata have: ")
665                 for _, h := range t.metadataCompletedChunks {
666                         fmt.Fprintf(w, "%c", func() rune {
667                                 if h {
668                                         return 'H'
669                                 } else {
670                                         return '.'
671                                 }
672                         }())
673                 }
674                 fmt.Fprintln(w)
675         }
676         fmt.Fprintf(w, "Piece length: %s\n",
677                 func() string {
678                         if t.haveInfo() {
679                                 return fmt.Sprintf("%v (%v chunks)",
680                                         t.usualPieceSize(),
681                                         float64(t.usualPieceSize())/float64(t.chunkSize))
682                         } else {
683                                 return "no info"
684                         }
685                 }(),
686         )
687         if t.info != nil {
688                 fmt.Fprintf(w, "Num Pieces: %d (%d completed)\n", t.numPieces(), t.numPiecesCompleted())
689                 fmt.Fprintf(w, "Piece States: %s\n", t.pieceStateRuns())
690                 // Generates a huge, unhelpful listing when piece availability is very scattered. Prefer
691                 // availability frequencies instead.
692                 if false {
693                         fmt.Fprintf(w, "Piece availability: %v\n", strings.Join(func() (ret []string) {
694                                 for _, run := range t.pieceAvailabilityRuns() {
695                                         ret = append(ret, run.String())
696                                 }
697                                 return
698                         }(), " "))
699                 }
700                 fmt.Fprintf(w, "Piece availability frequency: %v\n", strings.Join(
701                         func() (ret []string) {
702                                 for avail, freq := range t.pieceAvailabilityFrequencies() {
703                                         if freq == 0 {
704                                                 continue
705                                         }
706                                         ret = append(ret, fmt.Sprintf("%v: %v", avail, freq))
707                                 }
708                                 return
709                         }(),
710                         ", "))
711         }
712         fmt.Fprintf(w, "Reader Pieces:")
713         t.forReaderOffsetPieces(func(begin, end pieceIndex) (again bool) {
714                 fmt.Fprintf(w, " %d:%d", begin, end)
715                 return true
716         })
717         fmt.Fprintln(w)
718
719         fmt.Fprintf(w, "Enabled trackers:\n")
720         func() {
721                 tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
722                 fmt.Fprintf(tw, "    URL\tExtra\n")
723                 for _, ta := range slices.Sort(slices.FromMapElems(t.trackerAnnouncers), func(l, r torrentTrackerAnnouncer) bool {
724                         lu := l.URL()
725                         ru := r.URL()
726                         var luns, runs url.URL = *lu, *ru
727                         luns.Scheme = ""
728                         runs.Scheme = ""
729                         var ml missinggo.MultiLess
730                         ml.StrictNext(luns.String() == runs.String(), luns.String() < runs.String())
731                         ml.StrictNext(lu.String() == ru.String(), lu.String() < ru.String())
732                         return ml.Less()
733                 }).([]torrentTrackerAnnouncer) {
734                         fmt.Fprintf(tw, "    %q\t%v\n", ta.URL(), ta.statusLine())
735                 }
736                 tw.Flush()
737         }()
738
739         fmt.Fprintf(w, "DHT Announces: %d\n", t.numDHTAnnounces)
740
741         spew.NewDefaultConfig()
742         spew.Fdump(w, t.statsLocked())
743
744         peers := t.peersAsSlice()
745         sort.Slice(peers, func(_i, _j int) bool {
746                 i := peers[_i]
747                 j := peers[_j]
748                 if less, ok := multiless.New().EagerSameLess(
749                         i.downloadRate() == j.downloadRate(), i.downloadRate() < j.downloadRate(),
750                 ).LessOk(); ok {
751                         return less
752                 }
753                 return worseConn(i, j)
754         })
755         for i, c := range peers {
756                 fmt.Fprintf(w, "%2d. ", i+1)
757                 c.writeStatus(w, t)
758         }
759 }
760
761 func (t *Torrent) haveInfo() bool {
762         return t.info != nil
763 }
764
765 // Returns a run-time generated MetaInfo that includes the info bytes and
766 // announce-list as currently known to the client.
767 func (t *Torrent) newMetaInfo() metainfo.MetaInfo {
768         return metainfo.MetaInfo{
769                 CreationDate: time.Now().Unix(),
770                 Comment:      "dynamic metainfo from client",
771                 CreatedBy:    "go.torrent",
772                 AnnounceList: t.metainfo.UpvertedAnnounceList().Clone(),
773                 InfoBytes: func() []byte {
774                         if t.haveInfo() {
775                                 return t.metadataBytes
776                         } else {
777                                 return nil
778                         }
779                 }(),
780                 UrlList: func() []string {
781                         ret := make([]string, 0, len(t.webSeeds))
782                         for url := range t.webSeeds {
783                                 ret = append(ret, url)
784                         }
785                         return ret
786                 }(),
787         }
788 }
789
790 // Get bytes left
791 func (t *Torrent) BytesMissing() (n int64) {
792         t.cl.rLock()
793         n = t.bytesMissingLocked()
794         t.cl.rUnlock()
795         return
796 }
797
798 func (t *Torrent) bytesMissingLocked() int64 {
799         return t.bytesLeft()
800 }
801
802 func iterFlipped(b *roaring.Bitmap, end uint64, cb func(uint32) bool) {
803         roaring.Flip(b, 0, end).Iterate(cb)
804 }
805
806 func (t *Torrent) bytesLeft() (left int64) {
807         iterFlipped(&t._completedPieces, uint64(t.numPieces()), func(x uint32) bool {
808                 p := t.piece(pieceIndex(x))
809                 left += int64(p.length() - p.numDirtyBytes())
810                 return true
811         })
812         return
813 }
814
815 // Bytes left to give in tracker announces.
816 func (t *Torrent) bytesLeftAnnounce() int64 {
817         if t.haveInfo() {
818                 return t.bytesLeft()
819         } else {
820                 return -1
821         }
822 }
823
824 func (t *Torrent) piecePartiallyDownloaded(piece pieceIndex) bool {
825         if t.pieceComplete(piece) {
826                 return false
827         }
828         if t.pieceAllDirty(piece) {
829                 return false
830         }
831         return t.pieces[piece].hasDirtyChunks()
832 }
833
834 func (t *Torrent) usualPieceSize() int {
835         return int(t.info.PieceLength)
836 }
837
838 func (t *Torrent) numPieces() pieceIndex {
839         return pieceIndex(t.info.NumPieces())
840 }
841
842 func (t *Torrent) numPiecesCompleted() (num pieceIndex) {
843         return pieceIndex(t._completedPieces.GetCardinality())
844 }
845
846 func (t *Torrent) close(wg *sync.WaitGroup) (err error) {
847         t.closed.Set()
848         if t.storage != nil {
849                 wg.Add(1)
850                 go func() {
851                         defer wg.Done()
852                         t.storageLock.Lock()
853                         defer t.storageLock.Unlock()
854                         if f := t.storage.Close; f != nil {
855                                 err1 := f()
856                                 if err1 != nil {
857                                         t.logger.WithDefaultLevel(log.Warning).Printf("error closing storage: %v", err1)
858                                 }
859                         }
860                 }()
861         }
862         t.iterPeers(func(p *Peer) {
863                 p.close()
864         })
865         if t.storage != nil {
866                 t.deletePieceRequestOrder()
867         }
868         for i := range t.pieces {
869                 p := t.piece(i)
870                 if p.relativeAvailability != 0 {
871                         panic(fmt.Sprintf("piece %v has relative availability %v", i, p.relativeAvailability))
872                 }
873         }
874         t.pex.Reset()
875         t.cl.event.Broadcast()
876         t.pieceStateChanges.Close()
877         t.updateWantPeersEvent()
878         return
879 }
880
881 func (t *Torrent) requestOffset(r Request) int64 {
882         return torrentRequestOffset(*t.length, int64(t.usualPieceSize()), r)
883 }
884
885 // Return the request that would include the given offset into the torrent data. Returns !ok if
886 // there is no such request.
887 func (t *Torrent) offsetRequest(off int64) (req Request, ok bool) {
888         return torrentOffsetRequest(*t.length, t.info.PieceLength, int64(t.chunkSize), off)
889 }
890
891 func (t *Torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
892         defer perf.ScopeTimerErr(&err)()
893         n, err := t.pieces[piece].Storage().WriteAt(data, begin)
894         if err == nil && n != len(data) {
895                 err = io.ErrShortWrite
896         }
897         return err
898 }
899
900 func (t *Torrent) bitfield() (bf []bool) {
901         bf = make([]bool, t.numPieces())
902         t._completedPieces.Iterate(func(piece uint32) (again bool) {
903                 bf[piece] = true
904                 return true
905         })
906         return
907 }
908
909 func (t *Torrent) pieceNumChunks(piece pieceIndex) chunkIndexType {
910         return chunkIndexType((t.pieceLength(piece) + t.chunkSize - 1) / t.chunkSize)
911 }
912
913 func (t *Torrent) chunksPerRegularPiece() uint32 {
914         return uint32((pp.Integer(t.usualPieceSize()) + t.chunkSize - 1) / t.chunkSize)
915 }
916
917 func (t *Torrent) numRequests() RequestIndex {
918         if t.numPieces() == 0 {
919                 return 0
920         }
921         return uint32(t.numPieces()-1)*t.chunksPerRegularPiece() + t.pieceNumChunks(t.numPieces()-1)
922 }
923
924 func (t *Torrent) pendAllChunkSpecs(pieceIndex pieceIndex) {
925         t.dirtyChunks.RemoveRange(
926                 uint64(t.pieceRequestIndexOffset(pieceIndex)),
927                 uint64(t.pieceRequestIndexOffset(pieceIndex+1)))
928 }
929
930 func (t *Torrent) pieceLength(piece pieceIndex) pp.Integer {
931         if t.info.PieceLength == 0 {
932                 // There will be no variance amongst pieces. Only pain.
933                 return 0
934         }
935         if piece == t.numPieces()-1 {
936                 ret := pp.Integer(*t.length % t.info.PieceLength)
937                 if ret != 0 {
938                         return ret
939                 }
940         }
941         return pp.Integer(t.info.PieceLength)
942 }
943
944 func (t *Torrent) smartBanBlockCheckingWriter(piece pieceIndex) *blockCheckingWriter {
945         return &blockCheckingWriter{
946                 cache:        &t.smartBanCache,
947                 requestIndex: t.pieceRequestIndexOffset(piece),
948                 chunkSize:    t.chunkSize.Int(),
949         }
950 }
951
952 func (t *Torrent) hashPiece(piece pieceIndex) (
953         ret metainfo.Hash,
954         // These are peers that sent us blocks that differ from what we hash here.
955         differingPeers map[banPrefix]struct{},
956         err error,
957 ) {
958         p := t.piece(piece)
959         p.waitNoPendingWrites()
960         storagePiece := t.pieces[piece].Storage()
961
962         // Does the backend want to do its own hashing?
963         if i, ok := storagePiece.PieceImpl.(storage.SelfHashing); ok {
964                 var sum metainfo.Hash
965                 // log.Printf("A piece decided to self-hash: %d", piece)
966                 sum, err = i.SelfHash()
967                 missinggo.CopyExact(&ret, sum)
968                 return
969         }
970
971         hash := pieceHash.New()
972         const logPieceContents = false
973         smartBanWriter := t.smartBanBlockCheckingWriter(piece)
974         writers := []io.Writer{hash, smartBanWriter}
975         var examineBuf bytes.Buffer
976         if logPieceContents {
977                 writers = append(writers, &examineBuf)
978         }
979         _, err = storagePiece.WriteTo(io.MultiWriter(writers...))
980         if logPieceContents {
981                 log.Printf("hashed %q with copy err %v", examineBuf.Bytes(), err)
982         }
983         smartBanWriter.Flush()
984         differingPeers = smartBanWriter.badPeers
985         missinggo.CopyExact(&ret, hash.Sum(nil))
986         return
987 }
988
989 func (t *Torrent) haveAnyPieces() bool {
990         return !t._completedPieces.IsEmpty()
991 }
992
993 func (t *Torrent) haveAllPieces() bool {
994         if !t.haveInfo() {
995                 return false
996         }
997         return t._completedPieces.GetCardinality() == bitmap.BitRange(t.numPieces())
998 }
999
1000 func (t *Torrent) havePiece(index pieceIndex) bool {
1001         return t.haveInfo() && t.pieceComplete(index)
1002 }
1003
1004 func (t *Torrent) maybeDropMutuallyCompletePeer(
1005         // I'm not sure about taking peer here, not all peer implementations actually drop. Maybe that's
1006         // okay?
1007         p *Peer,
1008 ) {
1009         if !t.cl.config.DropMutuallyCompletePeers {
1010                 return
1011         }
1012         if !t.haveAllPieces() {
1013                 return
1014         }
1015         if all, known := p.peerHasAllPieces(); !(known && all) {
1016                 return
1017         }
1018         if p.useful() {
1019                 return
1020         }
1021         t.logger.WithDefaultLevel(log.Debug).Printf("dropping %v, which is mutually complete", p)
1022         p.drop()
1023 }
1024
1025 func (t *Torrent) haveChunk(r Request) (ret bool) {
1026         // defer func() {
1027         //      log.Println("have chunk", r, ret)
1028         // }()
1029         if !t.haveInfo() {
1030                 return false
1031         }
1032         if t.pieceComplete(pieceIndex(r.Index)) {
1033                 return true
1034         }
1035         p := &t.pieces[r.Index]
1036         return !p.pendingChunk(r.ChunkSpec, t.chunkSize)
1037 }
1038
1039 func chunkIndexFromChunkSpec(cs ChunkSpec, chunkSize pp.Integer) chunkIndexType {
1040         return chunkIndexType(cs.Begin / chunkSize)
1041 }
1042
1043 func (t *Torrent) wantPieceIndex(index pieceIndex) bool {
1044         return t._pendingPieces.Contains(uint32(index))
1045 }
1046
1047 // A pool of []*PeerConn, to reduce allocations in functions that need to index or sort Torrent
1048 // conns (which is a map).
1049 var peerConnSlices sync.Pool
1050
1051 func getPeerConnSlice(cap int) []*PeerConn {
1052         getInterface := peerConnSlices.Get()
1053         if getInterface == nil {
1054                 return make([]*PeerConn, 0, cap)
1055         } else {
1056                 return getInterface.([]*PeerConn)[:0]
1057         }
1058 }
1059
1060 // The worst connection is one that hasn't been sent, or sent anything useful for the longest. A bad
1061 // connection is one that usually sends us unwanted pieces, or has been in the worse half of the
1062 // established connections for more than a minute. This is O(n log n). If there was a way to not
1063 // consider the position of a conn relative to the total number, it could be reduced to O(n).
1064 func (t *Torrent) worstBadConn() (ret *PeerConn) {
1065         wcs := worseConnSlice{conns: t.appendUnclosedConns(getPeerConnSlice(len(t.conns)))}
1066         defer peerConnSlices.Put(wcs.conns)
1067         wcs.initKeys()
1068         heap.Init(&wcs)
1069         for wcs.Len() != 0 {
1070                 c := heap.Pop(&wcs).(*PeerConn)
1071                 if c._stats.ChunksReadWasted.Int64() >= 6 && c._stats.ChunksReadWasted.Int64() > c._stats.ChunksReadUseful.Int64() {
1072                         return c
1073                 }
1074                 // If the connection is in the worst half of the established
1075                 // connection quota and is older than a minute.
1076                 if wcs.Len() >= (t.maxEstablishedConns+1)/2 {
1077                         // Give connections 1 minute to prove themselves.
1078                         if time.Since(c.completedHandshake) > time.Minute {
1079                                 return c
1080                         }
1081                 }
1082         }
1083         return nil
1084 }
1085
1086 type PieceStateChange struct {
1087         Index int
1088         PieceState
1089 }
1090
1091 func (t *Torrent) publishPieceChange(piece pieceIndex) {
1092         t.cl._mu.Defer(func() {
1093                 cur := t.pieceState(piece)
1094                 p := &t.pieces[piece]
1095                 if cur != p.publicPieceState {
1096                         p.publicPieceState = cur
1097                         t.pieceStateChanges.Publish(PieceStateChange{
1098                                 int(piece),
1099                                 cur,
1100                         })
1101                 }
1102         })
1103 }
1104
1105 func (t *Torrent) pieceNumPendingChunks(piece pieceIndex) pp.Integer {
1106         if t.pieceComplete(piece) {
1107                 return 0
1108         }
1109         return pp.Integer(t.pieceNumChunks(piece) - t.pieces[piece].numDirtyChunks())
1110 }
1111
1112 func (t *Torrent) pieceAllDirty(piece pieceIndex) bool {
1113         return t.pieces[piece].allChunksDirty()
1114 }
1115
1116 func (t *Torrent) readersChanged() {
1117         t.updateReaderPieces()
1118         t.updateAllPiecePriorities("Torrent.readersChanged")
1119 }
1120
1121 func (t *Torrent) updateReaderPieces() {
1122         t._readerNowPieces, t._readerReadaheadPieces = t.readerPiecePriorities()
1123 }
1124
1125 func (t *Torrent) readerPosChanged(from, to pieceRange) {
1126         if from == to {
1127                 return
1128         }
1129         t.updateReaderPieces()
1130         // Order the ranges, high and low.
1131         l, h := from, to
1132         if l.begin > h.begin {
1133                 l, h = h, l
1134         }
1135         if l.end < h.begin {
1136                 // Two distinct ranges.
1137                 t.updatePiecePriorities(l.begin, l.end, "Torrent.readerPosChanged")
1138                 t.updatePiecePriorities(h.begin, h.end, "Torrent.readerPosChanged")
1139         } else {
1140                 // Ranges overlap.
1141                 end := l.end
1142                 if h.end > end {
1143                         end = h.end
1144                 }
1145                 t.updatePiecePriorities(l.begin, end, "Torrent.readerPosChanged")
1146         }
1147 }
1148
1149 func (t *Torrent) maybeNewConns() {
1150         // Tickle the accept routine.
1151         t.cl.event.Broadcast()
1152         t.openNewConns()
1153 }
1154
1155 func (t *Torrent) piecePriorityChanged(piece pieceIndex, reason string) {
1156         if t._pendingPieces.Contains(uint32(piece)) {
1157                 t.iterPeers(func(c *Peer) {
1158                         // if c.requestState.Interested {
1159                         //      return
1160                         // }
1161                         if !c.isLowOnRequests() {
1162                                 return
1163                         }
1164                         if !c.peerHasPiece(piece) {
1165                                 return
1166                         }
1167                         if c.requestState.Interested && c.peerChoking && !c.peerAllowedFast.Contains(uint32(piece)) {
1168                                 return
1169                         }
1170                         c.updateRequests(reason)
1171                 })
1172         }
1173         t.maybeNewConns()
1174         t.publishPieceChange(piece)
1175 }
1176
1177 func (t *Torrent) updatePiecePriority(piece pieceIndex, reason string) {
1178         if !t.closed.IsSet() {
1179                 // It would be possible to filter on pure-priority changes here to avoid churning the piece
1180                 // request order.
1181                 t.updatePieceRequestOrder(piece)
1182         }
1183         p := &t.pieces[piece]
1184         newPrio := p.uncachedPriority()
1185         // t.logger.Printf("torrent %p: piece %d: uncached priority: %v", t, piece, newPrio)
1186         if newPrio == PiecePriorityNone {
1187                 if !t._pendingPieces.CheckedRemove(uint32(piece)) {
1188                         return
1189                 }
1190         } else {
1191                 if !t._pendingPieces.CheckedAdd(uint32(piece)) {
1192                         return
1193                 }
1194         }
1195         t.piecePriorityChanged(piece, reason)
1196 }
1197
1198 func (t *Torrent) updateAllPiecePriorities(reason string) {
1199         t.updatePiecePriorities(0, t.numPieces(), reason)
1200 }
1201
1202 // Update all piece priorities in one hit. This function should have the same
1203 // output as updatePiecePriority, but across all pieces.
1204 func (t *Torrent) updatePiecePriorities(begin, end pieceIndex, reason string) {
1205         for i := begin; i < end; i++ {
1206                 t.updatePiecePriority(i, reason)
1207         }
1208 }
1209
1210 // Returns the range of pieces [begin, end) that contains the extent of bytes.
1211 func (t *Torrent) byteRegionPieces(off, size int64) (begin, end pieceIndex) {
1212         if off >= *t.length {
1213                 return
1214         }
1215         if off < 0 {
1216                 size += off
1217                 off = 0
1218         }
1219         if size <= 0 {
1220                 return
1221         }
1222         begin = pieceIndex(off / t.info.PieceLength)
1223         end = pieceIndex((off + size + t.info.PieceLength - 1) / t.info.PieceLength)
1224         if end > pieceIndex(t.info.NumPieces()) {
1225                 end = pieceIndex(t.info.NumPieces())
1226         }
1227         return
1228 }
1229
1230 // Returns true if all iterations complete without breaking. Returns the read regions for all
1231 // readers. The reader regions should not be merged as some callers depend on this method to
1232 // enumerate readers.
1233 func (t *Torrent) forReaderOffsetPieces(f func(begin, end pieceIndex) (more bool)) (all bool) {
1234         for r := range t.readers {
1235                 p := r.pieces
1236                 if p.begin >= p.end {
1237                         continue
1238                 }
1239                 if !f(p.begin, p.end) {
1240                         return false
1241                 }
1242         }
1243         return true
1244 }
1245
1246 func (t *Torrent) piecePriority(piece pieceIndex) piecePriority {
1247         return t.piece(piece).uncachedPriority()
1248 }
1249
1250 func (t *Torrent) pendRequest(req RequestIndex) {
1251         t.piece(int(req / t.chunksPerRegularPiece())).pendChunkIndex(req % t.chunksPerRegularPiece())
1252 }
1253
1254 func (t *Torrent) pieceCompletionChanged(piece pieceIndex, reason string) {
1255         t.cl.event.Broadcast()
1256         if t.pieceComplete(piece) {
1257                 t.onPieceCompleted(piece)
1258         } else {
1259                 t.onIncompletePiece(piece)
1260         }
1261         t.updatePiecePriority(piece, reason)
1262 }
1263
1264 func (t *Torrent) numReceivedConns() (ret int) {
1265         for c := range t.conns {
1266                 if c.Discovery == PeerSourceIncoming {
1267                         ret++
1268                 }
1269         }
1270         return
1271 }
1272
1273 func (t *Torrent) maxHalfOpen() int {
1274         // Note that if we somehow exceed the maximum established conns, we want
1275         // the negative value to have an effect.
1276         establishedHeadroom := int64(t.maxEstablishedConns - len(t.conns))
1277         extraIncoming := int64(t.numReceivedConns() - t.maxEstablishedConns/2)
1278         // We want to allow some experimentation with new peers, and to try to
1279         // upset an oversupply of received connections.
1280         return int(min(max(5, extraIncoming)+establishedHeadroom, int64(t.cl.config.HalfOpenConnsPerTorrent)))
1281 }
1282
1283 func (t *Torrent) openNewConns() (initiated int) {
1284         defer t.updateWantPeersEvent()
1285         for t.peers.Len() != 0 {
1286                 if !t.wantConns() {
1287                         return
1288                 }
1289                 if len(t.halfOpen) >= t.maxHalfOpen() {
1290                         return
1291                 }
1292                 if len(t.cl.dialers) == 0 {
1293                         return
1294                 }
1295                 if t.cl.numHalfOpen >= t.cl.config.TotalHalfOpenConns {
1296                         return
1297                 }
1298                 p := t.peers.PopMax()
1299                 t.initiateConn(p)
1300                 initiated++
1301         }
1302         return
1303 }
1304
1305 func (t *Torrent) updatePieceCompletion(piece pieceIndex) bool {
1306         p := t.piece(piece)
1307         uncached := t.pieceCompleteUncached(piece)
1308         cached := p.completion()
1309         changed := cached != uncached
1310         complete := uncached.Complete
1311         p.storageCompletionOk = uncached.Ok
1312         x := uint32(piece)
1313         if complete {
1314                 t._completedPieces.Add(x)
1315                 t.openNewConns()
1316         } else {
1317                 t._completedPieces.Remove(x)
1318         }
1319         p.t.updatePieceRequestOrder(piece)
1320         t.updateComplete()
1321         if complete && len(p.dirtiers) != 0 {
1322                 t.logger.Printf("marked piece %v complete but still has dirtiers", piece)
1323         }
1324         if changed {
1325                 log.Fstr("piece %d completion changed: %+v -> %+v", piece, cached, uncached).SetLevel(log.Debug).Log(t.logger)
1326                 t.pieceCompletionChanged(piece, "Torrent.updatePieceCompletion")
1327         }
1328         return changed
1329 }
1330
1331 // Non-blocking read. Client lock is not required.
1332 func (t *Torrent) readAt(b []byte, off int64) (n int, err error) {
1333         for len(b) != 0 {
1334                 p := &t.pieces[off/t.info.PieceLength]
1335                 p.waitNoPendingWrites()
1336                 var n1 int
1337                 n1, err = p.Storage().ReadAt(b, off-p.Info().Offset())
1338                 if n1 == 0 {
1339                         break
1340                 }
1341                 off += int64(n1)
1342                 n += n1
1343                 b = b[n1:]
1344         }
1345         return
1346 }
1347
1348 // Returns an error if the metadata was completed, but couldn't be set for some reason. Blame it on
1349 // the last peer to contribute. TODO: Actually we shouldn't blame peers for failure to open storage
1350 // etc. Also we should probably cached metadata pieces per-Peer, to isolate failure appropriately.
1351 func (t *Torrent) maybeCompleteMetadata() error {
1352         if t.haveInfo() {
1353                 // Nothing to do.
1354                 return nil
1355         }
1356         if !t.haveAllMetadataPieces() {
1357                 // Don't have enough metadata pieces.
1358                 return nil
1359         }
1360         err := t.setInfoBytesLocked(t.metadataBytes)
1361         if err != nil {
1362                 t.invalidateMetadata()
1363                 return fmt.Errorf("error setting info bytes: %s", err)
1364         }
1365         if t.cl.config.Debug {
1366                 t.logger.Printf("%s: got metadata from peers", t)
1367         }
1368         return nil
1369 }
1370
1371 func (t *Torrent) readerPiecePriorities() (now, readahead bitmap.Bitmap) {
1372         t.forReaderOffsetPieces(func(begin, end pieceIndex) bool {
1373                 if end > begin {
1374                         now.Add(bitmap.BitIndex(begin))
1375                         readahead.AddRange(bitmap.BitRange(begin)+1, bitmap.BitRange(end))
1376                 }
1377                 return true
1378         })
1379         return
1380 }
1381
1382 func (t *Torrent) needData() bool {
1383         if t.closed.IsSet() {
1384                 return false
1385         }
1386         if !t.haveInfo() {
1387                 return true
1388         }
1389         return !t._pendingPieces.IsEmpty()
1390 }
1391
1392 func appendMissingStrings(old, new []string) (ret []string) {
1393         ret = old
1394 new:
1395         for _, n := range new {
1396                 for _, o := range old {
1397                         if o == n {
1398                                 continue new
1399                         }
1400                 }
1401                 ret = append(ret, n)
1402         }
1403         return
1404 }
1405
1406 func appendMissingTrackerTiers(existing [][]string, minNumTiers int) (ret [][]string) {
1407         ret = existing
1408         for minNumTiers > len(ret) {
1409                 ret = append(ret, nil)
1410         }
1411         return
1412 }
1413
1414 func (t *Torrent) addTrackers(announceList [][]string) {
1415         fullAnnounceList := &t.metainfo.AnnounceList
1416         t.metainfo.AnnounceList = appendMissingTrackerTiers(*fullAnnounceList, len(announceList))
1417         for tierIndex, trackerURLs := range announceList {
1418                 (*fullAnnounceList)[tierIndex] = appendMissingStrings((*fullAnnounceList)[tierIndex], trackerURLs)
1419         }
1420         t.startMissingTrackerScrapers()
1421         t.updateWantPeersEvent()
1422 }
1423
1424 // Don't call this before the info is available.
1425 func (t *Torrent) bytesCompleted() int64 {
1426         if !t.haveInfo() {
1427                 return 0
1428         }
1429         return *t.length - t.bytesLeft()
1430 }
1431
1432 func (t *Torrent) SetInfoBytes(b []byte) (err error) {
1433         t.cl.lock()
1434         defer t.cl.unlock()
1435         return t.setInfoBytesLocked(b)
1436 }
1437
1438 // Returns true if connection is removed from torrent.Conns.
1439 func (t *Torrent) deletePeerConn(c *PeerConn) (ret bool) {
1440         if !c.closed.IsSet() {
1441                 panic("connection is not closed")
1442                 // There are behaviours prevented by the closed state that will fail
1443                 // if the connection has been deleted.
1444         }
1445         _, ret = t.conns[c]
1446         delete(t.conns, c)
1447         // Avoid adding a drop event more than once. Probably we should track whether we've generated
1448         // the drop event against the PexConnState instead.
1449         if ret {
1450                 if !t.cl.config.DisablePEX {
1451                         t.pex.Drop(c)
1452                 }
1453         }
1454         torrent.Add("deleted connections", 1)
1455         if !c.deleteAllRequests().IsEmpty() {
1456                 t.iterPeers(func(p *Peer) {
1457                         if p.isLowOnRequests() {
1458                                 p.updateRequests("Torrent.deletePeerConn")
1459                         }
1460                 })
1461         }
1462         t.assertPendingRequests()
1463         if t.numActivePeers() == 0 && len(t.connsWithAllPieces) != 0 {
1464                 panic(t.connsWithAllPieces)
1465         }
1466         return
1467 }
1468
1469 func (t *Torrent) decPeerPieceAvailability(p *Peer) {
1470         if t.deleteConnWithAllPieces(p) {
1471                 return
1472         }
1473         if !t.haveInfo() {
1474                 return
1475         }
1476         p.peerPieces().Iterate(func(i uint32) bool {
1477                 p.t.decPieceAvailability(pieceIndex(i))
1478                 return true
1479         })
1480 }
1481
1482 func (t *Torrent) assertPendingRequests() {
1483         if !check {
1484                 return
1485         }
1486         // var actual pendingRequests
1487         // if t.haveInfo() {
1488         //      actual.m = make([]int, t.numRequests())
1489         // }
1490         // t.iterPeers(func(p *Peer) {
1491         //      p.requestState.Requests.Iterate(func(x uint32) bool {
1492         //              actual.Inc(x)
1493         //              return true
1494         //      })
1495         // })
1496         // diff := cmp.Diff(actual.m, t.pendingRequests.m)
1497         // if diff != "" {
1498         //      panic(diff)
1499         // }
1500 }
1501
1502 func (t *Torrent) dropConnection(c *PeerConn) {
1503         t.cl.event.Broadcast()
1504         c.close()
1505         if t.deletePeerConn(c) {
1506                 t.openNewConns()
1507         }
1508 }
1509
1510 // Peers as in contact information for dialing out.
1511 func (t *Torrent) wantPeers() bool {
1512         if t.closed.IsSet() {
1513                 return false
1514         }
1515         if t.peers.Len() > t.cl.config.TorrentPeersLowWater {
1516                 return false
1517         }
1518         return t.wantConns()
1519 }
1520
1521 func (t *Torrent) updateWantPeersEvent() {
1522         if t.wantPeers() {
1523                 t.wantPeersEvent.Set()
1524         } else {
1525                 t.wantPeersEvent.Clear()
1526         }
1527 }
1528
1529 // Returns whether the client should make effort to seed the torrent.
1530 func (t *Torrent) seeding() bool {
1531         cl := t.cl
1532         if t.closed.IsSet() {
1533                 return false
1534         }
1535         if t.dataUploadDisallowed {
1536                 return false
1537         }
1538         if cl.config.NoUpload {
1539                 return false
1540         }
1541         if !cl.config.Seed {
1542                 return false
1543         }
1544         if cl.config.DisableAggressiveUpload && t.needData() {
1545                 return false
1546         }
1547         return true
1548 }
1549
1550 func (t *Torrent) onWebRtcConn(
1551         c datachannel.ReadWriteCloser,
1552         dcc webtorrent.DataChannelContext,
1553 ) {
1554         defer c.Close()
1555         netConn := webrtcNetConn{
1556                 ReadWriteCloser:    c,
1557                 DataChannelContext: dcc,
1558         }
1559         peerRemoteAddr := netConn.RemoteAddr()
1560         if t.cl.badPeerAddr(peerRemoteAddr) {
1561                 return
1562         }
1563         pc, err := t.cl.initiateProtocolHandshakes(
1564                 context.Background(),
1565                 netConn,
1566                 t,
1567                 dcc.LocalOffered,
1568                 false,
1569                 netConn.RemoteAddr(),
1570                 webrtcNetwork,
1571                 fmt.Sprintf("webrtc offer_id %x", dcc.OfferId),
1572         )
1573         if err != nil {
1574                 t.logger.WithDefaultLevel(log.Error).Printf("error in handshaking webrtc connection: %v", err)
1575                 return
1576         }
1577         if dcc.LocalOffered {
1578                 pc.Discovery = PeerSourceTracker
1579         } else {
1580                 pc.Discovery = PeerSourceIncoming
1581         }
1582         pc.conn.SetWriteDeadline(time.Time{})
1583         t.cl.lock()
1584         defer t.cl.unlock()
1585         err = t.cl.runHandshookConn(pc, t)
1586         if err != nil {
1587                 t.logger.WithDefaultLevel(log.Critical).Printf("error running handshook webrtc conn: %v", err)
1588         }
1589 }
1590
1591 func (t *Torrent) logRunHandshookConn(pc *PeerConn, logAll bool, level log.Level) {
1592         err := t.cl.runHandshookConn(pc, t)
1593         if err != nil || logAll {
1594                 t.logger.WithDefaultLevel(level).Printf("error running handshook conn: %v", err)
1595         }
1596 }
1597
1598 func (t *Torrent) runHandshookConnLoggingErr(pc *PeerConn) {
1599         t.logRunHandshookConn(pc, false, log.Debug)
1600 }
1601
1602 func (t *Torrent) startWebsocketAnnouncer(u url.URL) torrentTrackerAnnouncer {
1603         wtc, release := t.cl.websocketTrackers.Get(u.String())
1604         go func() {
1605                 <-t.closed.Done()
1606                 release()
1607         }()
1608         wst := websocketTrackerStatus{u, wtc}
1609         go func() {
1610                 err := wtc.Announce(tracker.Started, t.infoHash)
1611                 if err != nil {
1612                         t.logger.WithDefaultLevel(log.Warning).Printf(
1613                                 "error in initial announce to %q: %v",
1614                                 u.String(), err,
1615                         )
1616                 }
1617         }()
1618         return wst
1619 }
1620
1621 func (t *Torrent) startScrapingTracker(_url string) {
1622         if _url == "" {
1623                 return
1624         }
1625         u, err := url.Parse(_url)
1626         if err != nil {
1627                 // URLs with a leading '*' appear to be a uTorrent convention to
1628                 // disable trackers.
1629                 if _url[0] != '*' {
1630                         log.Str("error parsing tracker url").AddValues("url", _url).Log(t.logger)
1631                 }
1632                 return
1633         }
1634         if u.Scheme == "udp" {
1635                 u.Scheme = "udp4"
1636                 t.startScrapingTracker(u.String())
1637                 u.Scheme = "udp6"
1638                 t.startScrapingTracker(u.String())
1639                 return
1640         }
1641         if _, ok := t.trackerAnnouncers[_url]; ok {
1642                 return
1643         }
1644         sl := func() torrentTrackerAnnouncer {
1645                 switch u.Scheme {
1646                 case "ws", "wss":
1647                         if t.cl.config.DisableWebtorrent {
1648                                 return nil
1649                         }
1650                         return t.startWebsocketAnnouncer(*u)
1651                 case "udp4":
1652                         if t.cl.config.DisableIPv4Peers || t.cl.config.DisableIPv4 {
1653                                 return nil
1654                         }
1655                 case "udp6":
1656                         if t.cl.config.DisableIPv6 {
1657                                 return nil
1658                         }
1659                 }
1660                 newAnnouncer := &trackerScraper{
1661                         u:               *u,
1662                         t:               t,
1663                         lookupTrackerIp: t.cl.config.LookupTrackerIp,
1664                 }
1665                 go newAnnouncer.Run()
1666                 return newAnnouncer
1667         }()
1668         if sl == nil {
1669                 return
1670         }
1671         if t.trackerAnnouncers == nil {
1672                 t.trackerAnnouncers = make(map[string]torrentTrackerAnnouncer)
1673         }
1674         t.trackerAnnouncers[_url] = sl
1675 }
1676
1677 // Adds and starts tracker scrapers for tracker URLs that aren't already
1678 // running.
1679 func (t *Torrent) startMissingTrackerScrapers() {
1680         if t.cl.config.DisableTrackers {
1681                 return
1682         }
1683         t.startScrapingTracker(t.metainfo.Announce)
1684         for _, tier := range t.metainfo.AnnounceList {
1685                 for _, url := range tier {
1686                         t.startScrapingTracker(url)
1687                 }
1688         }
1689 }
1690
1691 // Returns an AnnounceRequest with fields filled out to defaults and current
1692 // values.
1693 func (t *Torrent) announceRequest(event tracker.AnnounceEvent) tracker.AnnounceRequest {
1694         // Note that IPAddress is not set. It's set for UDP inside the tracker code, since it's
1695         // dependent on the network in use.
1696         return tracker.AnnounceRequest{
1697                 Event: event,
1698                 NumWant: func() int32 {
1699                         if t.wantPeers() && len(t.cl.dialers) > 0 {
1700                                 return -1
1701                         } else {
1702                                 return 0
1703                         }
1704                 }(),
1705                 Port:     uint16(t.cl.incomingPeerPort()),
1706                 PeerId:   t.cl.peerID,
1707                 InfoHash: t.infoHash,
1708                 Key:      t.cl.announceKey(),
1709
1710                 // The following are vaguely described in BEP 3.
1711
1712                 Left:     t.bytesLeftAnnounce(),
1713                 Uploaded: t.stats.BytesWrittenData.Int64(),
1714                 // There's no mention of wasted or unwanted download in the BEP.
1715                 Downloaded: t.stats.BytesReadUsefulData.Int64(),
1716         }
1717 }
1718
1719 // Adds peers revealed in an announce until the announce ends, or we have
1720 // enough peers.
1721 func (t *Torrent) consumeDhtAnnouncePeers(pvs <-chan dht.PeersValues) {
1722         cl := t.cl
1723         for v := range pvs {
1724                 cl.lock()
1725                 added := 0
1726                 for _, cp := range v.Peers {
1727                         if cp.Port == 0 {
1728                                 // Can't do anything with this.
1729                                 continue
1730                         }
1731                         if t.addPeer(PeerInfo{
1732                                 Addr:   ipPortAddr{cp.IP, cp.Port},
1733                                 Source: PeerSourceDhtGetPeers,
1734                         }) {
1735                                 added++
1736                         }
1737                 }
1738                 cl.unlock()
1739                 // if added != 0 {
1740                 //      log.Printf("added %v peers from dht for %v", added, t.InfoHash().HexString())
1741                 // }
1742         }
1743 }
1744
1745 // Announce using the provided DHT server. Peers are consumed automatically. done is closed when the
1746 // announce ends. stop will force the announce to end.
1747 func (t *Torrent) AnnounceToDht(s DhtServer) (done <-chan struct{}, stop func(), err error) {
1748         ps, err := s.Announce(t.infoHash, t.cl.incomingPeerPort(), true)
1749         if err != nil {
1750                 return
1751         }
1752         _done := make(chan struct{})
1753         done = _done
1754         stop = ps.Close
1755         go func() {
1756                 t.consumeDhtAnnouncePeers(ps.Peers())
1757                 close(_done)
1758         }()
1759         return
1760 }
1761
1762 func (t *Torrent) timeboxedAnnounceToDht(s DhtServer) error {
1763         _, stop, err := t.AnnounceToDht(s)
1764         if err != nil {
1765                 return err
1766         }
1767         select {
1768         case <-t.closed.Done():
1769         case <-time.After(5 * time.Minute):
1770         }
1771         stop()
1772         return nil
1773 }
1774
1775 func (t *Torrent) dhtAnnouncer(s DhtServer) {
1776         cl := t.cl
1777         cl.lock()
1778         defer cl.unlock()
1779         for {
1780                 for {
1781                         if t.closed.IsSet() {
1782                                 return
1783                         }
1784                         // We're also announcing ourselves as a listener, so we don't just want peer addresses.
1785                         // TODO: We can include the announce_peer step depending on whether we can receive
1786                         // inbound connections. We should probably only announce once every 15 mins too.
1787                         if !t.wantConns() {
1788                                 goto wait
1789                         }
1790                         // TODO: Determine if there's a listener on the port we're announcing.
1791                         if len(cl.dialers) == 0 && len(cl.listeners) == 0 {
1792                                 goto wait
1793                         }
1794                         break
1795                 wait:
1796                         cl.event.Wait()
1797                 }
1798                 func() {
1799                         t.numDHTAnnounces++
1800                         cl.unlock()
1801                         defer cl.lock()
1802                         err := t.timeboxedAnnounceToDht(s)
1803                         if err != nil {
1804                                 t.logger.WithDefaultLevel(log.Warning).Printf("error announcing %q to DHT: %s", t, err)
1805                         }
1806                 }()
1807         }
1808 }
1809
1810 func (t *Torrent) addPeers(peers []PeerInfo) (added int) {
1811         for _, p := range peers {
1812                 if t.addPeer(p) {
1813                         added++
1814                 }
1815         }
1816         return
1817 }
1818
1819 // The returned TorrentStats may require alignment in memory. See
1820 // https://github.com/anacrolix/torrent/issues/383.
1821 func (t *Torrent) Stats() TorrentStats {
1822         t.cl.rLock()
1823         defer t.cl.rUnlock()
1824         return t.statsLocked()
1825 }
1826
1827 func (t *Torrent) statsLocked() (ret TorrentStats) {
1828         ret.ActivePeers = len(t.conns)
1829         ret.HalfOpenPeers = len(t.halfOpen)
1830         ret.PendingPeers = t.peers.Len()
1831         ret.TotalPeers = t.numTotalPeers()
1832         ret.ConnectedSeeders = 0
1833         for c := range t.conns {
1834                 if all, ok := c.peerHasAllPieces(); all && ok {
1835                         ret.ConnectedSeeders++
1836                 }
1837         }
1838         ret.ConnStats = t.stats.Copy()
1839         ret.PiecesComplete = t.numPiecesCompleted()
1840         return
1841 }
1842
1843 // The total number of peers in the torrent.
1844 func (t *Torrent) numTotalPeers() int {
1845         peers := make(map[string]struct{})
1846         for conn := range t.conns {
1847                 ra := conn.conn.RemoteAddr()
1848                 if ra == nil {
1849                         // It's been closed and doesn't support RemoteAddr.
1850                         continue
1851                 }
1852                 peers[ra.String()] = struct{}{}
1853         }
1854         for addr := range t.halfOpen {
1855                 peers[addr] = struct{}{}
1856         }
1857         t.peers.Each(func(peer PeerInfo) {
1858                 peers[peer.Addr.String()] = struct{}{}
1859         })
1860         return len(peers)
1861 }
1862
1863 // Reconcile bytes transferred before connection was associated with a
1864 // torrent.
1865 func (t *Torrent) reconcileHandshakeStats(c *PeerConn) {
1866         if c._stats != (ConnStats{
1867                 // Handshakes should only increment these fields:
1868                 BytesWritten: c._stats.BytesWritten,
1869                 BytesRead:    c._stats.BytesRead,
1870         }) {
1871                 panic("bad stats")
1872         }
1873         c.postHandshakeStats(func(cs *ConnStats) {
1874                 cs.BytesRead.Add(c._stats.BytesRead.Int64())
1875                 cs.BytesWritten.Add(c._stats.BytesWritten.Int64())
1876         })
1877         c.reconciledHandshakeStats = true
1878 }
1879
1880 // Returns true if the connection is added.
1881 func (t *Torrent) addPeerConn(c *PeerConn) (err error) {
1882         defer func() {
1883                 if err == nil {
1884                         torrent.Add("added connections", 1)
1885                 }
1886         }()
1887         if t.closed.IsSet() {
1888                 return errors.New("torrent closed")
1889         }
1890         for c0 := range t.conns {
1891                 if c.PeerID != c0.PeerID {
1892                         continue
1893                 }
1894                 if !t.cl.config.DropDuplicatePeerIds {
1895                         continue
1896                 }
1897                 if c.hasPreferredNetworkOver(c0) {
1898                         c0.close()
1899                         t.deletePeerConn(c0)
1900                 } else {
1901                         return errors.New("existing connection preferred")
1902                 }
1903         }
1904         if len(t.conns) >= t.maxEstablishedConns {
1905                 c := t.worstBadConn()
1906                 if c == nil {
1907                         return errors.New("don't want conns")
1908                 }
1909                 c.close()
1910                 t.deletePeerConn(c)
1911         }
1912         if len(t.conns) >= t.maxEstablishedConns {
1913                 panic(len(t.conns))
1914         }
1915         t.conns[c] = struct{}{}
1916         if !t.cl.config.DisablePEX && !c.PeerExtensionBytes.SupportsExtended() {
1917                 t.pex.Add(c) // as no further extended handshake expected
1918         }
1919         return nil
1920 }
1921
1922 func (t *Torrent) wantConns() bool {
1923         if !t.networkingEnabled.Bool() {
1924                 return false
1925         }
1926         if t.closed.IsSet() {
1927                 return false
1928         }
1929         if !t.needData() && (!t.seeding() || !t.haveAnyPieces()) {
1930                 return false
1931         }
1932         return len(t.conns) < t.maxEstablishedConns || t.worstBadConn() != nil
1933 }
1934
1935 func (t *Torrent) SetMaxEstablishedConns(max int) (oldMax int) {
1936         t.cl.lock()
1937         defer t.cl.unlock()
1938         oldMax = t.maxEstablishedConns
1939         t.maxEstablishedConns = max
1940         wcs := worseConnSlice{
1941                 conns: t.appendConns(nil, func(*PeerConn) bool {
1942                         return true
1943                 }),
1944         }
1945         wcs.initKeys()
1946         heap.Init(&wcs)
1947         for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 {
1948                 t.dropConnection(heap.Pop(&wcs).(*PeerConn))
1949         }
1950         t.openNewConns()
1951         return oldMax
1952 }
1953
1954 func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
1955         t.logger.Log(log.Fstr("hashed piece %d (passed=%t)", piece, passed).SetLevel(log.Debug))
1956         p := t.piece(piece)
1957         p.numVerifies++
1958         t.cl.event.Broadcast()
1959         if t.closed.IsSet() {
1960                 return
1961         }
1962
1963         // Don't score the first time a piece is hashed, it could be an initial check.
1964         if p.storageCompletionOk {
1965                 if passed {
1966                         pieceHashedCorrect.Add(1)
1967                 } else {
1968                         log.Fmsg(
1969                                 "piece %d failed hash: %d connections contributed", piece, len(p.dirtiers),
1970                         ).AddValues(t, p).SetLevel(log.Debug).Log(t.logger)
1971                         pieceHashedNotCorrect.Add(1)
1972                 }
1973         }
1974
1975         p.marking = true
1976         t.publishPieceChange(piece)
1977         defer func() {
1978                 p.marking = false
1979                 t.publishPieceChange(piece)
1980         }()
1981
1982         if passed {
1983                 if len(p.dirtiers) != 0 {
1984                         // Don't increment stats above connection-level for every involved connection.
1985                         t.allStats((*ConnStats).incrementPiecesDirtiedGood)
1986                 }
1987                 for c := range p.dirtiers {
1988                         c._stats.incrementPiecesDirtiedGood()
1989                 }
1990                 t.clearPieceTouchers(piece)
1991                 t.cl.unlock()
1992                 err := p.Storage().MarkComplete()
1993                 if err != nil {
1994                         t.logger.Printf("%T: error marking piece complete %d: %s", t.storage, piece, err)
1995                 }
1996                 t.cl.lock()
1997
1998                 if t.closed.IsSet() {
1999                         return
2000                 }
2001                 t.pendAllChunkSpecs(piece)
2002         } else {
2003                 if len(p.dirtiers) != 0 && p.allChunksDirty() && hashIoErr == nil {
2004                         // Peers contributed to all the data for this piece hash failure, and the failure was
2005                         // not due to errors in the storage (such as data being dropped in a cache).
2006
2007                         // Increment Torrent and above stats, and then specific connections.
2008                         t.allStats((*ConnStats).incrementPiecesDirtiedBad)
2009                         for c := range p.dirtiers {
2010                                 // Y u do dis peer?!
2011                                 c.stats().incrementPiecesDirtiedBad()
2012                         }
2013
2014                         bannableTouchers := make([]*Peer, 0, len(p.dirtiers))
2015                         for c := range p.dirtiers {
2016                                 if !c.trusted {
2017                                         bannableTouchers = append(bannableTouchers, c)
2018                                 }
2019                         }
2020                         t.clearPieceTouchers(piece)
2021                         slices.Sort(bannableTouchers, connLessTrusted)
2022
2023                         if t.cl.config.Debug {
2024                                 t.logger.Printf(
2025                                         "bannable conns by trust for piece %d: %v",
2026                                         piece,
2027                                         func() (ret []connectionTrust) {
2028                                                 for _, c := range bannableTouchers {
2029                                                         ret = append(ret, c.trust())
2030                                                 }
2031                                                 return
2032                                         }(),
2033                                 )
2034                         }
2035
2036                         if len(bannableTouchers) >= 1 {
2037                                 c := bannableTouchers[0]
2038                                 t.cl.banPeerIP(c.remoteIp())
2039                                 c.drop()
2040                         }
2041                 }
2042                 t.onIncompletePiece(piece)
2043                 p.Storage().MarkNotComplete()
2044         }
2045         t.updatePieceCompletion(piece)
2046 }
2047
2048 func (t *Torrent) cancelRequestsForPiece(piece pieceIndex) {
2049         for ri := t.pieceRequestIndexOffset(piece); ri < t.pieceRequestIndexOffset(piece+1); ri++ {
2050                 t.cancelRequest(ri)
2051         }
2052 }
2053
2054 func (t *Torrent) onPieceCompleted(piece pieceIndex) {
2055         t.pendAllChunkSpecs(piece)
2056         t.cancelRequestsForPiece(piece)
2057         t.piece(piece).readerCond.Broadcast()
2058         for conn := range t.conns {
2059                 conn.have(piece)
2060                 t.maybeDropMutuallyCompletePeer(&conn.Peer)
2061         }
2062 }
2063
2064 // Called when a piece is found to be not complete.
2065 func (t *Torrent) onIncompletePiece(piece pieceIndex) {
2066         if t.pieceAllDirty(piece) {
2067                 t.pendAllChunkSpecs(piece)
2068         }
2069         if !t.wantPieceIndex(piece) {
2070                 // t.logger.Printf("piece %d incomplete and unwanted", piece)
2071                 return
2072         }
2073         // We could drop any connections that we told we have a piece that we
2074         // don't here. But there's a test failure, and it seems clients don't care
2075         // if you request pieces that you already claim to have. Pruning bad
2076         // connections might just remove any connections that aren't treating us
2077         // favourably anyway.
2078
2079         // for c := range t.conns {
2080         //      if c.sentHave(piece) {
2081         //              c.drop()
2082         //      }
2083         // }
2084         t.iterPeers(func(conn *Peer) {
2085                 if conn.peerHasPiece(piece) {
2086                         conn.updateRequests("piece incomplete")
2087                 }
2088         })
2089 }
2090
2091 func (t *Torrent) tryCreateMorePieceHashers() {
2092         for !t.closed.IsSet() && t.activePieceHashes < 2 && t.tryCreatePieceHasher() {
2093         }
2094 }
2095
2096 func (t *Torrent) tryCreatePieceHasher() bool {
2097         if t.storage == nil {
2098                 return false
2099         }
2100         pi, ok := t.getPieceToHash()
2101         if !ok {
2102                 return false
2103         }
2104         p := t.piece(pi)
2105         t.piecesQueuedForHash.Remove(bitmap.BitIndex(pi))
2106         p.hashing = true
2107         t.publishPieceChange(pi)
2108         t.updatePiecePriority(pi, "Torrent.tryCreatePieceHasher")
2109         t.storageLock.RLock()
2110         t.activePieceHashes++
2111         go t.pieceHasher(pi)
2112         return true
2113 }
2114
2115 func (t *Torrent) getPieceToHash() (ret pieceIndex, ok bool) {
2116         t.piecesQueuedForHash.IterTyped(func(i pieceIndex) bool {
2117                 if t.piece(i).hashing {
2118                         return true
2119                 }
2120                 ret = i
2121                 ok = true
2122                 return false
2123         })
2124         return
2125 }
2126
2127 func (t *Torrent) pieceHasher(index pieceIndex) {
2128         p := t.piece(index)
2129         sum, failedPeers, copyErr := t.hashPiece(index)
2130         correct := sum == *p.hash
2131         if correct {
2132                 for peer := range failedPeers {
2133                         log.Printf("would smart ban %q for %v here", peer, p)
2134                         t.cl.banPrefix(peer)
2135                 }
2136         }
2137         switch copyErr {
2138         case nil, io.EOF:
2139         default:
2140                 log.Fmsg("piece %v (%s) hash failure copy error: %v", p, p.hash.HexString(), copyErr).Log(t.logger)
2141         }
2142         t.storageLock.RUnlock()
2143         t.cl.lock()
2144         defer t.cl.unlock()
2145         p.hashing = false
2146         t.pieceHashed(index, correct, copyErr)
2147         t.updatePiecePriority(index, "Torrent.pieceHasher")
2148         t.activePieceHashes--
2149         t.tryCreateMorePieceHashers()
2150 }
2151
2152 // Return the connections that touched a piece, and clear the entries while doing it.
2153 func (t *Torrent) clearPieceTouchers(pi pieceIndex) {
2154         p := t.piece(pi)
2155         for c := range p.dirtiers {
2156                 delete(c.peerTouchedPieces, pi)
2157                 delete(p.dirtiers, c)
2158         }
2159 }
2160
2161 func (t *Torrent) peersAsSlice() (ret []*Peer) {
2162         t.iterPeers(func(p *Peer) {
2163                 ret = append(ret, p)
2164         })
2165         return
2166 }
2167
2168 func (t *Torrent) queuePieceCheck(pieceIndex pieceIndex) {
2169         piece := t.piece(pieceIndex)
2170         if piece.queuedForHash() {
2171                 return
2172         }
2173         t.piecesQueuedForHash.Add(bitmap.BitIndex(pieceIndex))
2174         t.publishPieceChange(pieceIndex)
2175         t.updatePiecePriority(pieceIndex, "Torrent.queuePieceCheck")
2176         t.tryCreateMorePieceHashers()
2177 }
2178
2179 // Forces all the pieces to be re-hashed. See also Piece.VerifyData. This should not be called
2180 // before the Info is available.
2181 func (t *Torrent) VerifyData() {
2182         for i := pieceIndex(0); i < t.NumPieces(); i++ {
2183                 t.Piece(i).VerifyData()
2184         }
2185 }
2186
2187 // Start the process of connecting to the given peer for the given torrent if appropriate.
2188 func (t *Torrent) initiateConn(peer PeerInfo) {
2189         if peer.Id == t.cl.peerID {
2190                 return
2191         }
2192         if t.cl.badPeerAddr(peer.Addr) && !peer.Trusted {
2193                 return
2194         }
2195         addr := peer.Addr
2196         if t.addrActive(addr.String()) {
2197                 return
2198         }
2199         t.cl.numHalfOpen++
2200         t.halfOpen[addr.String()] = peer
2201         go t.cl.outgoingConnection(t, addr, peer.Source, peer.Trusted)
2202 }
2203
2204 // Adds a trusted, pending peer for each of the given Client's addresses. Typically used in tests to
2205 // quickly make one Client visible to the Torrent of another Client.
2206 func (t *Torrent) AddClientPeer(cl *Client) int {
2207         return t.AddPeers(func() (ps []PeerInfo) {
2208                 for _, la := range cl.ListenAddrs() {
2209                         ps = append(ps, PeerInfo{
2210                                 Addr:    la,
2211                                 Trusted: true,
2212                         })
2213                 }
2214                 return
2215         }())
2216 }
2217
2218 // All stats that include this Torrent. Useful when we want to increment ConnStats but not for every
2219 // connection.
2220 func (t *Torrent) allStats(f func(*ConnStats)) {
2221         f(&t.stats)
2222         f(&t.cl.stats)
2223 }
2224
2225 func (t *Torrent) hashingPiece(i pieceIndex) bool {
2226         return t.pieces[i].hashing
2227 }
2228
2229 func (t *Torrent) pieceQueuedForHash(i pieceIndex) bool {
2230         return t.piecesQueuedForHash.Get(bitmap.BitIndex(i))
2231 }
2232
2233 func (t *Torrent) dialTimeout() time.Duration {
2234         return reducedDialTimeout(t.cl.config.MinDialTimeout, t.cl.config.NominalDialTimeout, t.cl.config.HalfOpenConnsPerTorrent, t.peers.Len())
2235 }
2236
2237 func (t *Torrent) piece(i int) *Piece {
2238         return &t.pieces[i]
2239 }
2240
2241 func (t *Torrent) onWriteChunkErr(err error) {
2242         if t.userOnWriteChunkErr != nil {
2243                 go t.userOnWriteChunkErr(err)
2244                 return
2245         }
2246         t.logger.WithDefaultLevel(log.Critical).Printf("default chunk write error handler: disabling data download")
2247         t.disallowDataDownloadLocked()
2248 }
2249
2250 func (t *Torrent) DisallowDataDownload() {
2251         t.disallowDataDownloadLocked()
2252 }
2253
2254 func (t *Torrent) disallowDataDownloadLocked() {
2255         t.dataDownloadDisallowed.Set()
2256 }
2257
2258 func (t *Torrent) AllowDataDownload() {
2259         t.dataDownloadDisallowed.Clear()
2260 }
2261
2262 // Enables uploading data, if it was disabled.
2263 func (t *Torrent) AllowDataUpload() {
2264         t.cl.lock()
2265         defer t.cl.unlock()
2266         t.dataUploadDisallowed = false
2267         for c := range t.conns {
2268                 c.updateRequests("allow data upload")
2269         }
2270 }
2271
2272 // Disables uploading data, if it was enabled.
2273 func (t *Torrent) DisallowDataUpload() {
2274         t.cl.lock()
2275         defer t.cl.unlock()
2276         t.dataUploadDisallowed = true
2277         for c := range t.conns {
2278                 // TODO: This doesn't look right. Shouldn't we tickle writers to choke peers or something instead?
2279                 c.updateRequests("disallow data upload")
2280         }
2281 }
2282
2283 // Sets a handler that is called if there's an error writing a chunk to local storage. By default,
2284 // or if nil, a critical message is logged, and data download is disabled.
2285 func (t *Torrent) SetOnWriteChunkError(f func(error)) {
2286         t.cl.lock()
2287         defer t.cl.unlock()
2288         t.userOnWriteChunkErr = f
2289 }
2290
2291 func (t *Torrent) iterPeers(f func(p *Peer)) {
2292         for pc := range t.conns {
2293                 f(&pc.Peer)
2294         }
2295         for _, ws := range t.webSeeds {
2296                 f(ws)
2297         }
2298 }
2299
2300 func (t *Torrent) callbacks() *Callbacks {
2301         return &t.cl.config.Callbacks
2302 }
2303
2304 func (t *Torrent) addWebSeed(url string) {
2305         if t.cl.config.DisableWebseeds {
2306                 return
2307         }
2308         if _, ok := t.webSeeds[url]; ok {
2309                 return
2310         }
2311         // I don't think Go http supports pipelining requests. However, we can have more ready to go
2312         // right away. This value should be some multiple of the number of connections to a host. I
2313         // would expect that double maxRequests plus a bit would be appropriate. This value is based on
2314         // downloading Sintel (08ada5a7a6183aae1e09d831df6748d566095a10) from
2315         // "https://webtorrent.io/torrents/".
2316         const maxRequests = 16
2317         ws := webseedPeer{
2318                 peer: Peer{
2319                         t:                        t,
2320                         outgoing:                 true,
2321                         Network:                  "http",
2322                         reconciledHandshakeStats: true,
2323                         // This should affect how often we have to recompute requests for this peer. Note that
2324                         // because we can request more than 1 thing at a time over HTTP, we will hit the low
2325                         // requests mark more often, so recomputation is probably sooner than with regular peer
2326                         // conns. ~4x maxRequests would be about right.
2327                         PeerMaxRequests: 128,
2328                         // TODO: Set ban prefix?
2329                         RemoteAddr: remoteAddrFromUrl(url),
2330                         callbacks:  t.callbacks(),
2331                 },
2332                 client: webseed.Client{
2333                         HttpClient: t.cl.webseedHttpClient,
2334                         Url:        url,
2335                         ResponseBodyWrapper: func(r io.Reader) io.Reader {
2336                                 return &rateLimitedReader{
2337                                         l: t.cl.config.DownloadRateLimiter,
2338                                         r: r,
2339                                 }
2340                         },
2341                 },
2342                 activeRequests: make(map[Request]webseed.Request, maxRequests),
2343                 maxRequests:    maxRequests,
2344         }
2345         ws.peer.initUpdateRequestsTimer()
2346         ws.requesterCond.L = t.cl.locker()
2347         for i := 0; i < maxRequests; i += 1 {
2348                 go ws.requester(i)
2349         }
2350         for _, f := range t.callbacks().NewPeer {
2351                 f(&ws.peer)
2352         }
2353         ws.peer.logger = t.logger.WithContextValue(&ws)
2354         ws.peer.peerImpl = &ws
2355         if t.haveInfo() {
2356                 ws.onGotInfo(t.info)
2357         }
2358         t.webSeeds[url] = &ws.peer
2359 }
2360
2361 func (t *Torrent) peerIsActive(p *Peer) (active bool) {
2362         t.iterPeers(func(p1 *Peer) {
2363                 if p1 == p {
2364                         active = true
2365                 }
2366         })
2367         return
2368 }
2369
2370 func (t *Torrent) requestIndexToRequest(ri RequestIndex) Request {
2371         index := ri / t.chunksPerRegularPiece()
2372         return Request{
2373                 pp.Integer(index),
2374                 t.piece(int(index)).chunkIndexSpec(ri % t.chunksPerRegularPiece()),
2375         }
2376 }
2377
2378 func (t *Torrent) requestIndexFromRequest(r Request) RequestIndex {
2379         return t.pieceRequestIndexOffset(pieceIndex(r.Index)) + uint32(r.Begin/t.chunkSize)
2380 }
2381
2382 func (t *Torrent) pieceRequestIndexOffset(piece pieceIndex) RequestIndex {
2383         return RequestIndex(piece) * t.chunksPerRegularPiece()
2384 }
2385
2386 func (t *Torrent) updateComplete() {
2387         t.Complete.SetBool(t.haveAllPieces())
2388 }
2389
2390 func (t *Torrent) cancelRequest(r RequestIndex) *Peer {
2391         p := t.pendingRequests[r]
2392         if p != nil {
2393                 p.cancel(r)
2394         }
2395         delete(t.pendingRequests, r)
2396         return p
2397 }
2398
2399 func (t *Torrent) requestingPeer(r RequestIndex) *Peer {
2400         return t.pendingRequests[r]
2401 }
2402
2403 func (t *Torrent) addConnWithAllPieces(p *Peer) {
2404         if t.connsWithAllPieces == nil {
2405                 t.connsWithAllPieces = make(map[*Peer]struct{}, t.maxEstablishedConns)
2406         }
2407         t.connsWithAllPieces[p] = struct{}{}
2408 }
2409
2410 func (t *Torrent) deleteConnWithAllPieces(p *Peer) bool {
2411         _, ok := t.connsWithAllPieces[p]
2412         delete(t.connsWithAllPieces, p)
2413         return ok
2414 }
2415
2416 func (t *Torrent) numActivePeers() int {
2417         return len(t.conns) + len(t.webSeeds)
2418 }
2419
2420 func (t *Torrent) hasStorageCap() bool {
2421         f := t.storage.Capacity
2422         if f == nil {
2423                 return false
2424         }
2425         _, ok := (*f)()
2426         return ok
2427 }