client.go

   1 /*
   2 Package torrent implements a torrent client.
   3
   4 Simple example:
   5
   6         c := &Client{}
   7         c.Start()
   8         defer c.Stop()
   9         if err := c.AddTorrent(externalMetaInfoPackageSux); err != nil {
  10                 return fmt.Errors("error adding torrent: %s", err)
  11         }
  12         c.WaitAll()
  13         log.Print("erhmahgerd, torrent downloaded")
  14
  15 */
  16 package torrent
  17
  18 import (
  19         "bufio"
  20         "bytes"
  21         "container/heap"
  22         "crypto/rand"
  23         "crypto/sha1"
  24         "errors"
  25         "expvar"
  26         "fmt"
  27         "io"
  28         "log"
  29         "math/big"
  30         mathRand "math/rand"
  31         "net"
  32         "os"
  33         "path/filepath"
  34         "sort"
  35         "strings"
  36         "syscall"
  37         "time"
  38
  39         "bitbucket.org/anacrolix/go.torrent/data"
  40         filePkg "bitbucket.org/anacrolix/go.torrent/data/file"
  41         "bitbucket.org/anacrolix/go.torrent/dht"
  42         "bitbucket.org/anacrolix/go.torrent/internal/pieceordering"
  43         "bitbucket.org/anacrolix/go.torrent/iplist"
  44         pp "bitbucket.org/anacrolix/go.torrent/peer_protocol"
  45         "bitbucket.org/anacrolix/go.torrent/tracker"
  46         _ "bitbucket.org/anacrolix/go.torrent/tracker/udp"
  47         . "bitbucket.org/anacrolix/go.torrent/util"
  48         "bitbucket.org/anacrolix/sync"
  49         "bitbucket.org/anacrolix/utp"
  50         "github.com/anacrolix/libtorgo/bencode"
  51         "github.com/anacrolix/libtorgo/metainfo"
  52 )
  53
  54 var (
  55         unusedDownloadedChunksCount = expvar.NewInt("unusedDownloadedChunksCount")
  56         chunksDownloadedCount       = expvar.NewInt("chunksDownloadedCount")
  57         peersFoundByDHT             = expvar.NewInt("peersFoundByDHT")
  58         peersFoundByPEX             = expvar.NewInt("peersFoundByPEX")
  59         uploadChunksPosted          = expvar.NewInt("uploadChunksPosted")
  60         unexpectedCancels           = expvar.NewInt("unexpectedCancels")
  61         postedCancels               = expvar.NewInt("postedCancels")
  62         duplicateConnsAvoided       = expvar.NewInt("duplicateConnsAvoided")
  63         failedPieceHashes           = expvar.NewInt("failedPieceHashes")
  64         unsuccessfulDials           = expvar.NewInt("unsuccessfulDials")
  65         successfulDials             = expvar.NewInt("successfulDials")
  66         acceptedConns               = expvar.NewInt("acceptedConns")
  67         inboundConnsBlocked         = expvar.NewInt("inboundConnsBlocked")
  68 )
  69
  70 const (
  71         // Justification for set bits follows.
  72         //
  73         // Extension protocol: http://www.bittorrent.org/beps/bep_0010.html
  74         // DHT: http://www.bittorrent.org/beps/bep_0005.html
  75         extensionBytes = "\x00\x00\x00\x00\x00\x10\x00\x01"
  76
  77         socketsPerTorrent     = 40
  78         torrentPeersHighWater = 200
  79         torrentPeersLowWater  = 50
  80
  81         // Limit how long handshake can take. This is to reduce the lingering
  82         // impact of a few bad apples. 4s loses 1% of successful handshakes that
  83         // are obtained with 60s timeout, and 5% of unsuccessful handshakes.
  84         handshakeTimeout = 4 * time.Second
  85
  86         pruneInterval = 10 * time.Second
  87 )
  88
  89 // Currently doesn't really queue, but should in the future.
  90 func (cl *Client) queuePieceCheck(t *torrent, pieceIndex pp.Integer) {
  91         piece := t.Pieces[pieceIndex]
  92         if piece.QueuedForHash {
  93                 return
  94         }
  95         piece.QueuedForHash = true
  96         go cl.verifyPiece(t, pieceIndex)
  97 }
  98
  99 // Queue a piece check if one isn't already queued, and the piece has never
 100 // been checked before.
 101 func (cl *Client) queueFirstHash(t *torrent, piece int) {
 102         p := t.Pieces[piece]
 103         if p.EverHashed || p.Hashing || p.QueuedForHash || p.Complete() {
 104                 return
 105         }
 106         cl.queuePieceCheck(t, pp.Integer(piece))
 107 }
 108
 109 type Client struct {
 110         noUpload         bool
 111         dataDir          string
 112         halfOpenLimit    int
 113         peerID           [20]byte
 114         listeners        []net.Listener
 115         utpSock          *utp.Socket
 116         disableTrackers  bool
 117         downloadStrategy downloadStrategy
 118         dHT              *dht.Server
 119         disableUTP       bool
 120         disableTCP       bool
 121         ipBlockList      *iplist.IPList
 122         bannedTorrents   map[InfoHash]struct{}
 123         _configDir       string
 124         config           Config
 125         pruneTimer       *time.Timer
 126
 127         torrentDataOpener TorrentDataOpener
 128
 129         mu    sync.RWMutex
 130         event sync.Cond
 131         quit  chan struct{}
 132
 133         handshaking int
 134
 135         torrents map[InfoHash]*torrent
 136 }
 137
 138 func (me *Client) IPBlockList() *iplist.IPList {
 139         me.mu.Lock()
 140         defer me.mu.Unlock()
 141         return me.ipBlockList
 142 }
 143
 144 func (me *Client) SetIPBlockList(list *iplist.IPList) {
 145         me.mu.Lock()
 146         defer me.mu.Unlock()
 147         me.ipBlockList = list
 148         if me.dHT != nil {
 149                 me.dHT.SetIPBlockList(list)
 150         }
 151 }
 152
 153 func (me *Client) PeerID() string {
 154         return string(me.peerID[:])
 155 }
 156
 157 func (me *Client) ListenAddr() (addr net.Addr) {
 158         for _, l := range me.listeners {
 159                 if addr != nil && l.Addr().String() != addr.String() {
 160                         panic("listeners exist on different addresses")
 161                 }
 162                 addr = l.Addr()
 163         }
 164         return
 165 }
 166
 167 type hashSorter struct {
 168         Hashes []InfoHash
 169 }
 170
 171 func (me hashSorter) Len() int {
 172         return len(me.Hashes)
 173 }
 174
 175 func (me hashSorter) Less(a, b int) bool {
 176         return (&big.Int{}).SetBytes(me.Hashes[a][:]).Cmp((&big.Int{}).SetBytes(me.Hashes[b][:])) < 0
 177 }
 178
 179 func (me hashSorter) Swap(a, b int) {
 180         me.Hashes[a], me.Hashes[b] = me.Hashes[b], me.Hashes[a]
 181 }
 182
 183 func (cl *Client) sortedTorrents() (ret []*torrent) {
 184         var hs hashSorter
 185         for ih := range cl.torrents {
 186                 hs.Hashes = append(hs.Hashes, ih)
 187         }
 188         sort.Sort(hs)
 189         for _, ih := range hs.Hashes {
 190                 ret = append(ret, cl.torrent(ih))
 191         }
 192         return
 193 }
 194
 195 // Writes out a human readable status of the client, such as for writing to a
 196 // HTTP status page.
 197 func (cl *Client) WriteStatus(_w io.Writer) {
 198         cl.mu.RLock()
 199         defer cl.mu.RUnlock()
 200         w := bufio.NewWriter(_w)
 201         defer w.Flush()
 202         if addr := cl.ListenAddr(); addr != nil {
 203                 fmt.Fprintf(w, "Listening on %s\n", cl.ListenAddr())
 204         } else {
 205                 fmt.Fprintln(w, "Not listening!")
 206         }
 207         fmt.Fprintf(w, "Peer ID: %q\n", cl.peerID)
 208         fmt.Fprintf(w, "Handshaking: %d\n", cl.handshaking)
 209         if cl.dHT != nil {
 210                 dhtStats := cl.dHT.Stats()
 211                 fmt.Fprintf(w, "DHT nodes: %d (%d good)\n", dhtStats.NumNodes, dhtStats.NumGoodNodes)
 212                 fmt.Fprintf(w, "DHT Server ID: %x\n", cl.dHT.IDString())
 213                 fmt.Fprintf(w, "DHT port: %d\n", addrPort(cl.dHT.LocalAddr()))
 214                 fmt.Fprintf(w, "DHT announces: %d\n", cl.dHT.NumConfirmedAnnounces)
 215                 fmt.Fprintf(w, "Outstanding transactions: %d\n", dhtStats.NumOutstandingTransactions)
 216         }
 217         fmt.Fprintln(w)
 218         for _, t := range cl.sortedTorrents() {
 219                 if t.Name() == "" {
 220                         fmt.Fprint(w, "<unknown name>")
 221                 } else {
 222                         fmt.Fprint(w, t.Name())
 223                 }
 224                 fmt.Fprint(w, "\n")
 225                 if t.haveInfo() {
 226                         fmt.Fprintf(w, "%f%% of %d bytes", 100*(1-float32(t.bytesLeft())/float32(t.Length())), t.Length())
 227                 } else {
 228                         w.WriteString("<missing metainfo>")
 229                 }
 230                 fmt.Fprint(w, "\n")
 231                 t.WriteStatus(w)
 232                 fmt.Fprintln(w)
 233         }
 234 }
 235
 236 // Read torrent data at the given offset. Will block until it is available.
 237 func (cl *Client) torrentReadAt(t *torrent, off int64, p []byte) (n int, err error) {
 238         cl.mu.Lock()
 239         defer cl.mu.Unlock()
 240         index := int(off / int64(t.usualPieceSize()))
 241         // Reading outside the bounds of a file is an error.
 242         if index < 0 {
 243                 err = os.ErrInvalid
 244                 return
 245         }
 246         if int(index) >= len(t.Pieces) {
 247                 err = io.EOF
 248                 return
 249         }
 250         pieceOff := pp.Integer(off % int64(t.usualPieceSize()))
 251         pieceLeft := int(t.PieceLength(index) - pieceOff)
 252         if pieceLeft <= 0 {
 253                 err = io.EOF
 254                 return
 255         }
 256         if len(p) > pieceLeft {
 257                 p = p[:pieceLeft]
 258         }
 259         if len(p) == 0 {
 260                 panic(len(p))
 261         }
 262         cl.prepareRead(t, off)
 263         return dataReadAt(t.data, p, off)
 264 }
 265
 266 // Sets priorities to download from the given offset. Returns when the piece
 267 // at the given offset can be read. Returns the number of bytes that
 268 // immediately available from the offset.
 269 func (cl *Client) prepareRead(t *torrent, off int64) (n int64) {
 270         index := int(off / int64(t.usualPieceSize()))
 271         // Reading outside the bounds of a file is an error.
 272         if index < 0 || index >= t.numPieces() {
 273                 return
 274         }
 275         piece := t.Pieces[index]
 276         cl.readRaisePiecePriorities(t, off)
 277         for !piece.Complete() && !t.isClosed() {
 278                 piece.Event.Wait()
 279         }
 280         return t.Info.Piece(index).Length() - off%t.Info.PieceLength
 281 }
 282
 283 func (T Torrent) prepareRead(off int64) (avail int64) {
 284         T.cl.mu.Lock()
 285         defer T.cl.mu.Unlock()
 286         return T.cl.prepareRead(T.torrent, off)
 287 }
 288
 289 // Data implements a streaming interface that's more efficient than ReadAt.
 290 type SectionOpener interface {
 291         OpenSection(off, n int64) (io.ReadCloser, error)
 292 }
 293
 294 func dataReadAt(d data.Data, b []byte, off int64) (n int, err error) {
 295         if ra, ok := d.(io.ReaderAt); ok {
 296                 return ra.ReadAt(b, off)
 297         }
 298         if so, ok := d.(SectionOpener); ok {
 299                 var rc io.ReadCloser
 300                 rc, err = so.OpenSection(off, int64(len(b)))
 301                 if err != nil {
 302                         return
 303                 }
 304                 defer rc.Close()
 305                 return io.ReadFull(rc, b)
 306         }
 307         panic(fmt.Sprintf("can't read from %T", d))
 308 }
 309
 310 func readaheadPieces(readahead, pieceLength int64) int {
 311         return int((readahead+pieceLength-1)/pieceLength - 1)
 312 }
 313
 314 func (cl *Client) readRaisePiecePriorities(t *torrent, off int64) {
 315         index := int(off / int64(t.usualPieceSize()))
 316         cl.raisePiecePriority(t, index, piecePriorityNow)
 317         index++
 318         if index >= t.numPieces() {
 319                 return
 320         }
 321         cl.raisePiecePriority(t, index, piecePriorityNext)
 322         for i := 0; i < readaheadPieces(5*1024*1024, t.Info.PieceLength); i++ {
 323                 index++
 324                 if index >= t.numPieces() {
 325                         break
 326                 }
 327                 cl.raisePiecePriority(t, index, piecePriorityReadahead)
 328         }
 329 }
 330
 331 func (cl *Client) configDir() string {
 332         if cl._configDir == "" {
 333                 return filepath.Join(os.Getenv("HOME"), ".config/torrent")
 334         }
 335         return cl._configDir
 336 }
 337
 338 func (cl *Client) ConfigDir() string {
 339         return cl.configDir()
 340 }
 341
 342 func (t *torrent) connPendPiece(c *connection, piece int) {
 343         c.pendPiece(piece, t.Pieces[piece].Priority)
 344 }
 345
 346 func (cl *Client) raisePiecePriority(t *torrent, piece int, priority piecePriority) {
 347         if t.Pieces[piece].Priority < priority {
 348                 cl.prioritizePiece(t, piece, priority)
 349         }
 350 }
 351
 352 func (cl *Client) prioritizePiece(t *torrent, piece int, priority piecePriority) {
 353         if t.havePiece(piece) {
 354                 return
 355         }
 356         cl.queueFirstHash(t, piece)
 357         t.Pieces[piece].Priority = priority
 358         if t.wantPiece(piece) {
 359                 for _, c := range t.Conns {
 360                         if c.PeerHasPiece(pp.Integer(piece)) {
 361                                 t.connPendPiece(c, piece)
 362                                 cl.replenishConnRequests(t, c)
 363                         }
 364                 }
 365         }
 366 }
 367
 368 func (cl *Client) setEnvBlocklist() (err error) {
 369         filename := os.Getenv("TORRENT_BLOCKLIST_FILE")
 370         defaultBlocklist := filename == ""
 371         if defaultBlocklist {
 372                 filename = filepath.Join(cl.configDir(), "blocklist")
 373         }
 374         f, err := os.Open(filename)
 375         if err != nil {
 376                 if defaultBlocklist {
 377                         err = nil
 378                 }
 379                 return
 380         }
 381         defer f.Close()
 382         var ranges []iplist.Range
 383         uniqStrs := make(map[string]string)
 384         scanner := bufio.NewScanner(f)
 385         lineNum := 1
 386         for scanner.Scan() {
 387                 r, ok, lineErr := iplist.ParseBlocklistP2PLine(scanner.Bytes())
 388                 if lineErr != nil {
 389                         err = fmt.Errorf("error reading torrent blocklist line %d: %s", lineNum, lineErr)
 390                         return
 391                 }
 392                 lineNum++
 393                 if !ok {
 394                         continue
 395                 }
 396                 if s, ok := uniqStrs[r.Description]; ok {
 397                         r.Description = s
 398                 } else {
 399                         uniqStrs[r.Description] = r.Description
 400                 }
 401                 ranges = append(ranges, r)
 402         }
 403         err = scanner.Err()
 404         if err != nil {
 405                 err = fmt.Errorf("error reading torrent blocklist: %s", err)
 406                 return
 407         }
 408         cl.ipBlockList = iplist.New(ranges)
 409         return
 410 }
 411
 412 func (cl *Client) initBannedTorrents() error {
 413         f, err := os.Open(filepath.Join(cl.configDir(), "banned_infohashes"))
 414         if err != nil {
 415                 if os.IsNotExist(err) {
 416                         return nil
 417                 }
 418                 return fmt.Errorf("error opening banned infohashes file: %s", err)
 419         }
 420         defer f.Close()
 421         scanner := bufio.NewScanner(f)
 422         cl.bannedTorrents = make(map[InfoHash]struct{})
 423         for scanner.Scan() {
 424                 var ihs string
 425                 n, err := fmt.Sscanf(scanner.Text(), "%x", &ihs)
 426                 if err != nil {
 427                         return fmt.Errorf("error reading infohash: %s", err)
 428                 }
 429                 if n != 1 {
 430                         continue
 431                 }
 432                 if len(ihs) != 20 {
 433                         return errors.New("bad infohash")
 434                 }
 435                 var ih InfoHash
 436                 CopyExact(&ih, ihs)
 437                 cl.bannedTorrents[ih] = struct{}{}
 438         }
 439         if err := scanner.Err(); err != nil {
 440                 return fmt.Errorf("error scanning file: %s", err)
 441         }
 442         return nil
 443 }
 444
 445 func NewClient(cfg *Config) (cl *Client, err error) {
 446         if cfg == nil {
 447                 cfg = &Config{}
 448         }
 449
 450         cl = &Client{
 451                 noUpload:        cfg.NoUpload,
 452                 disableTrackers: cfg.DisableTrackers,
 453                 halfOpenLimit:   socketsPerTorrent,
 454                 dataDir:         cfg.DataDir,
 455                 disableUTP:      cfg.DisableUTP,
 456                 disableTCP:      cfg.DisableTCP,
 457                 _configDir:      cfg.ConfigDir,
 458                 config:          *cfg,
 459                 torrentDataOpener: func(md *metainfo.Info) data.Data {
 460                         return filePkg.TorrentData(md, cfg.DataDir)
 461                 },
 462
 463                 quit:     make(chan struct{}),
 464                 torrents: make(map[InfoHash]*torrent),
 465         }
 466         cl.event.L = &cl.mu
 467         if cfg.TorrentDataOpener != nil {
 468                 cl.torrentDataOpener = cfg.TorrentDataOpener
 469         }
 470
 471         if !cfg.NoDefaultBlocklist {
 472                 err = cl.setEnvBlocklist()
 473                 if err != nil {
 474                         return
 475                 }
 476         }
 477
 478         if err = cl.initBannedTorrents(); err != nil {
 479                 err = fmt.Errorf("error initing banned torrents: %s", err)
 480                 return
 481         }
 482
 483         if cfg.PeerID != "" {
 484                 CopyExact(&cl.peerID, cfg.PeerID)
 485         } else {
 486                 o := copy(cl.peerID[:], bep20)
 487                 _, err = rand.Read(cl.peerID[o:])
 488                 if err != nil {
 489                         panic("error generating peer id")
 490                 }
 491         }
 492
 493         // Returns the laddr string to listen on for the next Listen call.
 494         listenAddr := func() string {
 495                 if addr := cl.ListenAddr(); addr != nil {
 496                         return addr.String()
 497                 }
 498                 if cfg.ListenAddr == "" {
 499                         // IPv6 isn't well supported with blocklists, or with trackers and
 500                         // DHT.
 501                         return "0.0.0.0:50007"
 502                 }
 503                 return cfg.ListenAddr
 504         }
 505         if !cl.disableTCP {
 506                 var l net.Listener
 507                 l, err = net.Listen("tcp", listenAddr())
 508                 if err != nil {
 509                         return
 510                 }
 511                 cl.listeners = append(cl.listeners, l)
 512                 go cl.acceptConnections(l, false)
 513         }
 514         if !cl.disableUTP {
 515                 cl.utpSock, err = utp.NewSocket(listenAddr())
 516                 if err != nil {
 517                         return
 518                 }
 519                 cl.listeners = append(cl.listeners, cl.utpSock)
 520                 go cl.acceptConnections(cl.utpSock, true)
 521         }
 522         if !cfg.NoDHT {
 523                 dhtCfg := cfg.DHTConfig
 524                 if dhtCfg == nil {
 525                         dhtCfg = &dht.ServerConfig{}
 526                 }
 527                 if dhtCfg.Addr == "" {
 528                         dhtCfg.Addr = listenAddr()
 529                 }
 530                 if dhtCfg.Conn == nil && cl.utpSock != nil {
 531                         dhtCfg.Conn = cl.utpSock.PacketConn()
 532                 }
 533                 cl.dHT, err = dht.NewServer(dhtCfg)
 534                 if cl.ipBlockList != nil {
 535                         cl.dHT.SetIPBlockList(cl.ipBlockList)
 536                 }
 537                 if err != nil {
 538                         return
 539                 }
 540         }
 541
 542         return
 543 }
 544
 545 func (cl *Client) stopped() bool {
 546         select {
 547         case <-cl.quit:
 548                 return true
 549         default:
 550                 return false
 551         }
 552 }
 553
 554 // Stops the client. All connections to peers are closed and all activity will
 555 // come to a halt.
 556 func (me *Client) Close() {
 557         me.mu.Lock()
 558         defer me.mu.Unlock()
 559         close(me.quit)
 560         for _, l := range me.listeners {
 561                 l.Close()
 562         }
 563         me.event.Broadcast()
 564         for _, t := range me.torrents {
 565                 t.close()
 566         }
 567 }
 568
 569 var ipv6BlockRange = iplist.Range{Description: "non-IPv4 address"}
 570
 571 func (cl *Client) ipBlockRange(ip net.IP) (r *iplist.Range) {
 572         if cl.ipBlockList == nil {
 573                 return
 574         }
 575         ip = ip.To4()
 576         if ip == nil {
 577                 log.Printf("saw non-IPv4 address")
 578                 r = &ipv6BlockRange
 579                 return
 580         }
 581         r = cl.ipBlockList.Lookup(ip)
 582         return
 583 }
 584
 585 func (cl *Client) acceptConnections(l net.Listener, utp bool) {
 586         for {
 587                 // We accept all connections immediately, because we don't know what
 588                 // torrent they're for.
 589                 conn, err := l.Accept()
 590                 select {
 591                 case <-cl.quit:
 592                         if conn != nil {
 593                                 conn.Close()
 594                         }
 595                         return
 596                 default:
 597                 }
 598                 if err != nil {
 599                         log.Print(err)
 600                         return
 601                 }
 602                 acceptedConns.Add(1)
 603                 cl.mu.RLock()
 604                 blockRange := cl.ipBlockRange(AddrIP(conn.RemoteAddr()))
 605                 cl.mu.RUnlock()
 606                 if blockRange != nil {
 607                         inboundConnsBlocked.Add(1)
 608                         log.Printf("inbound connection from %s blocked by %s", conn.RemoteAddr(), blockRange)
 609                         conn.Close()
 610                         continue
 611                 }
 612                 go func() {
 613                         if err := cl.runConnection(conn, nil, peerSourceIncoming, utp); err != nil {
 614                                 log.Print(err)
 615                         }
 616                 }()
 617         }
 618 }
 619
 620 func (me *Client) torrent(ih InfoHash) *torrent {
 621         return me.torrents[ih]
 622 }
 623
 624 type dialResult struct {
 625         net.Conn
 626         UTP bool
 627 }
 628
 629 func doDial(dial func(addr string) (net.Conn, error), ch chan dialResult, utp bool, addr string) {
 630         conn, err := dial(addr)
 631         if err != nil {
 632                 if conn != nil {
 633                         conn.Close()
 634                 }
 635                 conn = nil // Pedantic
 636         }
 637         ch <- dialResult{conn, utp}
 638         if err == nil {
 639                 successfulDials.Add(1)
 640                 return
 641         }
 642         unsuccessfulDials.Add(1)
 643         if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
 644                 return
 645         }
 646         if netOpErr, ok := err.(*net.OpError); ok {
 647                 switch netOpErr.Err {
 648                 case syscall.ECONNREFUSED, syscall.EHOSTUNREACH:
 649                         return
 650                 }
 651         }
 652         if err != nil {
 653                 log.Printf("error dialing %s: %s", addr, err)
 654                 return
 655         }
 656 }
 657
 658 func reducedDialTimeout(max time.Duration, halfOpenLimit int, pendingPeers int) (ret time.Duration) {
 659         ret = max / time.Duration((pendingPeers+halfOpenLimit)/halfOpenLimit)
 660         if ret < minDialTimeout {
 661                 ret = minDialTimeout
 662         }
 663         return
 664 }
 665
 666 // Start the process of connecting to the given peer for the given torrent if
 667 // appropriate.
 668 func (me *Client) initiateConn(peer Peer, t *torrent) {
 669         if peer.Id == me.peerID {
 670                 return
 671         }
 672         addr := net.JoinHostPort(peer.IP.String(), fmt.Sprintf("%d", peer.Port))
 673         if t.addrActive(addr) {
 674                 duplicateConnsAvoided.Add(1)
 675                 return
 676         }
 677         if r := me.ipBlockRange(peer.IP); r != nil {
 678                 log.Printf("outbound connect to %s blocked by IP blocklist rule %s", peer.IP, r)
 679                 return
 680         }
 681         dialTimeout := reducedDialTimeout(nominalDialTimeout, me.halfOpenLimit, len(t.Peers))
 682         t.HalfOpen[addr] = struct{}{}
 683         go func() {
 684                 // Binding to the listen address and dialing via net.Dialer gives
 685                 // "address in use" error. It seems it's not possible to dial out from
 686                 // this address so that peers associate our local address with our
 687                 // listen address.
 688
 689                 // Initiate connections via TCP and UTP simultaneously. Use the first
 690                 // one that succeeds.
 691                 left := 0
 692                 if !me.disableUTP {
 693                         left++
 694                 }
 695                 if !me.disableTCP {
 696                         left++
 697                 }
 698                 resCh := make(chan dialResult, left)
 699                 if !me.disableUTP {
 700                         go doDial(func(addr string) (net.Conn, error) {
 701                                 return me.utpSock.DialTimeout(addr, dialTimeout)
 702                         }, resCh, true, addr)
 703                 }
 704                 if !me.disableTCP {
 705                         go doDial(func(addr string) (net.Conn, error) {
 706                                 // time.Sleep(time.Second) // Give uTP a bit of a head start.
 707                                 return net.DialTimeout("tcp", addr, dialTimeout)
 708                         }, resCh, false, addr)
 709                 }
 710                 var res dialResult
 711                 for ; left > 0 && res.Conn == nil; left-- {
 712                         res = <-resCh
 713                 }
 714                 // Whether or not the connection attempt succeeds, the half open
 715                 // counter should be decremented, and new connection attempts made.
 716                 go func() {
 717                         me.mu.Lock()
 718                         defer me.mu.Unlock()
 719                         if _, ok := t.HalfOpen[addr]; !ok {
 720                                 panic("invariant broken")
 721                         }
 722                         delete(t.HalfOpen, addr)
 723                         me.openNewConns(t)
 724                 }()
 725                 if res.Conn == nil {
 726                         return
 727                 }
 728                 if left > 0 {
 729                         go func() {
 730                                 for ; left > 0; left-- {
 731                                         conn := (<-resCh).Conn
 732                                         if conn != nil {
 733                                                 conn.Close()
 734                                         }
 735                                 }
 736                         }()
 737                 }
 738
 739                 // log.Printf("connected to %s", conn.RemoteAddr())
 740                 err := me.runConnection(res.Conn, t, peer.Source, res.UTP)
 741                 if err != nil {
 742                         log.Print(err)
 743                 }
 744         }()
 745 }
 746
 747 // The port number for incoming peer connections. 0 if the client isn't
 748 // listening.
 749 func (cl *Client) incomingPeerPort() int {
 750         listenAddr := cl.ListenAddr()
 751         if listenAddr == nil {
 752                 return 0
 753         }
 754         return addrPort(listenAddr)
 755 }
 756
 757 // Convert a net.Addr to its compact IP representation. Either 4 or 16 bytes
 758 // per "yourip" field of http://www.bittorrent.org/beps/bep_0010.html.
 759 func addrCompactIP(addr net.Addr) (string, error) {
 760         host, _, err := net.SplitHostPort(addr.String())
 761         if err != nil {
 762                 return "", err
 763         }
 764         ip := net.ParseIP(host)
 765         if v4 := ip.To4(); v4 != nil {
 766                 if len(v4) != 4 {
 767                         panic(v4)
 768                 }
 769                 return string(v4), nil
 770         }
 771         return string(ip.To16()), nil
 772 }
 773
 774 func handshakeWriter(w io.WriteCloser, bb <-chan []byte, done chan<- error) {
 775         var err error
 776         for b := range bb {
 777                 _, err = w.Write(b)
 778                 if err != nil {
 779                         w.Close()
 780                         break
 781                 }
 782         }
 783         done <- err
 784 }
 785
 786 type peerExtensionBytes [8]byte
 787 type peerID [20]byte
 788
 789 type handshakeResult struct {
 790         peerExtensionBytes
 791         peerID
 792         InfoHash
 793 }
 794
 795 // ih is nil if we expect the peer to declare the InfoHash, such as when the
 796 // peer initiated the connection. Returns ok if the handshake was successful,
 797 // and err if there was an unexpected condition other than the peer simply
 798 // abandoning the handshake.
 799 func handshake(sock io.ReadWriteCloser, ih *InfoHash, peerID [20]byte) (res handshakeResult, ok bool, err error) {
 800         // Bytes to be sent to the peer. Should never block the sender.
 801         postCh := make(chan []byte, 4)
 802         // A single error value sent when the writer completes.
 803         writeDone := make(chan error, 1)
 804         // Performs writes to the socket and ensures posts don't block.
 805         go handshakeWriter(sock, postCh, writeDone)
 806
 807         defer func() {
 808                 close(postCh) // Done writing.
 809                 if !ok {
 810                         return
 811                 }
 812                 if err != nil {
 813                         panic(err)
 814                 }
 815                 // Wait until writes complete before returning from handshake.
 816                 err = <-writeDone
 817                 if err != nil {
 818                         err = fmt.Errorf("error writing during handshake: %s", err)
 819                 }
 820         }()
 821
 822         post := func(bb []byte) {
 823                 select {
 824                 case postCh <- bb:
 825                 default:
 826                         panic("mustn't block while posting")
 827                 }
 828         }
 829
 830         post([]byte(pp.Protocol))
 831         post([]byte(extensionBytes))
 832         if ih != nil { // We already know what we want.
 833                 post(ih[:])
 834                 post(peerID[:])
 835         }
 836         var b [68]byte
 837         _, err = io.ReadFull(sock, b[:68])
 838         if err != nil {
 839                 err = nil
 840                 return
 841         }
 842         if string(b[:20]) != pp.Protocol {
 843                 return
 844         }
 845         CopyExact(&res.peerExtensionBytes, b[20:28])
 846         CopyExact(&res.InfoHash, b[28:48])
 847         CopyExact(&res.peerID, b[48:68])
 848
 849         if ih == nil { // We were waiting for the peer to tell us what they wanted.
 850                 post(res.InfoHash[:])
 851                 post(peerID[:])
 852         }
 853
 854         ok = true
 855         return
 856 }
 857
 858 // Wraps a raw connection and provides the interface we want for using the
 859 // connection in the message loop.
 860 type peerConn struct {
 861         net.Conn
 862 }
 863
 864 func (pc peerConn) Read(b []byte) (n int, err error) {
 865         // Keep-alives should be received every 2 mins. Give a bit of gracetime.
 866         err = pc.Conn.SetReadDeadline(time.Now().Add(150 * time.Second))
 867         if err != nil {
 868                 err = fmt.Errorf("error setting read deadline: %s", err)
 869         }
 870         n, err = pc.Conn.Read(b)
 871         // Convert common errors into io.EOF.
 872         if err != nil {
 873                 if opError, ok := err.(*net.OpError); ok && opError.Op == "read" && opError.Err == syscall.ECONNRESET {
 874                         err = io.EOF
 875                 } else if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
 876                         if n != 0 {
 877                                 panic(n)
 878                         }
 879                         err = io.EOF
 880                 }
 881         }
 882         return
 883 }
 884
 885 func (me *Client) runConnection(sock net.Conn, torrent *torrent, discovery peerSource, uTP bool) (err error) {
 886         if tcpConn, ok := sock.(*net.TCPConn); ok {
 887                 tcpConn.SetLinger(0)
 888         }
 889         defer sock.Close()
 890         err = sock.SetDeadline(time.Now().Add(handshakeTimeout))
 891         if err != nil {
 892                 err = fmt.Errorf("couldn't set handshake deadline: %s", err)
 893                 return
 894         }
 895         me.mu.Lock()
 896         me.handshaking++
 897         me.mu.Unlock()
 898         hsRes, ok, err := handshake(sock, func() *InfoHash {
 899                 if torrent == nil {
 900                         return nil
 901                 } else {
 902                         return &torrent.InfoHash
 903                 }
 904         }(), me.peerID)
 905         me.mu.Lock()
 906         defer me.mu.Unlock()
 907         if me.handshaking == 0 {
 908                 panic("handshake count invariant is broken")
 909         }
 910         me.handshaking--
 911         if err != nil {
 912                 err = fmt.Errorf("error during handshake: %s", err)
 913                 return
 914         }
 915         if !ok {
 916                 return
 917         }
 918         if hsRes.peerID == me.peerID {
 919                 return
 920         }
 921         torrent = me.torrent(hsRes.InfoHash)
 922         if torrent == nil {
 923                 return
 924         }
 925         sock.SetWriteDeadline(time.Time{})
 926         sock = peerConn{sock}
 927         conn := newConnection(sock, hsRes.peerExtensionBytes, hsRes.peerID, uTP)
 928         defer conn.Close()
 929         conn.Discovery = discovery
 930         if !me.addConnection(torrent, conn) {
 931                 return
 932         }
 933         if conn.PeerExtensionBytes[5]&0x10 != 0 {
 934                 conn.Post(pp.Message{
 935                         Type:       pp.Extended,
 936                         ExtendedID: pp.HandshakeExtendedID,
 937                         ExtendedPayload: func() []byte {
 938                                 d := map[string]interface{}{
 939                                         "m": map[string]int{
 940                                                 "ut_metadata": 1,
 941                                                 "ut_pex":      2,
 942                                         },
 943                                         "v": "go.torrent dev 20140825", // Just the date
 944                                         // No upload queue is implemented yet.
 945                                         "reqq": func() int {
 946                                                 if me.noUpload {
 947                                                         // No need to look strange if it costs us nothing.
 948                                                         return 250
 949                                                 } else {
 950                                                         return 1
 951                                                 }
 952                                         }(),
 953                                 }
 954                                 if torrent.metadataSizeKnown() {
 955                                         d["metadata_size"] = torrent.metadataSize()
 956                                 }
 957                                 if p := me.incomingPeerPort(); p != 0 {
 958                                         d["p"] = p
 959                                 }
 960                                 yourip, err := addrCompactIP(conn.Socket.RemoteAddr())
 961                                 if err != nil {
 962                                         log.Printf("error calculating yourip field value in extension handshake: %s", err)
 963                                 } else {
 964                                         d["yourip"] = yourip
 965                                 }
 966                                 // log.Printf("sending %v", d)
 967                                 b, err := bencode.Marshal(d)
 968                                 if err != nil {
 969                                         panic(err)
 970                                 }
 971                                 return b
 972                         }(),
 973                 })
 974         }
 975         if torrent.haveAnyPieces() {
 976                 conn.Post(pp.Message{
 977                         Type:     pp.Bitfield,
 978                         Bitfield: torrent.bitfield(),
 979                 })
 980         }
 981         if conn.PeerExtensionBytes[7]&0x01 != 0 && me.dHT != nil {
 982                 conn.Post(pp.Message{
 983                         Type: pp.Port,
 984                         Port: uint16(AddrPort(me.dHT.LocalAddr())),
 985                 })
 986         }
 987         if torrent.haveInfo() {
 988                 torrent.initRequestOrdering(conn)
 989                 me.replenishConnRequests(torrent, conn)
 990         }
 991         err = me.connectionLoop(torrent, conn)
 992         if err != nil {
 993                 err = fmt.Errorf("during Connection loop with peer %q: %s", conn.PeerID, err)
 994         }
 995         me.dropConnection(torrent, conn)
 996         return
 997 }
 998
 999 func (t *torrent) initRequestOrdering(c *connection) {
1000         if c.pieceRequestOrder != nil || c.piecePriorities != nil {
1001                 panic("double init of request ordering")
1002         }
1003         c.piecePriorities = mathRand.Perm(t.numPieces())
1004         c.pieceRequestOrder = pieceordering.New()
1005         for i := 0; i < t.numPieces(); i++ {
1006                 if !c.PeerHasPiece(pp.Integer(i)) {
1007                         continue
1008                 }
1009                 if !t.wantPiece(i) {
1010                         continue
1011                 }
1012                 t.connPendPiece(c, i)
1013         }
1014 }
1015
1016 func (me *Client) peerGotPiece(t *torrent, c *connection, piece int) {
1017         if t.haveInfo() {
1018                 if c.PeerPieces == nil {
1019                         c.PeerPieces = make([]bool, t.numPieces())
1020                 }
1021         } else {
1022                 for piece >= len(c.PeerPieces) {
1023                         c.PeerPieces = append(c.PeerPieces, false)
1024                 }
1025         }
1026         c.PeerPieces[piece] = true
1027         if t.wantPiece(piece) {
1028                 t.connPendPiece(c, piece)
1029                 me.replenishConnRequests(t, c)
1030         }
1031 }
1032
1033 func (me *Client) peerUnchoked(torrent *torrent, conn *connection) {
1034         me.replenishConnRequests(torrent, conn)
1035 }
1036
1037 func (cl *Client) connCancel(t *torrent, cn *connection, r request) (ok bool) {
1038         ok = cn.Cancel(r)
1039         if ok {
1040                 postedCancels.Add(1)
1041         }
1042         return
1043 }
1044
1045 func (cl *Client) connDeleteRequest(t *torrent, cn *connection, r request) {
1046         if !cn.RequestPending(r) {
1047                 return
1048         }
1049         delete(cn.Requests, r)
1050 }
1051
1052 func (cl *Client) requestPendingMetadata(t *torrent, c *connection) {
1053         if t.haveInfo() {
1054                 return
1055         }
1056         var pending []int
1057         for index := 0; index < t.metadataPieceCount(); index++ {
1058                 if !t.haveMetadataPiece(index) {
1059                         pending = append(pending, index)
1060                 }
1061         }
1062         for _, i := range mathRand.Perm(len(pending)) {
1063                 c.Post(pp.Message{
1064                         Type:       pp.Extended,
1065                         ExtendedID: byte(c.PeerExtensionIDs["ut_metadata"]),
1066                         ExtendedPayload: func() []byte {
1067                                 b, err := bencode.Marshal(map[string]int{
1068                                         "msg_type": 0,
1069                                         "piece":    pending[i],
1070                                 })
1071                                 if err != nil {
1072                                         panic(err)
1073                                 }
1074                                 return b
1075                         }(),
1076                 })
1077         }
1078 }
1079
1080 func (cl *Client) completedMetadata(t *torrent) {
1081         h := sha1.New()
1082         h.Write(t.MetaData)
1083         var ih InfoHash
1084         CopyExact(&ih, h.Sum(nil))
1085         if ih != t.InfoHash {
1086                 log.Print("bad metadata")
1087                 t.invalidateMetadata()
1088                 return
1089         }
1090         var info metainfo.Info
1091         err := bencode.Unmarshal(t.MetaData, &info)
1092         if err != nil {
1093                 log.Printf("error unmarshalling metadata: %s", err)
1094                 t.invalidateMetadata()
1095                 return
1096         }
1097         // TODO(anacrolix): If this fails, I think something harsher should be
1098         // done.
1099         err = cl.setMetaData(t, info, t.MetaData)
1100         if err != nil {
1101                 log.Printf("error setting metadata: %s", err)
1102                 t.invalidateMetadata()
1103                 return
1104         }
1105         log.Printf("%s: got metadata from peers", t)
1106 }
1107
1108 // Process incoming ut_metadata message.
1109 func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *torrent, c *connection) (err error) {
1110         var d map[string]int
1111         err = bencode.Unmarshal(payload, &d)
1112         if err != nil {
1113                 err = fmt.Errorf("error unmarshalling payload: %s: %q", err, payload)
1114                 return
1115         }
1116         msgType, ok := d["msg_type"]
1117         if !ok {
1118                 err = errors.New("missing msg_type field")
1119                 return
1120         }
1121         piece := d["piece"]
1122         switch msgType {
1123         case pp.DataMetadataExtensionMsgType:
1124                 if t.haveInfo() {
1125                         break
1126                 }
1127                 begin := len(payload) - metadataPieceSize(d["total_size"], piece)
1128                 if begin < 0 || begin >= len(payload) {
1129                         log.Printf("got bad metadata piece")
1130                         break
1131                 }
1132                 t.SaveMetadataPiece(piece, payload[begin:])
1133                 c.UsefulChunksReceived++
1134                 c.lastUsefulChunkReceived = time.Now()
1135                 if !t.haveAllMetadataPieces() {
1136                         break
1137                 }
1138                 cl.completedMetadata(t)
1139         case pp.RequestMetadataExtensionMsgType:
1140                 if !t.haveMetadataPiece(piece) {
1141                         c.Post(t.newMetadataExtensionMessage(c, pp.RejectMetadataExtensionMsgType, d["piece"], nil))
1142                         break
1143                 }
1144                 start := (1 << 14) * piece
1145                 c.Post(t.newMetadataExtensionMessage(c, pp.DataMetadataExtensionMsgType, piece, t.MetaData[start:start+t.metadataPieceSize(piece)]))
1146         case pp.RejectMetadataExtensionMsgType:
1147         default:
1148                 err = errors.New("unknown msg_type value")
1149         }
1150         return
1151 }
1152
1153 type peerExchangeMessage struct {
1154         Added      CompactPeers   `bencode:"added"`
1155         AddedFlags []byte         `bencode:"added.f"`
1156         Dropped    []tracker.Peer `bencode:"dropped"`
1157 }
1158
1159 // Extracts the port as an integer from an address string.
1160 func addrPort(addr net.Addr) int {
1161         return AddrPort(addr)
1162 }
1163
1164 // Processes incoming bittorrent messages. The client lock is held upon entry
1165 // and exit.
1166 func (me *Client) connectionLoop(t *torrent, c *connection) error {
1167         decoder := pp.Decoder{
1168                 R:         bufio.NewReader(c.Socket),
1169                 MaxLength: 256 * 1024,
1170         }
1171         for {
1172                 me.mu.Unlock()
1173                 var msg pp.Message
1174                 err := decoder.Decode(&msg)
1175                 me.mu.Lock()
1176                 c.lastMessageReceived = time.Now()
1177                 select {
1178                 case <-c.closing:
1179                         return nil
1180                 default:
1181                 }
1182                 if err != nil {
1183                         if me.stopped() || err == io.EOF {
1184                                 return nil
1185                         }
1186                         return err
1187                 }
1188                 if msg.Keepalive {
1189                         continue
1190                 }
1191                 switch msg.Type {
1192                 case pp.Choke:
1193                         c.PeerChoked = true
1194                         for r := range c.Requests {
1195                                 me.connDeleteRequest(t, c, r)
1196                         }
1197                         // We can then reset our interest.
1198                         me.replenishConnRequests(t, c)
1199                 case pp.Unchoke:
1200                         c.PeerChoked = false
1201                         me.peerUnchoked(t, c)
1202                 case pp.Interested:
1203                         c.PeerInterested = true
1204                         // TODO: This should be done from a dedicated unchoking routine.
1205                         if me.noUpload {
1206                                 break
1207                         }
1208                         c.Unchoke()
1209                 case pp.NotInterested:
1210                         c.PeerInterested = false
1211                         c.Choke()
1212                 case pp.Have:
1213                         me.peerGotPiece(t, c, int(msg.Index))
1214                 case pp.Request:
1215                         if me.noUpload {
1216                                 break
1217                         }
1218                         if c.PeerRequests == nil {
1219                                 c.PeerRequests = make(map[request]struct{}, maxRequests)
1220                         }
1221                         request := newRequest(msg.Index, msg.Begin, msg.Length)
1222                         // TODO: Requests should be satisfied from a dedicated upload
1223                         // routine.
1224                         // c.PeerRequests[request] = struct{}{}
1225                         p := make([]byte, msg.Length)
1226                         n, err := dataReadAt(t.data, p, int64(t.PieceLength(0))*int64(msg.Index)+int64(msg.Begin))
1227                         if err != nil {
1228                                 return fmt.Errorf("reading t data to serve request %q: %s", request, err)
1229                         }
1230                         if n != int(msg.Length) {
1231                                 return fmt.Errorf("bad request: %v", msg)
1232                         }
1233                         c.Post(pp.Message{
1234                                 Type:  pp.Piece,
1235                                 Index: msg.Index,
1236                                 Begin: msg.Begin,
1237                                 Piece: p,
1238                         })
1239                         uploadChunksPosted.Add(1)
1240                 case pp.Cancel:
1241                         req := newRequest(msg.Index, msg.Begin, msg.Length)
1242                         if !c.PeerCancel(req) {
1243                                 unexpectedCancels.Add(1)
1244                         }
1245                 case pp.Bitfield:
1246                         if c.PeerPieces != nil {
1247                                 err = errors.New("received unexpected bitfield")
1248                                 break
1249                         }
1250                         if t.haveInfo() {
1251                                 if len(msg.Bitfield) < t.numPieces() {
1252                                         err = errors.New("received invalid bitfield")
1253                                         break
1254                                 }
1255                                 msg.Bitfield = msg.Bitfield[:t.numPieces()]
1256                         }
1257                         c.PeerPieces = msg.Bitfield
1258                         for index, has := range c.PeerPieces {
1259                                 if has {
1260                                         me.peerGotPiece(t, c, index)
1261                                 }
1262                         }
1263                 case pp.Piece:
1264                         err = me.downloadedChunk(t, c, &msg)
1265                 case pp.Extended:
1266                         switch msg.ExtendedID {
1267                         case pp.HandshakeExtendedID:
1268                                 // TODO: Create a bencode struct for this.
1269                                 var d map[string]interface{}
1270                                 err = bencode.Unmarshal(msg.ExtendedPayload, &d)
1271                                 if err != nil {
1272                                         err = fmt.Errorf("error decoding extended message payload: %s", err)
1273                                         break
1274                                 }
1275                                 // log.Printf("got handshake from %q: %#v", c.Socket.RemoteAddr().String(), d)
1276                                 if reqq, ok := d["reqq"]; ok {
1277                                         if i, ok := reqq.(int64); ok {
1278                                                 c.PeerMaxRequests = int(i)
1279                                         }
1280                                 }
1281                                 if v, ok := d["v"]; ok {
1282                                         c.PeerClientName = v.(string)
1283                                 }
1284                                 m, ok := d["m"]
1285                                 if !ok {
1286                                         err = errors.New("handshake missing m item")
1287                                         break
1288                                 }
1289                                 mTyped, ok := m.(map[string]interface{})
1290                                 if !ok {
1291                                         err = errors.New("handshake m value is not dict")
1292                                         break
1293                                 }
1294                                 if c.PeerExtensionIDs == nil {
1295                                         c.PeerExtensionIDs = make(map[string]int64, len(mTyped))
1296                                 }
1297                                 for name, v := range mTyped {
1298                                         id, ok := v.(int64)
1299                                         if !ok {
1300                                                 log.Printf("bad handshake m item extension ID type: %T", v)
1301                                                 continue
1302                                         }
1303                                         if id == 0 {
1304                                                 delete(c.PeerExtensionIDs, name)
1305                                         } else {
1306                                                 c.PeerExtensionIDs[name] = id
1307                                         }
1308                                 }
1309                                 metadata_sizeUntyped, ok := d["metadata_size"]
1310                                 if ok {
1311                                         metadata_size, ok := metadata_sizeUntyped.(int64)
1312                                         if !ok {
1313                                                 log.Printf("bad metadata_size type: %T", metadata_sizeUntyped)
1314                                         } else {
1315                                                 t.SetMetadataSize(metadata_size)
1316                                         }
1317                                 }
1318                                 if _, ok := c.PeerExtensionIDs["ut_metadata"]; ok {
1319                                         me.requestPendingMetadata(t, c)
1320                                 }
1321                         case 1:
1322                                 err = me.gotMetadataExtensionMsg(msg.ExtendedPayload, t, c)
1323                                 if err != nil {
1324                                         err = fmt.Errorf("error handling metadata extension message: %s", err)
1325                                 }
1326                         case 2:
1327                                 var pexMsg peerExchangeMessage
1328                                 err := bencode.Unmarshal(msg.ExtendedPayload, &pexMsg)
1329                                 if err != nil {
1330                                         err = fmt.Errorf("error unmarshalling PEX message: %s", err)
1331                                         break
1332                                 }
1333                                 go func() {
1334                                         me.mu.Lock()
1335                                         me.addPeers(t, func() (ret []Peer) {
1336                                                 for _, cp := range pexMsg.Added {
1337                                                         p := Peer{
1338                                                                 IP:     make([]byte, 4),
1339                                                                 Port:   int(cp.Port),
1340                                                                 Source: peerSourcePEX,
1341                                                         }
1342                                                         if n := copy(p.IP, cp.IP[:]); n != 4 {
1343                                                                 panic(n)
1344                                                         }
1345                                                         ret = append(ret, p)
1346                                                 }
1347                                                 return
1348                                         }())
1349                                         me.mu.Unlock()
1350                                         peersFoundByPEX.Add(int64(len(pexMsg.Added)))
1351                                 }()
1352                         default:
1353                                 err = fmt.Errorf("unexpected extended message ID: %v", msg.ExtendedID)
1354                         }
1355                         if err != nil {
1356                                 // That client uses its own extension IDs for outgoing message
1357                                 // types, which is incorrect.
1358                                 if bytes.HasPrefix(c.PeerID[:], []byte("-SD0100-")) ||
1359                                         strings.HasPrefix(string(c.PeerID[:]), "-XL0012-") {
1360                                         return nil
1361                                 }
1362                                 // log.Printf("peer extension map: %#v", c.PeerExtensionIDs)
1363                         }
1364                 case pp.Port:
1365                         if me.dHT == nil {
1366                                 break
1367                         }
1368                         pingAddr, err := net.ResolveUDPAddr("", c.Socket.RemoteAddr().String())
1369                         if err != nil {
1370                                 panic(err)
1371                         }
1372                         if msg.Port != 0 {
1373                                 pingAddr.Port = int(msg.Port)
1374                         }
1375                         _, err = me.dHT.Ping(pingAddr)
1376                 default:
1377                         err = fmt.Errorf("received unknown message type: %#v", msg.Type)
1378                 }
1379                 if err != nil {
1380                         return err
1381                 }
1382         }
1383 }
1384
1385 func (me *Client) dropConnection(torrent *torrent, conn *connection) {
1386         for r := range conn.Requests {
1387                 me.connDeleteRequest(torrent, conn, r)
1388         }
1389         conn.Close()
1390         for i0, c := range torrent.Conns {
1391                 if c != conn {
1392                         continue
1393                 }
1394                 i1 := len(torrent.Conns) - 1
1395                 if i0 != i1 {
1396                         torrent.Conns[i0] = torrent.Conns[i1]
1397                 }
1398                 torrent.Conns = torrent.Conns[:i1]
1399                 me.openNewConns(torrent)
1400                 return
1401         }
1402         panic("connection not found")
1403 }
1404
1405 func (me *Client) addConnection(t *torrent, c *connection) bool {
1406         if me.stopped() {
1407                 return false
1408         }
1409         select {
1410         case <-t.ceasingNetworking:
1411                 return false
1412         default:
1413         }
1414         if !me.wantConns(t) {
1415                 return false
1416         }
1417         for _, c0 := range t.Conns {
1418                 if c.PeerID == c0.PeerID {
1419                         // Already connected to a client with that ID.
1420                         return false
1421                 }
1422         }
1423         t.Conns = append(t.Conns, c)
1424         // TODO: This should probably be done by a routine that kills off bad
1425         // connections, and extra connections killed here instead.
1426         if len(t.Conns) > socketsPerTorrent {
1427                 wcs := t.worstConnsHeap()
1428                 heap.Pop(wcs).(*connection).Close()
1429         }
1430         return true
1431 }
1432
1433 func (t *torrent) needData() bool {
1434         if !t.haveInfo() {
1435                 return true
1436         }
1437         for i := range t.Pieces {
1438                 if t.wantPiece(i) {
1439                         return true
1440                 }
1441         }
1442         return false
1443 }
1444
1445 // TODO: I'm sure there's something here to do with seeding.
1446 func (t *torrent) badConn(c *connection) bool {
1447         if time.Now().Sub(c.completedHandshake) < 30*time.Second {
1448                 return false
1449         }
1450         if !t.haveInfo() {
1451                 return !c.supportsExtension("ut_metadata")
1452         }
1453         return !t.connHasWantedPieces(c)
1454 }
1455
1456 func (t *torrent) numGoodConns() (num int) {
1457         for _, c := range t.Conns {
1458                 if !t.badConn(c) {
1459                         num++
1460                 }
1461         }
1462         return
1463 }
1464
1465 func (me *Client) wantConns(t *torrent) bool {
1466         if !t.needData() && me.noUpload {
1467                 return false
1468         }
1469         if t.numGoodConns() >= socketsPerTorrent {
1470                 return false
1471         }
1472         return true
1473 }
1474
1475 func (me *Client) openNewConns(t *torrent) {
1476         select {
1477         case <-t.ceasingNetworking:
1478                 return
1479         default:
1480         }
1481         for len(t.Peers) != 0 {
1482                 if !me.wantConns(t) {
1483                         return
1484                 }
1485                 if len(t.HalfOpen) >= me.halfOpenLimit {
1486                         return
1487                 }
1488                 var (
1489                         k peersKey
1490                         p Peer
1491                 )
1492                 for k, p = range t.Peers {
1493                         break
1494                 }
1495                 delete(t.Peers, k)
1496                 me.initiateConn(p, t)
1497         }
1498         t.wantPeers.Broadcast()
1499 }
1500
1501 func (me *Client) addPeers(t *torrent, peers []Peer) {
1502         blocked := 0
1503         for _, p := range peers {
1504                 if me.ipBlockRange(p.IP) != nil {
1505                         blocked++
1506                         continue
1507                 }
1508                 t.addPeer(p)
1509         }
1510         if blocked != 0 {
1511                 log.Printf("IP blocklist screened %d peers from being added", blocked)
1512         }
1513         me.openNewConns(t)
1514 }
1515
1516 func (cl *Client) torrentFileCachePath(ih InfoHash) string {
1517         return filepath.Join(cl.configDir(), "torrents", ih.HexString()+".torrent")
1518 }
1519
1520 func (cl *Client) saveTorrentFile(t *torrent) error {
1521         path := cl.torrentFileCachePath(t.InfoHash)
1522         os.MkdirAll(filepath.Dir(path), 0777)
1523         f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666)
1524         if err != nil {
1525                 return fmt.Errorf("error opening file: %s", err)
1526         }
1527         defer f.Close()
1528         e := bencode.NewEncoder(f)
1529         err = e.Encode(t.MetaInfo())
1530         if err != nil {
1531                 return fmt.Errorf("error marshalling metainfo: %s", err)
1532         }
1533         mi, err := cl.torrentCacheMetaInfo(t.InfoHash)
1534         if err != nil {
1535                 // For example, a script kiddy makes us load too many files, and we're
1536                 // able to save the torrent, but not load it again to check it.
1537                 return nil
1538         }
1539         if !bytes.Equal(mi.Info.Hash, t.InfoHash[:]) {
1540                 log.Fatalf("%x != %x", mi.Info.Hash, t.InfoHash[:])
1541         }
1542         return nil
1543 }
1544
1545 func (cl *Client) startTorrent(t *torrent) {
1546         if t.Info == nil || t.data == nil {
1547                 panic("nope")
1548         }
1549         // If the client intends to upload, it needs to know what state pieces are
1550         // in.
1551         if !cl.noUpload {
1552                 // Queue all pieces for hashing. This is done sequentially to avoid
1553                 // spamming goroutines.
1554                 for _, p := range t.Pieces {
1555                         p.QueuedForHash = true
1556                 }
1557                 go func() {
1558                         for i := range t.Pieces {
1559                                 cl.verifyPiece(t, pp.Integer(i))
1560                         }
1561                 }()
1562         }
1563 }
1564
1565 // Storage cannot be changed once it's set.
1566 func (cl *Client) setStorage(t *torrent, td data.Data) (err error) {
1567         err = t.setStorage(td)
1568         cl.event.Broadcast()
1569         if err != nil {
1570                 return
1571         }
1572         cl.startTorrent(t)
1573         return
1574 }
1575
1576 type TorrentDataOpener func(*metainfo.Info) data.Data
1577
1578 func (cl *Client) setMetaData(t *torrent, md metainfo.Info, bytes []byte) (err error) {
1579         err = t.setMetadata(md, bytes, &cl.mu)
1580         if err != nil {
1581                 return
1582         }
1583         if !cl.config.DisableMetainfoCache {
1584                 if err := cl.saveTorrentFile(t); err != nil {
1585                         log.Printf("error saving torrent file for %s: %s", t, err)
1586                 }
1587         }
1588         if strings.Contains(strings.ToLower(md.Name), "porn") {
1589                 cl.dropTorrent(t.InfoHash)
1590                 err = errors.New("no porn plx")
1591                 return
1592         }
1593         close(t.gotMetainfo)
1594         td := cl.torrentDataOpener(&md)
1595         err = cl.setStorage(t, td)
1596         return
1597 }
1598
1599 // Prepare a Torrent without any attachment to a Client. That means we can
1600 // initialize fields all fields that don't require the Client without locking
1601 // it.
1602 func newTorrent(ih InfoHash, announceList [][]string, halfOpenLimit int) (t *torrent, err error) {
1603         t = &torrent{
1604                 InfoHash: ih,
1605                 Peers:    make(map[peersKey]Peer),
1606
1607                 closing:           make(chan struct{}),
1608                 ceasingNetworking: make(chan struct{}),
1609
1610                 gotMetainfo: make(chan struct{}),
1611
1612                 HalfOpen: make(map[string]struct{}),
1613         }
1614         t.wantPeers.L = &t.stateMu
1615         t.GotMetainfo = t.gotMetainfo
1616         t.addTrackers(announceList)
1617         return
1618 }
1619
1620 func init() {
1621         // For shuffling the tracker tiers.
1622         mathRand.Seed(time.Now().Unix())
1623 }
1624
1625 // The trackers within each tier must be shuffled before use.
1626 // http://stackoverflow.com/a/12267471/149482
1627 // http://www.bittorrent.org/beps/bep_0012.html#order-of-processing
1628 func shuffleTier(tier []tracker.Client) {
1629         for i := range tier {
1630                 j := mathRand.Intn(i + 1)
1631                 tier[i], tier[j] = tier[j], tier[i]
1632         }
1633 }
1634
1635 func copyTrackers(base [][]tracker.Client) (copy [][]tracker.Client) {
1636         for _, tier := range base {
1637                 copy = append(copy, append([]tracker.Client{}, tier...))
1638         }
1639         return
1640 }
1641
1642 func mergeTier(tier []tracker.Client, newURLs []string) []tracker.Client {
1643 nextURL:
1644         for _, url := range newURLs {
1645                 for _, tr := range tier {
1646                         if tr.URL() == url {
1647                                 continue nextURL
1648                         }
1649                 }
1650                 tr, err := tracker.New(url)
1651                 if err != nil {
1652                         log.Printf("error creating tracker client for %q: %s", url, err)
1653                         continue
1654                 }
1655                 tier = append(tier, tr)
1656         }
1657         return tier
1658 }
1659
1660 func (t *torrent) addTrackers(announceList [][]string) {
1661         newTrackers := copyTrackers(t.Trackers)
1662         for tierIndex, tier := range announceList {
1663                 if tierIndex < len(newTrackers) {
1664                         newTrackers[tierIndex] = mergeTier(newTrackers[tierIndex], tier)
1665                 } else {
1666                         newTrackers = append(newTrackers, mergeTier(nil, tier))
1667                 }
1668                 shuffleTier(newTrackers[tierIndex])
1669         }
1670         t.Trackers = newTrackers
1671 }
1672
1673 type Torrent struct {
1674         cl *Client
1675         *torrent
1676 }
1677
1678 func (t Torrent) NumPieces() int {
1679         return t.numPieces()
1680 }
1681
1682 func (t Torrent) Drop() {
1683         t.cl.dropTorrent(t.InfoHash)
1684 }
1685
1686 type File struct {
1687         t      Torrent
1688         path   string
1689         offset int64
1690         length int64
1691         fi     metainfo.FileInfo
1692 }
1693
1694 func (f File) FileInfo() metainfo.FileInfo {
1695         return f.fi
1696 }
1697
1698 func (f File) Path() string {
1699         return f.path
1700 }
1701
1702 // A file-like handle to some torrent data resource.
1703 type Handle interface {
1704         io.Reader
1705         io.Seeker
1706         io.Closer
1707         io.ReaderAt
1708 }
1709
1710 // Implements a Handle within a subsection of another Handle.
1711 type sectionHandle struct {
1712         h           Handle
1713         off, n, cur int64
1714 }
1715
1716 func (me *sectionHandle) Seek(offset int64, whence int) (ret int64, err error) {
1717         if whence == 0 {
1718                 offset += me.off
1719         } else if whence == 2 {
1720                 whence = 0
1721                 offset += me.off + me.n
1722         }
1723         ret, err = me.h.Seek(offset, whence)
1724         me.cur = ret
1725         ret -= me.off
1726         return
1727 }
1728
1729 func (me *sectionHandle) Close() error {
1730         return me.h.Close()
1731 }
1732
1733 func (me *sectionHandle) Read(b []byte) (n int, err error) {
1734         max := me.off + me.n - me.cur
1735         if int64(len(b)) > max {
1736                 b = b[:max]
1737         }
1738         n, err = me.h.Read(b)
1739         me.cur += int64(n)
1740         if err != nil {
1741                 return
1742         }
1743         if me.cur == me.off+me.n {
1744                 err = io.EOF
1745         }
1746         return
1747 }
1748
1749 func (me *sectionHandle) ReadAt(b []byte, off int64) (n int, err error) {
1750         if off >= me.n {
1751                 err = io.EOF
1752                 return
1753         }
1754         if int64(len(b)) >= me.n-off {
1755                 b = b[:me.n-off]
1756         }
1757         return me.h.ReadAt(b, me.off+off)
1758 }
1759
1760 func (f File) Open() (h Handle, err error) {
1761         h = f.t.NewReadHandle()
1762         _, err = h.Seek(f.offset, os.SEEK_SET)
1763         if err != nil {
1764                 h.Close()
1765                 return
1766         }
1767         h = &sectionHandle{h, f.offset, f.Length(), f.offset}
1768         return
1769 }
1770
1771 func (f File) ReadAt(p []byte, off int64) (n int, err error) {
1772         maxLen := f.length - off
1773         if int64(len(p)) > maxLen {
1774                 p = p[:maxLen]
1775         }
1776         return f.t.ReadAt(p, off+f.offset)
1777 }
1778
1779 func (f *File) Length() int64 {
1780         return f.length
1781 }
1782
1783 type FilePieceState struct {
1784         Length int64
1785         State  byte
1786 }
1787
1788 func (f *File) Progress() (ret []FilePieceState) {
1789         pieceSize := int64(f.t.usualPieceSize())
1790         off := f.offset % pieceSize
1791         remaining := f.length
1792         for i := int(f.offset / pieceSize); ; i++ {
1793                 if remaining == 0 {
1794                         break
1795                 }
1796                 len1 := pieceSize - off
1797                 if len1 > remaining {
1798                         len1 = remaining
1799                 }
1800                 ret = append(ret, FilePieceState{len1, f.t.pieceStatusChar(i)})
1801                 off = 0
1802                 remaining -= len1
1803         }
1804         return
1805 }
1806
1807 func (f *File) PrioritizeRegion(off, len int64) {
1808         if off < 0 || off >= f.length {
1809                 return
1810         }
1811         if off+len > f.length {
1812                 len = f.length - off
1813         }
1814         off += f.offset
1815         f.t.SetRegionPriority(off, len)
1816 }
1817
1818 // Returns handles to the files in the torrent. This requires the metainfo is
1819 // available first.
1820 func (t Torrent) Files() (ret []File) {
1821         t.cl.mu.Lock()
1822         info := t.Info
1823         t.cl.mu.Unlock()
1824         if info == nil {
1825                 return
1826         }
1827         var offset int64
1828         for _, fi := range info.UpvertedFiles() {
1829                 ret = append(ret, File{
1830                         t,
1831                         strings.Join(append([]string{info.Name}, fi.Path...), "/"),
1832                         offset,
1833                         fi.Length,
1834                         fi,
1835                 })
1836                 offset += fi.Length
1837         }
1838         return
1839 }
1840
1841 func (t Torrent) SetRegionPriority(off, len int64) {
1842         t.cl.mu.Lock()
1843         defer t.cl.mu.Unlock()
1844         pieceSize := int64(t.usualPieceSize())
1845         for i := off / pieceSize; i*pieceSize < off+len; i++ {
1846                 t.cl.prioritizePiece(t.torrent, int(i), piecePriorityNormal)
1847         }
1848 }
1849
1850 func (t Torrent) MetainfoFilepath() string {
1851         return filepath.Join(t.cl.ConfigDir(), "torrents", t.InfoHash.HexString()+".torrent")
1852 }
1853
1854 func (t Torrent) AddPeers(pp []Peer) error {
1855         cl := t.cl
1856         cl.mu.Lock()
1857         defer cl.mu.Unlock()
1858         cl.addPeers(t.torrent, pp)
1859         return nil
1860 }
1861
1862 func (t Torrent) DownloadAll() {
1863         t.cl.mu.Lock()
1864         for i := 0; i < t.numPieces(); i++ {
1865                 // TODO: Leave higher priorities as they were?
1866                 t.cl.prioritizePiece(t.torrent, i, piecePriorityNormal)
1867         }
1868         // Nice to have the first and last pieces soon for various interactive
1869         // purposes.
1870         t.cl.prioritizePiece(t.torrent, 0, piecePriorityReadahead)
1871         t.cl.prioritizePiece(t.torrent, t.numPieces()-1, piecePriorityReadahead)
1872         t.cl.mu.Unlock()
1873 }
1874
1875 func (me Torrent) ReadAt(p []byte, off int64) (n int, err error) {
1876         return me.cl.torrentReadAt(me.torrent, off, p)
1877 }
1878
1879 // Returns nil metainfo if it isn't in the cache.
1880 func (cl *Client) torrentCacheMetaInfo(ih InfoHash) (mi *metainfo.MetaInfo, err error) {
1881         if cl.config.DisableMetainfoCache {
1882                 return
1883         }
1884         f, err := os.Open(cl.torrentFileCachePath(ih))
1885         if err != nil {
1886                 if os.IsNotExist(err) {
1887                         err = nil
1888                 }
1889                 return
1890         }
1891         defer f.Close()
1892         dec := bencode.NewDecoder(f)
1893         err = dec.Decode(&mi)
1894         if err != nil {
1895                 return
1896         }
1897         if !bytes.Equal(mi.Info.Hash, ih[:]) {
1898                 err = fmt.Errorf("cached torrent has wrong infohash: %x != %x", mi.Info.Hash, ih[:])
1899                 return
1900         }
1901         return
1902 }
1903
1904 func (cl *Client) AddMagnet(uri string) (T Torrent, err error) {
1905         m, err := ParseMagnetURI(uri)
1906         if err != nil {
1907                 return
1908         }
1909         mi, err := cl.torrentCacheMetaInfo(m.InfoHash)
1910         if err != nil {
1911                 log.Printf("error getting cached metainfo for %x: %s", m.InfoHash[:], err)
1912         } else if mi != nil {
1913                 _, err = cl.AddTorrent(mi)
1914                 if err != nil {
1915                         return
1916                 }
1917         }
1918         cl.mu.Lock()
1919         defer cl.mu.Unlock()
1920         T, err = cl.addOrMergeTorrent(m.InfoHash, [][]string{m.Trackers})
1921         if err != nil {
1922                 return
1923         }
1924         if m.DisplayName != "" {
1925                 T.DisplayName = m.DisplayName
1926         }
1927         return
1928 }
1929
1930 // Prunes unused connections. This is required to make space to dial for
1931 // replacements.
1932 func (cl *Client) pruneConnectionsUnlocked(t *torrent) {
1933         select {
1934         case <-t.ceasingNetworking:
1935                 return
1936         case <-t.closing:
1937                 return
1938         default:
1939         }
1940         cl.mu.Lock()
1941         license := len(t.Conns) - (socketsPerTorrent+1)/2
1942         for _, c := range t.Conns {
1943                 if license <= 0 {
1944                         break
1945                 }
1946                 if time.Now().Sub(c.lastUsefulChunkReceived) < time.Minute {
1947                         continue
1948                 }
1949                 if time.Now().Sub(c.completedHandshake) < time.Minute {
1950                         continue
1951                 }
1952                 c.Close()
1953                 license--
1954         }
1955         cl.mu.Unlock()
1956         t.pruneTimer.Reset(pruneInterval)
1957 }
1958
1959 func (me *Client) dropTorrent(infoHash InfoHash) (err error) {
1960         t, ok := me.torrents[infoHash]
1961         if !ok {
1962                 err = fmt.Errorf("no such torrent")
1963                 return
1964         }
1965         err = t.close()
1966         if err != nil {
1967                 panic(err)
1968         }
1969         delete(me.torrents, infoHash)
1970         return
1971 }
1972
1973 func (me *Client) addOrMergeTorrent(ih InfoHash, announceList [][]string) (T Torrent, err error) {
1974         if _, ok := me.bannedTorrents[ih]; ok {
1975                 err = errors.New("banned torrent")
1976                 return
1977         }
1978         T.cl = me
1979         var ok bool
1980         T.torrent, ok = me.torrents[ih]
1981         if ok {
1982                 T.torrent.addTrackers(announceList)
1983         } else {
1984                 T.torrent, err = newTorrent(ih, announceList, me.halfOpenLimit)
1985                 if err != nil {
1986                         return
1987                 }
1988                 me.torrents[ih] = T.torrent
1989                 if !me.disableTrackers {
1990                         go me.announceTorrentTrackers(T.torrent)
1991                 }
1992                 if me.dHT != nil {
1993                         go me.announceTorrentDHT(T.torrent, true)
1994                 }
1995                 T.torrent.pruneTimer = time.AfterFunc(0, func() {
1996                         me.pruneConnectionsUnlocked(T.torrent)
1997                 })
1998         }
1999         return
2000 }
2001
2002 // Adds a torrent to the client.
2003 func (me *Client) AddTorrent(metaInfo *metainfo.MetaInfo) (t Torrent, err error) {
2004         var ih InfoHash
2005         CopyExact(&ih, metaInfo.Info.Hash)
2006         me.mu.Lock()
2007         defer me.mu.Unlock()
2008         t, err = me.addOrMergeTorrent(ih, metaInfo.AnnounceList)
2009         if err != nil {
2010                 return
2011         }
2012         if !t.torrent.haveInfo() {
2013                 err = me.setMetaData(t.torrent, metaInfo.Info.Info, metaInfo.Info.Bytes)
2014                 if err != nil {
2015                         return
2016                 }
2017         }
2018         return
2019 }
2020
2021 func (me *Client) AddTorrentFromFile(name string) (t Torrent, err error) {
2022         mi, err := metainfo.LoadFromFile(name)
2023         if err != nil {
2024                 err = fmt.Errorf("error loading metainfo from file: %s", err)
2025                 return
2026         }
2027         return me.AddTorrent(mi)
2028 }
2029
2030 // Returns true when peers are required, or false if the torrent is closing.
2031 func (cl *Client) waitWantPeers(t *torrent) bool {
2032         cl.mu.Lock()
2033         defer cl.mu.Unlock()
2034         t.stateMu.Lock()
2035         defer t.stateMu.Unlock()
2036         for {
2037                 select {
2038                 case <-t.ceasingNetworking:
2039                         return false
2040                 default:
2041                 }
2042                 if len(t.Peers) < torrentPeersLowWater && t.needData() {
2043                         return true
2044                 }
2045                 cl.mu.Unlock()
2046                 t.wantPeers.Wait()
2047                 t.stateMu.Unlock()
2048                 cl.mu.Lock()
2049                 t.stateMu.Lock()
2050         }
2051 }
2052
2053 func (cl *Client) announceTorrentDHT(t *torrent, impliedPort bool) {
2054         for cl.waitWantPeers(t) {
2055                 log.Printf("getting peers for %q from DHT", t)
2056                 ps, err := cl.dHT.Announce(string(t.InfoHash[:]), cl.incomingPeerPort(), impliedPort)
2057                 if err != nil {
2058                         log.Printf("error getting peers from dht: %s", err)
2059                         return
2060                 }
2061                 allAddrs := make(map[string]struct{})
2062         getPeers:
2063                 for {
2064                         select {
2065                         case v, ok := <-ps.Values:
2066                                 if !ok {
2067                                         break getPeers
2068                                 }
2069                                 peersFoundByDHT.Add(int64(len(v.Peers)))
2070                                 for _, p := range v.Peers {
2071                                         allAddrs[(&net.UDPAddr{
2072                                                 IP:   p.IP[:],
2073                                                 Port: int(p.Port),
2074                                         }).String()] = struct{}{}
2075                                 }
2076                                 // log.Printf("%s: %d new peers from DHT", t, len(v.Peers))
2077                                 cl.mu.Lock()
2078                                 cl.addPeers(t, func() (ret []Peer) {
2079                                         for _, cp := range v.Peers {
2080                                                 ret = append(ret, Peer{
2081                                                         IP:     cp.IP[:],
2082                                                         Port:   int(cp.Port),
2083                                                         Source: peerSourceDHT,
2084                                                 })
2085                                         }
2086                                         return
2087                                 }())
2088                                 numPeers := len(t.Peers)
2089                                 cl.mu.Unlock()
2090                                 if numPeers >= torrentPeersHighWater {
2091                                         break getPeers
2092                                 }
2093                         case <-t.ceasingNetworking:
2094                                 ps.Close()
2095                                 return
2096                         }
2097                 }
2098                 ps.Close()
2099                 log.Printf("finished DHT peer scrape for %s: %d peers", t, len(allAddrs))
2100         }
2101 }
2102
2103 func (cl *Client) announceTorrentSingleTracker(tr tracker.Client, req *tracker.AnnounceRequest, t *torrent) error {
2104         if err := tr.Connect(); err != nil {
2105                 return fmt.Errorf("error connecting: %s", err)
2106         }
2107         resp, err := tr.Announce(req)
2108         if err != nil {
2109                 return fmt.Errorf("error announcing: %s", err)
2110         }
2111         var peers []Peer
2112         for _, peer := range resp.Peers {
2113                 peers = append(peers, Peer{
2114                         IP:   peer.IP,
2115                         Port: peer.Port,
2116                 })
2117         }
2118         cl.mu.Lock()
2119         cl.addPeers(t, peers)
2120         cl.mu.Unlock()
2121         log.Printf("%s: %d new peers from %s", t, len(peers), tr)
2122
2123         time.Sleep(time.Second * time.Duration(resp.Interval))
2124         return nil
2125 }
2126
2127 func (cl *Client) announceTorrentTrackersFastStart(req *tracker.AnnounceRequest, trackers [][]tracker.Client, t *torrent) (atLeastOne bool) {
2128         oks := make(chan bool)
2129         outstanding := 0
2130         for _, tier := range trackers {
2131                 for _, tr := range tier {
2132                         outstanding++
2133                         go func(tr tracker.Client) {
2134                                 err := cl.announceTorrentSingleTracker(tr, req, t)
2135                                 oks <- err == nil
2136                         }(tr)
2137                 }
2138         }
2139         for outstanding > 0 {
2140                 ok := <-oks
2141                 outstanding--
2142                 if ok {
2143                         atLeastOne = true
2144                 }
2145         }
2146         return
2147 }
2148
2149 // Announce torrent to its trackers.
2150 func (cl *Client) announceTorrentTrackers(t *torrent) {
2151         req := tracker.AnnounceRequest{
2152                 Event:    tracker.Started,
2153                 NumWant:  -1,
2154                 Port:     int16(cl.incomingPeerPort()),
2155                 PeerId:   cl.peerID,
2156                 InfoHash: t.InfoHash,
2157         }
2158         if !cl.waitWantPeers(t) {
2159                 return
2160         }
2161         cl.mu.RLock()
2162         req.Left = t.bytesLeft()
2163         trackers := t.Trackers
2164         cl.mu.RUnlock()
2165         if cl.announceTorrentTrackersFastStart(&req, trackers, t) {
2166                 req.Event = tracker.None
2167         }
2168 newAnnounce:
2169         for cl.waitWantPeers(t) {
2170                 cl.mu.RLock()
2171                 req.Left = t.bytesLeft()
2172                 trackers = t.Trackers
2173                 cl.mu.RUnlock()
2174                 numTrackersTried := 0
2175                 for _, tier := range trackers {
2176                         for trIndex, tr := range tier {
2177                                 numTrackersTried++
2178                                 err := cl.announceTorrentSingleTracker(tr, &req, t)
2179                                 if err != nil {
2180                                         continue
2181                                 }
2182                                 // Float the successful announce to the top of the tier. If
2183                                 // the trackers list has been changed, we'll be modifying an
2184                                 // old copy so it won't matter.
2185                                 cl.mu.Lock()
2186                                 tier[0], tier[trIndex] = tier[trIndex], tier[0]
2187                                 cl.mu.Unlock()
2188
2189                                 req.Event = tracker.None
2190                                 continue newAnnounce
2191                         }
2192                 }
2193                 if numTrackersTried != 0 {
2194                         log.Printf("%s: all trackers failed", t)
2195                 }
2196                 // TODO: Wait until trackers are added if there are none.
2197                 time.Sleep(10 * time.Second)
2198         }
2199 }
2200
2201 func (cl *Client) allTorrentsCompleted() bool {
2202         for _, t := range cl.torrents {
2203                 if !t.haveInfo() {
2204                         return false
2205                 }
2206                 if t.numPiecesCompleted() != t.numPieces() {
2207                         return false
2208                 }
2209         }
2210         return true
2211 }
2212
2213 // Returns true when all torrents are completely downloaded and false if the
2214 // client is stopped before that.
2215 func (me *Client) WaitAll() bool {
2216         me.mu.Lock()
2217         defer me.mu.Unlock()
2218         for !me.allTorrentsCompleted() {
2219                 if me.stopped() {
2220                         return false
2221                 }
2222                 me.event.Wait()
2223         }
2224         return true
2225 }
2226
2227 func (me *Client) replenishConnRequests(t *torrent, c *connection) {
2228         if !t.haveInfo() {
2229                 return
2230         }
2231         me.downloadStrategy.FillRequests(t, c)
2232         if len(c.Requests) == 0 && !c.PeerChoked {
2233                 c.SetInterested(false)
2234         }
2235 }
2236
2237 // Handle a received chunk from a peer.
2238 func (me *Client) downloadedChunk(t *torrent, c *connection, msg *pp.Message) error {
2239         chunksDownloadedCount.Add(1)
2240
2241         req := newRequest(msg.Index, msg.Begin, pp.Integer(len(msg.Piece)))
2242
2243         // Request has been satisfied.
2244         me.connDeleteRequest(t, c, req)
2245
2246         defer me.replenishConnRequests(t, c)
2247
2248         piece := t.Pieces[req.Index]
2249
2250         // Do we actually want this chunk?
2251         if _, ok := t.Pieces[req.Index].PendingChunkSpecs[req.chunkSpec]; !ok {
2252                 unusedDownloadedChunksCount.Add(1)
2253                 c.UnwantedChunksReceived++
2254                 return nil
2255         }
2256
2257         c.UsefulChunksReceived++
2258         c.lastUsefulChunkReceived = time.Now()
2259
2260         // Write the chunk out.
2261         err := t.writeChunk(int(msg.Index), int64(msg.Begin), msg.Piece)
2262         if err != nil {
2263                 return fmt.Errorf("error writing chunk: %s", err)
2264         }
2265
2266         // Record that we have the chunk.
2267         delete(piece.PendingChunkSpecs, req.chunkSpec)
2268         if len(piece.PendingChunkSpecs) == 0 {
2269                 for _, c := range t.Conns {
2270                         c.pieceRequestOrder.DeletePiece(int(req.Index))
2271                 }
2272                 me.queuePieceCheck(t, req.Index)
2273         }
2274
2275         // Cancel pending requests for this chunk.
2276         for _, c := range t.Conns {
2277                 if me.connCancel(t, c, req) {
2278                         me.replenishConnRequests(t, c)
2279                 }
2280         }
2281
2282         return nil
2283 }
2284
2285 func (me *Client) pieceHashed(t *torrent, piece pp.Integer, correct bool) {
2286         p := t.Pieces[piece]
2287         if p.EverHashed && !correct {
2288                 log.Printf("%s: piece %d failed hash", t, piece)
2289                 failedPieceHashes.Add(1)
2290         }
2291         p.EverHashed = true
2292         if correct {
2293                 if sd, ok := t.data.(StatefulData); ok {
2294                         err := sd.PieceCompleted(int(piece))
2295                         if err != nil {
2296                                 log.Printf("error completing piece: %s", err)
2297                                 correct = false
2298                         }
2299                 }
2300         }
2301         if correct {
2302                 p.Priority = piecePriorityNone
2303                 p.PendingChunkSpecs = nil
2304                 p.complete = true
2305                 p.Event.Broadcast()
2306         } else {
2307                 if len(p.PendingChunkSpecs) == 0 {
2308                         t.pendAllChunkSpecs(int(piece))
2309                 }
2310                 if p.Priority != piecePriorityNone {
2311                         me.openNewConns(t)
2312                 }
2313         }
2314         for _, conn := range t.Conns {
2315                 if correct {
2316                         conn.Post(pp.Message{
2317                                 Type:  pp.Have,
2318                                 Index: pp.Integer(piece),
2319                         })
2320                         // TODO: Cancel requests for this piece.
2321                         for r := range conn.Requests {
2322                                 if int(r.Index) == piece {
2323                                         panic("wat")
2324                                 }
2325                         }
2326                         conn.pieceRequestOrder.DeletePiece(int(piece))
2327                 }
2328                 if t.wantPiece(int(piece)) && conn.PeerHasPiece(pp.Integer(piece)) {
2329                         t.connPendPiece(conn, int(piece))
2330                         me.replenishConnRequests(t, conn)
2331                 }
2332         }
2333         if t.haveAllPieces() && me.noUpload {
2334                 t.ceaseNetworking()
2335         }
2336         me.event.Broadcast()
2337 }
2338
2339 func (cl *Client) verifyPiece(t *torrent, index pp.Integer) {
2340         cl.mu.Lock()
2341         defer cl.mu.Unlock()
2342         p := t.Pieces[index]
2343         for p.Hashing || t.data == nil {
2344                 cl.event.Wait()
2345         }
2346         p.QueuedForHash = false
2347         if t.isClosed() || p.complete {
2348                 return
2349         }
2350         p.Hashing = true
2351         cl.mu.Unlock()
2352         sum := t.hashPiece(index)
2353         cl.mu.Lock()
2354         select {
2355         case <-t.closing:
2356                 return
2357         default:
2358         }
2359         p.Hashing = false
2360         cl.pieceHashed(t, index, sum == p.Hash)
2361 }
2362
2363 // Returns handles to all the torrents loaded in the Client.
2364 func (me *Client) Torrents() (ret []Torrent) {
2365         me.mu.Lock()
2366         for _, t := range me.torrents {
2367                 ret = append(ret, Torrent{me, t})
2368         }
2369         me.mu.Unlock()
2370         return
2371 }