client.go

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