torrent.go

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