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/torrent/bencode"
  14         "github.com/anacrolix/torrent/metainfo"
  15         "github.com/bradfitz/iter"
  16
  17         "github.com/anacrolix/torrent/data"
  18         pp "github.com/anacrolix/torrent/peer_protocol"
  19         "github.com/anacrolix/torrent/tracker"
  20         "github.com/anacrolix/torrent/util"
  21 )
  22
  23 func (t *torrent) pieceNumPendingBytes(index int) (count pp.Integer) {
  24         if t.pieceComplete(index) {
  25                 return 0
  26         }
  27         piece := t.Pieces[index]
  28         if !piece.EverHashed {
  29                 return t.pieceLength(index)
  30         }
  31         pendingChunks := t.Pieces[index].PendingChunkSpecs
  32         count = pp.Integer(len(pendingChunks)) * chunkSize
  33         _lastChunkSpec := lastChunkSpec(t.pieceLength(index))
  34         if _lastChunkSpec.Length != chunkSize {
  35                 if _, ok := pendingChunks[_lastChunkSpec]; ok {
  36                         count += _lastChunkSpec.Length - chunkSize
  37                 }
  38         }
  39         return
  40 }
  41
  42 type peersKey struct {
  43         IPBytes string
  44         Port    int
  45 }
  46
  47 // Data maintains per-piece persistent state.
  48 type StatefulData interface {
  49         data.Data
  50         // We believe the piece data will pass a hash check.
  51         PieceCompleted(index int) error
  52         // Returns true if the piece is complete.
  53         PieceComplete(index int) bool
  54 }
  55
  56 // Is not aware of Client. Maintains state of torrent for with-in a Client.
  57 type torrent struct {
  58         stateMu sync.Mutex
  59         closing chan struct{}
  60
  61         // Closed when no more network activity is desired. This includes
  62         // announcing, and communicating with peers.
  63         ceasingNetworking chan struct{}
  64
  65         InfoHash InfoHash
  66         Pieces   []*piece
  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.
  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         GotMetainfo <-chan struct{}
 104
 105         pruneTimer *time.Timer
 106 }
 107
 108 func (t *torrent) pieceComplete(piece int) bool {
 109         // TODO: This is called when setting metadata, and before storage is
 110         // assigned, which doesn't seem right.
 111         return t.data != nil && t.data.PieceComplete(piece)
 112 }
 113
 114 func (t *torrent) numConnsUnchoked() (num int) {
 115         for _, c := range t.Conns {
 116                 if !c.PeerChoked {
 117                         num++
 118                 }
 119         }
 120         return
 121 }
 122
 123 // There's a connection to that address already.
 124 func (t *torrent) addrActive(addr string) bool {
 125         if _, ok := t.HalfOpen[addr]; ok {
 126                 return true
 127         }
 128         for _, c := range t.Conns {
 129                 if c.remoteAddr().String() == addr {
 130                         return true
 131                 }
 132         }
 133         return false
 134 }
 135
 136 func (t *torrent) worstConnsHeap() (wcs *worstConns) {
 137         wcs = &worstConns{
 138                 c: append([]*connection{}, t.Conns...),
 139                 t: t,
 140         }
 141         heap.Init(wcs)
 142         return
 143 }
 144
 145 func (t *torrent) ceaseNetworking() {
 146         t.stateMu.Lock()
 147         defer t.stateMu.Unlock()
 148         select {
 149         case <-t.ceasingNetworking:
 150                 return
 151         default:
 152         }
 153         close(t.ceasingNetworking)
 154         for _, c := range t.Conns {
 155                 c.Close()
 156         }
 157         if t.pruneTimer != nil {
 158                 t.pruneTimer.Stop()
 159         }
 160 }
 161
 162 func (t *torrent) addPeer(p Peer) {
 163         t.Peers[peersKey{string(p.IP), p.Port}] = p
 164 }
 165
 166 func (t *torrent) invalidateMetadata() {
 167         t.MetaData = nil
 168         t.metadataHave = nil
 169         t.Info = nil
 170 }
 171
 172 func (t *torrent) saveMetadataPiece(index int, data []byte) {
 173         if t.haveInfo() {
 174                 return
 175         }
 176         if index >= len(t.metadataHave) {
 177                 log.Printf("%s: ignoring metadata piece %d", t, index)
 178                 return
 179         }
 180         copy(t.MetaData[(1<<14)*index:], data)
 181         t.metadataHave[index] = true
 182 }
 183
 184 func (t *torrent) metadataPieceCount() int {
 185         return (len(t.MetaData) + (1 << 14) - 1) / (1 << 14)
 186 }
 187
 188 func (t *torrent) haveMetadataPiece(piece int) bool {
 189         if t.haveInfo() {
 190                 return (1<<14)*piece < len(t.MetaData)
 191         } else {
 192                 return piece < len(t.metadataHave) && t.metadataHave[piece]
 193         }
 194 }
 195
 196 func (t *torrent) metadataSizeKnown() bool {
 197         return t.MetaData != nil
 198 }
 199
 200 func (t *torrent) metadataSize() int {
 201         return len(t.MetaData)
 202 }
 203
 204 func infoPieceHashes(info *metainfo.Info) (ret []string) {
 205         for i := 0; i < len(info.Pieces); i += 20 {
 206                 ret = append(ret, string(info.Pieces[i:i+20]))
 207         }
 208         return
 209 }
 210
 211 // Called when metadata for a torrent becomes available.
 212 func (t *torrent) setMetadata(md *metainfo.Info, infoBytes []byte, eventLocker sync.Locker) (err error) {
 213         t.Info = md
 214         t.length = 0
 215         for _, f := range t.Info.UpvertedFiles() {
 216                 t.length += f.Length
 217         }
 218         t.MetaData = infoBytes
 219         t.metadataHave = nil
 220         for _, hash := range infoPieceHashes(md) {
 221                 piece := &piece{}
 222                 piece.Event.L = eventLocker
 223                 util.CopyExact(piece.Hash[:], hash)
 224                 t.Pieces = append(t.Pieces, piece)
 225         }
 226         for _, conn := range t.Conns {
 227                 t.initRequestOrdering(conn)
 228                 if err := conn.setNumPieces(t.numPieces()); err != nil {
 229                         log.Printf("closing connection: %s", err)
 230                         conn.Close()
 231                 }
 232         }
 233         return
 234 }
 235
 236 func (t *torrent) setStorage(td data.Data) (err error) {
 237         if c, ok := t.data.(io.Closer); ok {
 238                 c.Close()
 239         }
 240         if sd, ok := td.(StatefulData); ok {
 241                 t.data = sd
 242         } else {
 243                 t.data = &statelessDataWrapper{td, make([]bool, t.Info.NumPieces())}
 244         }
 245         return
 246 }
 247
 248 func (t *torrent) haveAllMetadataPieces() bool {
 249         if t.haveInfo() {
 250                 return true
 251         }
 252         if t.metadataHave == nil {
 253                 return false
 254         }
 255         for _, have := range t.metadataHave {
 256                 if !have {
 257                         return false
 258                 }
 259         }
 260         return true
 261 }
 262
 263 func (t *torrent) setMetadataSize(bytes int64, cl *Client) {
 264         if t.haveInfo() {
 265                 // We already know the correct metadata size.
 266                 return
 267         }
 268         if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
 269                 log.Printf("received bad metadata size: %d", bytes)
 270                 return
 271         }
 272         if t.MetaData != nil && len(t.MetaData) == int(bytes) {
 273                 return
 274         }
 275         t.MetaData = make([]byte, bytes)
 276         t.metadataHave = make([]bool, (bytes+(1<<14)-1)/(1<<14))
 277         for _, c := range t.Conns {
 278                 cl.requestPendingMetadata(t, c)
 279         }
 280
 281 }
 282
 283 // The current working name for the torrent. Either the name in the info dict,
 284 // or a display name given such as by the dn value in a magnet link, or "".
 285 func (t *torrent) Name() string {
 286         if t.haveInfo() {
 287                 return t.Info.Name
 288         }
 289         if t.DisplayName != "" {
 290                 return t.DisplayName
 291         }
 292         return ""
 293 }
 294
 295 func (t *torrent) pieceStatusChar(index int) byte {
 296         p := t.Pieces[index]
 297         switch {
 298         case t.pieceComplete(index):
 299                 return 'C'
 300         case p.QueuedForHash:
 301                 return 'Q'
 302         case p.Hashing:
 303                 return 'H'
 304         case !p.EverHashed:
 305                 return '?'
 306         case t.piecePartiallyDownloaded(index):
 307                 switch p.Priority {
 308                 case piecePriorityNone:
 309                         return 'F' // Forgotten
 310                 default:
 311                         return 'P'
 312                 }
 313         default:
 314                 switch p.Priority {
 315                 case piecePriorityNone:
 316                         return 'z'
 317                 case piecePriorityNow:
 318                         return '!'
 319                 case piecePriorityReadahead:
 320                         return 'R'
 321                 case piecePriorityNext:
 322                         return 'N'
 323                 default:
 324                         return '.'
 325                 }
 326         }
 327 }
 328
 329 func (t *torrent) metadataPieceSize(piece int) int {
 330         return metadataPieceSize(len(t.MetaData), piece)
 331 }
 332
 333 func (t *torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
 334         d := map[string]int{
 335                 "msg_type": msgType,
 336                 "piece":    piece,
 337         }
 338         if data != nil {
 339                 d["total_size"] = len(t.MetaData)
 340         }
 341         p, err := bencode.Marshal(d)
 342         if err != nil {
 343                 panic(err)
 344         }
 345         return pp.Message{
 346                 Type:            pp.Extended,
 347                 ExtendedID:      byte(c.PeerExtensionIDs["ut_metadata"]),
 348                 ExtendedPayload: append(p, data...),
 349         }
 350 }
 351
 352 type PieceStatusCharSequence struct {
 353         Char  byte // The state of this sequence of pieces.
 354         Count int  // How many consecutive pieces have this state.
 355 }
 356
 357 // Returns the state of pieces of the torrent. They are grouped into runs of
 358 // same state. The sum of the Counts of the sequences is the number of pieces
 359 // in the torrent. See the function torrent.pieceStatusChar for the possible
 360 // states.
 361 func (t *torrent) PieceStatusCharSequences() []PieceStatusCharSequence {
 362         t.stateMu.Lock()
 363         defer t.stateMu.Unlock()
 364         return t.pieceStatusCharSequences()
 365 }
 366
 367 // Returns the length of sequences of identical piece status chars.
 368 func (t *torrent) pieceStatusCharSequences() (ret []PieceStatusCharSequence) {
 369         var (
 370                 char  byte
 371                 count int
 372         )
 373         writeSequence := func() {
 374                 ret = append(ret, PieceStatusCharSequence{char, count})
 375         }
 376         if len(t.Pieces) != 0 {
 377                 char = t.pieceStatusChar(0)
 378         }
 379         for index := range t.Pieces {
 380                 char1 := t.pieceStatusChar(index)
 381                 if char1 == char {
 382                         count++
 383                 } else {
 384                         writeSequence()
 385                         char = char1
 386                         count = 1
 387                 }
 388         }
 389         if count != 0 {
 390                 writeSequence()
 391         }
 392         return
 393 }
 394
 395 func (t *torrent) writeStatus(w io.Writer) {
 396         fmt.Fprintf(w, "Infohash: %x\n", t.InfoHash)
 397         fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
 398         fmt.Fprintf(w, "Metadata have: ")
 399         for _, h := range t.metadataHave {
 400                 fmt.Fprintf(w, "%c", func() rune {
 401                         if h {
 402                                 return 'H'
 403                         } else {
 404                                 return '.'
 405                         }
 406                 }())
 407         }
 408         fmt.Fprintln(w)
 409         fmt.Fprintf(w, "Piece length: %s\n", func() string {
 410                 if t.haveInfo() {
 411                         return fmt.Sprint(t.usualPieceSize())
 412                 } else {
 413                         return "?"
 414                 }
 415         }())
 416         if t.haveInfo() {
 417                 fmt.Fprint(w, "Pieces: ")
 418                 for _, seq := range t.pieceStatusCharSequences() {
 419                         fmt.Fprintf(w, "%d%c ", seq.Count, seq.Char)
 420                 }
 421                 fmt.Fprintln(w)
 422         }
 423         fmt.Fprintf(w, "Urgent:")
 424         for req := range t.urgent {
 425                 fmt.Fprintf(w, " %s", req)
 426         }
 427         fmt.Fprintln(w)
 428         fmt.Fprintf(w, "Trackers: ")
 429         for _, tier := range t.Trackers {
 430                 for _, tr := range tier {
 431                         fmt.Fprintf(w, "%q ", tr.String())
 432                 }
 433         }
 434         fmt.Fprintf(w, "\n")
 435         fmt.Fprintf(w, "Pending peers: %d\n", len(t.Peers))
 436         fmt.Fprintf(w, "Half open: %d\n", len(t.HalfOpen))
 437         fmt.Fprintf(w, "Active peers: %d\n", len(t.Conns))
 438         sort.Sort(&worstConns{
 439                 c: t.Conns,
 440                 t: t,
 441         })
 442         for _, c := range t.Conns {
 443                 c.WriteStatus(w, t)
 444         }
 445 }
 446
 447 func (t *torrent) String() string {
 448         s := t.Name()
 449         if s == "" {
 450                 s = fmt.Sprintf("%x", t.InfoHash)
 451         }
 452         return s
 453 }
 454
 455 func (t *torrent) haveInfo() bool {
 456         return t.Info != nil
 457 }
 458
 459 // TODO: Include URIs that weren't converted to tracker clients.
 460 func (t *torrent) announceList() (al [][]string) {
 461         for _, tier := range t.Trackers {
 462                 var l []string
 463                 for _, tr := range tier {
 464                         l = append(l, tr.URL())
 465                 }
 466                 al = append(al, l)
 467         }
 468         return
 469 }
 470
 471 // Returns a run-time generated MetaInfo that includes the info bytes and
 472 // announce-list as currently known to the client.
 473 func (t *torrent) MetaInfo() *metainfo.MetaInfo {
 474         if t.MetaData == nil {
 475                 panic("info bytes not set")
 476         }
 477         return &metainfo.MetaInfo{
 478                 Info: metainfo.InfoEx{
 479                         Info:  *t.Info,
 480                         Bytes: t.MetaData,
 481                 },
 482                 CreationDate: time.Now().Unix(),
 483                 Comment:      "dynamic metainfo from client",
 484                 CreatedBy:    "go.torrent",
 485                 AnnounceList: t.announceList(),
 486         }
 487 }
 488
 489 func (t *torrent) bytesLeft() (left int64) {
 490         if !t.haveInfo() {
 491                 return -1
 492         }
 493         for i := 0; i < t.numPieces(); i++ {
 494                 left += int64(t.pieceNumPendingBytes(i))
 495         }
 496         return
 497 }
 498
 499 func (t *torrent) piecePartiallyDownloaded(index int) bool {
 500         return t.pieceNumPendingBytes(index) != t.pieceLength(index)
 501 }
 502
 503 func numChunksForPiece(chunkSize int, pieceSize int) int {
 504         return (pieceSize + chunkSize - 1) / chunkSize
 505 }
 506
 507 func (t *torrent) usualPieceSize() int {
 508         return int(t.Info.PieceLength)
 509 }
 510
 511 func (t *torrent) lastPieceSize() int {
 512         return int(t.pieceLength(t.numPieces() - 1))
 513 }
 514
 515 func (t *torrent) numPieces() int {
 516         return t.Info.NumPieces()
 517 }
 518
 519 func (t *torrent) numPiecesCompleted() (num int) {
 520         for i := range iter.N(t.Info.NumPieces()) {
 521                 if t.pieceComplete(i) {
 522                         num++
 523                 }
 524         }
 525         return
 526 }
 527
 528 func (t *torrent) Length() int64 {
 529         return t.length
 530 }
 531
 532 func (t *torrent) isClosed() bool {
 533         select {
 534         case <-t.closing:
 535                 return true
 536         default:
 537                 return false
 538         }
 539 }
 540
 541 func (t *torrent) close() (err error) {
 542         if t.isClosed() {
 543                 return
 544         }
 545         t.ceaseNetworking()
 546         close(t.closing)
 547         if c, ok := t.data.(io.Closer); ok {
 548                 c.Close()
 549         }
 550         for _, conn := range t.Conns {
 551                 conn.Close()
 552         }
 553         return
 554 }
 555
 556 func (t *torrent) requestOffset(r request) int64 {
 557         return torrentRequestOffset(t.Length(), int64(t.usualPieceSize()), r)
 558 }
 559
 560 // Return the request that would include the given offset into the torrent
 561 // data. Returns !ok if there is no such request.
 562 func (t *torrent) offsetRequest(off int64) (req request, ok bool) {
 563         return torrentOffsetRequest(t.Length(), t.Info.PieceLength, chunkSize, off)
 564 }
 565
 566 func (t *torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
 567         _, err = t.data.WriteAt(data, int64(piece)*t.Info.PieceLength+begin)
 568         return
 569 }
 570
 571 func (t *torrent) bitfield() (bf []bool) {
 572         for _, p := range t.Pieces {
 573                 // TODO: Check this logic.
 574                 bf = append(bf, p.EverHashed && len(p.PendingChunkSpecs) == 0)
 575         }
 576         return
 577 }
 578
 579 func (t *torrent) pieceChunks(piece int) (css []chunkSpec) {
 580         css = make([]chunkSpec, 0, (t.pieceLength(piece)+chunkSize-1)/chunkSize)
 581         var cs chunkSpec
 582         for left := t.pieceLength(piece); left != 0; left -= cs.Length {
 583                 cs.Length = left
 584                 if cs.Length > chunkSize {
 585                         cs.Length = chunkSize
 586                 }
 587                 css = append(css, cs)
 588                 cs.Begin += cs.Length
 589         }
 590         return
 591 }
 592
 593 func (t *torrent) pendAllChunkSpecs(index int) {
 594         piece := t.Pieces[index]
 595         if piece.PendingChunkSpecs == nil {
 596                 piece.PendingChunkSpecs = make(
 597                         map[chunkSpec]struct{},
 598                         (t.pieceLength(index)+chunkSize-1)/chunkSize)
 599         }
 600         pcss := piece.PendingChunkSpecs
 601         for _, cs := range t.pieceChunks(int(index)) {
 602                 pcss[cs] = struct{}{}
 603         }
 604         return
 605 }
 606
 607 type Peer struct {
 608         Id     [20]byte
 609         IP     net.IP
 610         Port   int
 611         Source peerSource
 612         // Peer is known to support encryption.
 613         SupportsEncryption bool
 614 }
 615
 616 func (t *torrent) pieceLength(piece int) (len_ pp.Integer) {
 617         if int(piece) == t.numPieces()-1 {
 618                 len_ = pp.Integer(t.Length() % t.Info.PieceLength)
 619         }
 620         if len_ == 0 {
 621                 len_ = pp.Integer(t.Info.PieceLength)
 622         }
 623         return
 624 }
 625
 626 func (t *torrent) hashPiece(piece pp.Integer) (ps pieceSum) {
 627         hash := pieceHash.New()
 628         t.data.WriteSectionTo(hash, int64(piece)*t.Info.PieceLength, t.Info.PieceLength)
 629         util.CopyExact(ps[:], hash.Sum(nil))
 630         return
 631 }
 632
 633 func (t *torrent) haveAllPieces() bool {
 634         if !t.haveInfo() {
 635                 return false
 636         }
 637         for i := range t.Pieces {
 638                 if !t.pieceComplete(i) {
 639                         return false
 640                 }
 641         }
 642         return true
 643 }
 644
 645 func (me *torrent) haveAnyPieces() bool {
 646         for i := range me.Pieces {
 647                 if me.pieceComplete(i) {
 648                         return true
 649                 }
 650         }
 651         return false
 652 }
 653
 654 func (t *torrent) havePiece(index int) bool {
 655         return t.haveInfo() && t.pieceComplete(index)
 656 }
 657
 658 func (t *torrent) haveChunk(r request) bool {
 659         if !t.haveInfo() {
 660                 return false
 661         }
 662         piece := t.Pieces[r.Index]
 663         _, ok := piece.PendingChunkSpecs[r.chunkSpec]
 664         // log.Println("have chunk", r, !ok)
 665         return !ok
 666 }
 667
 668 func (t *torrent) wantChunk(r request) bool {
 669         if !t.wantPiece(int(r.Index)) {
 670                 return false
 671         }
 672         _, ok := t.Pieces[r.Index].PendingChunkSpecs[r.chunkSpec]
 673         if ok {
 674                 return true
 675         }
 676         _, ok = t.urgent[r]
 677         return ok
 678 }
 679
 680 func (t *torrent) urgentChunkInPiece(piece int) bool {
 681         for req := range t.urgent {
 682                 if int(req.Index) == piece {
 683                         return true
 684                 }
 685         }
 686         return false
 687 }
 688
 689 func (t *torrent) wantPiece(index int) bool {
 690         if !t.haveInfo() {
 691                 return false
 692         }
 693         p := t.Pieces[index]
 694         if p.QueuedForHash {
 695                 return false
 696         }
 697         if p.Hashing {
 698                 return false
 699         }
 700         if p.Priority == piecePriorityNone {
 701                 if !t.urgentChunkInPiece(index) {
 702                         return false
 703                 }
 704         }
 705         // Put piece complete check last, since it's the slowest as it can involve
 706         // calling out into external data stores.
 707         return !t.pieceComplete(index)
 708 }
 709
 710 func (t *torrent) connHasWantedPieces(c *connection) bool {
 711         return c.pieceRequestOrder != nil && c.pieceRequestOrder.First() != nil
 712 }
 713
 714 func (t *torrent) extentPieces(off, _len int64) (pieces []int) {
 715         for i := off / int64(t.usualPieceSize()); i*int64(t.usualPieceSize()) < off+_len; i++ {
 716                 pieces = append(pieces, int(i))
 717         }
 718         return
 719 }