18 "github.com/anacrolix/missinggo"
19 "github.com/anacrolix/missinggo/bitmap"
20 "github.com/anacrolix/missinggo/itertools"
21 "github.com/anacrolix/missinggo/perf"
22 "github.com/anacrolix/missinggo/pubsub"
23 "github.com/anacrolix/missinggo/slices"
24 "github.com/bradfitz/iter"
26 "github.com/anacrolix/torrent/bencode"
27 "github.com/anacrolix/torrent/dht"
28 "github.com/anacrolix/torrent/metainfo"
29 pp "github.com/anacrolix/torrent/peer_protocol"
30 "github.com/anacrolix/torrent/storage"
31 "github.com/anacrolix/torrent/tracker"
34 func (t *Torrent) chunkIndexSpec(chunkIndex, piece int) chunkSpec {
35 return chunkIndexSpec(chunkIndex, t.pieceLength(piece), t.chunkSize)
38 type peersKey struct {
43 // Maintains state of torrent within a Client.
47 closed missinggo.Event
48 infoHash metainfo.Hash
50 // Values are the piece indices that changed.
51 pieceStateChanges *pubsub.PubSub
52 // The size of chunks to request from peers over the wire. This is
53 // normally 16KiB by convention these days.
56 // Total length of the torrent in bytes. Stored because it's not O(1) to
57 // get this from the info dict.
60 // The storage to open when the info dict becomes available.
61 storageOpener *storage.Client
62 // Storage for torrent data.
63 storage *storage.Torrent
65 metainfo metainfo.MetaInfo
67 // The info dict. nil if we don't have it (yet).
69 // Active peer connections, running message stream loops.
71 maxEstablishedConns int
72 // Set of addrs to which we're attempting to connect. Connections are
73 // half-open until all handshakes are completed.
74 halfOpen map[string]struct{}
76 // Reserve of peers to connect to. A peer can be both here and in the
77 // active connections if were told about the peer after connecting with
78 // them. That encourages us to reconnect to peers that are well known in
80 peers map[peersKey]Peer
81 wantPeersEvent missinggo.Event
82 // An announcer for each tracker URL.
83 trackerAnnouncers map[string]*trackerScraper
84 // How many times we've initiated a DHT announce. TODO: Move into stats.
87 // Name used if the info name isn't available. Should be cleared when the
88 // Info does become available.
90 // The bencoded bytes of the info dict. This is actively manipulated if
91 // the info bytes aren't initially available, and we try to fetch them
94 // Each element corresponds to the 16KiB metadata pieces. If true, we have
95 // received that piece.
96 metadataCompletedChunks []bool
98 // Set when .Info is obtained.
99 gotMetainfo missinggo.Event
101 readers map[*Reader]struct{}
102 readerNowPieces bitmap.Bitmap
103 readerReadaheadPieces bitmap.Bitmap
105 // The indexes of pieces we want with normal priority, that aren't
106 // currently available.
107 pendingPieces bitmap.Bitmap
108 // A cache of completed piece indices.
109 completedPieces bitmap.Bitmap
111 // A pool of piece priorities []int for assignment to new connections.
112 // These "inclinations" are used to give connections preference for
114 connPieceInclinationPool sync.Pool
115 // Torrent-level statistics.
119 func (t *Torrent) setChunkSize(size pp.Integer) {
121 t.chunkPool = &sync.Pool{
122 New: func() interface{} {
123 return make([]byte, size)
128 func (t *Torrent) setDisplayName(dn string) {
135 func (t *Torrent) pieceComplete(piece int) bool {
136 return t.completedPieces.Get(piece)
139 func (t *Torrent) pieceCompleteUncached(piece int) bool {
140 return t.pieces[piece].Storage().GetIsComplete()
143 func (t *Torrent) numConnsUnchoked() (num int) {
144 for _, c := range t.conns {
152 // There's a connection to that address already.
153 func (t *Torrent) addrActive(addr string) bool {
154 if _, ok := t.halfOpen[addr]; ok {
157 for _, c := range t.conns {
158 if c.remoteAddr().String() == addr {
165 func (t *Torrent) worstUnclosedConns() (ret []*connection) {
166 ret = make([]*connection, 0, len(t.conns))
167 for _, c := range t.conns {
168 if !c.closed.IsSet() {
175 func (t *Torrent) addPeer(p Peer) {
178 if len(t.peers) >= torrentPeersHighWater {
181 key := peersKey{string(p.IP), p.Port}
182 if _, ok := t.peers[key]; ok {
186 peersAddedBySource.Add(string(p.Source), 1)
191 func (t *Torrent) invalidateMetadata() {
192 for i := range t.metadataCompletedChunks {
193 t.metadataCompletedChunks[i] = false
198 func (t *Torrent) saveMetadataPiece(index int, data []byte) {
202 if index >= len(t.metadataCompletedChunks) {
203 log.Printf("%s: ignoring metadata piece %d", t, index)
206 copy(t.metadataBytes[(1<<14)*index:], data)
207 t.metadataCompletedChunks[index] = true
210 func (t *Torrent) metadataPieceCount() int {
211 return (len(t.metadataBytes) + (1 << 14) - 1) / (1 << 14)
214 func (t *Torrent) haveMetadataPiece(piece int) bool {
216 return (1<<14)*piece < len(t.metadataBytes)
218 return piece < len(t.metadataCompletedChunks) && t.metadataCompletedChunks[piece]
222 func (t *Torrent) metadataSizeKnown() bool {
223 return t.metadataBytes != nil
226 func (t *Torrent) metadataSize() int {
227 return len(t.metadataBytes)
230 func infoPieceHashes(info *metainfo.Info) (ret []string) {
231 for i := 0; i < len(info.Pieces); i += sha1.Size {
232 ret = append(ret, string(info.Pieces[i:i+sha1.Size]))
237 func (t *Torrent) makePieces() {
238 hashes := infoPieceHashes(t.info)
239 t.pieces = make([]piece, len(hashes))
240 for i, hash := range hashes {
241 piece := &t.pieces[i]
244 piece.noPendingWrites.L = &piece.pendingWritesMutex
245 missinggo.CopyExact(piece.Hash[:], hash)
249 // Called when metadata for a torrent becomes available.
250 func (t *Torrent) setInfoBytes(b []byte) error {
254 if metainfo.HashBytes(b) != t.infoHash {
255 return errors.New("info bytes have wrong hash")
257 var info metainfo.Info
258 err := bencode.Unmarshal(b, &info)
260 return fmt.Errorf("error unmarshalling info bytes: %s", err)
262 err = validateInfo(&info)
264 return fmt.Errorf("bad info: %s", err)
266 defer t.updateWantPeersEvent()
268 t.displayName = "" // Save a few bytes lol.
269 t.cl.event.Broadcast()
271 t.storage, err = t.storageOpener.OpenTorrent(t.info, t.infoHash)
273 return fmt.Errorf("error opening torrent storage: %s", err)
276 for _, f := range t.info.UpvertedFiles() {
280 t.metadataCompletedChunks = nil
282 for _, conn := range t.conns {
283 if err := conn.setNumPieces(t.numPieces()); err != nil {
284 log.Printf("closing connection: %s", err)
288 for i := range t.pieces {
289 t.updatePieceCompletion(i)
290 t.pieces[i].QueuedForHash = true
293 for i := range t.pieces {
300 func (t *Torrent) verifyPiece(piece int) {
301 t.cl.verifyPiece(t, piece)
304 func (t *Torrent) haveAllMetadataPieces() bool {
308 if t.metadataCompletedChunks == nil {
311 for _, have := range t.metadataCompletedChunks {
319 // TODO: Propagate errors to disconnect peer.
320 func (t *Torrent) setMetadataSize(bytes int64) (err error) {
322 // We already know the correct metadata size.
325 if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
326 return errors.New("bad size")
328 if t.metadataBytes != nil && len(t.metadataBytes) == int(bytes) {
331 t.metadataBytes = make([]byte, bytes)
332 t.metadataCompletedChunks = make([]bool, (bytes+(1<<14)-1)/(1<<14))
333 for _, c := range t.conns {
334 c.requestPendingMetadata()
339 // The current working name for the torrent. Either the name in the info dict,
340 // or a display name given such as by the dn value in a magnet link, or "".
341 func (t *Torrent) name() string {
348 func (t *Torrent) pieceState(index int) (ret PieceState) {
349 p := &t.pieces[index]
350 ret.Priority = t.piecePriority(index)
351 if t.pieceComplete(index) {
354 if p.QueuedForHash || p.Hashing {
357 if !ret.Complete && t.piecePartiallyDownloaded(index) {
363 func (t *Torrent) metadataPieceSize(piece int) int {
364 return metadataPieceSize(len(t.metadataBytes), piece)
367 func (t *Torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
373 d["total_size"] = len(t.metadataBytes)
375 p, err := bencode.Marshal(d)
381 ExtendedID: c.PeerExtensionIDs["ut_metadata"],
382 ExtendedPayload: append(p, data...),
386 func (t *Torrent) pieceStateRuns() (ret []PieceStateRun) {
387 rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
388 ret = append(ret, PieceStateRun{
389 PieceState: el.(PieceState),
393 for index := range t.pieces {
394 rle.Append(t.pieceState(index), 1)
400 // Produces a small string representing a PieceStateRun.
401 func pieceStateRunStatusChars(psr PieceStateRun) (ret string) {
402 ret = fmt.Sprintf("%d", psr.Length)
403 ret += func() string {
404 switch psr.Priority {
405 case PiecePriorityNext:
407 case PiecePriorityNormal:
409 case PiecePriorityReadahead:
411 case PiecePriorityNow:
429 func (t *Torrent) writeStatus(w io.Writer) {
430 fmt.Fprintf(w, "Infohash: %x\n", t.infoHash)
431 fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
433 fmt.Fprintf(w, "Metadata have: ")
434 for _, h := range t.metadataCompletedChunks {
435 fmt.Fprintf(w, "%c", func() rune {
445 fmt.Fprintf(w, "Piece length: %s\n", func() string {
447 return fmt.Sprint(t.usualPieceSize())
453 fmt.Fprintf(w, "Num Pieces: %d\n", t.numPieces())
454 fmt.Fprint(w, "Piece States:")
455 for _, psr := range t.PieceStateRuns() {
457 w.Write([]byte(pieceStateRunStatusChars(psr)))
461 fmt.Fprintf(w, "Reader Pieces:")
462 t.forReaderOffsetPieces(func(begin, end int) (again bool) {
463 fmt.Fprintf(w, " %d:%d", begin, end)
468 fmt.Fprintf(w, "Trackers:\n")
470 tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
471 fmt.Fprintf(tw, " URL\tNext announce\tLast announce\n")
472 for _, ta := range slices.Sort(slices.FromMapElems(t.trackerAnnouncers), func(l, r *trackerScraper) bool {
474 }).([]*trackerScraper) {
475 fmt.Fprintf(tw, " %s\n", ta.statusLine())
480 fmt.Fprintf(w, "DHT Announces: %d\n", t.numDHTAnnounces)
482 fmt.Fprintf(w, "Pending peers: %d\n", len(t.peers))
483 fmt.Fprintf(w, "Half open: %d\n", len(t.halfOpen))
484 fmt.Fprintf(w, "Active peers: %d\n", len(t.conns))
485 slices.Sort(t.conns, worseConn)
486 for i, c := range t.conns {
487 fmt.Fprintf(w, "%2d. ", i+1)
492 func (t *Torrent) haveInfo() bool {
496 // TODO: Include URIs that weren't converted to tracker clients.
497 func (t *Torrent) announceList() (al [][]string) {
498 return t.metainfo.AnnounceList
501 // Returns a run-time generated MetaInfo that includes the info bytes and
502 // announce-list as currently known to the client.
503 func (t *Torrent) newMetaInfo() metainfo.MetaInfo {
504 return metainfo.MetaInfo{
505 CreationDate: time.Now().Unix(),
506 Comment: "dynamic metainfo from client",
507 CreatedBy: "go.torrent",
508 AnnounceList: t.announceList(),
509 InfoBytes: t.metadataBytes,
513 func (t *Torrent) BytesMissing() int64 {
515 defer t.mu().RUnlock()
519 func (t *Torrent) bytesLeft() (left int64) {
520 for i := 0; i < t.numPieces(); i++ {
521 left += int64(t.pieces[i].bytesLeft())
526 // Bytes left to give in tracker announces.
527 func (t *Torrent) bytesLeftAnnounce() uint64 {
529 return uint64(t.bytesLeft())
531 return math.MaxUint64
535 func (t *Torrent) piecePartiallyDownloaded(piece int) bool {
536 if t.pieceComplete(piece) {
539 if t.pieceAllDirty(piece) {
542 return t.pieces[piece].hasDirtyChunks()
545 func (t *Torrent) usualPieceSize() int {
546 return int(t.info.PieceLength)
549 func (t *Torrent) lastPieceSize() int {
550 return int(t.pieceLength(t.numPieces() - 1))
553 func (t *Torrent) numPieces() int {
554 return t.info.NumPieces()
557 func (t *Torrent) numPiecesCompleted() (num int) {
558 return t.completedPieces.Len()
561 func (t *Torrent) close() (err error) {
563 if t.storage != nil {
566 for _, conn := range t.conns {
569 t.pieceStateChanges.Close()
570 t.updateWantPeersEvent()
574 func (t *Torrent) requestOffset(r request) int64 {
575 return torrentRequestOffset(t.length, int64(t.usualPieceSize()), r)
578 // Return the request that would include the given offset into the torrent
579 // data. Returns !ok if there is no such request.
580 func (t *Torrent) offsetRequest(off int64) (req request, ok bool) {
581 return torrentOffsetRequest(t.length, t.info.PieceLength, int64(t.chunkSize), off)
584 func (t *Torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
585 tr := perf.NewTimer()
587 n, err := t.pieces[piece].Storage().WriteAt(data, begin)
588 if err == nil && n != len(data) {
589 err = io.ErrShortWrite
592 tr.Stop("write chunk")
597 func (t *Torrent) bitfield() (bf []bool) {
598 bf = make([]bool, t.numPieces())
599 t.completedPieces.IterTyped(func(piece int) (again bool) {
606 func (t *Torrent) validOutgoingRequest(r request) bool {
607 if r.Index >= pp.Integer(t.info.NumPieces()) {
610 if r.Begin%t.chunkSize != 0 {
613 if r.Length > t.chunkSize {
616 pieceLength := t.pieceLength(int(r.Index))
617 if r.Begin+r.Length > pieceLength {
620 return r.Length == t.chunkSize || r.Begin+r.Length == pieceLength
623 func (t *Torrent) pieceChunks(piece int) (css []chunkSpec) {
624 css = make([]chunkSpec, 0, (t.pieceLength(piece)+t.chunkSize-1)/t.chunkSize)
626 for left := t.pieceLength(piece); left != 0; left -= cs.Length {
628 if cs.Length > t.chunkSize {
629 cs.Length = t.chunkSize
631 css = append(css, cs)
632 cs.Begin += cs.Length
637 func (t *Torrent) pieceNumChunks(piece int) int {
638 return int((t.pieceLength(piece) + t.chunkSize - 1) / t.chunkSize)
641 func (t *Torrent) pendAllChunkSpecs(pieceIndex int) {
642 t.pieces[pieceIndex].DirtyChunks.Clear()
650 // Peer is known to support encryption.
651 SupportsEncryption bool
654 func (t *Torrent) pieceLength(piece int) (len_ pp.Integer) {
655 if piece < 0 || piece >= t.info.NumPieces() {
658 if piece == t.numPieces()-1 {
659 len_ = pp.Integer(t.length % t.info.PieceLength)
662 len_ = pp.Integer(t.info.PieceLength)
667 func (t *Torrent) hashPiece(piece int) (ret metainfo.Hash) {
668 hash := pieceHash.New()
669 p := &t.pieces[piece]
670 p.waitNoPendingWrites()
671 ip := t.info.Piece(piece)
673 n, err := io.Copy(hash, io.NewSectionReader(t.pieces[piece].Storage(), 0, pl))
675 missinggo.CopyExact(&ret, hash.Sum(nil))
678 if err != io.ErrUnexpectedEOF && !os.IsNotExist(err) {
679 log.Printf("unexpected error hashing piece with %T: %s", t.storage, err)
684 func (t *Torrent) haveAllPieces() bool {
688 return t.completedPieces.Len() == t.numPieces()
691 func (t *Torrent) haveAnyPieces() bool {
692 for i := range t.pieces {
693 if t.pieceComplete(i) {
700 func (t *Torrent) havePiece(index int) bool {
701 return t.haveInfo() && t.pieceComplete(index)
704 func (t *Torrent) haveChunk(r request) (ret bool) {
706 // log.Println("have chunk", r, ret)
711 if t.pieceComplete(int(r.Index)) {
714 p := &t.pieces[r.Index]
715 return !p.pendingChunk(r.chunkSpec, t.chunkSize)
718 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
719 return int(cs.Begin / chunkSize)
722 func (t *Torrent) wantPiece(r request) bool {
723 if !t.wantPieceIndex(int(r.Index)) {
726 if t.pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) {
729 // TODO: What about pieces that were wanted, but aren't now, and aren't
730 // completed either? That used to be done here.
734 func (t *Torrent) wantPieceIndex(index int) bool {
738 p := &t.pieces[index]
745 if t.pieceComplete(index) {
748 if t.pendingPieces.Contains(index) {
751 return !t.forReaderOffsetPieces(func(begin, end int) bool {
752 return index < begin || index >= end
756 func (t *Torrent) forNeededPieces(f func(piece int) (more bool)) (all bool) {
757 return t.forReaderOffsetPieces(func(begin, end int) (more bool) {
758 for i := begin; begin < end; i++ {
767 func (t *Torrent) connHasWantedPieces(c *connection) bool {
768 return !c.pieceRequestOrder.IsEmpty()
771 func (t *Torrent) extentPieces(off, _len int64) (pieces []int) {
772 for i := off / int64(t.usualPieceSize()); i*int64(t.usualPieceSize()) < off+_len; i++ {
773 pieces = append(pieces, int(i))
778 // The worst connection is one that hasn't been sent, or sent anything useful
779 // for the longest. A bad connection is one that usually sends us unwanted
780 // pieces, or has been in worser half of the established connections for more
782 func (t *Torrent) worstBadConn() *connection {
783 wcs := slices.HeapInterface(t.worstUnclosedConns(), worseConn)
785 c := heap.Pop(wcs).(*connection)
786 if c.UnwantedChunksReceived >= 6 && c.UnwantedChunksReceived > c.UsefulChunksReceived {
789 if wcs.Len() >= (t.maxEstablishedConns+1)/2 {
790 // Give connections 1 minute to prove themselves.
791 if time.Since(c.completedHandshake) > time.Minute {
799 type PieceStateChange struct {
804 func (t *Torrent) publishPieceChange(piece int) {
805 cur := t.pieceState(piece)
806 p := &t.pieces[piece]
807 if cur != p.PublicPieceState {
808 p.PublicPieceState = cur
809 t.pieceStateChanges.Publish(PieceStateChange{
816 func (t *Torrent) pieceNumPendingChunks(piece int) int {
817 if t.pieceComplete(piece) {
820 return t.pieceNumChunks(piece) - t.pieces[piece].numDirtyChunks()
823 func (t *Torrent) pieceAllDirty(piece int) bool {
824 return t.pieces[piece].DirtyChunks.Len() == t.pieceNumChunks(piece)
827 func (t *Torrent) forUrgentPieces(f func(piece int) (again bool)) (all bool) {
828 return t.forReaderOffsetPieces(func(begin, end int) (again bool) {
838 func (t *Torrent) readersChanged() {
839 t.readerNowPieces, t.readerReadaheadPieces = t.readerPiecePriorities()
840 t.updatePiecePriorities()
843 func (t *Torrent) maybeNewConns() {
844 // Tickle the accept routine.
845 t.cl.event.Broadcast()
849 func (t *Torrent) piecePriorityChanged(piece int) {
850 for _, c := range t.conns {
851 c.updatePiecePriority(piece)
854 t.publishPieceChange(piece)
857 func (t *Torrent) updatePiecePriority(piece int) {
858 p := &t.pieces[piece]
859 newPrio := t.piecePriorityUncached(piece)
860 if newPrio == p.priority {
864 t.piecePriorityChanged(piece)
867 // Update all piece priorities in one hit. This function should have the same
868 // output as updatePiecePriority, but across all pieces.
869 func (t *Torrent) updatePiecePriorities() {
870 for i := range t.pieces {
871 t.updatePiecePriority(i)
875 // Returns the range of pieces [begin, end) that contains the extent of bytes.
876 func (t *Torrent) byteRegionPieces(off, size int64) (begin, end int) {
887 begin = int(off / t.info.PieceLength)
888 end = int((off + size + t.info.PieceLength - 1) / t.info.PieceLength)
889 if end > t.info.NumPieces() {
890 end = t.info.NumPieces()
895 // Returns true if all iterations complete without breaking. Returns the read
896 // regions for all readers. The reader regions should not be merged as some
897 // callers depend on this method to enumerate readers.
898 func (t *Torrent) forReaderOffsetPieces(f func(begin, end int) (more bool)) (all bool) {
899 for r := range t.readers {
901 if p.begin >= p.end {
904 if !f(p.begin, p.end) {
911 func (t *Torrent) piecePriority(piece int) piecePriority {
913 return PiecePriorityNone
915 return t.pieces[piece].priority
918 func (t *Torrent) piecePriorityUncached(piece int) piecePriority {
919 if t.pieceComplete(piece) {
920 return PiecePriorityNone
922 if t.readerNowPieces.Contains(piece) {
923 return PiecePriorityNow
925 // if t.readerNowPieces.Contains(piece - 1) {
926 // return PiecePriorityNext
928 if t.readerReadaheadPieces.Contains(piece) {
929 return PiecePriorityReadahead
931 if t.pendingPieces.Contains(piece) {
932 return PiecePriorityNormal
934 return PiecePriorityNone
937 func (t *Torrent) pendPiece(piece int) {
938 if t.pendingPieces.Contains(piece) {
941 if t.havePiece(piece) {
944 t.pendingPieces.Add(piece)
945 t.updatePiecePriority(piece)
948 func (t *Torrent) getCompletedPieces() (ret bitmap.Bitmap) {
949 return t.completedPieces.Copy()
952 func (t *Torrent) unpendPieces(unpend *bitmap.Bitmap) {
953 t.pendingPieces.Sub(unpend)
954 t.updatePiecePriorities()
957 func (t *Torrent) pendPieceRange(begin, end int) {
958 for i := begin; i < end; i++ {
963 func (t *Torrent) unpendPieceRange(begin, end int) {
965 bm.AddRange(begin, end)
969 func (t *Torrent) connRequestPiecePendingChunks(c *connection, piece int) (more bool) {
970 if !c.PeerHasPiece(piece) {
973 chunkIndices := t.pieces[piece].undirtiedChunkIndices().ToSortedSlice()
974 return itertools.ForPerm(len(chunkIndices), func(i int) bool {
975 req := request{pp.Integer(piece), t.chunkIndexSpec(chunkIndices[i], piece)}
976 return c.Request(req)
980 func (t *Torrent) pendRequest(req request) {
981 ci := chunkIndex(req.chunkSpec, t.chunkSize)
982 t.pieces[req.Index].pendChunkIndex(ci)
985 func (t *Torrent) pieceChanged(piece int) {
986 t.cl.pieceChanged(t, piece)
989 func (t *Torrent) openNewConns() {
993 func (t *Torrent) getConnPieceInclination() []int {
994 _ret := t.connPieceInclinationPool.Get()
996 pieceInclinationsNew.Add(1)
997 return rand.Perm(t.numPieces())
999 pieceInclinationsReused.Add(1)
1003 func (t *Torrent) putPieceInclination(pi []int) {
1004 t.connPieceInclinationPool.Put(pi)
1005 pieceInclinationsPut.Add(1)
1008 func (t *Torrent) updatePieceCompletion(piece int) {
1009 pcu := t.pieceCompleteUncached(piece)
1010 changed := t.completedPieces.Get(piece) != pcu
1011 t.completedPieces.Set(piece, pcu)
1013 t.pieceChanged(piece)
1017 // Non-blocking read. Client lock is not required.
1018 func (t *Torrent) readAt(b []byte, off int64) (n int, err error) {
1019 p := &t.pieces[off/t.info.PieceLength]
1020 p.waitNoPendingWrites()
1021 return p.Storage().ReadAt(b, off-p.Info().Offset())
1024 func (t *Torrent) updateAllPieceCompletions() {
1025 for i := range iter.N(t.numPieces()) {
1026 t.updatePieceCompletion(i)
1030 // Returns an error if the metadata was completed, but couldn't be set for
1031 // some reason. Blame it on the last peer to contribute.
1032 func (t *Torrent) maybeCompleteMetadata() error {
1037 if !t.haveAllMetadataPieces() {
1038 // Don't have enough metadata pieces.
1041 err := t.setInfoBytes(t.metadataBytes)
1043 t.invalidateMetadata()
1044 return fmt.Errorf("error setting info bytes: %s", err)
1046 if t.cl.config.Debug {
1047 log.Printf("%s: got metadata from peers", t)
1052 func (t *Torrent) readerPieces() (ret bitmap.Bitmap) {
1053 t.forReaderOffsetPieces(func(begin, end int) bool {
1054 ret.AddRange(begin, end)
1060 func (t *Torrent) readerPiecePriorities() (now, readahead bitmap.Bitmap) {
1061 t.forReaderOffsetPieces(func(begin, end int) bool {
1064 readahead.AddRange(begin+1, end)
1071 func (t *Torrent) needData() bool {
1072 if t.closed.IsSet() {
1078 if t.pendingPieces.Len() != 0 {
1081 // Read as "not all complete".
1082 return !t.readerPieces().IterTyped(func(piece int) bool {
1083 return t.pieceComplete(piece)
1087 func appendMissingStrings(old, new []string) (ret []string) {
1090 for _, n := range new {
1091 for _, o := range old {
1096 ret = append(ret, n)
1101 func appendMissingTrackerTiers(existing [][]string, minNumTiers int) (ret [][]string) {
1103 for minNumTiers > len(ret) {
1104 ret = append(ret, nil)
1109 func (t *Torrent) addTrackers(announceList [][]string) {
1110 fullAnnounceList := &t.metainfo.AnnounceList
1111 t.metainfo.AnnounceList = appendMissingTrackerTiers(*fullAnnounceList, len(announceList))
1112 for tierIndex, trackerURLs := range announceList {
1113 (*fullAnnounceList)[tierIndex] = appendMissingStrings((*fullAnnounceList)[tierIndex], trackerURLs)
1115 t.startMissingTrackerScrapers()
1116 t.updateWantPeersEvent()
1119 // Don't call this before the info is available.
1120 func (t *Torrent) bytesCompleted() int64 {
1124 return t.info.TotalLength() - t.bytesLeft()
1127 func (t *Torrent) SetInfoBytes(b []byte) (err error) {
1129 defer t.cl.mu.Unlock()
1130 return t.setInfoBytes(b)
1133 // Returns true if connection is removed from torrent.Conns.
1134 func (t *Torrent) deleteConnection(c *connection) bool {
1135 for i0, _c := range t.conns {
1139 i1 := len(t.conns) - 1
1141 t.conns[i0] = t.conns[i1]
1143 t.conns = t.conns[:i1]
1149 func (t *Torrent) dropConnection(c *connection) {
1150 t.cl.event.Broadcast()
1152 if t.deleteConnection(c) {
1157 func (t *Torrent) wantPeers() bool {
1158 if t.closed.IsSet() {
1161 if len(t.peers) > torrentPeersLowWater {
1164 return t.needData() || t.seeding()
1167 func (t *Torrent) updateWantPeersEvent() {
1169 t.wantPeersEvent.Set()
1171 t.wantPeersEvent.Clear()
1175 // Returns whether the client should make effort to seed the torrent.
1176 func (t *Torrent) seeding() bool {
1178 if t.closed.IsSet() {
1181 if cl.config.NoUpload {
1184 if !cl.config.Seed {
1193 // Adds and starts tracker scrapers for tracker URLs that aren't already
1195 func (t *Torrent) startMissingTrackerScrapers() {
1196 if t.cl.config.DisableTrackers {
1199 for _, tier := range t.announceList() {
1200 for _, trackerURL := range tier {
1201 if _, ok := t.trackerAnnouncers[trackerURL]; ok {
1204 newAnnouncer := &trackerScraper{
1208 if t.trackerAnnouncers == nil {
1209 t.trackerAnnouncers = make(map[string]*trackerScraper)
1211 t.trackerAnnouncers[trackerURL] = newAnnouncer
1212 go newAnnouncer.Run()
1217 // Returns an AnnounceRequest with fields filled out to defaults and current
1219 func (t *Torrent) announceRequest() tracker.AnnounceRequest {
1220 return tracker.AnnounceRequest{
1221 Event: tracker.None,
1223 Port: uint16(t.cl.incomingPeerPort()),
1224 PeerId: t.cl.peerID,
1225 InfoHash: t.infoHash,
1226 Left: t.bytesLeftAnnounce(),
1230 // Adds peers revealed in an announce until the announce ends, or we have
1232 func (t *Torrent) consumeDHTAnnounce(pvs <-chan dht.PeersValues) {
1234 // Count all the unique addresses we got during this announce.
1235 allAddrs := make(map[string]struct{})
1238 case v, ok := <-pvs:
1242 addPeers := make([]Peer, 0, len(v.Peers))
1243 for _, cp := range v.Peers {
1245 // Can't do anything with this.
1248 addPeers = append(addPeers, Peer{
1251 Source: peerSourceDHT,
1253 key := (&net.UDPAddr{
1257 allAddrs[key] = struct{}{}
1260 t.addPeers(addPeers)
1261 numPeers := len(t.peers)
1263 if numPeers >= torrentPeersHighWater {
1266 case <-t.closed.LockedChan(&cl.mu):
1272 func (t *Torrent) announceDHT(impliedPort bool) (err error) {
1274 ps, err := cl.dHT.Announce(string(t.infoHash[:]), cl.incomingPeerPort(), impliedPort)
1278 t.consumeDHTAnnounce(ps.Peers)
1283 func (t *Torrent) dhtAnnouncer() {
1287 case <-t.wantPeersEvent.LockedChan(&cl.mu):
1288 case <-t.closed.LockedChan(&cl.mu):
1291 err := t.announceDHT(true)
1297 log.Printf("error announcing %q to DHT: %s", t, err)
1300 case <-t.closed.LockedChan(&cl.mu):
1302 case <-time.After(5 * time.Minute):
1307 func (t *Torrent) addPeers(peers []Peer) {
1308 for _, p := range peers {
1309 if t.cl.badPeerIPPort(p.IP, p.Port) {
1316 func (t *Torrent) Stats() TorrentStats {
1318 defer t.cl.mu.Unlock()
1322 // Returns true if the connection is added.
1323 func (t *Torrent) addConnection(c *connection) bool {
1324 if t.cl.closed.IsSet() {
1330 for _, c0 := range t.conns {
1331 if c.PeerID == c0.PeerID {
1332 // Already connected to a client with that ID.
1333 duplicateClientConns.Add(1)
1337 if len(t.conns) >= t.maxEstablishedConns {
1338 c := t.worstBadConn()
1342 if t.cl.config.Debug && missinggo.CryHeard() {
1343 log.Printf("%s: dropping connection to make room for new one:\n %s", t, c)
1346 t.deleteConnection(c)
1348 if len(t.conns) >= t.maxEstablishedConns {
1351 t.conns = append(t.conns, c)
1353 panic("connection already associated with a torrent")
1355 // Reconcile bytes transferred before connection was associated with a
1357 t.stats.wroteBytes(c.stats.BytesWritten)
1358 t.stats.readBytes(c.stats.BytesRead)
1363 func (t *Torrent) wantConns() bool {
1364 if t.closed.IsSet() {
1367 if !t.seeding() && !t.needData() {
1370 if len(t.conns) < t.maxEstablishedConns {
1373 return t.worstBadConn() != nil
1376 func (t *Torrent) SetMaxEstablishedConns(max int) (oldMax int) {
1378 defer t.cl.mu.Unlock()
1379 oldMax = t.maxEstablishedConns
1380 t.maxEstablishedConns = max
1381 wcs := slices.HeapInterface(append([]*connection(nil), t.conns...), worseConn)
1382 for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 {
1383 t.dropConnection(wcs.Pop().(*connection))
1389 func (t *Torrent) mu() missinggo.RWLocker {