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