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) newConnPiecePriorities() []int {
 107         _ret := t.connPiecePriorites.Get()
 108         if _ret != nil {
 109                 piecePrioritiesReused.Add(1)
 110                 return _ret.([]int)
 111         }
 112         piecePrioritiesNew.Add(1)
 113         return rand.Perm(t.numPieces())
 114 }
 115
 116 func (t *torrent) pieceComplete(piece int) bool {
 117         // TODO: This is called when setting metadata, and before storage is
 118         // assigned, which doesn't seem right.
 119         return t.data != nil && t.data.PieceComplete(piece)
 120 }
 121
 122 func (t *torrent) numConnsUnchoked() (num int) {
 123         for _, c := range t.Conns {
 124                 if !c.PeerChoked {
 125                         num++
 126                 }
 127         }
 128         return
 129 }
 130
 131 // There's a connection to that address already.
 132 func (t *torrent) addrActive(addr string) bool {
 133         if _, ok := t.HalfOpen[addr]; ok {
 134                 return true
 135         }
 136         for _, c := range t.Conns {
 137                 if c.remoteAddr().String() == addr {
 138                         return true
 139                 }
 140         }
 141         return false
 142 }
 143
 144 func (t *torrent) worstConns(cl *Client) (wcs *worstConns) {
 145         wcs = &worstConns{
 146                 c:  make([]*connection, 0, len(t.Conns)),
 147                 t:  t,
 148                 cl: cl,
 149         }
 150         for _, c := range t.Conns {
 151                 select {
 152                 case <-c.closing:
 153                 default:
 154                         wcs.c = append(wcs.c, c)
 155                 }
 156         }
 157         return
 158 }
 159
 160 func (t *torrent) ceaseNetworking() {
 161         t.stateMu.Lock()
 162         defer t.stateMu.Unlock()
 163         select {
 164         case <-t.ceasingNetworking:
 165                 return
 166         default:
 167         }
 168         close(t.ceasingNetworking)
 169         for _, c := range t.Conns {
 170                 c.Close()
 171         }
 172 }
 173
 174 func (t *torrent) addPeer(p Peer, cl *Client) {
 175         cl.openNewConns(t)
 176         if len(t.Peers) >= torrentPeersHighWater {
 177                 return
 178         }
 179         key := peersKey{string(p.IP), p.Port}
 180         if _, ok := t.Peers[key]; ok {
 181                 return
 182         }
 183         t.Peers[key] = p
 184         peersAddedBySource.Add(string(p.Source), 1)
 185         cl.openNewConns(t)
 186
 187 }
 188
 189 func (t *torrent) invalidateMetadata() {
 190         t.MetaData = nil
 191         t.metadataHave = nil
 192         t.Info = nil
 193 }
 194
 195 func (t *torrent) saveMetadataPiece(index int, data []byte) {
 196         if t.haveInfo() {
 197                 return
 198         }
 199         if index >= len(t.metadataHave) {
 200                 log.Printf("%s: ignoring metadata piece %d", t, index)
 201                 return
 202         }
 203         copy(t.MetaData[(1<<14)*index:], data)
 204         t.metadataHave[index] = true
 205 }
 206
 207 func (t *torrent) metadataPieceCount() int {
 208         return (len(t.MetaData) + (1 << 14) - 1) / (1 << 14)
 209 }
 210
 211 func (t *torrent) haveMetadataPiece(piece int) bool {
 212         if t.haveInfo() {
 213                 return (1<<14)*piece < len(t.MetaData)
 214         } else {
 215                 return piece < len(t.metadataHave) && t.metadataHave[piece]
 216         }
 217 }
 218
 219 func (t *torrent) metadataSizeKnown() bool {
 220         return t.MetaData != nil
 221 }
 222
 223 func (t *torrent) metadataSize() int {
 224         return len(t.MetaData)
 225 }
 226
 227 func infoPieceHashes(info *metainfo.Info) (ret []string) {
 228         for i := 0; i < len(info.Pieces); i += 20 {
 229                 ret = append(ret, string(info.Pieces[i:i+20]))
 230         }
 231         return
 232 }
 233
 234 // Called when metadata for a torrent becomes available.
 235 func (t *torrent) setMetadata(md *metainfo.Info, infoBytes []byte, eventLocker sync.Locker) (err error) {
 236         err = validateInfo(md)
 237         if err != nil {
 238                 err = fmt.Errorf("bad info: %s", err)
 239                 return
 240         }
 241         t.Info = md
 242         t.length = 0
 243         for _, f := range t.Info.UpvertedFiles() {
 244                 t.length += f.Length
 245         }
 246         t.MetaData = infoBytes
 247         t.metadataHave = nil
 248         hashes := infoPieceHashes(md)
 249         t.Pieces = make([]piece, len(hashes))
 250         for i, hash := range hashes {
 251                 piece := &t.Pieces[i]
 252                 piece.Event.L = eventLocker
 253                 piece.noPendingWrites.L = &piece.pendingWritesMutex
 254                 missinggo.CopyExact(piece.Hash[:], hash)
 255         }
 256         for _, conn := range t.Conns {
 257                 t.initRequestOrdering(conn)
 258                 if err := conn.setNumPieces(t.numPieces()); err != nil {
 259                         log.Printf("closing connection: %s", err)
 260                         conn.Close()
 261                 }
 262         }
 263         return
 264 }
 265
 266 func (t *torrent) setStorage(td Data) (err error) {
 267         if t.data != nil {
 268                 t.data.Close()
 269         }
 270         t.data = td
 271         return
 272 }
 273
 274 func (t *torrent) haveAllMetadataPieces() bool {
 275         if t.haveInfo() {
 276                 return true
 277         }
 278         if t.metadataHave == nil {
 279                 return false
 280         }
 281         for _, have := range t.metadataHave {
 282                 if !have {
 283                         return false
 284                 }
 285         }
 286         return true
 287 }
 288
 289 func (t *torrent) setMetadataSize(bytes int64, cl *Client) {
 290         if t.haveInfo() {
 291                 // We already know the correct metadata size.
 292                 return
 293         }
 294         if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
 295                 log.Printf("received bad metadata size: %d", bytes)
 296                 return
 297         }
 298         if t.MetaData != nil && len(t.MetaData) == int(bytes) {
 299                 return
 300         }
 301         t.MetaData = make([]byte, bytes)
 302         t.metadataHave = make([]bool, (bytes+(1<<14)-1)/(1<<14))
 303         for _, c := range t.Conns {
 304                 cl.requestPendingMetadata(t, c)
 305         }
 306
 307 }
 308
 309 // The current working name for the torrent. Either the name in the info dict,
 310 // or a display name given such as by the dn value in a magnet link, or "".
 311 func (t *torrent) Name() string {
 312         if t.haveInfo() {
 313                 return t.Info.Name
 314         }
 315         return t.DisplayName
 316 }
 317
 318 func (t *torrent) pieceState(index int) (ret PieceState) {
 319         p := &t.Pieces[index]
 320         ret.Priority = p.Priority
 321         if t.pieceComplete(index) {
 322                 ret.Complete = true
 323         }
 324         if p.QueuedForHash || p.Hashing {
 325                 ret.Checking = true
 326         }
 327         if !ret.Complete && t.piecePartiallyDownloaded(index) {
 328                 ret.Partial = true
 329         }
 330         return
 331 }
 332
 333 func (t *torrent) metadataPieceSize(piece int) int {
 334         return metadataPieceSize(len(t.MetaData), piece)
 335 }
 336
 337 func (t *torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
 338         d := map[string]int{
 339                 "msg_type": msgType,
 340                 "piece":    piece,
 341         }
 342         if data != nil {
 343                 d["total_size"] = len(t.MetaData)
 344         }
 345         p, err := bencode.Marshal(d)
 346         if err != nil {
 347                 panic(err)
 348         }
 349         return pp.Message{
 350                 Type:            pp.Extended,
 351                 ExtendedID:      byte(c.PeerExtensionIDs["ut_metadata"]),
 352                 ExtendedPayload: append(p, data...),
 353         }
 354 }
 355
 356 func (t *torrent) pieceStateRuns() (ret []PieceStateRun) {
 357         rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
 358                 ret = append(ret, PieceStateRun{
 359                         PieceState: el.(PieceState),
 360                         Length:     int(count),
 361                 })
 362         })
 363         for index := range t.Pieces {
 364                 rle.Append(t.pieceState(index), 1)
 365         }
 366         rle.Flush()
 367         return
 368 }
 369
 370 // Produces a small string representing a PieceStateRun.
 371 func pieceStateRunStatusChars(psr PieceStateRun) (ret string) {
 372         ret = fmt.Sprintf("%d", psr.Length)
 373         ret += func() string {
 374                 switch psr.Priority {
 375                 case PiecePriorityNext:
 376                         return "N"
 377                 case PiecePriorityNormal:
 378                         return "."
 379                 case PiecePriorityReadahead:
 380                         return "R"
 381                 case PiecePriorityNow:
 382                         return "!"
 383                 default:
 384                         return ""
 385                 }
 386         }()
 387         if psr.Checking {
 388                 ret += "H"
 389         }
 390         if psr.Partial {
 391                 ret += "P"
 392         }
 393         if psr.Complete {
 394                 ret += "C"
 395         }
 396         return
 397 }
 398
 399 func (t *torrent) writeStatus(w io.Writer, cl *Client) {
 400         fmt.Fprintf(w, "Infohash: %x\n", t.InfoHash)
 401         fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
 402         if !t.haveInfo() {
 403                 fmt.Fprintf(w, "Metadata have: ")
 404                 for _, h := range t.metadataHave {
 405                         fmt.Fprintf(w, "%c", func() rune {
 406                                 if h {
 407                                         return 'H'
 408                                 } else {
 409                                         return '.'
 410                                 }
 411                         }())
 412                 }
 413                 fmt.Fprintln(w)
 414         }
 415         fmt.Fprintf(w, "Piece length: %s\n", func() string {
 416                 if t.haveInfo() {
 417                         return fmt.Sprint(t.usualPieceSize())
 418                 } else {
 419                         return "?"
 420                 }
 421         }())
 422         if t.haveInfo() {
 423                 fmt.Fprintf(w, "Num Pieces: %d\n", t.numPieces())
 424                 fmt.Fprint(w, "Piece States:")
 425                 for _, psr := range t.pieceStateRuns() {
 426                         w.Write([]byte(" "))
 427                         w.Write([]byte(pieceStateRunStatusChars(psr)))
 428                 }
 429                 fmt.Fprintln(w)
 430         }
 431         fmt.Fprintf(w, "Urgent:")
 432         for req := range t.urgent {
 433                 fmt.Fprintf(w, " %v", req)
 434         }
 435         fmt.Fprintln(w)
 436         fmt.Fprintf(w, "Trackers: ")
 437         for _, tier := range t.Trackers {
 438                 for _, tr := range tier {
 439                         fmt.Fprintf(w, "%q ", tr.String())
 440                 }
 441         }
 442         fmt.Fprintf(w, "\n")
 443         fmt.Fprintf(w, "Pending peers: %d\n", len(t.Peers))
 444         fmt.Fprintf(w, "Half open: %d\n", len(t.HalfOpen))
 445         fmt.Fprintf(w, "Active peers: %d\n", len(t.Conns))
 446         sort.Sort(&worstConns{
 447                 c:  t.Conns,
 448                 t:  t,
 449                 cl: cl,
 450         })
 451         for i, c := range t.Conns {
 452                 fmt.Fprintf(w, "%2d. ", i+1)
 453                 c.WriteStatus(w, t)
 454         }
 455 }
 456
 457 func (t *torrent) String() string {
 458         s := t.Name()
 459         if s == "" {
 460                 s = fmt.Sprintf("%x", t.InfoHash)
 461         }
 462         return s
 463 }
 464
 465 func (t *torrent) haveInfo() bool {
 466         return t != nil && t.Info != nil
 467 }
 468
 469 // TODO: Include URIs that weren't converted to tracker clients.
 470 func (t *torrent) announceList() (al [][]string) {
 471         for _, tier := range t.Trackers {
 472                 var l []string
 473                 for _, tr := range tier {
 474                         l = append(l, tr.URL())
 475                 }
 476                 al = append(al, l)
 477         }
 478         return
 479 }
 480
 481 // Returns a run-time generated MetaInfo that includes the info bytes and
 482 // announce-list as currently known to the client.
 483 func (t *torrent) MetaInfo() *metainfo.MetaInfo {
 484         if t.MetaData == nil {
 485                 panic("info bytes not set")
 486         }
 487         return &metainfo.MetaInfo{
 488                 Info: metainfo.InfoEx{
 489                         Info:  *t.Info,
 490                         Bytes: t.MetaData,
 491                 },
 492                 CreationDate: time.Now().Unix(),
 493                 Comment:      "dynamic metainfo from client",
 494                 CreatedBy:    "go.torrent",
 495                 AnnounceList: t.announceList(),
 496         }
 497 }
 498
 499 func (t *torrent) bytesLeft() (left int64) {
 500         if !t.haveInfo() {
 501                 return -1
 502         }
 503         for i := 0; i < t.numPieces(); i++ {
 504                 left += int64(t.pieceNumPendingBytes(i))
 505         }
 506         return
 507 }
 508
 509 func (t *torrent) piecePartiallyDownloaded(index int) bool {
 510         pendingBytes := t.pieceNumPendingBytes(index)
 511         return pendingBytes != 0 && pendingBytes != t.pieceLength(index)
 512 }
 513
 514 func numChunksForPiece(chunkSize int, pieceSize int) int {
 515         return (pieceSize + chunkSize - 1) / chunkSize
 516 }
 517
 518 func (t *torrent) usualPieceSize() int {
 519         return int(t.Info.PieceLength)
 520 }
 521
 522 func (t *torrent) lastPieceSize() int {
 523         return int(t.pieceLength(t.numPieces() - 1))
 524 }
 525
 526 func (t *torrent) numPieces() int {
 527         return t.Info.NumPieces()
 528 }
 529
 530 func (t *torrent) numPiecesCompleted() (num int) {
 531         for i := range iter.N(t.Info.NumPieces()) {
 532                 if t.pieceComplete(i) {
 533                         num++
 534                 }
 535         }
 536         return
 537 }
 538
 539 func (t *torrent) Length() int64 {
 540         return t.length
 541 }
 542
 543 func (t *torrent) isClosed() bool {
 544         select {
 545         case <-t.closing:
 546                 return true
 547         default:
 548                 return false
 549         }
 550 }
 551
 552 func (t *torrent) close() (err error) {
 553         if t.isClosed() {
 554                 return
 555         }
 556         t.ceaseNetworking()
 557         close(t.closing)
 558         if c, ok := t.data.(io.Closer); ok {
 559                 c.Close()
 560         }
 561         for _, conn := range t.Conns {
 562                 conn.Close()
 563         }
 564         t.pieceStateChanges.Close()
 565         return
 566 }
 567
 568 func (t *torrent) requestOffset(r request) int64 {
 569         return torrentRequestOffset(t.Length(), int64(t.usualPieceSize()), r)
 570 }
 571
 572 // Return the request that would include the given offset into the torrent
 573 // data. Returns !ok if there is no such request.
 574 func (t *torrent) offsetRequest(off int64) (req request, ok bool) {
 575         return torrentOffsetRequest(t.Length(), t.Info.PieceLength, int64(t.chunkSize), off)
 576 }
 577
 578 func (t *torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
 579         n, err := t.data.WriteAt(data, int64(piece)*t.Info.PieceLength+begin)
 580         if err == nil && n != len(data) {
 581                 err = io.ErrShortWrite
 582         }
 583         return
 584 }
 585
 586 func (t *torrent) bitfield() (bf []bool) {
 587         for i := range t.Pieces {
 588                 p := &t.Pieces[i]
 589                 // TODO: Check this logic.
 590                 bf = append(bf, p.EverHashed && p.numPendingChunks() == 0)
 591         }
 592         return
 593 }
 594
 595 func (t *torrent) validOutgoingRequest(r request) bool {
 596         if r.Index >= pp.Integer(t.Info.NumPieces()) {
 597                 return false
 598         }
 599         if r.Begin%t.chunkSize != 0 {
 600                 return false
 601         }
 602         if r.Length > t.chunkSize {
 603                 return false
 604         }
 605         pieceLength := t.pieceLength(int(r.Index))
 606         if r.Begin+r.Length > pieceLength {
 607                 return false
 608         }
 609         return r.Length == t.chunkSize || r.Begin+r.Length == pieceLength
 610 }
 611
 612 func (t *torrent) pieceChunks(piece int) (css []chunkSpec) {
 613         css = make([]chunkSpec, 0, (t.pieceLength(piece)+t.chunkSize-1)/t.chunkSize)
 614         var cs chunkSpec
 615         for left := t.pieceLength(piece); left != 0; left -= cs.Length {
 616                 cs.Length = left
 617                 if cs.Length > t.chunkSize {
 618                         cs.Length = t.chunkSize
 619                 }
 620                 css = append(css, cs)
 621                 cs.Begin += cs.Length
 622         }
 623         return
 624 }
 625
 626 func (t *torrent) pendAllChunkSpecs(pieceIndex int) {
 627         piece := &t.Pieces[pieceIndex]
 628         if piece.PendingChunkSpecs == nil {
 629                 // Allocate to exact size.
 630                 piece.PendingChunkSpecs = make([]bool, (t.pieceLength(pieceIndex)+t.chunkSize-1)/t.chunkSize)
 631         }
 632         // Pend all the chunks.
 633         pcss := piece.PendingChunkSpecs
 634         for i := range pcss {
 635                 pcss[i] = true
 636         }
 637         return
 638 }
 639
 640 type Peer struct {
 641         Id     [20]byte
 642         IP     net.IP
 643         Port   int
 644         Source peerSource
 645         // Peer is known to support encryption.
 646         SupportsEncryption bool
 647 }
 648
 649 func (t *torrent) pieceLength(piece int) (len_ pp.Integer) {
 650         if int(piece) == t.numPieces()-1 {
 651                 len_ = pp.Integer(t.Length() % t.Info.PieceLength)
 652         }
 653         if len_ == 0 {
 654                 len_ = pp.Integer(t.Info.PieceLength)
 655         }
 656         return
 657 }
 658
 659 func (t *torrent) hashPiece(piece pp.Integer) (ps pieceSum) {
 660         hash := pieceHash.New()
 661         p := &t.Pieces[piece]
 662         p.pendingWritesMutex.Lock()
 663         for p.pendingWrites != 0 {
 664                 p.noPendingWrites.Wait()
 665         }
 666         p.pendingWritesMutex.Unlock()
 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) bool {
 708         if !t.haveInfo() {
 709                 return false
 710         }
 711         if t.pieceComplete(int(r.Index)) {
 712                 return true
 713         }
 714         p := &t.Pieces[r.Index]
 715         if p.PendingChunkSpecs == nil {
 716                 return false
 717         }
 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         _, ok := t.urgent[r]
 734         return ok
 735 }
 736
 737 func (t *torrent) urgentChunkInPiece(piece int) bool {
 738         p := pp.Integer(piece)
 739         for req := range t.urgent {
 740                 if req.Index == p {
 741                         return true
 742                 }
 743         }
 744         return false
 745 }
 746
 747 // TODO: This should be called wantPieceIndex.
 748 func (t *torrent) wantPiece(index int) bool {
 749         if !t.haveInfo() {
 750                 return false
 751         }
 752         p := &t.Pieces[index]
 753         if p.QueuedForHash {
 754                 return false
 755         }
 756         if p.Hashing {
 757                 return false
 758         }
 759         if p.Priority == PiecePriorityNone {
 760                 if !t.urgentChunkInPiece(index) {
 761                         return false
 762                 }
 763         }
 764         // Put piece complete check last, since it's the slowest as it can involve
 765         // calling out into external data stores.
 766         return !t.pieceComplete(index)
 767 }
 768
 769 func (t *torrent) connHasWantedPieces(c *connection) bool {
 770         return c.pieceRequestOrder != nil && !c.pieceRequestOrder.Empty()
 771 }
 772
 773 func (t *torrent) extentPieces(off, _len int64) (pieces []int) {
 774         for i := off / int64(t.usualPieceSize()); i*int64(t.usualPieceSize()) < off+_len; i++ {
 775                 pieces = append(pieces, int(i))
 776         }
 777         return
 778 }
 779
 780 func (t *torrent) worstBadConn(cl *Client) *connection {
 781         wcs := t.worstConns(cl)
 782         heap.Init(wcs)
 783         for wcs.Len() != 0 {
 784                 c := heap.Pop(wcs).(*connection)
 785                 if c.UnwantedChunksReceived >= 6 && c.UnwantedChunksReceived > c.UsefulChunksReceived {
 786                         return c
 787                 }
 788                 if wcs.Len() >= (socketsPerTorrent+1)/2 {
 789                         // Give connections 1 minute to prove themselves.
 790                         if time.Since(c.completedHandshake) > time.Minute {
 791                                 return c
 792                         }
 793                 }
 794         }
 795         return nil
 796 }
 797
 798 func (t *torrent) publishPieceChange(piece int) {
 799         cur := t.pieceState(piece)
 800         p := &t.Pieces[piece]
 801         if cur != p.PublicPieceState {
 802                 t.pieceStateChanges.Publish(piece)
 803         }
 804         p.PublicPieceState = cur
 805 }