torrent.go

   1 package torrent
   2
   3 import (
   4         "container/heap"
   5         "expvar"
   6         "fmt"
   7         "io"
   8         "log"
   9         "math/rand"
  10         "net"
  11         "sort"
  12         "sync"
  13         "time"
  14
  15         "github.com/anacrolix/missinggo"
  16         "github.com/anacrolix/missinggo/bitmap"
  17         "github.com/anacrolix/missinggo/itertools"
  18         "github.com/anacrolix/missinggo/perf"
  19         "github.com/anacrolix/missinggo/pubsub"
  20         "github.com/bradfitz/iter"
  21
  22         "github.com/anacrolix/torrent/bencode"
  23         "github.com/anacrolix/torrent/metainfo"
  24         pp "github.com/anacrolix/torrent/peer_protocol"
  25         "github.com/anacrolix/torrent/tracker"
  26 )
  27
  28 func (t *torrent) chunkIndexSpec(chunkIndex, piece int) chunkSpec {
  29         return chunkIndexSpec(chunkIndex, t.pieceLength(piece), t.chunkSize)
  30 }
  31
  32 func (t *torrent) pieceNumPendingBytes(index int) (count pp.Integer) {
  33         if t.pieceComplete(index) {
  34                 return
  35         }
  36         piece := &t.Pieces[index]
  37         count = t.pieceLength(index)
  38         if !piece.EverHashed {
  39                 return
  40         }
  41         for i, dirty := range piece.DirtyChunks {
  42                 if dirty {
  43                         count -= t.chunkIndexSpec(i, index).Length
  44                 }
  45         }
  46         return
  47 }
  48
  49 type peersKey struct {
  50         IPBytes string
  51         Port    int
  52 }
  53
  54 // Is not aware of Client. Maintains state of torrent for with-in a Client.
  55 type torrent struct {
  56         cl *Client
  57
  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         // Values are the piece indices that changed.
  68         pieceStateChanges *pubsub.PubSub
  69         chunkSize         pp.Integer
  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 Data
  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         readers map[*Reader]struct{}
 105
 106         pendingPieces bitmap.Bitmap
 107
 108         connPieceInclinationPool sync.Pool
 109 }
 110
 111 var (
 112         pieceInclinationsReused = expvar.NewInt("pieceInclinationsReused")
 113         pieceInclinationsNew    = expvar.NewInt("pieceInclinationsNew")
 114         pieceInclinationsPut    = expvar.NewInt("pieceInclinationsPut")
 115 )
 116
 117 func (t *torrent) setDisplayName(dn string) {
 118         t.displayName = dn
 119 }
 120
 121 func (t *torrent) pieceComplete(piece int) bool {
 122         // TODO: This is called when setting metadata, and before storage is
 123         // assigned, which doesn't seem right.
 124         return t.data != nil && t.data.PieceComplete(piece)
 125 }
 126
 127 func (t *torrent) numConnsUnchoked() (num int) {
 128         for _, c := range t.Conns {
 129                 if !c.PeerChoked {
 130                         num++
 131                 }
 132         }
 133         return
 134 }
 135
 136 // There's a connection to that address already.
 137 func (t *torrent) addrActive(addr string) bool {
 138         if _, ok := t.HalfOpen[addr]; ok {
 139                 return true
 140         }
 141         for _, c := range t.Conns {
 142                 if c.remoteAddr().String() == addr {
 143                         return true
 144                 }
 145         }
 146         return false
 147 }
 148
 149 func (t *torrent) worstConns(cl *Client) (wcs *worstConns) {
 150         wcs = &worstConns{
 151                 c:  make([]*connection, 0, len(t.Conns)),
 152                 t:  t,
 153                 cl: cl,
 154         }
 155         for _, c := range t.Conns {
 156                 if !c.closed.IsSet() {
 157                         wcs.c = append(wcs.c, c)
 158                 }
 159         }
 160         return
 161 }
 162
 163 func (t *torrent) ceaseNetworking() {
 164         t.stateMu.Lock()
 165         defer t.stateMu.Unlock()
 166         select {
 167         case <-t.ceasingNetworking:
 168                 return
 169         default:
 170         }
 171         close(t.ceasingNetworking)
 172         for _, c := range t.Conns {
 173                 c.Close()
 174         }
 175 }
 176
 177 func (t *torrent) addPeer(p Peer, cl *Client) {
 178         cl.openNewConns(t)
 179         if len(t.Peers) >= torrentPeersHighWater {
 180                 return
 181         }
 182         key := peersKey{string(p.IP), p.Port}
 183         if _, ok := t.Peers[key]; ok {
 184                 return
 185         }
 186         t.Peers[key] = p
 187         peersAddedBySource.Add(string(p.Source), 1)
 188         cl.openNewConns(t)
 189
 190 }
 191
 192 func (t *torrent) invalidateMetadata() {
 193         t.MetaData = nil
 194         t.metadataHave = nil
 195         t.Info = nil
 196 }
 197
 198 func (t *torrent) saveMetadataPiece(index int, data []byte) {
 199         if t.haveInfo() {
 200                 return
 201         }
 202         if index >= len(t.metadataHave) {
 203                 log.Printf("%s: ignoring metadata piece %d", t, index)
 204                 return
 205         }
 206         copy(t.MetaData[(1<<14)*index:], data)
 207         t.metadataHave[index] = true
 208 }
 209
 210 func (t *torrent) metadataPieceCount() int {
 211         return (len(t.MetaData) + (1 << 14) - 1) / (1 << 14)
 212 }
 213
 214 func (t *torrent) haveMetadataPiece(piece int) bool {
 215         if t.haveInfo() {
 216                 return (1<<14)*piece < len(t.MetaData)
 217         } else {
 218                 return piece < len(t.metadataHave) && t.metadataHave[piece]
 219         }
 220 }
 221
 222 func (t *torrent) metadataSizeKnown() bool {
 223         return t.MetaData != nil
 224 }
 225
 226 func (t *torrent) metadataSize() int {
 227         return len(t.MetaData)
 228 }
 229
 230 func infoPieceHashes(info *metainfo.Info) (ret []string) {
 231         for i := 0; i < len(info.Pieces); i += 20 {
 232                 ret = append(ret, string(info.Pieces[i:i+20]))
 233         }
 234         return
 235 }
 236
 237 // Called when metadata for a torrent becomes available.
 238 func (t *torrent) setMetadata(md *metainfo.Info, infoBytes []byte) (err error) {
 239         err = validateInfo(md)
 240         if err != nil {
 241                 err = fmt.Errorf("bad info: %s", err)
 242                 return
 243         }
 244         t.Info = md
 245         t.length = 0
 246         for _, f := range t.Info.UpvertedFiles() {
 247                 t.length += f.Length
 248         }
 249         t.MetaData = infoBytes
 250         t.metadataHave = nil
 251         hashes := infoPieceHashes(md)
 252         t.Pieces = make([]piece, len(hashes))
 253         for i, hash := range hashes {
 254                 piece := &t.Pieces[i]
 255                 piece.noPendingWrites.L = &piece.pendingWritesMutex
 256                 missinggo.CopyExact(piece.Hash[:], hash)
 257         }
 258         for _, conn := range t.Conns {
 259                 if err := conn.setNumPieces(t.numPieces()); err != nil {
 260                         log.Printf("closing connection: %s", err)
 261                         conn.Close()
 262                 }
 263         }
 264         return
 265 }
 266
 267 func (t *torrent) setStorage(td Data) (err error) {
 268         if t.data != nil {
 269                 t.data.Close()
 270         }
 271         t.data = td
 272         return
 273 }
 274
 275 func (t *torrent) haveAllMetadataPieces() bool {
 276         if t.haveInfo() {
 277                 return true
 278         }
 279         if t.metadataHave == nil {
 280                 return false
 281         }
 282         for _, have := range t.metadataHave {
 283                 if !have {
 284                         return false
 285                 }
 286         }
 287         return true
 288 }
 289
 290 func (t *torrent) setMetadataSize(bytes int64, cl *Client) {
 291         if t.haveInfo() {
 292                 // We already know the correct metadata size.
 293                 return
 294         }
 295         if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
 296                 log.Printf("received bad metadata size: %d", bytes)
 297                 return
 298         }
 299         if t.MetaData != nil && len(t.MetaData) == int(bytes) {
 300                 return
 301         }
 302         t.MetaData = make([]byte, bytes)
 303         t.metadataHave = make([]bool, (bytes+(1<<14)-1)/(1<<14))
 304         for _, c := range t.Conns {
 305                 cl.requestPendingMetadata(t, c)
 306         }
 307
 308 }
 309
 310 // The current working name for the torrent. Either the name in the info dict,
 311 // or a display name given such as by the dn value in a magnet link, or "".
 312 func (t *torrent) Name() string {
 313         if t.haveInfo() {
 314                 return t.Info.Name
 315         }
 316         return t.displayName
 317 }
 318
 319 func (t *torrent) pieceState(index int) (ret PieceState) {
 320         p := &t.Pieces[index]
 321         ret.Priority = t.piecePriority(index)
 322         if t.pieceComplete(index) {
 323                 ret.Complete = true
 324         }
 325         if p.QueuedForHash || p.Hashing {
 326                 ret.Checking = true
 327         }
 328         if !ret.Complete && t.piecePartiallyDownloaded(index) {
 329                 ret.Partial = true
 330         }
 331         return
 332 }
 333
 334 func (t *torrent) metadataPieceSize(piece int) int {
 335         return metadataPieceSize(len(t.MetaData), piece)
 336 }
 337
 338 func (t *torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
 339         d := map[string]int{
 340                 "msg_type": msgType,
 341                 "piece":    piece,
 342         }
 343         if data != nil {
 344                 d["total_size"] = len(t.MetaData)
 345         }
 346         p, err := bencode.Marshal(d)
 347         if err != nil {
 348                 panic(err)
 349         }
 350         return pp.Message{
 351                 Type:            pp.Extended,
 352                 ExtendedID:      byte(c.PeerExtensionIDs["ut_metadata"]),
 353                 ExtendedPayload: append(p, data...),
 354         }
 355 }
 356
 357 func (t *torrent) pieceStateRuns() (ret []PieceStateRun) {
 358         rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
 359                 ret = append(ret, PieceStateRun{
 360                         PieceState: el.(PieceState),
 361                         Length:     int(count),
 362                 })
 363         })
 364         for index := range t.Pieces {
 365                 rle.Append(t.pieceState(index), 1)
 366         }
 367         rle.Flush()
 368         return
 369 }
 370
 371 // Produces a small string representing a PieceStateRun.
 372 func pieceStateRunStatusChars(psr PieceStateRun) (ret string) {
 373         ret = fmt.Sprintf("%d", psr.Length)
 374         ret += func() string {
 375                 switch psr.Priority {
 376                 case PiecePriorityNext:
 377                         return "N"
 378                 case PiecePriorityNormal:
 379                         return "."
 380                 case PiecePriorityReadahead:
 381                         return "R"
 382                 case PiecePriorityNow:
 383                         return "!"
 384                 default:
 385                         return ""
 386                 }
 387         }()
 388         if psr.Checking {
 389                 ret += "H"
 390         }
 391         if psr.Partial {
 392                 ret += "P"
 393         }
 394         if psr.Complete {
 395                 ret += "C"
 396         }
 397         return
 398 }
 399
 400 func (t *torrent) writeStatus(w io.Writer, cl *Client) {
 401         fmt.Fprintf(w, "Infohash: %x\n", t.InfoHash)
 402         fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
 403         if !t.haveInfo() {
 404                 fmt.Fprintf(w, "Metadata have: ")
 405                 for _, h := range t.metadataHave {
 406                         fmt.Fprintf(w, "%c", func() rune {
 407                                 if h {
 408                                         return 'H'
 409                                 } else {
 410                                         return '.'
 411                                 }
 412                         }())
 413                 }
 414                 fmt.Fprintln(w)
 415         }
 416         fmt.Fprintf(w, "Piece length: %s\n", func() string {
 417                 if t.haveInfo() {
 418                         return fmt.Sprint(t.usualPieceSize())
 419                 } else {
 420                         return "?"
 421                 }
 422         }())
 423         if t.haveInfo() {
 424                 fmt.Fprintf(w, "Num Pieces: %d\n", t.numPieces())
 425                 fmt.Fprint(w, "Piece States:")
 426                 for _, psr := range t.pieceStateRuns() {
 427                         w.Write([]byte(" "))
 428                         w.Write([]byte(pieceStateRunStatusChars(psr)))
 429                 }
 430                 fmt.Fprintln(w)
 431         }
 432         fmt.Fprintf(w, "Reader Pieces:")
 433         t.forReaderOffsetPieces(func(begin, end int) (again bool) {
 434                 fmt.Fprintf(w, " %d:%d", begin, end)
 435                 return true
 436         })
 437         fmt.Fprintln(w)
 438         fmt.Fprintf(w, "Trackers: ")
 439         for _, tier := range t.Trackers {
 440                 for _, tr := range tier {
 441                         fmt.Fprintf(w, "%q ", tr.String())
 442                 }
 443         }
 444         fmt.Fprintf(w, "\n")
 445         fmt.Fprintf(w, "Pending peers: %d\n", len(t.Peers))
 446         fmt.Fprintf(w, "Half open: %d\n", len(t.HalfOpen))
 447         fmt.Fprintf(w, "Active peers: %d\n", len(t.Conns))
 448         sort.Sort(&worstConns{
 449                 c:  t.Conns,
 450                 t:  t,
 451                 cl: cl,
 452         })
 453         for i, c := range t.Conns {
 454                 fmt.Fprintf(w, "%2d. ", i+1)
 455                 c.WriteStatus(w, t)
 456         }
 457 }
 458
 459 func (t *torrent) String() string {
 460         s := t.Name()
 461         if s == "" {
 462                 s = fmt.Sprintf("%x", t.InfoHash)
 463         }
 464         return s
 465 }
 466
 467 func (t *torrent) haveInfo() bool {
 468         return t.Info != nil
 469 }
 470
 471 // TODO: Include URIs that weren't converted to tracker clients.
 472 func (t *torrent) announceList() (al [][]string) {
 473         for _, tier := range t.Trackers {
 474                 var l []string
 475                 for _, tr := range tier {
 476                         l = append(l, tr.URL())
 477                 }
 478                 al = append(al, l)
 479         }
 480         return
 481 }
 482
 483 // Returns a run-time generated MetaInfo that includes the info bytes and
 484 // announce-list as currently known to the client.
 485 func (t *torrent) MetaInfo() *metainfo.MetaInfo {
 486         if t.MetaData == nil {
 487                 panic("info bytes not set")
 488         }
 489         return &metainfo.MetaInfo{
 490                 Info: metainfo.InfoEx{
 491                         Info:  *t.Info,
 492                         Bytes: t.MetaData,
 493                 },
 494                 CreationDate: time.Now().Unix(),
 495                 Comment:      "dynamic metainfo from client",
 496                 CreatedBy:    "go.torrent",
 497                 AnnounceList: t.announceList(),
 498         }
 499 }
 500
 501 func (t *torrent) bytesLeft() (left int64) {
 502         if !t.haveInfo() {
 503                 return -1
 504         }
 505         for i := 0; i < t.numPieces(); i++ {
 506                 left += int64(t.pieceNumPendingBytes(i))
 507         }
 508         return
 509 }
 510
 511 func (t *torrent) piecePartiallyDownloaded(index int) bool {
 512         pendingBytes := t.pieceNumPendingBytes(index)
 513         return pendingBytes != 0 && pendingBytes != t.pieceLength(index)
 514 }
 515
 516 func numChunksForPiece(chunkSize int, pieceSize int) int {
 517         return (pieceSize + chunkSize - 1) / chunkSize
 518 }
 519
 520 func (t *torrent) usualPieceSize() int {
 521         return int(t.Info.PieceLength)
 522 }
 523
 524 func (t *torrent) lastPieceSize() int {
 525         return int(t.pieceLength(t.numPieces() - 1))
 526 }
 527
 528 func (t *torrent) numPieces() int {
 529         return t.Info.NumPieces()
 530 }
 531
 532 func (t *torrent) numPiecesCompleted() (num int) {
 533         for i := range iter.N(t.Info.NumPieces()) {
 534                 if t.pieceComplete(i) {
 535                         num++
 536                 }
 537         }
 538         return
 539 }
 540
 541 func (t *torrent) isClosed() bool {
 542         select {
 543         case <-t.closing:
 544                 return true
 545         default:
 546                 return false
 547         }
 548 }
 549
 550 func (t *torrent) close() (err error) {
 551         if t.isClosed() {
 552                 return
 553         }
 554         t.ceaseNetworking()
 555         close(t.closing)
 556         if c, ok := t.data.(io.Closer); ok {
 557                 c.Close()
 558         }
 559         for _, conn := range t.Conns {
 560                 conn.Close()
 561         }
 562         t.pieceStateChanges.Close()
 563         return
 564 }
 565
 566 func (t *torrent) requestOffset(r request) int64 {
 567         return torrentRequestOffset(t.length, int64(t.usualPieceSize()), r)
 568 }
 569
 570 // Return the request that would include the given offset into the torrent
 571 // data. Returns !ok if there is no such request.
 572 func (t *torrent) offsetRequest(off int64) (req request, ok bool) {
 573         return torrentOffsetRequest(t.length, t.Info.PieceLength, int64(t.chunkSize), off)
 574 }
 575
 576 func (t *torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
 577         tr := perf.NewTimer()
 578         n, err := t.data.WriteAt(data, int64(piece)*t.Info.PieceLength+begin)
 579         if err == nil && n != len(data) {
 580                 err = io.ErrShortWrite
 581         }
 582         if err == nil {
 583                 tr.Stop("write chunk")
 584         }
 585         return
 586 }
 587
 588 func (t *torrent) bitfield() (bf []bool) {
 589         for i := range t.Pieces {
 590                 bf = append(bf, t.havePiece(i))
 591         }
 592         return
 593 }
 594
 595 func (t *torrent) validOutgoingRequest(r request) bool {
 596         if r.Index >= pp.Integer(t.Info.NumPieces()) {
 597                 return false
 598         }
 599         if r.Begin%t.chunkSize != 0 {
 600                 return false
 601         }
 602         if r.Length > t.chunkSize {
 603                 return false
 604         }
 605         pieceLength := t.pieceLength(int(r.Index))
 606         if r.Begin+r.Length > pieceLength {
 607                 return false
 608         }
 609         return r.Length == t.chunkSize || r.Begin+r.Length == pieceLength
 610 }
 611
 612 func (t *torrent) pieceChunks(piece int) (css []chunkSpec) {
 613         css = make([]chunkSpec, 0, (t.pieceLength(piece)+t.chunkSize-1)/t.chunkSize)
 614         var cs chunkSpec
 615         for left := t.pieceLength(piece); left != 0; left -= cs.Length {
 616                 cs.Length = left
 617                 if cs.Length > t.chunkSize {
 618                         cs.Length = t.chunkSize
 619                 }
 620                 css = append(css, cs)
 621                 cs.Begin += cs.Length
 622         }
 623         return
 624 }
 625
 626 func (t *torrent) pieceNumChunks(piece int) int {
 627         return int((t.pieceLength(piece) + t.chunkSize - 1) / t.chunkSize)
 628 }
 629
 630 func (t *torrent) pendAllChunkSpecs(pieceIndex int) {
 631         t.Pieces[pieceIndex].DirtyChunks = nil
 632 }
 633
 634 type Peer struct {
 635         Id     [20]byte
 636         IP     net.IP
 637         Port   int
 638         Source peerSource
 639         // Peer is known to support encryption.
 640         SupportsEncryption bool
 641 }
 642
 643 func (t *torrent) pieceLength(piece int) (len_ pp.Integer) {
 644         if piece < 0 || piece > t.Info.NumPieces() {
 645                 return
 646         }
 647         if int(piece) == t.numPieces()-1 {
 648                 len_ = pp.Integer(t.length % t.Info.PieceLength)
 649         }
 650         if len_ == 0 {
 651                 len_ = pp.Integer(t.Info.PieceLength)
 652         }
 653         return
 654 }
 655
 656 func (t *torrent) hashPiece(piece int) (ps pieceSum) {
 657         hash := pieceHash.New()
 658         p := &t.Pieces[piece]
 659         p.waitNoPendingWrites()
 660         pl := t.Info.Piece(int(piece)).Length()
 661         n, err := t.data.WriteSectionTo(hash, int64(piece)*t.Info.PieceLength, pl)
 662         if err != nil {
 663                 if err != io.ErrUnexpectedEOF {
 664                         log.Printf("error hashing piece with %T: %s", t.data, err)
 665                 }
 666                 return
 667         }
 668         if n != pl {
 669                 panic(fmt.Sprintf("%T: %d != %d", t.data, n, pl))
 670         }
 671         missinggo.CopyExact(ps[:], hash.Sum(nil))
 672         return
 673 }
 674
 675 func (t *torrent) haveAllPieces() bool {
 676         if !t.haveInfo() {
 677                 return false
 678         }
 679         for i := range t.Pieces {
 680                 if !t.pieceComplete(i) {
 681                         return false
 682                 }
 683         }
 684         return true
 685 }
 686
 687 func (me *torrent) haveAnyPieces() bool {
 688         for i := range me.Pieces {
 689                 if me.pieceComplete(i) {
 690                         return true
 691                 }
 692         }
 693         return false
 694 }
 695
 696 func (t *torrent) havePiece(index int) bool {
 697         return t.haveInfo() && t.pieceComplete(index)
 698 }
 699
 700 func (t *torrent) haveChunk(r request) (ret bool) {
 701         // defer func() {
 702         //      log.Println("have chunk", r, ret)
 703         // }()
 704         if !t.haveInfo() {
 705                 return false
 706         }
 707         if t.pieceComplete(int(r.Index)) {
 708                 return true
 709         }
 710         p := &t.Pieces[r.Index]
 711         return !p.pendingChunk(r.chunkSpec, t.chunkSize)
 712 }
 713
 714 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
 715         return int(cs.Begin / chunkSize)
 716 }
 717
 718 // TODO: This should probably be called wantPiece.
 719 func (t *torrent) wantChunk(r request) bool {
 720         if !t.wantPiece(int(r.Index)) {
 721                 return false
 722         }
 723         if t.Pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) {
 724                 return true
 725         }
 726         // TODO: What about pieces that were wanted, but aren't now, and aren't
 727         // completed either? That used to be done here.
 728         return false
 729 }
 730
 731 // TODO: This should be called wantPieceIndex.
 732 func (t *torrent) wantPiece(index int) bool {
 733         if !t.haveInfo() {
 734                 return false
 735         }
 736         p := &t.Pieces[index]
 737         if p.QueuedForHash {
 738                 return false
 739         }
 740         if p.Hashing {
 741                 return false
 742         }
 743         if t.pieceComplete(index) {
 744                 return false
 745         }
 746         if t.pendingPieces.Contains(index) {
 747                 return true
 748         }
 749         return !t.forReaderOffsetPieces(func(begin, end int) bool {
 750                 return index < begin || index >= end
 751         })
 752 }
 753
 754 func (t *torrent) forNeededPieces(f func(piece int) (more bool)) (all bool) {
 755         return t.forReaderOffsetPieces(func(begin, end int) (more bool) {
 756                 for i := begin; begin < end; i++ {
 757                         if !f(i) {
 758                                 return false
 759                         }
 760                 }
 761                 return true
 762         })
 763 }
 764
 765 func (t *torrent) connHasWantedPieces(c *connection) bool {
 766         for it := t.pendingPieces.IterTyped(); it.Next(); {
 767                 if c.PeerHasPiece(it.ValueInt()) {
 768                         it.Stop()
 769                         return true
 770                 }
 771         }
 772         return !t.forReaderOffsetPieces(func(begin, end int) (again bool) {
 773                 for i := begin; i < end; i++ {
 774                         if c.PeerHasPiece(i) {
 775                                 return false
 776                         }
 777                 }
 778                 return true
 779         })
 780 }
 781
 782 func (t *torrent) extentPieces(off, _len int64) (pieces []int) {
 783         for i := off / int64(t.usualPieceSize()); i*int64(t.usualPieceSize()) < off+_len; i++ {
 784                 pieces = append(pieces, int(i))
 785         }
 786         return
 787 }
 788
 789 func (t *torrent) worstBadConn(cl *Client) *connection {
 790         wcs := t.worstConns(cl)
 791         heap.Init(wcs)
 792         for wcs.Len() != 0 {
 793                 c := heap.Pop(wcs).(*connection)
 794                 if c.UnwantedChunksReceived >= 6 && c.UnwantedChunksReceived > c.UsefulChunksReceived {
 795                         return c
 796                 }
 797                 if wcs.Len() >= (socketsPerTorrent+1)/2 {
 798                         // Give connections 1 minute to prove themselves.
 799                         if time.Since(c.completedHandshake) > time.Minute {
 800                                 return c
 801                         }
 802                 }
 803         }
 804         return nil
 805 }
 806
 807 func (t *torrent) publishPieceChange(piece int) {
 808         cur := t.pieceState(piece)
 809         p := &t.Pieces[piece]
 810         if cur != p.PublicPieceState {
 811                 t.pieceStateChanges.Publish(piece)
 812         }
 813         p.PublicPieceState = cur
 814 }
 815
 816 func (t *torrent) pieceNumPendingChunks(piece int) int {
 817         if t.pieceComplete(piece) {
 818                 return 0
 819         }
 820         return t.pieceNumChunks(piece) - t.Pieces[piece].numDirtyChunks()
 821 }
 822
 823 func (t *torrent) pieceAllDirty(piece int) bool {
 824         p := &t.Pieces[piece]
 825         if len(p.DirtyChunks) != t.pieceNumChunks(piece) {
 826                 return false
 827         }
 828         for _, dirty := range p.DirtyChunks {
 829                 if !dirty {
 830                         return false
 831                 }
 832         }
 833         return true
 834 }
 835
 836 func (t *torrent) forUrgentPieces(f func(piece int) (again bool)) (all bool) {
 837         return t.forReaderOffsetPieces(func(begin, end int) (again bool) {
 838                 if begin < end {
 839                         if !f(begin) {
 840                                 return false
 841                         }
 842                 }
 843                 return true
 844         })
 845 }
 846
 847 func (t *torrent) readersChanged() {
 848         t.updatePiecePriorities()
 849 }
 850
 851 func (t *torrent) maybeNewConns() {
 852         // Tickle the accept routine.
 853         t.cl.event.Broadcast()
 854         t.openNewConns()
 855 }
 856
 857 func (t *torrent) piecePriorityChanged(piece int) {
 858         for _, c := range t.Conns {
 859                 c.updatePiecePriority(piece)
 860         }
 861         t.maybeNewConns()
 862 }
 863
 864 func (t *torrent) updatePiecePriority(piece int) bool {
 865         p := &t.Pieces[piece]
 866         newPrio := t.piecePriorityUncached(piece)
 867         if newPrio == p.priority {
 868                 return false
 869         }
 870         p.priority = newPrio
 871         return true
 872 }
 873
 874 func (t *torrent) updatePiecePriorities() {
 875         newPrios := make([]piecePriority, t.numPieces())
 876         itertools.ForIterable(&t.pendingPieces, func(value interface{}) (next bool) {
 877                 newPrios[value.(int)] = PiecePriorityNormal
 878                 return true
 879         })
 880         t.forReaderOffsetPieces(func(begin, end int) (next bool) {
 881                 if begin < end {
 882                         newPrios[begin].Raise(PiecePriorityNow)
 883                 }
 884                 for i := begin + 1; i < end; i++ {
 885                         newPrios[begin].Raise(PiecePriorityReadahead)
 886                 }
 887                 return true
 888         })
 889         for i, prio := range newPrios {
 890                 if prio != t.Pieces[i].priority {
 891                         t.Pieces[i].priority = prio
 892                         t.piecePriorityChanged(i)
 893                 }
 894         }
 895 }
 896
 897 func (t *torrent) byteRegionPieces(off, size int64) (begin, end int) {
 898         if off >= t.length {
 899                 return
 900         }
 901         if off < 0 {
 902                 size += off
 903                 off = 0
 904         }
 905         if size <= 0 {
 906                 return
 907         }
 908         begin = int(off / t.Info.PieceLength)
 909         end = int((off + size + t.Info.PieceLength - 1) / t.Info.PieceLength)
 910         if end > t.Info.NumPieces() {
 911                 end = t.Info.NumPieces()
 912         }
 913         return
 914 }
 915
 916 // Returns true if all iterations complete without breaking.
 917 func (t *torrent) forReaderOffsetPieces(f func(begin, end int) (more bool)) (all bool) {
 918         for r := range t.readers {
 919                 r.mu.Lock()
 920                 pos, readahead := r.pos, r.readahead
 921                 r.mu.Unlock()
 922                 if readahead < 1 {
 923                         readahead = 1
 924                 }
 925                 begin, end := t.byteRegionPieces(pos, readahead)
 926                 if begin >= end {
 927                         continue
 928                 }
 929                 if !f(begin, end) {
 930                         return false
 931                 }
 932         }
 933         return true
 934 }
 935
 936 func (t *torrent) piecePriority(piece int) piecePriority {
 937         if !t.haveInfo() {
 938                 return PiecePriorityNone
 939         }
 940         return t.Pieces[piece].priority
 941 }
 942
 943 func (t *torrent) piecePriorityUncached(piece int) (ret piecePriority) {
 944         ret = PiecePriorityNone
 945         if t.pieceComplete(piece) {
 946                 return
 947         }
 948         if t.pendingPieces.Contains(piece) {
 949                 ret = PiecePriorityNormal
 950         }
 951         raiseRet := ret.Raise
 952         t.forReaderOffsetPieces(func(begin, end int) (again bool) {
 953                 if piece == begin {
 954                         raiseRet(PiecePriorityNow)
 955                 }
 956                 if begin <= piece && piece < end {
 957                         raiseRet(PiecePriorityReadahead)
 958                 }
 959                 return true
 960         })
 961         return
 962 }
 963
 964 func (t *torrent) pendPiece(piece int, cl *Client) {
 965         if t.pendingPieces.Contains(piece) {
 966                 return
 967         }
 968         if t.havePiece(piece) {
 969                 return
 970         }
 971         t.pendingPieces.Add(piece)
 972         if !t.updatePiecePriority(piece) {
 973                 return
 974         }
 975         t.piecePriorityChanged(piece)
 976 }
 977
 978 func (t *torrent) connRequestPiecePendingChunks(c *connection, piece int) (more bool) {
 979         if !c.PeerHasPiece(piece) {
 980                 return true
 981         }
 982         for _, cs := range t.Pieces[piece].shuffledPendingChunkSpecs(t, piece) {
 983                 req := request{pp.Integer(piece), cs}
 984                 if !c.Request(req) {
 985                         return false
 986                 }
 987         }
 988         return true
 989 }
 990
 991 func (t *torrent) pendRequest(req request) {
 992         ci := chunkIndex(req.chunkSpec, t.chunkSize)
 993         t.Pieces[req.Index].pendChunkIndex(ci)
 994 }
 995
 996 func (t *torrent) pieceChanged(piece int) {
 997         t.cl.pieceChanged(t, piece)
 998 }
 999
1000 func (t *torrent) openNewConns() {
1001         t.cl.openNewConns(t)
1002 }
1003
1004 func (t *torrent) getConnPieceInclination() []int {
1005         _ret := t.connPieceInclinationPool.Get()
1006         if _ret == nil {
1007                 pieceInclinationsNew.Add(1)
1008                 return rand.Perm(t.numPieces())
1009         }
1010         pieceInclinationsReused.Add(1)
1011         return _ret.([]int)
1012 }
1013
1014 func (t *torrent) putPieceInclination(pi []int) {
1015         t.connPieceInclinationPool.Put(pi)
1016         pieceInclinationsPut.Add(1)
1017 }