13 "github.com/anacrolix/missinggo"
14 "github.com/bradfitz/iter"
16 "github.com/anacrolix/torrent/bencode"
17 "github.com/anacrolix/torrent/data"
18 "github.com/anacrolix/torrent/metainfo"
19 pp "github.com/anacrolix/torrent/peer_protocol"
20 "github.com/anacrolix/torrent/tracker"
21 "github.com/anacrolix/torrent/util"
24 func (t *torrent) pieceNumPendingBytes(index int) (count pp.Integer) {
25 if t.pieceComplete(index) {
28 piece := t.Pieces[index]
29 pieceLength := t.pieceLength(index)
30 if !piece.EverHashed {
33 for i, pending := range piece.PendingChunkSpecs {
35 count += chunkIndexSpec(i, pieceLength, t.chunkSize).Length
41 type peersKey struct {
46 // Data maintains per-piece persistent state.
47 type StatefulData interface {
49 // We believe the piece data will pass a hash check.
50 PieceCompleted(index int) error
51 // Returns true if the piece is complete.
52 PieceComplete(index int) bool
55 // Is not aware of Client. Maintains state of torrent for with-in a Client.
60 // Closed when no more network activity is desired. This includes
61 // announcing, and communicating with peers.
62 ceasingNetworking chan struct{}
67 // Chunks that are wanted before all others. This is for
68 // responsive/streaming readers that want to unblock ASAP.
69 urgent map[request]struct{}
70 // Total length of the torrent in bytes. Stored because it's not O(1) to
71 // get this from the info dict.
76 // The info dict. Nil if we don't have it (yet).
78 // Active peer connections, running message stream loops.
80 // Set of addrs to which we're attempting to connect. Connections are
81 // half-open until all handshakes are completed.
82 HalfOpen map[string]struct{}
84 // Reserve of peers to connect to. A peer can be both here and in the
85 // active connections if were told about the peer after connecting with
86 // them. That encourages us to reconnect to peers that are well known.
87 Peers map[peersKey]Peer
90 // BEP 12 Multitracker Metadata Extension. The tracker.Client instances
91 // mirror their respective URLs from the announce-list metainfo key.
92 Trackers [][]tracker.Client
93 // Name used if the info name isn't available.
95 // The bencoded bytes of the info dict.
97 // Each element corresponds to the 16KiB metadata pieces. If true, we have
98 // received that piece.
101 // Closed when .Info is set.
102 gotMetainfo chan struct{}
105 func (t *torrent) pieceComplete(piece int) bool {
106 // TODO: This is called when setting metadata, and before storage is
107 // assigned, which doesn't seem right.
108 return t.data != nil && t.data.PieceComplete(piece)
111 func (t *torrent) numConnsUnchoked() (num int) {
112 for _, c := range t.Conns {
120 // There's a connection to that address already.
121 func (t *torrent) addrActive(addr string) bool {
122 if _, ok := t.HalfOpen[addr]; ok {
125 for _, c := range t.Conns {
126 if c.remoteAddr().String() == addr {
133 func (t *torrent) worstConns(cl *Client) (wcs *worstConns) {
135 c: make([]*connection, 0, len(t.Conns)),
139 for _, c := range t.Conns {
143 wcs.c = append(wcs.c, c)
149 func (t *torrent) ceaseNetworking() {
151 defer t.stateMu.Unlock()
153 case <-t.ceasingNetworking:
157 close(t.ceasingNetworking)
158 for _, c := range t.Conns {
163 func (t *torrent) addPeer(p Peer) {
164 t.Peers[peersKey{string(p.IP), p.Port}] = p
167 func (t *torrent) invalidateMetadata() {
173 func (t *torrent) saveMetadataPiece(index int, data []byte) {
177 if index >= len(t.metadataHave) {
178 log.Printf("%s: ignoring metadata piece %d", t, index)
181 copy(t.MetaData[(1<<14)*index:], data)
182 t.metadataHave[index] = true
185 func (t *torrent) metadataPieceCount() int {
186 return (len(t.MetaData) + (1 << 14) - 1) / (1 << 14)
189 func (t *torrent) haveMetadataPiece(piece int) bool {
191 return (1<<14)*piece < len(t.MetaData)
193 return piece < len(t.metadataHave) && t.metadataHave[piece]
197 func (t *torrent) metadataSizeKnown() bool {
198 return t.MetaData != nil
201 func (t *torrent) metadataSize() int {
202 return len(t.MetaData)
205 func infoPieceHashes(info *metainfo.Info) (ret []string) {
206 for i := 0; i < len(info.Pieces); i += 20 {
207 ret = append(ret, string(info.Pieces[i:i+20]))
212 // Called when metadata for a torrent becomes available.
213 func (t *torrent) setMetadata(md *metainfo.Info, infoBytes []byte, eventLocker sync.Locker) (err error) {
214 err = validateInfo(md)
216 err = fmt.Errorf("bad info: %s", err)
221 for _, f := range t.Info.UpvertedFiles() {
224 t.MetaData = infoBytes
226 for _, hash := range infoPieceHashes(md) {
228 piece.Event.L = eventLocker
229 util.CopyExact(piece.Hash[:], hash)
230 t.Pieces = append(t.Pieces, piece)
232 for _, conn := range t.Conns {
233 t.initRequestOrdering(conn)
234 if err := conn.setNumPieces(t.numPieces()); err != nil {
235 log.Printf("closing connection: %s", err)
242 func (t *torrent) setStorage(td data.Data) (err error) {
243 if c, ok := t.data.(io.Closer); ok {
246 if sd, ok := td.(StatefulData); ok {
249 t.data = &statelessDataWrapper{td, make([]bool, t.Info.NumPieces())}
254 func (t *torrent) haveAllMetadataPieces() bool {
258 if t.metadataHave == nil {
261 for _, have := range t.metadataHave {
269 func (t *torrent) setMetadataSize(bytes int64, cl *Client) {
271 // We already know the correct metadata size.
274 if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
275 log.Printf("received bad metadata size: %d", bytes)
278 if t.MetaData != nil && len(t.MetaData) == int(bytes) {
281 t.MetaData = make([]byte, bytes)
282 t.metadataHave = make([]bool, (bytes+(1<<14)-1)/(1<<14))
283 for _, c := range t.Conns {
284 cl.requestPendingMetadata(t, c)
289 // The current working name for the torrent. Either the name in the info dict,
290 // or a display name given such as by the dn value in a magnet link, or "".
291 func (t *torrent) Name() string {
295 if t.DisplayName != "" {
301 func (t *torrent) pieceState(index int) (ret PieceState) {
303 ret.Priority = p.Priority
304 if t.pieceComplete(index) {
307 if p.QueuedForHash || p.Hashing {
310 if t.piecePartiallyDownloaded(index) {
316 func (t *torrent) metadataPieceSize(piece int) int {
317 return metadataPieceSize(len(t.MetaData), piece)
320 func (t *torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
326 d["total_size"] = len(t.MetaData)
328 p, err := bencode.Marshal(d)
334 ExtendedID: byte(c.PeerExtensionIDs["ut_metadata"]),
335 ExtendedPayload: append(p, data...),
339 func (t *torrent) pieceStateRuns() (ret []PieceStateRun) {
340 rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
341 ret = append(ret, PieceStateRun{
342 PieceState: el.(PieceState),
346 for index := range t.Pieces {
347 rle.Append(t.pieceState(index), 1)
353 // Produces a small string representing a PieceStateRun.
354 func pieceStateRunStatusChars(psr PieceStateRun) (ret string) {
355 ret = fmt.Sprintf("%d", psr.Length)
356 ret += func() string {
357 switch psr.Priority {
358 case PiecePriorityNext:
360 case PiecePriorityNormal:
362 case PiecePriorityReadahead:
364 case PiecePriorityNow:
382 func (t *torrent) writeStatus(w io.Writer, cl *Client) {
383 fmt.Fprintf(w, "Infohash: %x\n", t.InfoHash)
384 fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
386 fmt.Fprintf(w, "Metadata have: ")
387 for _, h := range t.metadataHave {
388 fmt.Fprintf(w, "%c", func() rune {
398 fmt.Fprintf(w, "Piece length: %s\n", func() string {
400 return fmt.Sprint(t.usualPieceSize())
406 fmt.Fprint(w, "Pieces:")
407 for _, psr := range t.pieceStateRuns() {
409 w.Write([]byte(pieceStateRunStatusChars(psr)))
413 fmt.Fprintf(w, "Urgent:")
414 for req := range t.urgent {
415 fmt.Fprintf(w, " %v", req)
418 fmt.Fprintf(w, "Trackers: ")
419 for _, tier := range t.Trackers {
420 for _, tr := range tier {
421 fmt.Fprintf(w, "%q ", tr.String())
425 fmt.Fprintf(w, "Pending peers: %d\n", len(t.Peers))
426 fmt.Fprintf(w, "Half open: %d\n", len(t.HalfOpen))
427 fmt.Fprintf(w, "Active peers: %d\n", len(t.Conns))
428 sort.Sort(&worstConns{
433 for i, c := range t.Conns {
434 fmt.Fprintf(w, "%2d. ", i+1)
439 func (t *torrent) String() string {
442 s = fmt.Sprintf("%x", t.InfoHash)
447 func (t *torrent) haveInfo() bool {
448 return t != nil && t.Info != nil
451 // TODO: Include URIs that weren't converted to tracker clients.
452 func (t *torrent) announceList() (al [][]string) {
453 for _, tier := range t.Trackers {
455 for _, tr := range tier {
456 l = append(l, tr.URL())
463 // Returns a run-time generated MetaInfo that includes the info bytes and
464 // announce-list as currently known to the client.
465 func (t *torrent) MetaInfo() *metainfo.MetaInfo {
466 if t.MetaData == nil {
467 panic("info bytes not set")
469 return &metainfo.MetaInfo{
470 Info: metainfo.InfoEx{
474 CreationDate: time.Now().Unix(),
475 Comment: "dynamic metainfo from client",
476 CreatedBy: "go.torrent",
477 AnnounceList: t.announceList(),
481 func (t *torrent) bytesLeft() (left int64) {
485 for i := 0; i < t.numPieces(); i++ {
486 left += int64(t.pieceNumPendingBytes(i))
491 func (t *torrent) piecePartiallyDownloaded(index int) bool {
492 pendingBytes := t.pieceNumPendingBytes(index)
493 return pendingBytes != 0 && pendingBytes != t.pieceLength(index)
496 func numChunksForPiece(chunkSize int, pieceSize int) int {
497 return (pieceSize + chunkSize - 1) / chunkSize
500 func (t *torrent) usualPieceSize() int {
501 return int(t.Info.PieceLength)
504 func (t *torrent) lastPieceSize() int {
505 return int(t.pieceLength(t.numPieces() - 1))
508 func (t *torrent) numPieces() int {
509 return t.Info.NumPieces()
512 func (t *torrent) numPiecesCompleted() (num int) {
513 for i := range iter.N(t.Info.NumPieces()) {
514 if t.pieceComplete(i) {
521 func (t *torrent) Length() int64 {
525 func (t *torrent) isClosed() bool {
534 func (t *torrent) close() (err error) {
540 if c, ok := t.data.(io.Closer); ok {
543 for _, conn := range t.Conns {
549 func (t *torrent) requestOffset(r request) int64 {
550 return torrentRequestOffset(t.Length(), int64(t.usualPieceSize()), r)
553 // Return the request that would include the given offset into the torrent
554 // data. Returns !ok if there is no such request.
555 func (t *torrent) offsetRequest(off int64) (req request, ok bool) {
556 return torrentOffsetRequest(t.Length(), t.Info.PieceLength, int64(t.chunkSize), off)
559 func (t *torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
560 n, err := t.data.WriteAt(data, int64(piece)*t.Info.PieceLength+begin)
561 if err == nil && n != len(data) {
562 err = io.ErrShortWrite
567 func (t *torrent) bitfield() (bf []bool) {
568 for _, p := range t.Pieces {
569 // TODO: Check this logic.
570 bf = append(bf, p.EverHashed && p.numPendingChunks() == 0)
575 func (t *torrent) validOutgoingRequest(r request) bool {
576 if r.Index >= pp.Integer(t.Info.NumPieces()) {
579 if r.Begin%t.chunkSize != 0 {
582 if r.Length > t.chunkSize {
585 pieceLength := t.pieceLength(int(r.Index))
586 if r.Begin+r.Length > pieceLength {
589 return r.Length == t.chunkSize || r.Begin+r.Length == pieceLength
592 func (t *torrent) pieceChunks(piece int) (css []chunkSpec) {
593 css = make([]chunkSpec, 0, (t.pieceLength(piece)+t.chunkSize-1)/t.chunkSize)
595 for left := t.pieceLength(piece); left != 0; left -= cs.Length {
597 if cs.Length > t.chunkSize {
598 cs.Length = t.chunkSize
600 css = append(css, cs)
601 cs.Begin += cs.Length
606 func (t *torrent) pendAllChunkSpecs(pieceIndex int) {
607 piece := t.Pieces[pieceIndex]
608 if piece.PendingChunkSpecs == nil {
609 // Allocate to exact size.
610 piece.PendingChunkSpecs = make([]bool, (t.pieceLength(pieceIndex)+t.chunkSize-1)/t.chunkSize)
612 // Pend all the chunks.
613 pcss := piece.PendingChunkSpecs
614 for i := range pcss {
625 // Peer is known to support encryption.
626 SupportsEncryption bool
629 func (t *torrent) pieceLength(piece int) (len_ pp.Integer) {
630 if int(piece) == t.numPieces()-1 {
631 len_ = pp.Integer(t.Length() % t.Info.PieceLength)
634 len_ = pp.Integer(t.Info.PieceLength)
639 func (t *torrent) hashPiece(piece pp.Integer) (ps pieceSum) {
640 hash := pieceHash.New()
641 t.data.WriteSectionTo(hash, int64(piece)*t.Info.PieceLength, t.Info.PieceLength)
642 util.CopyExact(ps[:], hash.Sum(nil))
646 func (t *torrent) haveAllPieces() bool {
650 for i := range t.Pieces {
651 if !t.pieceComplete(i) {
658 func (me *torrent) haveAnyPieces() bool {
659 for i := range me.Pieces {
660 if me.pieceComplete(i) {
667 func (t *torrent) havePiece(index int) bool {
668 return t.haveInfo() && t.pieceComplete(index)
671 func (t *torrent) haveChunk(r request) bool {
675 p := t.Pieces[r.Index]
676 return !p.pendingChunk(r.chunkSpec, t.chunkSize)
679 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
680 return int(cs.Begin / chunkSize)
683 // TODO: This should probably be called wantPiece.
684 func (t *torrent) wantChunk(r request) bool {
685 if !t.wantPiece(int(r.Index)) {
688 if t.Pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) {
695 func (t *torrent) urgentChunkInPiece(piece int) bool {
696 p := pp.Integer(piece)
697 for req := range t.urgent {
705 // TODO: This should be called wantPieceIndex.
706 func (t *torrent) wantPiece(index int) bool {
717 if p.Priority == PiecePriorityNone {
718 if !t.urgentChunkInPiece(index) {
722 // Put piece complete check last, since it's the slowest as it can involve
723 // calling out into external data stores.
724 return !t.pieceComplete(index)
727 func (t *torrent) connHasWantedPieces(c *connection) bool {
728 return c.pieceRequestOrder != nil && c.pieceRequestOrder.First() != nil
731 func (t *torrent) extentPieces(off, _len int64) (pieces []int) {
732 for i := off / int64(t.usualPieceSize()); i*int64(t.usualPieceSize()) < off+_len; i++ {
733 pieces = append(pieces, int(i))
738 func (t *torrent) worstBadConn(cl *Client) *connection {
739 wcs := t.worstConns(cl)
741 // A connection can only be bad if it's in the worst half, rounded down.
742 for wcs.Len() > (socketsPerTorrent+1)/2 {
743 c := heap.Pop(wcs).(*connection)
744 // Give connections 1 minute to prove themselves.
745 if time.Since(c.completedHandshake) < time.Minute {