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) (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.noPendingWrites.L = &piece.pendingWritesMutex
 253                 missinggo.CopyExact(piece.Hash[:], hash)
 254         }
 255         for _, conn := range t.Conns {
 256                 t.initRequestOrdering(conn)
 257                 if err := conn.setNumPieces(t.numPieces()); err != nil {
 258                         log.Printf("closing connection: %s", err)
 259                         conn.Close()
 260                 }
 261         }
 262         return
 263 }
 264
 265 func (t *torrent) setStorage(td Data) (err error) {
 266         if t.data != nil {
 267                 t.data.Close()
 268         }
 269         t.data = td
 270         return
 271 }
 272
 273 func (t *torrent) haveAllMetadataPieces() bool {
 274         if t.haveInfo() {
 275                 return true
 276         }
 277         if t.metadataHave == nil {
 278                 return false
 279         }
 280         for _, have := range t.metadataHave {
 281                 if !have {
 282                         return false
 283                 }
 284         }
 285         return true
 286 }
 287
 288 func (t *torrent) setMetadataSize(bytes int64, cl *Client) {
 289         if t.haveInfo() {
 290                 // We already know the correct metadata size.
 291                 return
 292         }
 293         if bytes <= 0 || bytes > 10000000 { // 10MB, pulled from my ass.
 294                 log.Printf("received bad metadata size: %d", bytes)
 295                 return
 296         }
 297         if t.MetaData != nil && len(t.MetaData) == int(bytes) {
 298                 return
 299         }
 300         t.MetaData = make([]byte, bytes)
 301         t.metadataHave = make([]bool, (bytes+(1<<14)-1)/(1<<14))
 302         for _, c := range t.Conns {
 303                 cl.requestPendingMetadata(t, c)
 304         }
 305
 306 }
 307
 308 // The current working name for the torrent. Either the name in the info dict,
 309 // or a display name given such as by the dn value in a magnet link, or "".
 310 func (t *torrent) Name() string {
 311         if t.haveInfo() {
 312                 return t.Info.Name
 313         }
 314         return t.DisplayName
 315 }
 316
 317 func (t *torrent) pieceState(index int) (ret PieceState) {
 318         p := &t.Pieces[index]
 319         ret.Priority = p.Priority
 320         if t.pieceComplete(index) {
 321                 ret.Complete = true
 322         }
 323         if p.QueuedForHash || p.Hashing {
 324                 ret.Checking = true
 325         }
 326         if !ret.Complete && t.piecePartiallyDownloaded(index) {
 327                 ret.Partial = true
 328         }
 329         return
 330 }
 331
 332 func (t *torrent) metadataPieceSize(piece int) int {
 333         return metadataPieceSize(len(t.MetaData), piece)
 334 }
 335
 336 func (t *torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
 337         d := map[string]int{
 338                 "msg_type": msgType,
 339                 "piece":    piece,
 340         }
 341         if data != nil {
 342                 d["total_size"] = len(t.MetaData)
 343         }
 344         p, err := bencode.Marshal(d)
 345         if err != nil {
 346                 panic(err)
 347         }
 348         return pp.Message{
 349                 Type:            pp.Extended,
 350                 ExtendedID:      byte(c.PeerExtensionIDs["ut_metadata"]),
 351                 ExtendedPayload: append(p, data...),
 352         }
 353 }
 354
 355 func (t *torrent) pieceStateRuns() (ret []PieceStateRun) {
 356         rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) {
 357                 ret = append(ret, PieceStateRun{
 358                         PieceState: el.(PieceState),
 359                         Length:     int(count),
 360                 })
 361         })
 362         for index := range t.Pieces {
 363                 rle.Append(t.pieceState(index), 1)
 364         }
 365         rle.Flush()
 366         return
 367 }
 368
 369 // Produces a small string representing a PieceStateRun.
 370 func pieceStateRunStatusChars(psr PieceStateRun) (ret string) {
 371         ret = fmt.Sprintf("%d", psr.Length)
 372         ret += func() string {
 373                 switch psr.Priority {
 374                 case PiecePriorityNext:
 375                         return "N"
 376                 case PiecePriorityNormal:
 377                         return "."
 378                 case PiecePriorityReadahead:
 379                         return "R"
 380                 case PiecePriorityNow:
 381                         return "!"
 382                 default:
 383                         return ""
 384                 }
 385         }()
 386         if psr.Checking {
 387                 ret += "H"
 388         }
 389         if psr.Partial {
 390                 ret += "P"
 391         }
 392         if psr.Complete {
 393                 ret += "C"
 394         }
 395         return
 396 }
 397
 398 func (t *torrent) writeStatus(w io.Writer, cl *Client) {
 399         fmt.Fprintf(w, "Infohash: %x\n", t.InfoHash)
 400         fmt.Fprintf(w, "Metadata length: %d\n", t.metadataSize())
 401         if !t.haveInfo() {
 402                 fmt.Fprintf(w, "Metadata have: ")
 403                 for _, h := range t.metadataHave {
 404                         fmt.Fprintf(w, "%c", func() rune {
 405                                 if h {
 406                                         return 'H'
 407                                 } else {
 408                                         return '.'
 409                                 }
 410                         }())
 411                 }
 412                 fmt.Fprintln(w)
 413         }
 414         fmt.Fprintf(w, "Piece length: %s\n", func() string {
 415                 if t.haveInfo() {
 416                         return fmt.Sprint(t.usualPieceSize())
 417                 } else {
 418                         return "?"
 419                 }
 420         }())
 421         if t.haveInfo() {
 422                 fmt.Fprintf(w, "Num Pieces: %d\n", t.numPieces())
 423                 fmt.Fprint(w, "Piece States:")
 424                 for _, psr := range t.pieceStateRuns() {
 425                         w.Write([]byte(" "))
 426                         w.Write([]byte(pieceStateRunStatusChars(psr)))
 427                 }
 428                 fmt.Fprintln(w)
 429         }
 430         fmt.Fprintf(w, "Urgent:")
 431         for req := range t.urgent {
 432                 fmt.Fprintf(w, " %v", req)
 433         }
 434         fmt.Fprintln(w)
 435         fmt.Fprintf(w, "Trackers: ")
 436         for _, tier := range t.Trackers {
 437                 for _, tr := range tier {
 438                         fmt.Fprintf(w, "%q ", tr.String())
 439                 }
 440         }
 441         fmt.Fprintf(w, "\n")
 442         fmt.Fprintf(w, "Pending peers: %d\n", len(t.Peers))
 443         fmt.Fprintf(w, "Half open: %d\n", len(t.HalfOpen))
 444         fmt.Fprintf(w, "Active peers: %d\n", len(t.Conns))
 445         sort.Sort(&worstConns{
 446                 c:  t.Conns,
 447                 t:  t,
 448                 cl: cl,
 449         })
 450         for i, c := range t.Conns {
 451                 fmt.Fprintf(w, "%2d. ", i+1)
 452                 c.WriteStatus(w, t)
 453         }
 454 }
 455
 456 func (t *torrent) String() string {
 457         s := t.Name()
 458         if s == "" {
 459                 s = fmt.Sprintf("%x", t.InfoHash)
 460         }
 461         return s
 462 }
 463
 464 func (t *torrent) haveInfo() bool {
 465         return t != nil && t.Info != nil
 466 }
 467
 468 // TODO: Include URIs that weren't converted to tracker clients.
 469 func (t *torrent) announceList() (al [][]string) {
 470         for _, tier := range t.Trackers {
 471                 var l []string
 472                 for _, tr := range tier {
 473                         l = append(l, tr.URL())
 474                 }
 475                 al = append(al, l)
 476         }
 477         return
 478 }
 479
 480 // Returns a run-time generated MetaInfo that includes the info bytes and
 481 // announce-list as currently known to the client.
 482 func (t *torrent) MetaInfo() *metainfo.MetaInfo {
 483         if t.MetaData == nil {
 484                 panic("info bytes not set")
 485         }
 486         return &metainfo.MetaInfo{
 487                 Info: metainfo.InfoEx{
 488                         Info:  *t.Info,
 489                         Bytes: t.MetaData,
 490                 },
 491                 CreationDate: time.Now().Unix(),
 492                 Comment:      "dynamic metainfo from client",
 493                 CreatedBy:    "go.torrent",
 494                 AnnounceList: t.announceList(),
 495         }
 496 }
 497
 498 func (t *torrent) bytesLeft() (left int64) {
 499         if !t.haveInfo() {
 500                 return -1
 501         }
 502         for i := 0; i < t.numPieces(); i++ {
 503                 left += int64(t.pieceNumPendingBytes(i))
 504         }
 505         return
 506 }
 507
 508 func (t *torrent) piecePartiallyDownloaded(index int) bool {
 509         pendingBytes := t.pieceNumPendingBytes(index)
 510         return pendingBytes != 0 && pendingBytes != t.pieceLength(index)
 511 }
 512
 513 func numChunksForPiece(chunkSize int, pieceSize int) int {
 514         return (pieceSize + chunkSize - 1) / chunkSize
 515 }
 516
 517 func (t *torrent) usualPieceSize() int {
 518         return int(t.Info.PieceLength)
 519 }
 520
 521 func (t *torrent) lastPieceSize() int {
 522         return int(t.pieceLength(t.numPieces() - 1))
 523 }
 524
 525 func (t *torrent) numPieces() int {
 526         return t.Info.NumPieces()
 527 }
 528
 529 func (t *torrent) numPiecesCompleted() (num int) {
 530         for i := range iter.N(t.Info.NumPieces()) {
 531                 if t.pieceComplete(i) {
 532                         num++
 533                 }
 534         }
 535         return
 536 }
 537
 538 func (t *torrent) Length() int64 {
 539         return t.length
 540 }
 541
 542 func (t *torrent) isClosed() bool {
 543         select {
 544         case <-t.closing:
 545                 return true
 546         default:
 547                 return false
 548         }
 549 }
 550
 551 func (t *torrent) close() (err error) {
 552         if t.isClosed() {
 553                 return
 554         }
 555         t.ceaseNetworking()
 556         close(t.closing)
 557         if c, ok := t.data.(io.Closer); ok {
 558                 c.Close()
 559         }
 560         for _, conn := range t.Conns {
 561                 conn.Close()
 562         }
 563         t.pieceStateChanges.Close()
 564         return
 565 }
 566
 567 func (t *torrent) requestOffset(r request) int64 {
 568         return torrentRequestOffset(t.Length(), int64(t.usualPieceSize()), r)
 569 }
 570
 571 // Return the request that would include the given offset into the torrent
 572 // data. Returns !ok if there is no such request.
 573 func (t *torrent) offsetRequest(off int64) (req request, ok bool) {
 574         return torrentOffsetRequest(t.Length(), t.Info.PieceLength, int64(t.chunkSize), off)
 575 }
 576
 577 func (t *torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
 578         n, err := t.data.WriteAt(data, int64(piece)*t.Info.PieceLength+begin)
 579         if err == nil && n != len(data) {
 580                 err = io.ErrShortWrite
 581         }
 582         return
 583 }
 584
 585 func (t *torrent) bitfield() (bf []bool) {
 586         for i := range t.Pieces {
 587                 p := &t.Pieces[i]
 588                 // TODO: Check this logic.
 589                 bf = append(bf, p.EverHashed && p.numPendingChunks() == 0)
 590         }
 591         return
 592 }
 593
 594 func (t *torrent) validOutgoingRequest(r request) bool {
 595         if r.Index >= pp.Integer(t.Info.NumPieces()) {
 596                 return false
 597         }
 598         if r.Begin%t.chunkSize != 0 {
 599                 return false
 600         }
 601         if r.Length > t.chunkSize {
 602                 return false
 603         }
 604         pieceLength := t.pieceLength(int(r.Index))
 605         if r.Begin+r.Length > pieceLength {
 606                 return false
 607         }
 608         return r.Length == t.chunkSize || r.Begin+r.Length == pieceLength
 609 }
 610
 611 func (t *torrent) pieceChunks(piece int) (css []chunkSpec) {
 612         css = make([]chunkSpec, 0, (t.pieceLength(piece)+t.chunkSize-1)/t.chunkSize)
 613         var cs chunkSpec
 614         for left := t.pieceLength(piece); left != 0; left -= cs.Length {
 615                 cs.Length = left
 616                 if cs.Length > t.chunkSize {
 617                         cs.Length = t.chunkSize
 618                 }
 619                 css = append(css, cs)
 620                 cs.Begin += cs.Length
 621         }
 622         return
 623 }
 624
 625 func (t *torrent) pendAllChunkSpecs(pieceIndex int) {
 626         piece := &t.Pieces[pieceIndex]
 627         if piece.PendingChunkSpecs == nil {
 628                 // Allocate to exact size.
 629                 piece.PendingChunkSpecs = make([]bool, (t.pieceLength(pieceIndex)+t.chunkSize-1)/t.chunkSize)
 630         }
 631         // Pend all the chunks.
 632         pcss := piece.PendingChunkSpecs
 633         for i := range pcss {
 634                 pcss[i] = true
 635         }
 636         return
 637 }
 638
 639 type Peer struct {
 640         Id     [20]byte
 641         IP     net.IP
 642         Port   int
 643         Source peerSource
 644         // Peer is known to support encryption.
 645         SupportsEncryption bool
 646 }
 647
 648 func (t *torrent) pieceLength(piece int) (len_ pp.Integer) {
 649         if int(piece) == t.numPieces()-1 {
 650                 len_ = pp.Integer(t.Length() % t.Info.PieceLength)
 651         }
 652         if len_ == 0 {
 653                 len_ = pp.Integer(t.Info.PieceLength)
 654         }
 655         return
 656 }
 657
 658 func (t *torrent) hashPiece(piece pp.Integer) (ps pieceSum) {
 659         hash := pieceHash.New()
 660         p := &t.Pieces[piece]
 661         p.pendingWritesMutex.Lock()
 662         for p.pendingWrites != 0 {
 663                 p.noPendingWrites.Wait()
 664         }
 665         p.pendingWritesMutex.Unlock()
 666         pl := t.Info.Piece(int(piece)).Length()
 667         n, err := t.data.WriteSectionTo(hash, int64(piece)*t.Info.PieceLength, pl)
 668         if err != nil {
 669                 if err != io.ErrUnexpectedEOF {
 670                         log.Printf("error hashing piece with %T: %s", t.data, err)
 671                 }
 672                 return
 673         }
 674         if n != pl {
 675                 panic(fmt.Sprintf("%T: %d != %d", t.data, n, pl))
 676         }
 677         missinggo.CopyExact(ps[:], hash.Sum(nil))
 678         return
 679 }
 680
 681 func (t *torrent) haveAllPieces() bool {
 682         if !t.haveInfo() {
 683                 return false
 684         }
 685         for i := range t.Pieces {
 686                 if !t.pieceComplete(i) {
 687                         return false
 688                 }
 689         }
 690         return true
 691 }
 692
 693 func (me *torrent) haveAnyPieces() bool {
 694         for i := range me.Pieces {
 695                 if me.pieceComplete(i) {
 696                         return true
 697                 }
 698         }
 699         return false
 700 }
 701
 702 func (t *torrent) havePiece(index int) bool {
 703         return t.haveInfo() && t.pieceComplete(index)
 704 }
 705
 706 func (t *torrent) haveChunk(r request) bool {
 707         if !t.haveInfo() {
 708                 return false
 709         }
 710         if t.pieceComplete(int(r.Index)) {
 711                 return true
 712         }
 713         p := &t.Pieces[r.Index]
 714         if p.PendingChunkSpecs == nil {
 715                 return false
 716         }
 717         return !p.pendingChunk(r.chunkSpec, t.chunkSize)
 718 }
 719
 720 func chunkIndex(cs chunkSpec, chunkSize pp.Integer) int {
 721         return int(cs.Begin / chunkSize)
 722 }
 723
 724 // TODO: This should probably be called wantPiece.
 725 func (t *torrent) wantChunk(r request) bool {
 726         if !t.wantPiece(int(r.Index)) {
 727                 return false
 728         }
 729         if t.Pieces[r.Index].pendingChunk(r.chunkSpec, t.chunkSize) {
 730                 return true
 731         }
 732         _, ok := t.urgent[r]
 733         return ok
 734 }
 735
 736 func (t *torrent) urgentChunkInPiece(piece int) bool {
 737         p := pp.Integer(piece)
 738         for req := range t.urgent {
 739                 if req.Index == p {
 740                         return true
 741                 }
 742         }
 743         return false
 744 }
 745
 746 // TODO: This should be called wantPieceIndex.
 747 func (t *torrent) wantPiece(index int) bool {
 748         if !t.haveInfo() {
 749                 return false
 750         }
 751         p := &t.Pieces[index]
 752         if p.QueuedForHash {
 753                 return false
 754         }
 755         if p.Hashing {
 756                 return false
 757         }
 758         if p.Priority == PiecePriorityNone {
 759                 if !t.urgentChunkInPiece(index) {
 760                         return false
 761                 }
 762         }
 763         // Put piece complete check last, since it's the slowest as it can involve
 764         // calling out into external data stores.
 765         return !t.pieceComplete(index)
 766 }
 767
 768 func (t *torrent) connHasWantedPieces(c *connection) bool {
 769         return c.pieceRequestOrder != nil && !c.pieceRequestOrder.Empty()
 770 }
 771
 772 func (t *torrent) extentPieces(off, _len int64) (pieces []int) {
 773         for i := off / int64(t.usualPieceSize()); i*int64(t.usualPieceSize()) < off+_len; i++ {
 774                 pieces = append(pieces, int(i))
 775         }
 776         return
 777 }
 778
 779 func (t *torrent) worstBadConn(cl *Client) *connection {
 780         wcs := t.worstConns(cl)
 781         heap.Init(wcs)
 782         for wcs.Len() != 0 {
 783                 c := heap.Pop(wcs).(*connection)
 784                 if c.UnwantedChunksReceived >= 6 && c.UnwantedChunksReceived > c.UsefulChunksReceived {
 785                         return c
 786                 }
 787                 if wcs.Len() >= (socketsPerTorrent+1)/2 {
 788                         // Give connections 1 minute to prove themselves.
 789                         if time.Since(c.completedHandshake) > time.Minute {
 790                                 return c
 791                         }
 792                 }
 793         }
 794         return nil
 795 }
 796
 797 func (t *torrent) publishPieceChange(piece int) {
 798         cur := t.pieceState(piece)
 799         p := &t.Pieces[piece]
 800         if cur != p.PublicPieceState {
 801                 t.pieceStateChanges.Publish(piece)
 802         }
 803         p.PublicPieceState = cur
 804 }