]> Sergey Matveev's repositories - btrtrc.git/blob - torrent.go
Allow chunk size to be specified per torrent
[btrtrc.git] / torrent.go
1 package torrent
2
3 import (
4         "container/heap"
5         "fmt"
6         "io"
7         "log"
8         "net"
9         "sort"
10         "sync"
11         "time"
12
13         "github.com/anacrolix/missinggo"
14         "github.com/bradfitz/iter"
15
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"
22 )
23
24 func (t *torrent) pieceNumPendingBytes(index int) (count pp.Integer) {
25         if t.pieceComplete(index) {
26                 return 0
27         }
28         piece := t.Pieces[index]
29         pieceLength := t.pieceLength(index)
30         if !piece.EverHashed {
31                 return pieceLength
32         }
33         for i, pending := range piece.PendingChunkSpecs {
34                 if pending {
35                         count += chunkIndexSpec(i, pieceLength, t.chunkSize).Length
36                 }
37         }
38         return
39 }
40
41 type peersKey struct {
42         IPBytes string
43         Port    int
44 }
45
46 // Data maintains per-piece persistent state.
47 type StatefulData interface {
48         data.Data
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
53 }
54
55 // Is not aware of Client. Maintains state of torrent for with-in a Client.
56 type torrent struct {
57         stateMu sync.Mutex
58         closing chan struct{}
59
60         // Closed when no more network activity is desired. This includes
61         // announcing, and communicating with peers.
62         ceasingNetworking chan struct{}
63
64         InfoHash  InfoHash
65         Pieces    []*piece
66         chunkSize pp.Integer
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.
72         length int64
73
74         data StatefulData
75
76         // The info dict. Nil if we don't have it (yet).
77         Info *metainfo.Info
78         // Active peer connections, running message stream loops.
79         Conns []*connection
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{}
83
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
88         wantPeers sync.Cond
89
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.
94         DisplayName string
95         // The bencoded bytes of the info dict.
96         MetaData []byte
97         // Each element corresponds to the 16KiB metadata pieces. If true, we have
98         // received that piece.
99         metadataHave []bool
100
101         // Closed when .Info is set.
102         gotMetainfo chan struct{}
103 }
104
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)
109 }
110
111 func (t *torrent) numConnsUnchoked() (num int) {
112         for _, c := range t.Conns {
113                 if !c.PeerChoked {
114                         num++
115                 }
116         }
117         return
118 }
119
120 // There's a connection to that address already.
121 func (t *torrent) addrActive(addr string) bool {
122         if _, ok := t.HalfOpen[addr]; ok {
123                 return true
124         }
125         for _, c := range t.Conns {
126                 if c.remoteAddr().String() == addr {
127                         return true
128                 }
129         }
130         return false
131 }
132
133 func (t *torrent) worstConns(cl *Client) (wcs *worstConns) {
134         wcs = &worstConns{
135                 c:  make([]*connection, 0, len(t.Conns)),
136                 t:  t,
137                 cl: cl,
138         }
139         for _, c := range t.Conns {
140                 select {
141                 case <-c.closing:
142                 default:
143                         wcs.c = append(wcs.c, c)
144                 }
145         }
146         return
147 }
148
149 func (t *torrent) ceaseNetworking() {
150         t.stateMu.Lock()
151         defer t.stateMu.Unlock()
152         select {
153         case <-t.ceasingNetworking:
154                 return
155         default:
156         }
157         close(t.ceasingNetworking)
158         for _, c := range t.Conns {
159                 c.Close()
160         }
161 }
162
163 func (t *torrent) addPeer(p Peer) {
164         t.Peers[peersKey{string(p.IP), p.Port}] = p
165 }
166
167 func (t *torrent) invalidateMetadata() {
168         t.MetaData = nil
169         t.metadataHave = nil
170         t.Info = nil
171 }
172
173 func (t *torrent) saveMetadataPiece(index int, data []byte) {
174         if t.haveInfo() {
175                 return
176         }
177         if index >= len(t.metadataHave) {
178                 log.Printf("%s: ignoring metadata piece %d", t, index)
179                 return
180         }
181         copy(t.MetaData[(1<<14)*index:], data)
182         t.metadataHave[index] = true
183 }
184
185 func (t *torrent) metadataPieceCount() int {
186         return (len(t.MetaData) + (1 << 14) - 1) / (1 << 14)
187 }
188
189 func (t *torrent) haveMetadataPiece(piece int) bool {
190         if t.haveInfo() {
191                 return (1<<14)*piece < len(t.MetaData)
192         } else {
193                 return piece < len(t.metadataHave) && t.metadataHave[piece]
194         }
195 }
196
197 func (t *torrent) metadataSizeKnown() bool {
198         return t.MetaData != nil
199 }
200
201 func (t *torrent) metadataSize() int {
202         return len(t.MetaData)
203 }
204
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]))
208         }
209         return
210 }
211
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)
215         if err != nil {
216                 err = fmt.Errorf("bad info: %s", err)
217                 return
218         }
219         t.Info = md
220         t.length = 0
221         for _, f := range t.Info.UpvertedFiles() {
222                 t.length += f.Length
223         }
224         t.MetaData = infoBytes
225         t.metadataHave = nil
226         for _, hash := range infoPieceHashes(md) {
227                 piece := &piece{}
228                 piece.Event.L = eventLocker
229                 util.CopyExact(piece.Hash[:], hash)
230                 t.Pieces = append(t.Pieces, piece)
231         }
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)
236                         conn.Close()
237                 }
238         }
239         return
240 }
241
242 func (t *torrent) setStorage(td data.Data) (err error) {
243         if c, ok := t.data.(io.Closer); ok {
244                 c.Close()
245         }
246         if sd, ok := td.(StatefulData); ok {
247                 t.data = sd
248         } else {
249                 t.data = &statelessDataWrapper{td, make([]bool, t.Info.NumPieces())}
250         }
251         return
252 }
253
254 func (t *torrent) haveAllMetadataPieces() bool {
255         if t.haveInfo() {
256                 return true
257         }
258         if t.metadataHave == nil {
259                 return false
260         }
261         for _, have := range t.metadataHave {
262                 if !have {
263                         return false
264                 }
265         }
266         return true
267 }
268
269 func (t *torrent) setMetadataSize(bytes int64, cl *Client) {
270         if t.haveInfo() {
271                 // We already know the correct metadata size.
272                 return
273         }
274         if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
275                 log.Printf("received bad metadata size: %d", bytes)
276                 return
277         }
278         if t.MetaData != nil && len(t.MetaData) == int(bytes) {
279                 return
280         }
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)
285         }
286
287 }
288
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 {
292         if t.haveInfo() {
293                 return t.Info.Name
294         }
295         if t.DisplayName != "" {
296                 return t.DisplayName
297         }
298         return ""
299 }
300
301 func (t *torrent) pieceState(index int) (ret PieceState) {
302         p := t.Pieces[index]
303         ret.Priority = p.Priority
304         if t.pieceComplete(index) {
305                 ret.Complete = true
306         }
307         if p.QueuedForHash || p.Hashing {
308                 ret.Checking = true
309         }
310         if t.piecePartiallyDownloaded(index) {
311                 ret.Partial = true
312         }
313         return
314 }
315
316 func (t *torrent) metadataPieceSize(piece int) int {
317         return metadataPieceSize(len(t.MetaData), piece)
318 }
319
320 func (t *torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
321         d := map[string]int{
322                 "msg_type": msgType,
323                 "piece":    piece,
324         }
325         if data != nil {
326                 d["total_size"] = len(t.MetaData)
327         }
328         p, err := bencode.Marshal(d)
329         if err != nil {
330                 panic(err)
331         }
332         return pp.Message{
333                 Type:            pp.Extended,
334                 ExtendedID:      byte(c.PeerExtensionIDs["ut_metadata"]),
335                 ExtendedPayload: append(p, data...),
336         }
337 }
338
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),
343                         Length:     int(count),
344                 })
345         })
346         for index := range t.Pieces {
347                 rle.Append(t.pieceState(index), 1)
348         }
349         rle.Flush()
350         return
351 }
352
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:
359                         return "N"
360                 case PiecePriorityNormal:
361                         return "."
362                 case PiecePriorityReadahead:
363                         return "R"
364                 case PiecePriorityNow:
365                         return "!"
366                 default:
367                         return ""
368                 }
369         }()
370         if psr.Checking {
371                 ret += "H"
372         }
373         if psr.Partial {
374                 ret += "P"
375         }
376         if psr.Complete {
377                 ret += "C"
378         }
379         return
380 }
381
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())
385         if !t.haveInfo() {
386                 fmt.Fprintf(w, "Metadata have: ")
387                 for _, h := range t.metadataHave {
388                         fmt.Fprintf(w, "%c", func() rune {
389                                 if h {
390                                         return 'H'
391                                 } else {
392                                         return '.'
393                                 }
394                         }())
395                 }
396                 fmt.Fprintln(w)
397         }
398         fmt.Fprintf(w, "Piece length: %s\n", func() string {
399                 if t.haveInfo() {
400                         return fmt.Sprint(t.usualPieceSize())
401                 } else {
402                         return "?"
403                 }
404         }())
405         if t.haveInfo() {
406                 fmt.Fprint(w, "Pieces:")
407                 for _, psr := range t.pieceStateRuns() {
408                         w.Write([]byte(" "))
409                         w.Write([]byte(pieceStateRunStatusChars(psr)))
410                 }
411                 fmt.Fprintln(w)
412         }
413         fmt.Fprintf(w, "Urgent:")
414         for req := range t.urgent {
415                 fmt.Fprintf(w, " %v", req)
416         }
417         fmt.Fprintln(w)
418         fmt.Fprintf(w, "Trackers: ")
419         for _, tier := range t.Trackers {
420                 for _, tr := range tier {
421                         fmt.Fprintf(w, "%q ", tr.String())
422                 }
423         }
424         fmt.Fprintf(w, "\n")
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{
429                 c:  t.Conns,
430                 t:  t,
431                 cl: cl,
432         })
433         for i, c := range t.Conns {
434                 fmt.Fprintf(w, "%2d. ", i+1)
435                 c.WriteStatus(w, t)
436         }
437 }
438
439 func (t *torrent) String() string {
440         s := t.Name()
441         if s == "" {
442                 s = fmt.Sprintf("%x", t.InfoHash)
443         }
444         return s
445 }
446
447 func (t *torrent) haveInfo() bool {
448         return t != nil && t.Info != nil
449 }
450
451 // TODO: Include URIs that weren't converted to tracker clients.
452 func (t *torrent) announceList() (al [][]string) {
453         for _, tier := range t.Trackers {
454                 var l []string
455                 for _, tr := range tier {
456                         l = append(l, tr.URL())
457                 }
458                 al = append(al, l)
459         }
460         return
461 }
462
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")
468         }
469         return &metainfo.MetaInfo{
470                 Info: metainfo.InfoEx{
471                         Info:  *t.Info,
472                         Bytes: t.MetaData,
473                 },
474                 CreationDate: time.Now().Unix(),
475                 Comment:      "dynamic metainfo from client",
476                 CreatedBy:    "go.torrent",
477                 AnnounceList: t.announceList(),
478         }
479 }
480
481 func (t *torrent) bytesLeft() (left int64) {
482         if !t.haveInfo() {
483                 return -1
484         }
485         for i := 0; i < t.numPieces(); i++ {
486                 left += int64(t.pieceNumPendingBytes(i))
487         }
488         return
489 }
490
491 func (t *torrent) piecePartiallyDownloaded(index int) bool {
492         pendingBytes := t.pieceNumPendingBytes(index)
493         return pendingBytes != 0 && pendingBytes != t.pieceLength(index)
494 }
495
496 func numChunksForPiece(chunkSize int, pieceSize int) int {
497         return (pieceSize + chunkSize - 1) / chunkSize
498 }
499
500 func (t *torrent) usualPieceSize() int {
501         return int(t.Info.PieceLength)
502 }
503
504 func (t *torrent) lastPieceSize() int {
505         return int(t.pieceLength(t.numPieces() - 1))
506 }
507
508 func (t *torrent) numPieces() int {
509         return t.Info.NumPieces()
510 }
511
512 func (t *torrent) numPiecesCompleted() (num int) {
513         for i := range iter.N(t.Info.NumPieces()) {
514                 if t.pieceComplete(i) {
515                         num++
516                 }
517         }
518         return
519 }
520
521 func (t *torrent) Length() int64 {
522         return t.length
523 }
524
525 func (t *torrent) isClosed() bool {
526         select {
527         case <-t.closing:
528                 return true
529         default:
530                 return false
531         }
532 }
533
534 func (t *torrent) close() (err error) {
535         if t.isClosed() {
536                 return
537         }
538         t.ceaseNetworking()
539         close(t.closing)
540         if c, ok := t.data.(io.Closer); ok {
541                 c.Close()
542         }
543         for _, conn := range t.Conns {
544                 conn.Close()
545         }
546         return
547 }
548
549 func (t *torrent) requestOffset(r request) int64 {
550         return torrentRequestOffset(t.Length(), int64(t.usualPieceSize()), r)
551 }
552
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)
557 }
558
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
563         }
564         return
565 }
566
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)
571         }
572         return
573 }
574
575 func (t *torrent) validOutgoingRequest(r request) bool {
576         if r.Index >= pp.Integer(t.Info.NumPieces()) {
577                 return false
578         }
579         if r.Begin%t.chunkSize != 0 {
580                 return false
581         }
582         if r.Length > t.chunkSize {
583                 return false
584         }
585         pieceLength := t.pieceLength(int(r.Index))
586         if r.Begin+r.Length > pieceLength {
587                 return false
588         }
589         return r.Length == t.chunkSize || r.Begin+r.Length == pieceLength
590 }
591
592 func (t *torrent) pieceChunks(piece int) (css []chunkSpec) {
593         css = make([]chunkSpec, 0, (t.pieceLength(piece)+t.chunkSize-1)/t.chunkSize)
594         var cs chunkSpec
595         for left := t.pieceLength(piece); left != 0; left -= cs.Length {
596                 cs.Length = left
597                 if cs.Length > t.chunkSize {
598                         cs.Length = t.chunkSize
599                 }
600                 css = append(css, cs)
601                 cs.Begin += cs.Length
602         }
603         return
604 }
605
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)
611         }
612         // Pend all the chunks.
613         pcss := piece.PendingChunkSpecs
614         for i := range pcss {
615                 pcss[i] = true
616         }
617         return
618 }
619
620 type Peer struct {
621         Id     [20]byte
622         IP     net.IP
623         Port   int
624         Source peerSource
625         // Peer is known to support encryption.
626         SupportsEncryption bool
627 }
628
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)
632         }
633         if len_ == 0 {
634                 len_ = pp.Integer(t.Info.PieceLength)
635         }
636         return
637 }
638
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))
643         return
644 }
645
646 func (t *torrent) haveAllPieces() bool {
647         if !t.haveInfo() {
648                 return false
649         }
650         for i := range t.Pieces {
651                 if !t.pieceComplete(i) {
652                         return false
653                 }
654         }
655         return true
656 }
657
658 func (me *torrent) haveAnyPieces() bool {
659         for i := range me.Pieces {
660                 if me.pieceComplete(i) {
661                         return true
662                 }
663         }
664         return false
665 }
666
667 func (t *torrent) havePiece(index int) bool {
668         return t.haveInfo() && t.pieceComplete(index)
669 }
670
671 func (t *torrent) haveChunk(r request) bool {
672         if !t.haveInfo() {
673                 return false
674         }
675         p := t.Pieces[r.Index]
676         return !p.pendingChunk(r.chunkSpec, t.chunkSize)
677 }
678
679 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
680         return int(cs.Begin / chunkSize)
681 }
682
683 // TODO: This should probably be called wantPiece.
684 func (t *torrent) wantChunk(r request) bool {
685         if !t.wantPiece(int(r.Index)) {
686                 return false
687         }
688         if t.Pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) {
689                 return true
690         }
691         _, ok := t.urgent[r]
692         return ok
693 }
694
695 func (t *torrent) urgentChunkInPiece(piece int) bool {
696         p := pp.Integer(piece)
697         for req := range t.urgent {
698                 if req.Index == p {
699                         return true
700                 }
701         }
702         return false
703 }
704
705 // TODO: This should be called wantPieceIndex.
706 func (t *torrent) wantPiece(index int) bool {
707         if !t.haveInfo() {
708                 return false
709         }
710         p := t.Pieces[index]
711         if p.QueuedForHash {
712                 return false
713         }
714         if p.Hashing {
715                 return false
716         }
717         if p.Priority == PiecePriorityNone {
718                 if !t.urgentChunkInPiece(index) {
719                         return false
720                 }
721         }
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)
725 }
726
727 func (t *torrent) connHasWantedPieces(c *connection) bool {
728         return c.pieceRequestOrder != nil && c.pieceRequestOrder.First() != nil
729 }
730
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))
734         }
735         return
736 }
737
738 func (t *torrent) worstBadConn(cl *Client) *connection {
739         wcs := t.worstConns(cl)
740         heap.Init(wcs)
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 {
746                         continue
747                 }
748                 return c
749         }
750         return nil
751 }