torrent.go

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