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/pubsub"
  17         "github.com/bradfitz/iter"
  18
  19         "github.com/anacrolix/torrent/bencode"
  20         "github.com/anacrolix/torrent/metainfo"
  21         pp "github.com/anacrolix/torrent/peer_protocol"
  22         "github.com/anacrolix/torrent/tracker"
  23 )
  24
  25 func (t *torrent) pieceNumPendingBytes(index int) (count pp.Integer) {
  26         if t.pieceComplete(index) {
  27                 return 0
  28         }
  29         piece := &t.Pieces[index]
  30         pieceLength := t.pieceLength(index)
  31         if !piece.EverHashed {
  32                 return pieceLength
  33         }
  34         for i, pending := range piece.PendingChunkSpecs {
  35                 if pending {
  36                         count += chunkIndexSpec(i, pieceLength, t.chunkSize).Length
  37                 }
  38         }
  39         return
  40 }
  41
  42 type PeersKey struct {
  43         IPBytes string
  44         Port    int
  45 }
  46
  47 // Is not aware of Client. Maintains state of torrent for with-in a Client.
  48 type torrent struct {
  49         stateMu sync.Mutex
  50         closing chan struct{}
  51
  52         // Closed when no more network activity is desired. This includes
  53         // announcing, and communicating with peers.
  54         ceasingNetworking chan struct{}
  55
  56         InfoHash InfoHash
  57         Pieces   []piece
  58         // Values are the piece indices that changed.
  59         pieceStateChanges *pubsub.PubSub
  60         chunkSize         pp.Integer
  61         // Chunks that are wanted before all others. This is for
  62         // responsive/streaming readers that want to unblock ASAP.
  63         urgent map[request]struct{}
  64         // Total length of the torrent in bytes. Stored because it's not O(1) to
  65         // get this from the info dict.
  66         length int64
  67
  68         data Data
  69
  70         // The info dict. Nil if we don't have it (yet).
  71         Info *metainfo.Info
  72         // Active peer connections, running message stream loops.
  73         Conns []*connection
  74         // Set of addrs to which we're attempting to connect. Connections are
  75         // half-open until all handshakes are completed.
  76         HalfOpen map[string]struct{}
  77
  78         // Reserve of peers to connect to. A peer can be both here and in the
  79         // active connections if were told about the peer after connecting with
  80         // them. That encourages us to reconnect to peers that are well known.
  81         Peers     map[PeersKey]Peer
  82         wantPeers sync.Cond
  83
  84         // BEP 12 Multitracker Metadata Extension. The tracker.Client instances
  85         // mirror their respective URLs from the announce-list metainfo key.
  86         Trackers [][]tracker.Client
  87         // Name used if the info name isn't available.
  88         displayName string
  89         // The bencoded bytes of the info dict.
  90         MetaData []byte
  91         // Each element corresponds to the 16KiB metadata pieces. If true, we have
  92         // received that piece.
  93         metadataHave []bool
  94
  95         // Closed when .Info is set.
  96         gotMetainfo chan struct{}
  97
  98         connPiecePriorites sync.Pool
  99 }
 100
 101 var (
 102         piecePrioritiesReused = expvar.NewInt("piecePrioritiesReused")
 103         piecePrioritiesNew    = expvar.NewInt("piecePrioritiesNew")
 104 )
 105
 106 func (t *torrent) setDisplayName(dn string) {
 107         t.displayName = dn
 108 }
 109
 110 func (t *torrent) newConnPiecePriorities() []int {
 111         _ret := t.connPiecePriorites.Get()
 112         if _ret != nil {
 113                 piecePrioritiesReused.Add(1)
 114                 return _ret.([]int)
 115         }
 116         piecePrioritiesNew.Add(1)
 117         return rand.Perm(t.numPieces())
 118 }
 119
 120 func (t *torrent) pieceComplete(piece int) bool {
 121         // TODO: This is called when setting metadata, and before storage is
 122         // assigned, which doesn't seem right.
 123         return t.data != nil && t.data.PieceComplete(piece)
 124 }
 125
 126 func (t *torrent) numConnsUnchoked() (num int) {
 127         for _, c := range t.Conns {
 128                 if !c.PeerChoked {
 129                         num++
 130                 }
 131         }
 132         return
 133 }
 134
 135 // There's a connection to that address already.
 136 func (t *torrent) addrActive(addr string) bool {
 137         if _, ok := t.HalfOpen[addr]; ok {
 138                 return true
 139         }
 140         for _, c := range t.Conns {
 141                 if c.remoteAddr().String() == addr {
 142                         return true
 143                 }
 144         }
 145         return false
 146 }
 147
 148 func (t *torrent) worstConns(cl *Client) (wcs *worstConns) {
 149         wcs = &worstConns{
 150                 c:  make([]*connection, 0, len(t.Conns)),
 151                 t:  t,
 152                 cl: cl,
 153         }
 154         for _, c := range t.Conns {
 155                 select {
 156                 case <-c.closing:
 157                 default:
 158                         wcs.c = append(wcs.c, c)
 159                 }
 160         }
 161         return
 162 }
 163
 164 func (t *torrent) ceaseNetworking() {
 165         t.stateMu.Lock()
 166         defer t.stateMu.Unlock()
 167         select {
 168         case <-t.ceasingNetworking:
 169                 return
 170         default:
 171         }
 172         close(t.ceasingNetworking)
 173         for _, c := range t.Conns {
 174                 c.Close()
 175         }
 176 }
 177
 178 func (t *torrent) addPeer(p Peer, cl *Client) {
 179         cl.openNewConns(t)
 180         if len(t.Peers) >= torrentPeersHighWater {
 181                 return
 182         }
 183         key := PeersKey{string(p.IP), p.Port}
 184         if _, ok := t.Peers[key]; ok {
 185                 return
 186         }
 187         t.Peers[key] = p
 188         peersAddedBySource.Add(string(p.Source), 1)
 189         cl.openNewConns(t)
 190
 191 }
 192
 193 func (t *torrent) invalidateMetadata() {
 194         t.MetaData = nil
 195         t.metadataHave = nil
 196         t.Info = nil
 197 }
 198
 199 func (t *torrent) saveMetadataPiece(index int, data []byte) {
 200         if t.haveInfo() {
 201                 return
 202         }
 203         if index >= len(t.metadataHave) {
 204                 log.Printf("%s: ignoring metadata piece %d", t, index)
 205                 return
 206         }
 207         copy(t.MetaData[(1<<14)*index:], data)
 208         t.metadataHave[index] = true
 209 }
 210
 211 func (t *torrent) metadataPieceCount() int {
 212         return (len(t.MetaData) + (1 << 14) - 1) / (1 << 14)
 213 }
 214
 215 func (t *torrent) haveMetadataPiece(piece int) bool {
 216         if t.haveInfo() {
 217                 return (1<<14)*piece < len(t.MetaData)
 218         } else {
 219                 return piece < len(t.metadataHave) && t.metadataHave[piece]
 220         }
 221 }
 222
 223 func (t *torrent) metadataSizeKnown() bool {
 224         return t.MetaData != nil
 225 }
 226
 227 func (t *torrent) metadataSize() int {
 228         return len(t.MetaData)
 229 }
 230
 231 func infoPieceHashes(info *metainfo.Info) (ret []string) {
 232         for i := 0; i < len(info.Pieces); i += 20 {
 233                 ret = append(ret, string(info.Pieces[i:i+20]))
 234         }
 235         return
 236 }
 237
 238 // Called when metadata for a torrent becomes available.
 239 func (t *torrent) setMetadata(md *metainfo.Info, infoBytes []byte) (err error) {
 240         err = validateInfo(md)
 241         if err != nil {
 242                 err = fmt.Errorf("bad info: %s", err)
 243                 return
 244         }
 245         t.Info = md
 246         t.length = 0
 247         for _, f := range t.Info.UpvertedFiles() {
 248                 t.length += f.Length
 249         }
 250         t.MetaData = infoBytes
 251         t.metadataHave = nil
 252         hashes := infoPieceHashes(md)
 253         t.Pieces = make([]piece, len(hashes))
 254         for i, hash := range hashes {
 255                 piece := &t.Pieces[i]
 256                 piece.noPendingWrites.L = &piece.pendingWritesMutex
 257                 missinggo.CopyExact(piece.Hash[:], hash)
 258         }
 259         for _, conn := range t.Conns {
 260                 t.initRequestOrdering(conn)
 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 = p.Priority
 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, "Urgent:")
 435         for req := range t.urgent {
 436                 fmt.Fprintf(w, " %v", req)
 437         }
 438         fmt.Fprintln(w)
 439         fmt.Fprintf(w, "Trackers: ")
 440         for _, tier := range t.Trackers {
 441                 for _, tr := range tier {
 442                         fmt.Fprintf(w, "%q ", tr.String())
 443                 }
 444         }
 445         fmt.Fprintf(w, "\n")
 446         fmt.Fprintf(w, "Pending peers: %d\n", len(t.Peers))
 447         fmt.Fprintf(w, "Half open: %d\n", len(t.HalfOpen))
 448         fmt.Fprintf(w, "Active peers: %d\n", len(t.Conns))
 449         sort.Sort(&worstConns{
 450                 c:  t.Conns,
 451                 t:  t,
 452                 cl: cl,
 453         })
 454         for i, c := range t.Conns {
 455                 fmt.Fprintf(w, "%2d. ", i+1)
 456                 c.WriteStatus(w, t)
 457         }
 458 }
 459
 460 func (t *torrent) String() string {
 461         s := t.Name()
 462         if s == "" {
 463                 s = fmt.Sprintf("%x", t.InfoHash)
 464         }
 465         return s
 466 }
 467
 468 func (t *torrent) haveInfo() bool {
 469         return t != nil && t.Info != nil
 470 }
 471
 472 // TODO: Include URIs that weren't converted to tracker clients.
 473 func (t *torrent) announceList() (al [][]string) {
 474         for _, tier := range t.Trackers {
 475                 var l []string
 476                 for _, tr := range tier {
 477                         l = append(l, tr.URL())
 478                 }
 479                 al = append(al, l)
 480         }
 481         return
 482 }
 483
 484 // Returns a run-time generated MetaInfo that includes the info bytes and
 485 // announce-list as currently known to the client.
 486 func (t *torrent) MetaInfo() *metainfo.MetaInfo {
 487         if t.MetaData == nil {
 488                 panic("info bytes not set")
 489         }
 490         return &metainfo.MetaInfo{
 491                 Info: metainfo.InfoEx{
 492                         Info:  *t.Info,
 493                         Bytes: t.MetaData,
 494                 },
 495                 CreationDate: time.Now().Unix(),
 496                 Comment:      "dynamic metainfo from client",
 497                 CreatedBy:    "go.torrent",
 498                 AnnounceList: t.announceList(),
 499         }
 500 }
 501
 502 func (t *torrent) bytesLeft() (left int64) {
 503         if !t.haveInfo() {
 504                 return -1
 505         }
 506         for i := 0; i < t.numPieces(); i++ {
 507                 left += int64(t.pieceNumPendingBytes(i))
 508         }
 509         return
 510 }
 511
 512 func (t *torrent) piecePartiallyDownloaded(index int) bool {
 513         pendingBytes := t.pieceNumPendingBytes(index)
 514         return pendingBytes != 0 && pendingBytes != t.pieceLength(index)
 515 }
 516
 517 func numChunksForPiece(chunkSize int, pieceSize int) int {
 518         return (pieceSize + chunkSize - 1) / chunkSize
 519 }
 520
 521 func (t *torrent) usualPieceSize() int {
 522         return int(t.Info.PieceLength)
 523 }
 524
 525 func (t *torrent) lastPieceSize() int {
 526         return int(t.pieceLength(t.numPieces() - 1))
 527 }
 528
 529 func (t *torrent) numPieces() int {
 530         return t.Info.NumPieces()
 531 }
 532
 533 func (t *torrent) numPiecesCompleted() (num int) {
 534         for i := range iter.N(t.Info.NumPieces()) {
 535                 if t.pieceComplete(i) {
 536                         num++
 537                 }
 538         }
 539         return
 540 }
 541
 542 func (t *torrent) Length() int64 {
 543         return t.length
 544 }
 545
 546 func (t *torrent) isClosed() bool {
 547         select {
 548         case <-t.closing:
 549                 return true
 550         default:
 551                 return false
 552         }
 553 }
 554
 555 func (t *torrent) close() (err error) {
 556         if t.isClosed() {
 557                 return
 558         }
 559         t.ceaseNetworking()
 560         close(t.closing)
 561         if c, ok := t.data.(io.Closer); ok {
 562                 c.Close()
 563         }
 564         for _, conn := range t.Conns {
 565                 conn.Close()
 566         }
 567         t.pieceStateChanges.Close()
 568         return
 569 }
 570
 571 func (t *torrent) requestOffset(r request) int64 {
 572         return torrentRequestOffset(t.Length(), int64(t.usualPieceSize()), r)
 573 }
 574
 575 // Return the request that would include the given offset into the torrent
 576 // data. Returns !ok if there is no such request.
 577 func (t *torrent) offsetRequest(off int64) (req request, ok bool) {
 578         return torrentOffsetRequest(t.Length(), t.Info.PieceLength, int64(t.chunkSize), off)
 579 }
 580
 581 func (t *torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
 582         n, err := t.data.WriteAt(data, int64(piece)*t.Info.PieceLength+begin)
 583         if err == nil && n != len(data) {
 584                 err = io.ErrShortWrite
 585         }
 586         return
 587 }
 588
 589 func (t *torrent) bitfield() (bf []bool) {
 590         for i := range t.Pieces {
 591                 p := &t.Pieces[i]
 592                 // TODO: Check this logic.
 593                 bf = append(bf, p.EverHashed && p.numPendingChunks() == 0)
 594         }
 595         return
 596 }
 597
 598 func (t *torrent) validOutgoingRequest(r request) bool {
 599         if r.Index >= pp.Integer(t.Info.NumPieces()) {
 600                 return false
 601         }
 602         if r.Begin%t.chunkSize != 0 {
 603                 return false
 604         }
 605         if r.Length > t.chunkSize {
 606                 return false
 607         }
 608         pieceLength := t.pieceLength(int(r.Index))
 609         if r.Begin+r.Length > pieceLength {
 610                 return false
 611         }
 612         return r.Length == t.chunkSize || r.Begin+r.Length == pieceLength
 613 }
 614
 615 func (t *torrent) pieceChunks(piece int) (css []chunkSpec) {
 616         css = make([]chunkSpec, 0, (t.pieceLength(piece)+t.chunkSize-1)/t.chunkSize)
 617         var cs chunkSpec
 618         for left := t.pieceLength(piece); left != 0; left -= cs.Length {
 619                 cs.Length = left
 620                 if cs.Length > t.chunkSize {
 621                         cs.Length = t.chunkSize
 622                 }
 623                 css = append(css, cs)
 624                 cs.Begin += cs.Length
 625         }
 626         return
 627 }
 628
 629 func (t *torrent) pendAllChunkSpecs(pieceIndex int) {
 630         piece := &t.Pieces[pieceIndex]
 631         if piece.PendingChunkSpecs == nil {
 632                 // Allocate to exact size.
 633                 piece.PendingChunkSpecs = make([]bool, (t.pieceLength(pieceIndex)+t.chunkSize-1)/t.chunkSize)
 634         }
 635         // Pend all the chunks.
 636         pcss := piece.PendingChunkSpecs
 637         for i := range pcss {
 638                 pcss[i] = true
 639         }
 640         return
 641 }
 642
 643 type Peer struct {
 644         Id     [20]byte
 645         IP     net.IP
 646         Port   int
 647         Source peerSource
 648         // Peer is known to support encryption.
 649         SupportsEncryption bool
 650 }
 651
 652 func (t *torrent) pieceLength(piece int) (len_ pp.Integer) {
 653         if int(piece) == t.numPieces()-1 {
 654                 len_ = pp.Integer(t.Length() % t.Info.PieceLength)
 655         }
 656         if len_ == 0 {
 657                 len_ = pp.Integer(t.Info.PieceLength)
 658         }
 659         return
 660 }
 661
 662 func (t *torrent) hashPiece(piece int) (ps pieceSum) {
 663         hash := pieceHash.New()
 664         p := &t.Pieces[piece]
 665         p.pendingWritesMutex.Lock()
 666         for p.pendingWrites != 0 {
 667                 p.noPendingWrites.Wait()
 668         }
 669         p.pendingWritesMutex.Unlock()
 670         pl := t.Info.Piece(int(piece)).Length()
 671         n, err := t.data.WriteSectionTo(hash, int64(piece)*t.Info.PieceLength, pl)
 672         if err != nil {
 673                 if err != io.ErrUnexpectedEOF {
 674                         log.Printf("error hashing piece with %T: %s", t.data, err)
 675                 }
 676                 return
 677         }
 678         if n != pl {
 679                 panic(fmt.Sprintf("%T: %d != %d", t.data, n, pl))
 680         }
 681         missinggo.CopyExact(ps[:], hash.Sum(nil))
 682         return
 683 }
 684
 685 func (t *torrent) haveAllPieces() bool {
 686         if !t.haveInfo() {
 687                 return false
 688         }
 689         for i := range t.Pieces {
 690                 if !t.pieceComplete(i) {
 691                         return false
 692                 }
 693         }
 694         return true
 695 }
 696
 697 func (me *torrent) haveAnyPieces() bool {
 698         for i := range me.Pieces {
 699                 if me.pieceComplete(i) {
 700                         return true
 701                 }
 702         }
 703         return false
 704 }
 705
 706 func (t *torrent) havePiece(index int) bool {
 707         return t.haveInfo() && t.pieceComplete(index)
 708 }
 709
 710 func (t *torrent) haveChunk(r request) bool {
 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         if p.PendingChunkSpecs == nil {
 719                 return false
 720         }
 721         return !p.pendingChunk(r.chunkSpec, t.chunkSize)
 722 }
 723
 724 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
 725         return int(cs.Begin / chunkSize)
 726 }
 727
 728 // TODO: This should probably be called wantPiece.
 729 func (t *torrent) wantChunk(r request) bool {
 730         if !t.wantPiece(int(r.Index)) {
 731                 return false
 732         }
 733         if t.Pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) {
 734                 return true
 735         }
 736         _, ok := t.urgent[r]
 737         return ok
 738 }
 739
 740 func (t *torrent) urgentChunkInPiece(piece int) bool {
 741         p := pp.Integer(piece)
 742         for req := range t.urgent {
 743                 if req.Index == p {
 744                         return true
 745                 }
 746         }
 747         return false
 748 }
 749
 750 // TODO: This should be called wantPieceIndex.
 751 func (t *torrent) wantPiece(index int) bool {
 752         if !t.haveInfo() {
 753                 return false
 754         }
 755         p := &t.Pieces[index]
 756         if p.QueuedForHash {
 757                 return false
 758         }
 759         if p.Hashing {
 760                 return false
 761         }
 762         if p.Priority == PiecePriorityNone {
 763                 if !t.urgentChunkInPiece(index) {
 764                         return false
 765                 }
 766         }
 767         // Put piece complete check last, since it's the slowest as it can involve
 768         // calling out into external data stores.
 769         return !t.pieceComplete(index)
 770 }
 771
 772 func (t *torrent) connHasWantedPieces(c *connection) bool {
 773         return c.pieceRequestOrder != nil && !c.pieceRequestOrder.Empty()
 774 }
 775
 776 func (t *torrent) extentPieces(off, _len int64) (pieces []int) {
 777         for i := off / int64(t.usualPieceSize()); i*int64(t.usualPieceSize()) < off+_len; i++ {
 778                 pieces = append(pieces, int(i))
 779         }
 780         return
 781 }
 782
 783 func (t *torrent) worstBadConn(cl *Client) *connection {
 784         wcs := t.worstConns(cl)
 785         heap.Init(wcs)
 786         for wcs.Len() != 0 {
 787                 c := heap.Pop(wcs).(*connection)
 788                 if c.UnwantedChunksReceived >= 6 && c.UnwantedChunksReceived > c.UsefulChunksReceived {
 789                         return c
 790                 }
 791                 if wcs.Len() >= (socketsPerTorrent+1)/2 {
 792                         // Give connections 1 minute to prove themselves.
 793                         if time.Since(c.completedHandshake) > time.Minute {
 794                                 return c
 795                         }
 796                 }
 797         }
 798         return nil
 799 }
 800
 801 func (t *torrent) publishPieceChange(piece int) {
 802         cur := t.pieceState(piece)
 803         p := &t.Pieces[piece]
 804         if cur != p.PublicPieceState {
 805                 t.pieceStateChanges.Publish(piece)
 806         }
 807         p.PublicPieceState = cur
 808 }