client.go

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