peerconn.go

   1 package torrent
   2
   3 import (
   4         "bufio"
   5         "bytes"
   6         "fmt"
   7         "io"
   8         "math/rand"
   9         "net"
  10         "strconv"
  11         "strings"
  12         "sync"
  13         "time"
  14
  15         "github.com/anacrolix/log"
  16         "github.com/anacrolix/missinggo"
  17         "github.com/anacrolix/missinggo/iter"
  18         "github.com/anacrolix/missinggo/v2/bitmap"
  19         "github.com/anacrolix/missinggo/v2/prioritybitmap"
  20         "github.com/anacrolix/multiless"
  21         "github.com/anacrolix/torrent/metainfo"
  22         "github.com/pkg/errors"
  23
  24         "github.com/anacrolix/torrent/bencode"
  25         "github.com/anacrolix/torrent/mse"
  26         pp "github.com/anacrolix/torrent/peer_protocol"
  27 )
  28
  29 type PeerSource string
  30
  31 const (
  32         PeerSourceTracker         = "Tr"
  33         PeerSourceIncoming        = "I"
  34         PeerSourceDhtGetPeers     = "Hg" // Peers we found by searching a DHT.
  35         PeerSourceDhtAnnouncePeer = "Ha" // Peers that were announced to us by a DHT.
  36         PeerSourcePex             = "X"
  37         // The peer was given directly, such as through a magnet link.
  38         PeerSourceDirect = "M"
  39 )
  40
  41 type peerRequestState struct {
  42         data []byte
  43 }
  44
  45 type PeerRemoteAddr interface {
  46         String() string
  47 }
  48
  49 type Peer struct {
  50         // First to ensure 64-bit alignment for atomics. See #262.
  51         _stats ConnStats
  52
  53         t *Torrent
  54
  55         peerImpl
  56         callbacks *Callbacks
  57
  58         outgoing   bool
  59         Network    string
  60         RemoteAddr PeerRemoteAddr
  61         // True if the connection is operating over MSE obfuscation.
  62         headerEncrypted bool
  63         cryptoMethod    mse.CryptoMethod
  64         Discovery       PeerSource
  65         trusted         bool
  66         closed          missinggo.Event
  67         // Set true after we've added our ConnStats generated during handshake to
  68         // other ConnStat instances as determined when the *Torrent became known.
  69         reconciledHandshakeStats bool
  70
  71         lastMessageReceived     time.Time
  72         completedHandshake      time.Time
  73         lastUsefulChunkReceived time.Time
  74         lastChunkSent           time.Time
  75
  76         // Stuff controlled by the local peer.
  77         interested           bool
  78         lastBecameInterested time.Time
  79         priorInterest        time.Duration
  80
  81         lastStartedExpectingToReceiveChunks time.Time
  82         cumulativeExpectedToReceiveChunks   time.Duration
  83         _chunksReceivedWhileExpecting       int64
  84
  85         choking          bool
  86         requests         map[Request]struct{}
  87         requestsLowWater int
  88         // Chunks that we might reasonably expect to receive from the peer. Due to
  89         // latency, buffering, and implementation differences, we may receive
  90         // chunks that are no longer in the set of requests actually want.
  91         validReceiveChunks map[Request]int
  92         // Indexed by metadata piece, set to true if posted and pending a
  93         // response.
  94         metadataRequests []bool
  95         sentHaves        bitmap.Bitmap
  96
  97         // Stuff controlled by the remote peer.
  98         peerInterested        bool
  99         peerChoking           bool
 100         peerRequests          map[Request]*peerRequestState
 101         PeerPrefersEncryption bool // as indicated by 'e' field in extension handshake
 102         PeerListenPort        int
 103         // The pieces the peer has claimed to have.
 104         _peerPieces bitmap.Bitmap
 105         // The peer has everything. This can occur due to a special message, when
 106         // we may not even know the number of pieces in the torrent yet.
 107         peerSentHaveAll bool
 108         // The highest possible number of pieces the torrent could have based on
 109         // communication with the peer. Generally only useful until we have the
 110         // torrent info.
 111         peerMinPieces pieceIndex
 112         // Pieces we've accepted chunks for from the peer.
 113         peerTouchedPieces map[pieceIndex]struct{}
 114         peerAllowedFast   bitmap.Bitmap
 115
 116         PeerMaxRequests  int // Maximum pending requests the peer allows.
 117         PeerExtensionIDs map[pp.ExtensionName]pp.ExtensionNumber
 118         PeerClientName   string
 119
 120         pieceInclination   []int
 121         _pieceRequestOrder prioritybitmap.PriorityBitmap
 122
 123         logger log.Logger
 124 }
 125
 126 // Maintains the state of a BitTorrent-protocol based connection with a peer.
 127 type PeerConn struct {
 128         Peer
 129
 130         // A string that should identify the PeerConn's net.Conn endpoints. The net.Conn could
 131         // be wrapping WebRTC, uTP, or TCP etc. Used in writing the conn status for peers.
 132         connString string
 133
 134         // See BEP 3 etc.
 135         PeerID             PeerID
 136         PeerExtensionBytes pp.PeerExtensionBits
 137
 138         // The actual Conn, used for closing, and setting socket options.
 139         conn net.Conn
 140         // The Reader and Writer for this Conn, with hooks installed for stats,
 141         // limiting, deadlines etc.
 142         w io.Writer
 143         r io.Reader
 144
 145         writeBuffer *bytes.Buffer
 146         uploadTimer *time.Timer
 147         writerCond  sync.Cond
 148
 149         pex pexConnState
 150 }
 151
 152 func (cn *PeerConn) connStatusString() string {
 153         return fmt.Sprintf("%+-55q %s %s", cn.PeerID, cn.PeerExtensionBytes, cn.connString)
 154 }
 155
 156 func (cn *Peer) updateExpectingChunks() {
 157         if cn.expectingChunks() {
 158                 if cn.lastStartedExpectingToReceiveChunks.IsZero() {
 159                         cn.lastStartedExpectingToReceiveChunks = time.Now()
 160                 }
 161         } else {
 162                 if !cn.lastStartedExpectingToReceiveChunks.IsZero() {
 163                         cn.cumulativeExpectedToReceiveChunks += time.Since(cn.lastStartedExpectingToReceiveChunks)
 164                         cn.lastStartedExpectingToReceiveChunks = time.Time{}
 165                 }
 166         }
 167 }
 168
 169 func (cn *Peer) expectingChunks() bool {
 170         return cn.interested && !cn.peerChoking
 171 }
 172
 173 // Returns true if the connection is over IPv6.
 174 func (cn *PeerConn) ipv6() bool {
 175         ip := cn.remoteIp()
 176         if ip.To4() != nil {
 177                 return false
 178         }
 179         return len(ip) == net.IPv6len
 180 }
 181
 182 // Returns true the if the dialer/initiator has the lower client peer ID. TODO: Find the
 183 // specification for this.
 184 func (cn *PeerConn) isPreferredDirection() bool {
 185         return bytes.Compare(cn.t.cl.peerID[:], cn.PeerID[:]) < 0 == cn.outgoing
 186 }
 187
 188 // Returns whether the left connection should be preferred over the right one,
 189 // considering only their networking properties. If ok is false, we can't
 190 // decide.
 191 func (l *PeerConn) hasPreferredNetworkOver(r *PeerConn) (left, ok bool) {
 192         var ml multiLess
 193         ml.NextBool(l.isPreferredDirection(), r.isPreferredDirection())
 194         ml.NextBool(!l.utp(), !r.utp())
 195         ml.NextBool(l.ipv6(), r.ipv6())
 196         return ml.FinalOk()
 197 }
 198
 199 func (cn *Peer) cumInterest() time.Duration {
 200         ret := cn.priorInterest
 201         if cn.interested {
 202                 ret += time.Since(cn.lastBecameInterested)
 203         }
 204         return ret
 205 }
 206
 207 func (cn *Peer) peerHasAllPieces() (all bool, known bool) {
 208         if cn.peerSentHaveAll {
 209                 return true, true
 210         }
 211         if !cn.t.haveInfo() {
 212                 return false, false
 213         }
 214         return bitmap.Flip(cn._peerPieces, 0, bitmap.BitIndex(cn.t.numPieces())).IsEmpty(), true
 215 }
 216
 217 func (cn *PeerConn) locker() *lockWithDeferreds {
 218         return cn.t.cl.locker()
 219 }
 220
 221 func (cn *Peer) supportsExtension(ext pp.ExtensionName) bool {
 222         _, ok := cn.PeerExtensionIDs[ext]
 223         return ok
 224 }
 225
 226 // The best guess at number of pieces in the torrent for this peer.
 227 func (cn *Peer) bestPeerNumPieces() pieceIndex {
 228         if cn.t.haveInfo() {
 229                 return cn.t.numPieces()
 230         }
 231         return cn.peerMinPieces
 232 }
 233
 234 func (cn *Peer) completedString() string {
 235         have := pieceIndex(cn._peerPieces.Len())
 236         if cn.peerSentHaveAll {
 237                 have = cn.bestPeerNumPieces()
 238         }
 239         return fmt.Sprintf("%d/%d", have, cn.bestPeerNumPieces())
 240 }
 241
 242 func (cn *PeerConn) onGotInfo(info *metainfo.Info) {
 243         cn.setNumPieces(info.NumPieces())
 244 }
 245
 246 // Correct the PeerPieces slice length. Return false if the existing slice is invalid, such as by
 247 // receiving badly sized BITFIELD, or invalid HAVE messages.
 248 func (cn *PeerConn) setNumPieces(num pieceIndex) {
 249         cn._peerPieces.RemoveRange(bitmap.BitIndex(num), bitmap.ToEnd)
 250         cn.peerPiecesChanged()
 251 }
 252
 253 func eventAgeString(t time.Time) string {
 254         if t.IsZero() {
 255                 return "never"
 256         }
 257         return fmt.Sprintf("%.2fs ago", time.Since(t).Seconds())
 258 }
 259
 260 func (cn *PeerConn) connectionFlags() (ret string) {
 261         c := func(b byte) {
 262                 ret += string([]byte{b})
 263         }
 264         if cn.cryptoMethod == mse.CryptoMethodRC4 {
 265                 c('E')
 266         } else if cn.headerEncrypted {
 267                 c('e')
 268         }
 269         ret += string(cn.Discovery)
 270         if cn.utp() {
 271                 c('U')
 272         }
 273         return
 274 }
 275
 276 func (cn *PeerConn) utp() bool {
 277         return parseNetworkString(cn.Network).Udp
 278 }
 279
 280 // Inspired by https://github.com/transmission/transmission/wiki/Peer-Status-Text.
 281 func (cn *Peer) statusFlags() (ret string) {
 282         c := func(b byte) {
 283                 ret += string([]byte{b})
 284         }
 285         if cn.interested {
 286                 c('i')
 287         }
 288         if cn.choking {
 289                 c('c')
 290         }
 291         c('-')
 292         ret += cn.connectionFlags()
 293         c('-')
 294         if cn.peerInterested {
 295                 c('i')
 296         }
 297         if cn.peerChoking {
 298                 c('c')
 299         }
 300         return
 301 }
 302
 303 // func (cn *connection) String() string {
 304 //      var buf bytes.Buffer
 305 //      cn.writeStatus(&buf, nil)
 306 //      return buf.String()
 307 // }
 308
 309 func (cn *Peer) downloadRate() float64 {
 310         return float64(cn._stats.BytesReadUsefulData.Int64()) / cn.cumInterest().Seconds()
 311 }
 312
 313 func (cn *Peer) writeStatus(w io.Writer, t *Torrent) {
 314         // \t isn't preserved in <pre> blocks?
 315         fmt.Fprintln(w, cn.connStatusString())
 316         fmt.Fprintf(w, "    last msg: %s, connected: %s, last helpful: %s, itime: %s, etime: %s\n",
 317                 eventAgeString(cn.lastMessageReceived),
 318                 eventAgeString(cn.completedHandshake),
 319                 eventAgeString(cn.lastHelpful()),
 320                 cn.cumInterest(),
 321                 cn.totalExpectingTime(),
 322         )
 323         fmt.Fprintf(w,
 324                 "    %s completed, %d pieces touched, good chunks: %v/%v-%v reqq: (%d,%d,%d]-%d, flags: %s, dr: %.1f KiB/s\n",
 325                 cn.completedString(),
 326                 len(cn.peerTouchedPieces),
 327                 &cn._stats.ChunksReadUseful,
 328                 &cn._stats.ChunksRead,
 329                 &cn._stats.ChunksWritten,
 330                 cn.requestsLowWater,
 331                 cn.numLocalRequests(),
 332                 cn.nominalMaxRequests(),
 333                 len(cn.peerRequests),
 334                 cn.statusFlags(),
 335                 cn.downloadRate()/(1<<10),
 336         )
 337         fmt.Fprintf(w, "    next pieces: %v%s\n",
 338                 iter.ToSlice(iter.Head(10, cn.iterPendingPiecesUntyped)),
 339                 func() string {
 340                         if cn == t.fastestPeer {
 341                                 return " (fastest)"
 342                         } else {
 343                                 return ""
 344                         }
 345                 }(),
 346         )
 347 }
 348
 349 func (cn *Peer) close() {
 350         if !cn.closed.Set() {
 351                 return
 352         }
 353         cn.discardPieceInclination()
 354         cn._pieceRequestOrder.Clear()
 355         cn.peerImpl.onClose()
 356         for _, f := range cn.callbacks.PeerClosed {
 357                 f(cn)
 358         }
 359 }
 360
 361 func (cn *PeerConn) onClose() {
 362         if cn.pex.IsEnabled() {
 363                 cn.pex.Close()
 364         }
 365         cn.tickleWriter()
 366         if cn.conn != nil {
 367                 cn.conn.Close()
 368         }
 369         if cb := cn.callbacks.PeerConnClosed; cb != nil {
 370                 cb(cn)
 371         }
 372 }
 373
 374 func (cn *Peer) peerHasPiece(piece pieceIndex) bool {
 375         return cn.peerSentHaveAll || cn._peerPieces.Contains(bitmap.BitIndex(piece))
 376 }
 377
 378 // 64KiB, but temporarily less to work around an issue with WebRTC. TODO: Update when
 379 // https://github.com/pion/datachannel/issues/59 is fixed.
 380 const writeBufferHighWaterLen = 1 << 15
 381
 382 // Writes a message into the write buffer. Returns whether it's okay to keep writing. Posting is
 383 // done asynchronously, so it may be that we're not able to honour backpressure from this method. It
 384 // might be possible to merge this with PeerConn.write down the track? They seem to be very similar.
 385 func (cn *PeerConn) post(msg pp.Message) bool {
 386         torrent.Add(fmt.Sprintf("messages posted of type %s", msg.Type.String()), 1)
 387         // We don't need to track bytes here because a connection.w Writer wrapper takes care of that
 388         // (although there's some delay between us recording the message, and the connection writer
 389         // flushing it out.).
 390         cn.writeBuffer.Write(msg.MustMarshalBinary())
 391         // Last I checked only Piece messages affect stats, and we don't post those.
 392         cn.wroteMsg(&msg)
 393         cn.tickleWriter()
 394         return cn.writeBuffer.Len() < writeBufferHighWaterLen
 395 }
 396
 397 // Returns true if there's room to write more.
 398 func (cn *PeerConn) write(msg pp.Message) bool {
 399         cn.wroteMsg(&msg)
 400         cn.writeBuffer.Write(msg.MustMarshalBinary())
 401         torrent.Add(fmt.Sprintf("messages filled of type %s", msg.Type.String()), 1)
 402         return cn.writeBuffer.Len() < writeBufferHighWaterLen
 403 }
 404
 405 func (cn *PeerConn) requestMetadataPiece(index int) {
 406         eID := cn.PeerExtensionIDs[pp.ExtensionNameMetadata]
 407         if eID == 0 {
 408                 return
 409         }
 410         if index < len(cn.metadataRequests) && cn.metadataRequests[index] {
 411                 return
 412         }
 413         cn.logger.WithDefaultLevel(log.Debug).Printf("requesting metadata piece %d", index)
 414         cn.post(pp.Message{
 415                 Type:       pp.Extended,
 416                 ExtendedID: eID,
 417                 ExtendedPayload: func() []byte {
 418                         b, err := bencode.Marshal(map[string]int{
 419                                 "msg_type": pp.RequestMetadataExtensionMsgType,
 420                                 "piece":    index,
 421                         })
 422                         if err != nil {
 423                                 panic(err)
 424                         }
 425                         return b
 426                 }(),
 427         })
 428         for index >= len(cn.metadataRequests) {
 429                 cn.metadataRequests = append(cn.metadataRequests, false)
 430         }
 431         cn.metadataRequests[index] = true
 432 }
 433
 434 func (cn *PeerConn) requestedMetadataPiece(index int) bool {
 435         return index < len(cn.metadataRequests) && cn.metadataRequests[index]
 436 }
 437
 438 // The actual value to use as the maximum outbound requests.
 439 func (cn *Peer) nominalMaxRequests() (ret int) {
 440         return int(clamp(
 441                 1,
 442                 int64(cn.PeerMaxRequests),
 443                 int64(cn.t.requestStrategy.nominalMaxRequests(cn.requestStrategyConnection())),
 444         ))
 445 }
 446
 447 func (cn *Peer) totalExpectingTime() (ret time.Duration) {
 448         ret = cn.cumulativeExpectedToReceiveChunks
 449         if !cn.lastStartedExpectingToReceiveChunks.IsZero() {
 450                 ret += time.Since(cn.lastStartedExpectingToReceiveChunks)
 451         }
 452         return
 453
 454 }
 455
 456 func (cn *PeerConn) onPeerSentCancel(r Request) {
 457         if _, ok := cn.peerRequests[r]; !ok {
 458                 torrent.Add("unexpected cancels received", 1)
 459                 return
 460         }
 461         if cn.fastEnabled() {
 462                 cn.reject(r)
 463         } else {
 464                 delete(cn.peerRequests, r)
 465         }
 466 }
 467
 468 func (cn *PeerConn) choke(msg messageWriter) (more bool) {
 469         if cn.choking {
 470                 return true
 471         }
 472         cn.choking = true
 473         more = msg(pp.Message{
 474                 Type: pp.Choke,
 475         })
 476         if cn.fastEnabled() {
 477                 for r := range cn.peerRequests {
 478                         // TODO: Don't reject pieces in allowed fast set.
 479                         cn.reject(r)
 480                 }
 481         } else {
 482                 cn.peerRequests = nil
 483         }
 484         return
 485 }
 486
 487 func (cn *PeerConn) unchoke(msg func(pp.Message) bool) bool {
 488         if !cn.choking {
 489                 return true
 490         }
 491         cn.choking = false
 492         return msg(pp.Message{
 493                 Type: pp.Unchoke,
 494         })
 495 }
 496
 497 func (cn *Peer) setInterested(interested bool) bool {
 498         if cn.interested == interested {
 499                 return true
 500         }
 501         cn.interested = interested
 502         if interested {
 503                 cn.lastBecameInterested = time.Now()
 504         } else if !cn.lastBecameInterested.IsZero() {
 505                 cn.priorInterest += time.Since(cn.lastBecameInterested)
 506         }
 507         cn.updateExpectingChunks()
 508         // log.Printf("%p: setting interest: %v", cn, interested)
 509         return cn.writeInterested(interested)
 510 }
 511
 512 func (pc *PeerConn) writeInterested(interested bool) bool {
 513         return pc.write(pp.Message{
 514                 Type: func() pp.MessageType {
 515                         if interested {
 516                                 return pp.Interested
 517                         } else {
 518                                 return pp.NotInterested
 519                         }
 520                 }(),
 521         })
 522 }
 523
 524 // The function takes a message to be sent, and returns true if more messages
 525 // are okay.
 526 type messageWriter func(pp.Message) bool
 527
 528 func (cn *Peer) request(r Request) bool {
 529         if _, ok := cn.requests[r]; ok {
 530                 panic("chunk already requested")
 531         }
 532         if !cn.peerHasPiece(pieceIndex(r.Index)) {
 533                 panic("requesting piece peer doesn't have")
 534         }
 535         if !cn.t.peerIsActive(cn) {
 536                 panic("requesting but not in active conns")
 537         }
 538         if cn.closed.IsSet() {
 539                 panic("requesting when connection is closed")
 540         }
 541         if cn.peerChoking {
 542                 if cn.peerAllowedFast.Get(int(r.Index)) {
 543                         torrent.Add("allowed fast requests sent", 1)
 544                 } else {
 545                         panic("requesting while choking and not allowed fast")
 546                 }
 547         }
 548         if cn.t.hashingPiece(pieceIndex(r.Index)) {
 549                 panic("piece is being hashed")
 550         }
 551         if cn.t.pieceQueuedForHash(pieceIndex(r.Index)) {
 552                 panic("piece is queued for hash")
 553         }
 554         if cn.requests == nil {
 555                 cn.requests = make(map[Request]struct{})
 556         }
 557         cn.requests[r] = struct{}{}
 558         if cn.validReceiveChunks == nil {
 559                 cn.validReceiveChunks = make(map[Request]int)
 560         }
 561         cn.validReceiveChunks[r]++
 562         cn.t.pendingRequests[r]++
 563         cn.t.requestStrategy.hooks().sentRequest(r)
 564         cn.updateExpectingChunks()
 565         for _, f := range cn.callbacks.SentRequest {
 566                 f(PeerRequestEvent{cn, r})
 567         }
 568         return cn.peerImpl.request(r)
 569 }
 570
 571 func (me *PeerConn) request(r Request) bool {
 572         return me.write(pp.Message{
 573                 Type:   pp.Request,
 574                 Index:  r.Index,
 575                 Begin:  r.Begin,
 576                 Length: r.Length,
 577         })
 578 }
 579
 580 func (me *PeerConn) cancel(r Request) bool {
 581         return me.write(makeCancelMessage(r))
 582 }
 583
 584 func (cn *Peer) doRequestState() bool {
 585         if !cn.t.networkingEnabled || cn.t.dataDownloadDisallowed {
 586                 if !cn.setInterested(false) {
 587                         return false
 588                 }
 589                 if len(cn.requests) != 0 {
 590                         for r := range cn.requests {
 591                                 cn.deleteRequest(r)
 592                                 // log.Printf("%p: cancelling request: %v", cn, r)
 593                                 if !cn.peerImpl.cancel(r) {
 594                                         return false
 595                                 }
 596                         }
 597                 }
 598         } else if len(cn.requests) <= cn.requestsLowWater {
 599                 filledBuffer := false
 600                 cn.iterPendingPieces(func(pieceIndex pieceIndex) bool {
 601                         cn.iterPendingRequests(pieceIndex, func(r Request) bool {
 602                                 if !cn.setInterested(true) {
 603                                         filledBuffer = true
 604                                         return false
 605                                 }
 606                                 if len(cn.requests) >= cn.nominalMaxRequests() {
 607                                         return false
 608                                 }
 609                                 // Choking is looked at here because our interest is dependent
 610                                 // on whether we'd make requests in its absence.
 611                                 if cn.peerChoking {
 612                                         if !cn.peerAllowedFast.Get(bitmap.BitIndex(r.Index)) {
 613                                                 return false
 614                                         }
 615                                 }
 616                                 if _, ok := cn.requests[r]; ok {
 617                                         return true
 618                                 }
 619                                 filledBuffer = !cn.request(r)
 620                                 return !filledBuffer
 621                         })
 622                         return !filledBuffer
 623                 })
 624                 if filledBuffer {
 625                         // If we didn't completely top up the requests, we shouldn't mark
 626                         // the low water, since we'll want to top up the requests as soon
 627                         // as we have more write buffer space.
 628                         return false
 629                 }
 630                 cn.requestsLowWater = len(cn.requests) / 2
 631         }
 632         return true
 633 }
 634
 635 func (cn *PeerConn) fillWriteBuffer() {
 636         if !cn.doRequestState() {
 637                 return
 638         }
 639         if cn.pex.IsEnabled() {
 640                 if flow := cn.pex.Share(cn.write); !flow {
 641                         return
 642                 }
 643         }
 644         cn.upload(cn.write)
 645 }
 646
 647 // Routine that writes to the peer. Some of what to write is buffered by
 648 // activity elsewhere in the Client, and some is determined locally when the
 649 // connection is writable.
 650 func (cn *PeerConn) writer(keepAliveTimeout time.Duration) {
 651         var (
 652                 lastWrite      time.Time = time.Now()
 653                 keepAliveTimer *time.Timer
 654         )
 655         keepAliveTimer = time.AfterFunc(keepAliveTimeout, func() {
 656                 cn.locker().Lock()
 657                 defer cn.locker().Unlock()
 658                 if time.Since(lastWrite) >= keepAliveTimeout {
 659                         cn.tickleWriter()
 660                 }
 661                 keepAliveTimer.Reset(keepAliveTimeout)
 662         })
 663         cn.locker().Lock()
 664         defer cn.locker().Unlock()
 665         defer cn.close()
 666         defer keepAliveTimer.Stop()
 667         frontBuf := new(bytes.Buffer)
 668         for {
 669                 if cn.closed.IsSet() {
 670                         return
 671                 }
 672                 if cn.writeBuffer.Len() == 0 {
 673                         cn.fillWriteBuffer()
 674                 }
 675                 if cn.writeBuffer.Len() == 0 && time.Since(lastWrite) >= keepAliveTimeout && cn.useful() {
 676                         cn.writeBuffer.Write(pp.Message{Keepalive: true}.MustMarshalBinary())
 677                         postedKeepalives.Add(1)
 678                 }
 679                 if cn.writeBuffer.Len() == 0 {
 680                         // TODO: Minimize wakeups....
 681                         cn.writerCond.Wait()
 682                         continue
 683                 }
 684                 // Flip the buffers.
 685                 frontBuf, cn.writeBuffer = cn.writeBuffer, frontBuf
 686                 cn.locker().Unlock()
 687                 n, err := cn.w.Write(frontBuf.Bytes())
 688                 cn.locker().Lock()
 689                 if n != 0 {
 690                         lastWrite = time.Now()
 691                         keepAliveTimer.Reset(keepAliveTimeout)
 692                 }
 693                 if err != nil {
 694                         cn.logger.WithDefaultLevel(log.Debug).Printf("error writing: %v", err)
 695                         return
 696                 }
 697                 if n != frontBuf.Len() {
 698                         panic("short write")
 699                 }
 700                 frontBuf.Reset()
 701         }
 702 }
 703
 704 func (cn *PeerConn) have(piece pieceIndex) {
 705         if cn.sentHaves.Get(bitmap.BitIndex(piece)) {
 706                 return
 707         }
 708         cn.post(pp.Message{
 709                 Type:  pp.Have,
 710                 Index: pp.Integer(piece),
 711         })
 712         cn.sentHaves.Add(bitmap.BitIndex(piece))
 713 }
 714
 715 func (cn *PeerConn) postBitfield() {
 716         if cn.sentHaves.Len() != 0 {
 717                 panic("bitfield must be first have-related message sent")
 718         }
 719         if !cn.t.haveAnyPieces() {
 720                 return
 721         }
 722         cn.post(pp.Message{
 723                 Type:     pp.Bitfield,
 724                 Bitfield: cn.t.bitfield(),
 725         })
 726         cn.sentHaves = cn.t._completedPieces.Copy()
 727 }
 728
 729 func (cn *PeerConn) updateRequests() {
 730         // log.Print("update requests")
 731         cn.tickleWriter()
 732 }
 733
 734 // Emits the indices in the Bitmaps bms in order, never repeating any index.
 735 // skip is mutated during execution, and its initial values will never be
 736 // emitted.
 737 func iterBitmapsDistinct(skip *bitmap.Bitmap, bms ...bitmap.Bitmap) iter.Func {
 738         return func(cb iter.Callback) {
 739                 for _, bm := range bms {
 740                         if !iter.All(
 741                                 func(i interface{}) bool {
 742                                         skip.Add(i.(int))
 743                                         return cb(i)
 744                                 },
 745                                 bitmap.Sub(bm, *skip).Iter,
 746                         ) {
 747                                 return
 748                         }
 749                 }
 750         }
 751 }
 752
 753 func iterUnbiasedPieceRequestOrder(cn requestStrategyConnection, f func(piece pieceIndex) bool) bool {
 754         now, readahead := cn.torrent().readerPiecePriorities()
 755         skip := bitmap.Flip(cn.peerPieces(), 0, cn.torrent().numPieces())
 756         skip.Union(cn.torrent().ignorePieces())
 757         // Return an iterator over the different priority classes, minus the skip pieces.
 758         return iter.All(
 759                 func(_piece interface{}) bool {
 760                         return f(pieceIndex(_piece.(bitmap.BitIndex)))
 761                 },
 762                 iterBitmapsDistinct(&skip, now, readahead),
 763                 // We have to iterate _pendingPieces separately because it isn't a Bitmap.
 764                 func(cb iter.Callback) {
 765                         cn.torrent().pendingPieces().IterTyped(func(piece int) bool {
 766                                 if skip.Contains(piece) {
 767                                         return true
 768                                 }
 769                                 more := cb(piece)
 770                                 skip.Add(piece)
 771                                 return more
 772                         })
 773                 },
 774         )
 775 }
 776
 777 // The connection should download highest priority pieces first, without any inclination toward
 778 // avoiding wastage. Generally we might do this if there's a single connection, or this is the
 779 // fastest connection, and we have active readers that signal an ordering preference. It's
 780 // conceivable that the best connection should do this, since it's least likely to waste our time if
 781 // assigned to the highest priority pieces, and assigning more than one this role would cause
 782 // significant wasted bandwidth.
 783 func (cn *Peer) shouldRequestWithoutBias() bool {
 784         return cn.t.requestStrategy.shouldRequestWithoutBias(cn.requestStrategyConnection())
 785 }
 786
 787 func (cn *Peer) iterPendingPieces(f func(pieceIndex) bool) bool {
 788         if !cn.t.haveInfo() {
 789                 return false
 790         }
 791         return cn.t.requestStrategy.iterPendingPieces(cn, f)
 792 }
 793 func (cn *Peer) iterPendingPiecesUntyped(f iter.Callback) {
 794         cn.iterPendingPieces(func(i pieceIndex) bool { return f(i) })
 795 }
 796
 797 func (cn *Peer) iterPendingRequests(piece pieceIndex, f func(Request) bool) bool {
 798         return cn.t.requestStrategy.iterUndirtiedChunks(
 799                 cn.t.piece(piece).requestStrategyPiece(),
 800                 func(cs ChunkSpec) bool {
 801                         return f(Request{pp.Integer(piece), cs})
 802                 },
 803         )
 804 }
 805
 806 // check callers updaterequests
 807 func (cn *Peer) stopRequestingPiece(piece pieceIndex) bool {
 808         return cn._pieceRequestOrder.Remove(bitmap.BitIndex(piece))
 809 }
 810
 811 // This is distinct from Torrent piece priority, which is the user's
 812 // preference. Connection piece priority is specific to a connection and is
 813 // used to pseudorandomly avoid connections always requesting the same pieces
 814 // and thus wasting effort.
 815 func (cn *Peer) updatePiecePriority(piece pieceIndex) bool {
 816         tpp := cn.t.piecePriority(piece)
 817         if !cn.peerHasPiece(piece) {
 818                 tpp = PiecePriorityNone
 819         }
 820         if tpp == PiecePriorityNone {
 821                 return cn.stopRequestingPiece(piece)
 822         }
 823         prio := cn.getPieceInclination()[piece]
 824         prio = cn.t.requestStrategy.piecePriority(cn, piece, tpp, prio)
 825         return cn._pieceRequestOrder.Set(bitmap.BitIndex(piece), prio) || cn.shouldRequestWithoutBias()
 826 }
 827
 828 func (cn *Peer) getPieceInclination() []int {
 829         if cn.pieceInclination == nil {
 830                 cn.pieceInclination = cn.t.getConnPieceInclination()
 831         }
 832         return cn.pieceInclination
 833 }
 834
 835 func (cn *Peer) discardPieceInclination() {
 836         if cn.pieceInclination == nil {
 837                 return
 838         }
 839         cn.t.putPieceInclination(cn.pieceInclination)
 840         cn.pieceInclination = nil
 841 }
 842
 843 func (cn *PeerConn) peerPiecesChanged() {
 844         if cn.t.haveInfo() {
 845                 prioritiesChanged := false
 846                 for i := pieceIndex(0); i < cn.t.numPieces(); i++ {
 847                         if cn.updatePiecePriority(i) {
 848                                 prioritiesChanged = true
 849                         }
 850                 }
 851                 if prioritiesChanged {
 852                         cn.updateRequests()
 853                 }
 854         }
 855         cn.t.maybeDropMutuallyCompletePeer(&cn.Peer)
 856 }
 857
 858 func (cn *PeerConn) raisePeerMinPieces(newMin pieceIndex) {
 859         if newMin > cn.peerMinPieces {
 860                 cn.peerMinPieces = newMin
 861         }
 862 }
 863
 864 func (cn *PeerConn) peerSentHave(piece pieceIndex) error {
 865         if cn.t.haveInfo() && piece >= cn.t.numPieces() || piece < 0 {
 866                 return errors.New("invalid piece")
 867         }
 868         if cn.peerHasPiece(piece) {
 869                 return nil
 870         }
 871         cn.raisePeerMinPieces(piece + 1)
 872         cn._peerPieces.Set(bitmap.BitIndex(piece), true)
 873         cn.t.maybeDropMutuallyCompletePeer(&cn.Peer)
 874         if cn.updatePiecePriority(piece) {
 875                 cn.updateRequests()
 876         }
 877         return nil
 878 }
 879
 880 func (cn *PeerConn) peerSentBitfield(bf []bool) error {
 881         cn.peerSentHaveAll = false
 882         if len(bf)%8 != 0 {
 883                 panic("expected bitfield length divisible by 8")
 884         }
 885         // We know that the last byte means that at most the last 7 bits are
 886         // wasted.
 887         cn.raisePeerMinPieces(pieceIndex(len(bf) - 7))
 888         if cn.t.haveInfo() && len(bf) > int(cn.t.numPieces()) {
 889                 // Ignore known excess pieces.
 890                 bf = bf[:cn.t.numPieces()]
 891         }
 892         for i, have := range bf {
 893                 if have {
 894                         cn.raisePeerMinPieces(pieceIndex(i) + 1)
 895                 }
 896                 cn._peerPieces.Set(i, have)
 897         }
 898         cn.peerPiecesChanged()
 899         return nil
 900 }
 901
 902 func (cn *PeerConn) onPeerSentHaveAll() error {
 903         cn.peerSentHaveAll = true
 904         cn._peerPieces.Clear()
 905         cn.peerPiecesChanged()
 906         return nil
 907 }
 908
 909 func (cn *PeerConn) peerSentHaveNone() error {
 910         cn._peerPieces.Clear()
 911         cn.peerSentHaveAll = false
 912         cn.peerPiecesChanged()
 913         return nil
 914 }
 915
 916 func (c *PeerConn) requestPendingMetadata() {
 917         if c.t.haveInfo() {
 918                 return
 919         }
 920         if c.PeerExtensionIDs[pp.ExtensionNameMetadata] == 0 {
 921                 // Peer doesn't support this.
 922                 return
 923         }
 924         // Request metadata pieces that we don't have in a random order.
 925         var pending []int
 926         for index := 0; index < c.t.metadataPieceCount(); index++ {
 927                 if !c.t.haveMetadataPiece(index) && !c.requestedMetadataPiece(index) {
 928                         pending = append(pending, index)
 929                 }
 930         }
 931         rand.Shuffle(len(pending), func(i, j int) { pending[i], pending[j] = pending[j], pending[i] })
 932         for _, i := range pending {
 933                 c.requestMetadataPiece(i)
 934         }
 935 }
 936
 937 func (cn *PeerConn) wroteMsg(msg *pp.Message) {
 938         torrent.Add(fmt.Sprintf("messages written of type %s", msg.Type.String()), 1)
 939         if msg.Type == pp.Extended {
 940                 for name, id := range cn.PeerExtensionIDs {
 941                         if id != msg.ExtendedID {
 942                                 continue
 943                         }
 944                         torrent.Add(fmt.Sprintf("Extended messages written for protocol %q", name), 1)
 945                 }
 946         }
 947         cn.allStats(func(cs *ConnStats) { cs.wroteMsg(msg) })
 948 }
 949
 950 func (cn *PeerConn) readMsg(msg *pp.Message) {
 951         cn.allStats(func(cs *ConnStats) { cs.readMsg(msg) })
 952 }
 953
 954 // After handshake, we know what Torrent and Client stats to include for a
 955 // connection.
 956 func (cn *Peer) postHandshakeStats(f func(*ConnStats)) {
 957         t := cn.t
 958         f(&t.stats)
 959         f(&t.cl.stats)
 960 }
 961
 962 // All ConnStats that include this connection. Some objects are not known
 963 // until the handshake is complete, after which it's expected to reconcile the
 964 // differences.
 965 func (cn *Peer) allStats(f func(*ConnStats)) {
 966         f(&cn._stats)
 967         if cn.reconciledHandshakeStats {
 968                 cn.postHandshakeStats(f)
 969         }
 970 }
 971
 972 func (cn *PeerConn) wroteBytes(n int64) {
 973         cn.allStats(add(n, func(cs *ConnStats) *Count { return &cs.BytesWritten }))
 974 }
 975
 976 func (cn *PeerConn) readBytes(n int64) {
 977         cn.allStats(add(n, func(cs *ConnStats) *Count { return &cs.BytesRead }))
 978 }
 979
 980 // Returns whether the connection could be useful to us. We're seeding and
 981 // they want data, we don't have metainfo and they can provide it, etc.
 982 func (c *Peer) useful() bool {
 983         t := c.t
 984         if c.closed.IsSet() {
 985                 return false
 986         }
 987         if !t.haveInfo() {
 988                 return c.supportsExtension("ut_metadata")
 989         }
 990         if t.seeding() && c.peerInterested {
 991                 return true
 992         }
 993         if c.peerHasWantedPieces() {
 994                 return true
 995         }
 996         return false
 997 }
 998
 999 func (c *Peer) lastHelpful() (ret time.Time) {
1000         ret = c.lastUsefulChunkReceived
1001         if c.t.seeding() && c.lastChunkSent.After(ret) {
1002                 ret = c.lastChunkSent
1003         }
1004         return
1005 }
1006
1007 func (c *PeerConn) fastEnabled() bool {
1008         return c.PeerExtensionBytes.SupportsFast() && c.t.cl.config.Extensions.SupportsFast()
1009 }
1010
1011 func (c *PeerConn) reject(r Request) {
1012         if !c.fastEnabled() {
1013                 panic("fast not enabled")
1014         }
1015         c.post(r.ToMsg(pp.Reject))
1016         delete(c.peerRequests, r)
1017 }
1018
1019 func (c *PeerConn) onReadRequest(r Request) error {
1020         requestedChunkLengths.Add(strconv.FormatUint(r.Length.Uint64(), 10), 1)
1021         if _, ok := c.peerRequests[r]; ok {
1022                 torrent.Add("duplicate requests received", 1)
1023                 return nil
1024         }
1025         if c.choking {
1026                 torrent.Add("requests received while choking", 1)
1027                 if c.fastEnabled() {
1028                         torrent.Add("requests rejected while choking", 1)
1029                         c.reject(r)
1030                 }
1031                 return nil
1032         }
1033         if len(c.peerRequests) >= maxRequests {
1034                 torrent.Add("requests received while queue full", 1)
1035                 if c.fastEnabled() {
1036                         c.reject(r)
1037                 }
1038                 // BEP 6 says we may close here if we choose.
1039                 return nil
1040         }
1041         if !c.t.havePiece(pieceIndex(r.Index)) {
1042                 // This isn't necessarily them screwing up. We can drop pieces
1043                 // from our storage, and can't communicate this to peers
1044                 // except by reconnecting.
1045                 requestsReceivedForMissingPieces.Add(1)
1046                 return fmt.Errorf("peer requested piece we don't have: %v", r.Index.Int())
1047         }
1048         // Check this after we know we have the piece, so that the piece length will be known.
1049         if r.Begin+r.Length > c.t.pieceLength(pieceIndex(r.Index)) {
1050                 torrent.Add("bad requests received", 1)
1051                 return errors.New("bad Request")
1052         }
1053         if c.peerRequests == nil {
1054                 c.peerRequests = make(map[Request]*peerRequestState, maxRequests)
1055         }
1056         value := &peerRequestState{}
1057         c.peerRequests[r] = value
1058         go c.peerRequestDataReader(r, value)
1059         //c.tickleWriter()
1060         return nil
1061 }
1062
1063 func (c *PeerConn) peerRequestDataReader(r Request, prs *peerRequestState) {
1064         b, err := readPeerRequestData(r, c)
1065         c.locker().Lock()
1066         defer c.locker().Unlock()
1067         if err != nil {
1068                 c.peerRequestDataReadFailed(err, r)
1069         } else {
1070                 if b == nil {
1071                         panic("data must be non-nil to trigger send")
1072                 }
1073                 prs.data = b
1074                 c.tickleWriter()
1075         }
1076 }
1077
1078 // If this is maintained correctly, we might be able to support optional synchronous reading for
1079 // chunk sending, the way it used to work.
1080 func (c *PeerConn) peerRequestDataReadFailed(err error, r Request) {
1081         c.logger.WithDefaultLevel(log.Warning).Printf("error reading chunk for peer Request %v: %v", r, err)
1082         i := pieceIndex(r.Index)
1083         if c.t.pieceComplete(i) {
1084                 // There used to be more code here that just duplicated the following break. Piece
1085                 // completions are currently cached, so I'm not sure how helpful this update is, except to
1086                 // pull any completion changes pushed to the storage backend in failed reads that got us
1087                 // here.
1088                 c.t.updatePieceCompletion(i)
1089         }
1090         // If we failed to send a chunk, choke the peer to ensure they flush all their requests. We've
1091         // probably dropped a piece from storage, but there's no way to communicate this to the peer. If
1092         // they ask for it again, we'll kick them to allow us to send them an updated bitfield on the
1093         // next connect. TODO: Support rejecting here too.
1094         if c.choking {
1095                 c.logger.WithDefaultLevel(log.Warning).Printf("already choking peer, requests might not be rejected correctly")
1096         }
1097         c.choke(c.post)
1098 }
1099
1100 func readPeerRequestData(r Request, c *PeerConn) ([]byte, error) {
1101         b := make([]byte, r.Length)
1102         p := c.t.info.Piece(int(r.Index))
1103         n, err := c.t.readAt(b, p.Offset()+int64(r.Begin))
1104         if n == len(b) {
1105                 if err == io.EOF {
1106                         err = nil
1107                 }
1108         } else {
1109                 if err == nil {
1110                         panic("expected error")
1111                 }
1112         }
1113         return b, err
1114 }
1115
1116 func runSafeExtraneous(f func()) {
1117         if true {
1118                 go f()
1119         } else {
1120                 f()
1121         }
1122 }
1123
1124 // Processes incoming BitTorrent wire-protocol messages. The client lock is held upon entry and
1125 // exit. Returning will end the connection.
1126 func (c *PeerConn) mainReadLoop() (err error) {
1127         defer func() {
1128                 if err != nil {
1129                         torrent.Add("connection.mainReadLoop returned with error", 1)
1130                 } else {
1131                         torrent.Add("connection.mainReadLoop returned with no error", 1)
1132                 }
1133         }()
1134         t := c.t
1135         cl := t.cl
1136
1137         decoder := pp.Decoder{
1138                 R:         bufio.NewReaderSize(c.r, 1<<17),
1139                 MaxLength: 256 * 1024,
1140                 Pool:      t.chunkPool,
1141         }
1142         for {
1143                 var msg pp.Message
1144                 func() {
1145                         cl.unlock()
1146                         defer cl.lock()
1147                         err = decoder.Decode(&msg)
1148                 }()
1149                 if cb := c.callbacks.ReadMessage; cb != nil && err == nil {
1150                         cb(c, &msg)
1151                 }
1152                 if t.closed.IsSet() || c.closed.IsSet() {
1153                         return nil
1154                 }
1155                 if err != nil {
1156                         return err
1157                 }
1158                 c.readMsg(&msg)
1159                 c.lastMessageReceived = time.Now()
1160                 if msg.Keepalive {
1161                         receivedKeepalives.Add(1)
1162                         continue
1163                 }
1164                 messageTypesReceived.Add(msg.Type.String(), 1)
1165                 if msg.Type.FastExtension() && !c.fastEnabled() {
1166                         runSafeExtraneous(func() { torrent.Add("fast messages received when extension is disabled", 1) })
1167                         return fmt.Errorf("received fast extension message (type=%v) but extension is disabled", msg.Type)
1168                 }
1169                 switch msg.Type {
1170                 case pp.Choke:
1171                         c.peerChoking = true
1172                         if !c.fastEnabled() {
1173                                 c.deleteAllRequests()
1174                         }
1175                         // We can then reset our interest.
1176                         c.updateRequests()
1177                         c.updateExpectingChunks()
1178                 case pp.Unchoke:
1179                         c.peerChoking = false
1180                         c.tickleWriter()
1181                         c.updateExpectingChunks()
1182                 case pp.Interested:
1183                         c.peerInterested = true
1184                         c.tickleWriter()
1185                 case pp.NotInterested:
1186                         c.peerInterested = false
1187                         // We don't clear their requests since it isn't clear in the spec.
1188                         // We'll probably choke them for this, which will clear them if
1189                         // appropriate, and is clearly specified.
1190                 case pp.Have:
1191                         err = c.peerSentHave(pieceIndex(msg.Index))
1192                 case pp.Bitfield:
1193                         err = c.peerSentBitfield(msg.Bitfield)
1194                 case pp.Request:
1195                         r := newRequestFromMessage(&msg)
1196                         err = c.onReadRequest(r)
1197                 case pp.Piece:
1198                         err = c.receiveChunk(&msg)
1199                         if len(msg.Piece) == int(t.chunkSize) {
1200                                 t.chunkPool.Put(&msg.Piece)
1201                         }
1202                         if err != nil {
1203                                 err = fmt.Errorf("receiving chunk: %s", err)
1204                         }
1205                 case pp.Cancel:
1206                         req := newRequestFromMessage(&msg)
1207                         c.onPeerSentCancel(req)
1208                 case pp.Port:
1209                         ipa, ok := tryIpPortFromNetAddr(c.RemoteAddr)
1210                         if !ok {
1211                                 break
1212                         }
1213                         pingAddr := net.UDPAddr{
1214                                 IP:   ipa.IP,
1215                                 Port: ipa.Port,
1216                         }
1217                         if msg.Port != 0 {
1218                                 pingAddr.Port = int(msg.Port)
1219                         }
1220                         cl.eachDhtServer(func(s DhtServer) {
1221                                 go s.Ping(&pingAddr)
1222                         })
1223                 case pp.Suggest:
1224                         torrent.Add("suggests received", 1)
1225                         log.Fmsg("peer suggested piece %d", msg.Index).AddValues(c, msg.Index).SetLevel(log.Debug).Log(c.t.logger)
1226                         c.updateRequests()
1227                 case pp.HaveAll:
1228                         err = c.onPeerSentHaveAll()
1229                 case pp.HaveNone:
1230                         err = c.peerSentHaveNone()
1231                 case pp.Reject:
1232                         c.remoteRejectedRequest(newRequestFromMessage(&msg))
1233                 case pp.AllowedFast:
1234                         torrent.Add("allowed fasts received", 1)
1235                         log.Fmsg("peer allowed fast: %d", msg.Index).AddValues(c).SetLevel(log.Debug).Log(c.t.logger)
1236                         c.peerAllowedFast.Add(int(msg.Index))
1237                         c.updateRequests()
1238                 case pp.Extended:
1239                         err = c.onReadExtendedMsg(msg.ExtendedID, msg.ExtendedPayload)
1240                 default:
1241                         err = fmt.Errorf("received unknown message type: %#v", msg.Type)
1242                 }
1243                 if err != nil {
1244                         return err
1245                 }
1246         }
1247 }
1248
1249 func (c *Peer) remoteRejectedRequest(r Request) {
1250         if c.deleteRequest(r) {
1251                 c.decExpectedChunkReceive(r)
1252         }
1253 }
1254
1255 func (c *Peer) decExpectedChunkReceive(r Request) {
1256         count := c.validReceiveChunks[r]
1257         if count == 1 {
1258                 delete(c.validReceiveChunks, r)
1259         } else if count > 1 {
1260                 c.validReceiveChunks[r] = count - 1
1261         } else {
1262                 panic(r)
1263         }
1264 }
1265
1266 func (c *PeerConn) onReadExtendedMsg(id pp.ExtensionNumber, payload []byte) (err error) {
1267         defer func() {
1268                 // TODO: Should we still do this?
1269                 if err != nil {
1270                         // These clients use their own extension IDs for outgoing message
1271                         // types, which is incorrect.
1272                         if bytes.HasPrefix(c.PeerID[:], []byte("-SD0100-")) || strings.HasPrefix(string(c.PeerID[:]), "-XL0012-") {
1273                                 err = nil
1274                         }
1275                 }
1276         }()
1277         t := c.t
1278         cl := t.cl
1279         switch id {
1280         case pp.HandshakeExtendedID:
1281                 var d pp.ExtendedHandshakeMessage
1282                 if err := bencode.Unmarshal(payload, &d); err != nil {
1283                         c.logger.Printf("error parsing extended handshake message %q: %s", payload, err)
1284                         return errors.Wrap(err, "unmarshalling extended handshake payload")
1285                 }
1286                 if cb := c.callbacks.ReadExtendedHandshake; cb != nil {
1287                         cb(c, &d)
1288                 }
1289                 //c.logger.WithDefaultLevel(log.Debug).Printf("received extended handshake message:\n%s", spew.Sdump(d))
1290                 if d.Reqq != 0 {
1291                         c.PeerMaxRequests = d.Reqq
1292                 }
1293                 c.PeerClientName = d.V
1294                 if c.PeerExtensionIDs == nil {
1295                         c.PeerExtensionIDs = make(map[pp.ExtensionName]pp.ExtensionNumber, len(d.M))
1296                 }
1297                 c.PeerListenPort = d.Port
1298                 c.PeerPrefersEncryption = d.Encryption
1299                 for name, id := range d.M {
1300                         if _, ok := c.PeerExtensionIDs[name]; !ok {
1301                                 torrent.Add(fmt.Sprintf("peers supporting extension %q", name), 1)
1302                         }
1303                         c.PeerExtensionIDs[name] = id
1304                 }
1305                 if d.MetadataSize != 0 {
1306                         if err = t.setMetadataSize(d.MetadataSize); err != nil {
1307                                 return errors.Wrapf(err, "setting metadata size to %d", d.MetadataSize)
1308                         }
1309                 }
1310                 c.requestPendingMetadata()
1311                 if !t.cl.config.DisablePEX {
1312                         t.pex.Add(c) // we learnt enough now
1313                         c.pex.Init(c)
1314                 }
1315                 return nil
1316         case metadataExtendedId:
1317                 err := cl.gotMetadataExtensionMsg(payload, t, c)
1318                 if err != nil {
1319                         return fmt.Errorf("handling metadata extension message: %w", err)
1320                 }
1321                 return nil
1322         case pexExtendedId:
1323                 if !c.pex.IsEnabled() {
1324                         return nil // or hang-up maybe?
1325                 }
1326                 return c.pex.Recv(payload)
1327         default:
1328                 return fmt.Errorf("unexpected extended message ID: %v", id)
1329         }
1330 }
1331
1332 // Set both the Reader and Writer for the connection from a single ReadWriter.
1333 func (cn *PeerConn) setRW(rw io.ReadWriter) {
1334         cn.r = rw
1335         cn.w = rw
1336 }
1337
1338 // Returns the Reader and Writer as a combined ReadWriter.
1339 func (cn *PeerConn) rw() io.ReadWriter {
1340         return struct {
1341                 io.Reader
1342                 io.Writer
1343         }{cn.r, cn.w}
1344 }
1345
1346 // Handle a received chunk from a peer.
1347 func (c *Peer) receiveChunk(msg *pp.Message) error {
1348         t := c.t
1349         cl := t.cl
1350         torrent.Add("chunks received", 1)
1351
1352         req := newRequestFromMessage(msg)
1353
1354         if c.peerChoking {
1355                 torrent.Add("chunks received while choking", 1)
1356         }
1357
1358         if c.validReceiveChunks[req] <= 0 {
1359                 torrent.Add("chunks received unexpected", 1)
1360                 return errors.New("received unexpected chunk")
1361         }
1362         c.decExpectedChunkReceive(req)
1363
1364         if c.peerChoking && c.peerAllowedFast.Get(int(req.Index)) {
1365                 torrent.Add("chunks received due to allowed fast", 1)
1366         }
1367
1368         // TODO: This needs to happen immediately, to prevent cancels occurring asynchronously when have
1369         // actually already received the piece, while we have the Client unlocked to write the data out.
1370         {
1371                 if _, ok := c.requests[req]; ok {
1372                         for _, f := range c.callbacks.ReceivedRequested {
1373                                 f(PeerMessageEvent{c, msg})
1374                         }
1375                 }
1376                 // Request has been satisfied.
1377                 if c.deleteRequest(req) {
1378                         if c.expectingChunks() {
1379                                 c._chunksReceivedWhileExpecting++
1380                         }
1381                 } else {
1382                         torrent.Add("chunks received unwanted", 1)
1383                 }
1384         }
1385
1386         // Do we actually want this chunk?
1387         if t.haveChunk(req) {
1388                 torrent.Add("chunks received wasted", 1)
1389                 c.allStats(add(1, func(cs *ConnStats) *Count { return &cs.ChunksReadWasted }))
1390                 return nil
1391         }
1392
1393         piece := &t.pieces[req.Index]
1394
1395         c.allStats(add(1, func(cs *ConnStats) *Count { return &cs.ChunksReadUseful }))
1396         c.allStats(add(int64(len(msg.Piece)), func(cs *ConnStats) *Count { return &cs.BytesReadUsefulData }))
1397         for _, f := range c.t.cl.config.Callbacks.ReceivedUsefulData {
1398                 f(ReceivedUsefulDataEvent{c, msg})
1399         }
1400         c.lastUsefulChunkReceived = time.Now()
1401         // if t.fastestPeer != c {
1402         // log.Printf("setting fastest connection %p", c)
1403         // }
1404         t.fastestPeer = c
1405
1406         // Need to record that it hasn't been written yet, before we attempt to do
1407         // anything with it.
1408         piece.incrementPendingWrites()
1409         // Record that we have the chunk, so we aren't trying to download it while
1410         // waiting for it to be written to storage.
1411         piece.unpendChunkIndex(chunkIndex(req.ChunkSpec, t.chunkSize))
1412
1413         // Cancel pending requests for this chunk from *other* peers.
1414         t.iterPeers(func(p *Peer) {
1415                 if p == c {
1416                         return
1417                 }
1418                 p.postCancel(req)
1419         })
1420
1421         err := func() error {
1422                 cl.unlock()
1423                 defer cl.lock()
1424                 concurrentChunkWrites.Add(1)
1425                 defer concurrentChunkWrites.Add(-1)
1426                 // Write the chunk out. Note that the upper bound on chunk writing concurrency will be the
1427                 // number of connections. We write inline with receiving the chunk (with this lock dance),
1428                 // because we want to handle errors synchronously and I haven't thought of a nice way to
1429                 // defer any concurrency to the storage and have that notify the client of errors. TODO: Do
1430                 // that instead.
1431                 return t.writeChunk(int(msg.Index), int64(msg.Begin), msg.Piece)
1432         }()
1433
1434         piece.decrementPendingWrites()
1435
1436         if err != nil {
1437                 c.logger.WithDefaultLevel(log.Error).Printf("writing received chunk %v: %v", req, err)
1438                 t.pendRequest(req)
1439                 //t.updatePieceCompletion(pieceIndex(msg.Index))
1440                 t.onWriteChunkErr(err)
1441                 return nil
1442         }
1443
1444         c.onDirtiedPiece(pieceIndex(req.Index))
1445
1446         // We need to ensure the piece is only queued once, so only the last chunk writer gets this job.
1447         if t.pieceAllDirty(pieceIndex(req.Index)) && piece.pendingWrites == 0 {
1448                 t.queuePieceCheck(pieceIndex(req.Index))
1449                 // We don't pend all chunks here anymore because we don't want code dependent on the dirty
1450                 // chunk status (such as the haveChunk call above) to have to check all the various other
1451                 // piece states like queued for hash, hashing etc. This does mean that we need to be sure
1452                 // that chunk pieces are pended at an appropriate time later however.
1453         }
1454
1455         cl.event.Broadcast()
1456         // We do this because we've written a chunk, and may change PieceState.Partial.
1457         t.publishPieceChange(pieceIndex(req.Index))
1458
1459         return nil
1460 }
1461
1462 func (c *Peer) onDirtiedPiece(piece pieceIndex) {
1463         if c.peerTouchedPieces == nil {
1464                 c.peerTouchedPieces = make(map[pieceIndex]struct{})
1465         }
1466         c.peerTouchedPieces[piece] = struct{}{}
1467         ds := &c.t.pieces[piece].dirtiers
1468         if *ds == nil {
1469                 *ds = make(map[*Peer]struct{})
1470         }
1471         (*ds)[c] = struct{}{}
1472 }
1473
1474 func (c *PeerConn) uploadAllowed() bool {
1475         if c.t.cl.config.NoUpload {
1476                 return false
1477         }
1478         if c.t.dataUploadDisallowed {
1479                 return false
1480         }
1481         if c.t.seeding() {
1482                 return true
1483         }
1484         if !c.peerHasWantedPieces() {
1485                 return false
1486         }
1487         // Don't upload more than 100 KiB more than we download.
1488         if c._stats.BytesWrittenData.Int64() >= c._stats.BytesReadData.Int64()+100<<10 {
1489                 return false
1490         }
1491         return true
1492 }
1493
1494 func (c *PeerConn) setRetryUploadTimer(delay time.Duration) {
1495         if c.uploadTimer == nil {
1496                 c.uploadTimer = time.AfterFunc(delay, c.writerCond.Broadcast)
1497         } else {
1498                 c.uploadTimer.Reset(delay)
1499         }
1500 }
1501
1502 // Also handles choking and unchoking of the remote peer.
1503 func (c *PeerConn) upload(msg func(pp.Message) bool) bool {
1504         // Breaking or completing this loop means we don't want to upload to the
1505         // peer anymore, and we choke them.
1506 another:
1507         for c.uploadAllowed() {
1508                 // We want to upload to the peer.
1509                 if !c.unchoke(msg) {
1510                         return false
1511                 }
1512                 for r, state := range c.peerRequests {
1513                         if state.data == nil {
1514                                 continue
1515                         }
1516                         res := c.t.cl.config.UploadRateLimiter.ReserveN(time.Now(), int(r.Length))
1517                         if !res.OK() {
1518                                 panic(fmt.Sprintf("upload rate limiter burst size < %d", r.Length))
1519                         }
1520                         delay := res.Delay()
1521                         if delay > 0 {
1522                                 res.Cancel()
1523                                 c.setRetryUploadTimer(delay)
1524                                 // Hard to say what to return here.
1525                                 return true
1526                         }
1527                         more := c.sendChunk(r, msg, state)
1528                         delete(c.peerRequests, r)
1529                         if !more {
1530                                 return false
1531                         }
1532                         goto another
1533                 }
1534                 return true
1535         }
1536         return c.choke(msg)
1537 }
1538
1539 func (cn *PeerConn) drop() {
1540         cn.t.dropConnection(cn)
1541 }
1542
1543 func (cn *Peer) netGoodPiecesDirtied() int64 {
1544         return cn._stats.PiecesDirtiedGood.Int64() - cn._stats.PiecesDirtiedBad.Int64()
1545 }
1546
1547 func (c *Peer) peerHasWantedPieces() bool {
1548         return !c._pieceRequestOrder.IsEmpty()
1549 }
1550
1551 func (c *Peer) numLocalRequests() int {
1552         return len(c.requests)
1553 }
1554
1555 func (c *Peer) deleteRequest(r Request) bool {
1556         if _, ok := c.requests[r]; !ok {
1557                 return false
1558         }
1559         delete(c.requests, r)
1560         for _, f := range c.callbacks.DeletedRequest {
1561                 f(PeerRequestEvent{c, r})
1562         }
1563         c.updateExpectingChunks()
1564         c.t.requestStrategy.hooks().deletedRequest(r)
1565         pr := c.t.pendingRequests
1566         pr[r]--
1567         n := pr[r]
1568         if n == 0 {
1569                 delete(pr, r)
1570         }
1571         if n < 0 {
1572                 panic(n)
1573         }
1574         // If a request fails, updating the requests for the current peer first may miss the opportunity
1575         // to try other peers for that request instead, depending on the request strategy. This might
1576         // only affect webseed peers though, since they synchronously issue new requests: PeerConns do
1577         // it in the writer routine.
1578         const updateCurrentConnRequestsFirst = false
1579         if updateCurrentConnRequestsFirst {
1580                 c.updateRequests()
1581         }
1582         // Give other conns a chance to pick up the request.
1583         c.t.iterPeers(func(_c *Peer) {
1584                 // We previously checked that the peer wasn't interested to to only wake connections that
1585                 // were unable to issue requests due to starvation by the request strategy. There could be
1586                 // performance ramifications.
1587                 if _c != c && c.peerHasPiece(pieceIndex(r.Index)) {
1588                         _c.updateRequests()
1589                 }
1590         })
1591         if !updateCurrentConnRequestsFirst {
1592                 c.updateRequests()
1593         }
1594         return true
1595 }
1596
1597 func (c *Peer) deleteAllRequests() {
1598         for r := range c.requests {
1599                 c.deleteRequest(r)
1600         }
1601         if len(c.requests) != 0 {
1602                 panic(len(c.requests))
1603         }
1604         // for c := range c.t.conns {
1605         //      c.tickleWriter()
1606         // }
1607 }
1608
1609 // This is called when something has changed that should wake the writer, such as putting stuff into
1610 // the writeBuffer, or changing some state that the writer can act on.
1611 func (c *PeerConn) tickleWriter() {
1612         c.writerCond.Broadcast()
1613 }
1614
1615 func (c *Peer) postCancel(r Request) bool {
1616         if !c.deleteRequest(r) {
1617                 return false
1618         }
1619         c.peerImpl._postCancel(r)
1620         return true
1621 }
1622
1623 func (c *PeerConn) _postCancel(r Request) {
1624         c.post(makeCancelMessage(r))
1625 }
1626
1627 func (c *PeerConn) sendChunk(r Request, msg func(pp.Message) bool, state *peerRequestState) (more bool) {
1628         c.lastChunkSent = time.Now()
1629         return msg(pp.Message{
1630                 Type:  pp.Piece,
1631                 Index: r.Index,
1632                 Begin: r.Begin,
1633                 Piece: state.data,
1634         })
1635 }
1636
1637 func (c *PeerConn) setTorrent(t *Torrent) {
1638         if c.t != nil {
1639                 panic("connection already associated with a torrent")
1640         }
1641         c.t = t
1642         c.logger.WithDefaultLevel(log.Debug).Printf("set torrent=%v", t)
1643         t.reconcileHandshakeStats(c)
1644 }
1645
1646 func (c *Peer) peerPriority() (peerPriority, error) {
1647         return bep40Priority(c.remoteIpPort(), c.t.cl.publicAddr(c.remoteIp()))
1648 }
1649
1650 func (c *Peer) remoteIp() net.IP {
1651         host, _, _ := net.SplitHostPort(c.RemoteAddr.String())
1652         return net.ParseIP(host)
1653 }
1654
1655 func (c *Peer) remoteIpPort() IpPort {
1656         ipa, _ := tryIpPortFromNetAddr(c.RemoteAddr)
1657         return IpPort{ipa.IP, uint16(ipa.Port)}
1658 }
1659
1660 func (c *PeerConn) pexPeerFlags() pp.PexPeerFlags {
1661         f := pp.PexPeerFlags(0)
1662         if c.PeerPrefersEncryption {
1663                 f |= pp.PexPrefersEncryption
1664         }
1665         if c.outgoing {
1666                 f |= pp.PexOutgoingConn
1667         }
1668         if c.utp() {
1669                 f |= pp.PexSupportsUtp
1670         }
1671         return f
1672 }
1673
1674 // This returns the address to use if we want to dial the peer again. It incorporates the peer's
1675 // advertised listen port.
1676 func (c *PeerConn) dialAddr() PeerRemoteAddr {
1677         if !c.outgoing && c.PeerListenPort != 0 {
1678                 switch addr := c.RemoteAddr.(type) {
1679                 case *net.TCPAddr:
1680                         dialAddr := *addr
1681                         dialAddr.Port = c.PeerListenPort
1682                         return &dialAddr
1683                 case *net.UDPAddr:
1684                         dialAddr := *addr
1685                         dialAddr.Port = c.PeerListenPort
1686                         return &dialAddr
1687                 }
1688         }
1689         return c.RemoteAddr
1690 }
1691
1692 func (c *PeerConn) pexEvent(t pexEventType) pexEvent {
1693         f := c.pexPeerFlags()
1694         addr := c.dialAddr()
1695         return pexEvent{t, addr, f}
1696 }
1697
1698 func (c *PeerConn) String() string {
1699         return fmt.Sprintf("connection %p", c)
1700 }
1701
1702 func (c *Peer) trust() connectionTrust {
1703         return connectionTrust{c.trusted, c.netGoodPiecesDirtied()}
1704 }
1705
1706 type connectionTrust struct {
1707         Implicit            bool
1708         NetGoodPiecesDirted int64
1709 }
1710
1711 func (l connectionTrust) Less(r connectionTrust) bool {
1712         return multiless.New().Bool(l.Implicit, r.Implicit).Int64(l.NetGoodPiecesDirted, r.NetGoodPiecesDirted).Less()
1713 }
1714
1715 func (cn *Peer) requestStrategyConnection() requestStrategyConnection {
1716         return cn
1717 }
1718
1719 func (cn *Peer) chunksReceivedWhileExpecting() int64 {
1720         return cn._chunksReceivedWhileExpecting
1721 }
1722
1723 func (cn *Peer) fastest() bool {
1724         return cn == cn.t.fastestPeer
1725 }
1726
1727 func (cn *Peer) peerMaxRequests() int {
1728         return cn.PeerMaxRequests
1729 }
1730
1731 // Returns the pieces the peer has claimed to have.
1732 func (cn *PeerConn) PeerPieces() bitmap.Bitmap {
1733         cn.locker().RLock()
1734         defer cn.locker().RUnlock()
1735         return cn.peerPieces()
1736 }
1737
1738 func (cn *Peer) peerPieces() bitmap.Bitmap {
1739         ret := cn._peerPieces.Copy()
1740         if cn.peerSentHaveAll {
1741                 ret.AddRange(0, cn.t.numPieces())
1742         }
1743         return ret
1744 }
1745
1746 func (cn *Peer) pieceRequestOrder() *prioritybitmap.PriorityBitmap {
1747         return &cn._pieceRequestOrder
1748 }
1749
1750 func (cn *Peer) stats() *ConnStats {
1751         return &cn._stats
1752 }
1753
1754 func (cn *Peer) torrent() requestStrategyTorrent {
1755         return cn.t.requestStrategyTorrent()
1756 }
1757
1758 func (p *Peer) TryAsPeerConn() (*PeerConn, bool) {
1759         pc, ok := p.peerImpl.(*PeerConn)
1760         return pc, ok
1761 }