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 len(cn.requests) != 0 && !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                 if len(cn.requests) == 0 {
 632                         return cn.setInterested(false)
 633                 }
 634         }
 635         return true
 636 }
 637
 638 func (cn *PeerConn) fillWriteBuffer() {
 639         if !cn.doRequestState() {
 640                 return
 641         }
 642         if cn.pex.IsEnabled() {
 643                 if flow := cn.pex.Share(cn.write); !flow {
 644                         return
 645                 }
 646         }
 647         cn.upload(cn.write)
 648 }
 649
 650 // Routine that writes to the peer. Some of what to write is buffered by
 651 // activity elsewhere in the Client, and some is determined locally when the
 652 // connection is writable.
 653 func (cn *PeerConn) writer(keepAliveTimeout time.Duration) {
 654         var (
 655                 lastWrite      time.Time = time.Now()
 656                 keepAliveTimer *time.Timer
 657         )
 658         keepAliveTimer = time.AfterFunc(keepAliveTimeout, func() {
 659                 cn.locker().Lock()
 660                 defer cn.locker().Unlock()
 661                 if time.Since(lastWrite) >= keepAliveTimeout {
 662                         cn.tickleWriter()
 663                 }
 664                 keepAliveTimer.Reset(keepAliveTimeout)
 665         })
 666         cn.locker().Lock()
 667         defer cn.locker().Unlock()
 668         defer cn.close()
 669         defer keepAliveTimer.Stop()
 670         frontBuf := new(bytes.Buffer)
 671         for {
 672                 if cn.closed.IsSet() {
 673                         return
 674                 }
 675                 if cn.writeBuffer.Len() == 0 {
 676                         cn.fillWriteBuffer()
 677                 }
 678                 if cn.writeBuffer.Len() == 0 && time.Since(lastWrite) >= keepAliveTimeout && cn.useful() {
 679                         cn.writeBuffer.Write(pp.Message{Keepalive: true}.MustMarshalBinary())
 680                         postedKeepalives.Add(1)
 681                 }
 682                 if cn.writeBuffer.Len() == 0 {
 683                         // TODO: Minimize wakeups....
 684                         cn.writerCond.Wait()
 685                         continue
 686                 }
 687                 // Flip the buffers.
 688                 frontBuf, cn.writeBuffer = cn.writeBuffer, frontBuf
 689                 cn.locker().Unlock()
 690                 n, err := cn.w.Write(frontBuf.Bytes())
 691                 cn.locker().Lock()
 692                 if n != 0 {
 693                         lastWrite = time.Now()
 694                         keepAliveTimer.Reset(keepAliveTimeout)
 695                 }
 696                 if err != nil {
 697                         cn.logger.WithDefaultLevel(log.Debug).Printf("error writing: %v", err)
 698                         return
 699                 }
 700                 if n != frontBuf.Len() {
 701                         panic("short write")
 702                 }
 703                 frontBuf.Reset()
 704         }
 705 }
 706
 707 func (cn *PeerConn) have(piece pieceIndex) {
 708         if cn.sentHaves.Get(bitmap.BitIndex(piece)) {
 709                 return
 710         }
 711         cn.post(pp.Message{
 712                 Type:  pp.Have,
 713                 Index: pp.Integer(piece),
 714         })
 715         cn.sentHaves.Add(bitmap.BitIndex(piece))
 716 }
 717
 718 func (cn *PeerConn) postBitfield() {
 719         if cn.sentHaves.Len() != 0 {
 720                 panic("bitfield must be first have-related message sent")
 721         }
 722         if !cn.t.haveAnyPieces() {
 723                 return
 724         }
 725         cn.post(pp.Message{
 726                 Type:     pp.Bitfield,
 727                 Bitfield: cn.t.bitfield(),
 728         })
 729         cn.sentHaves = cn.t._completedPieces.Copy()
 730 }
 731
 732 func (cn *PeerConn) updateRequests() {
 733         // log.Print("update requests")
 734         cn.tickleWriter()
 735 }
 736
 737 // Emits the indices in the Bitmaps bms in order, never repeating any index.
 738 // skip is mutated during execution, and its initial values will never be
 739 // emitted.
 740 func iterBitmapsDistinct(skip *bitmap.Bitmap, bms ...bitmap.Bitmap) iter.Func {
 741         return func(cb iter.Callback) {
 742                 for _, bm := range bms {
 743                         if !iter.All(
 744                                 func(i interface{}) bool {
 745                                         skip.Add(i.(int))
 746                                         return cb(i)
 747                                 },
 748                                 bitmap.Sub(bm, *skip).Iter,
 749                         ) {
 750                                 return
 751                         }
 752                 }
 753         }
 754 }
 755
 756 func iterUnbiasedPieceRequestOrder(cn requestStrategyConnection, f func(piece pieceIndex) bool) bool {
 757         now, readahead := cn.torrent().readerPiecePriorities()
 758         skip := bitmap.Flip(cn.peerPieces(), 0, cn.torrent().numPieces())
 759         skip.Union(cn.torrent().ignorePieces())
 760         // Return an iterator over the different priority classes, minus the skip pieces.
 761         return iter.All(
 762                 func(_piece interface{}) bool {
 763                         return f(pieceIndex(_piece.(bitmap.BitIndex)))
 764                 },
 765                 iterBitmapsDistinct(&skip, now, readahead),
 766                 // We have to iterate _pendingPieces separately because it isn't a Bitmap.
 767                 func(cb iter.Callback) {
 768                         cn.torrent().pendingPieces().IterTyped(func(piece int) bool {
 769                                 if skip.Contains(piece) {
 770                                         return true
 771                                 }
 772                                 more := cb(piece)
 773                                 skip.Add(piece)
 774                                 return more
 775                         })
 776                 },
 777         )
 778 }
 779
 780 // The connection should download highest priority pieces first, without any inclination toward
 781 // avoiding wastage. Generally we might do this if there's a single connection, or this is the
 782 // fastest connection, and we have active readers that signal an ordering preference. It's
 783 // conceivable that the best connection should do this, since it's least likely to waste our time if
 784 // assigned to the highest priority pieces, and assigning more than one this role would cause
 785 // significant wasted bandwidth.
 786 func (cn *Peer) shouldRequestWithoutBias() bool {
 787         return cn.t.requestStrategy.shouldRequestWithoutBias(cn.requestStrategyConnection())
 788 }
 789
 790 func (cn *Peer) iterPendingPieces(f func(pieceIndex) bool) bool {
 791         if !cn.t.haveInfo() {
 792                 return false
 793         }
 794         return cn.t.requestStrategy.iterPendingPieces(cn, f)
 795 }
 796 func (cn *Peer) iterPendingPiecesUntyped(f iter.Callback) {
 797         cn.iterPendingPieces(func(i pieceIndex) bool { return f(i) })
 798 }
 799
 800 func (cn *Peer) iterPendingRequests(piece pieceIndex, f func(Request) bool) bool {
 801         return cn.t.requestStrategy.iterUndirtiedChunks(
 802                 cn.t.piece(piece).requestStrategyPiece(),
 803                 func(cs ChunkSpec) bool {
 804                         return f(Request{pp.Integer(piece), cs})
 805                 },
 806         )
 807 }
 808
 809 // check callers updaterequests
 810 func (cn *Peer) stopRequestingPiece(piece pieceIndex) bool {
 811         return cn._pieceRequestOrder.Remove(bitmap.BitIndex(piece))
 812 }
 813
 814 // This is distinct from Torrent piece priority, which is the user's
 815 // preference. Connection piece priority is specific to a connection and is
 816 // used to pseudorandomly avoid connections always requesting the same pieces
 817 // and thus wasting effort.
 818 func (cn *Peer) updatePiecePriority(piece pieceIndex) bool {
 819         tpp := cn.t.piecePriority(piece)
 820         if !cn.peerHasPiece(piece) {
 821                 tpp = PiecePriorityNone
 822         }
 823         if tpp == PiecePriorityNone {
 824                 return cn.stopRequestingPiece(piece)
 825         }
 826         prio := cn.getPieceInclination()[piece]
 827         prio = cn.t.requestStrategy.piecePriority(cn, piece, tpp, prio)
 828         return cn._pieceRequestOrder.Set(bitmap.BitIndex(piece), prio) || cn.shouldRequestWithoutBias()
 829 }
 830
 831 func (cn *Peer) getPieceInclination() []int {
 832         if cn.pieceInclination == nil {
 833                 cn.pieceInclination = cn.t.getConnPieceInclination()
 834         }
 835         return cn.pieceInclination
 836 }
 837
 838 func (cn *Peer) discardPieceInclination() {
 839         if cn.pieceInclination == nil {
 840                 return
 841         }
 842         cn.t.putPieceInclination(cn.pieceInclination)
 843         cn.pieceInclination = nil
 844 }
 845
 846 func (cn *PeerConn) peerPiecesChanged() {
 847         if cn.t.haveInfo() {
 848                 prioritiesChanged := false
 849                 for i := pieceIndex(0); i < cn.t.numPieces(); i++ {
 850                         if cn.updatePiecePriority(i) {
 851                                 prioritiesChanged = true
 852                         }
 853                 }
 854                 if prioritiesChanged {
 855                         cn.updateRequests()
 856                 }
 857         }
 858         cn.t.maybeDropMutuallyCompletePeer(&cn.Peer)
 859 }
 860
 861 func (cn *PeerConn) raisePeerMinPieces(newMin pieceIndex) {
 862         if newMin > cn.peerMinPieces {
 863                 cn.peerMinPieces = newMin
 864         }
 865 }
 866
 867 func (cn *PeerConn) peerSentHave(piece pieceIndex) error {
 868         if cn.t.haveInfo() && piece >= cn.t.numPieces() || piece < 0 {
 869                 return errors.New("invalid piece")
 870         }
 871         if cn.peerHasPiece(piece) {
 872                 return nil
 873         }
 874         cn.raisePeerMinPieces(piece + 1)
 875         cn._peerPieces.Set(bitmap.BitIndex(piece), true)
 876         cn.t.maybeDropMutuallyCompletePeer(&cn.Peer)
 877         if cn.updatePiecePriority(piece) {
 878                 cn.updateRequests()
 879         }
 880         return nil
 881 }
 882
 883 func (cn *PeerConn) peerSentBitfield(bf []bool) error {
 884         cn.peerSentHaveAll = false
 885         if len(bf)%8 != 0 {
 886                 panic("expected bitfield length divisible by 8")
 887         }
 888         // We know that the last byte means that at most the last 7 bits are
 889         // wasted.
 890         cn.raisePeerMinPieces(pieceIndex(len(bf) - 7))
 891         if cn.t.haveInfo() && len(bf) > int(cn.t.numPieces()) {
 892                 // Ignore known excess pieces.
 893                 bf = bf[:cn.t.numPieces()]
 894         }
 895         for i, have := range bf {
 896                 if have {
 897                         cn.raisePeerMinPieces(pieceIndex(i) + 1)
 898                 }
 899                 cn._peerPieces.Set(i, have)
 900         }
 901         cn.peerPiecesChanged()
 902         return nil
 903 }
 904
 905 func (cn *PeerConn) onPeerSentHaveAll() error {
 906         cn.peerSentHaveAll = true
 907         cn._peerPieces.Clear()
 908         cn.peerPiecesChanged()
 909         return nil
 910 }
 911
 912 func (cn *PeerConn) peerSentHaveNone() error {
 913         cn._peerPieces.Clear()
 914         cn.peerSentHaveAll = false
 915         cn.peerPiecesChanged()
 916         return nil
 917 }
 918
 919 func (c *PeerConn) requestPendingMetadata() {
 920         if c.t.haveInfo() {
 921                 return
 922         }
 923         if c.PeerExtensionIDs[pp.ExtensionNameMetadata] == 0 {
 924                 // Peer doesn't support this.
 925                 return
 926         }
 927         // Request metadata pieces that we don't have in a random order.
 928         var pending []int
 929         for index := 0; index < c.t.metadataPieceCount(); index++ {
 930                 if !c.t.haveMetadataPiece(index) && !c.requestedMetadataPiece(index) {
 931                         pending = append(pending, index)
 932                 }
 933         }
 934         rand.Shuffle(len(pending), func(i, j int) { pending[i], pending[j] = pending[j], pending[i] })
 935         for _, i := range pending {
 936                 c.requestMetadataPiece(i)
 937         }
 938 }
 939
 940 func (cn *PeerConn) wroteMsg(msg *pp.Message) {
 941         torrent.Add(fmt.Sprintf("messages written of type %s", msg.Type.String()), 1)
 942         if msg.Type == pp.Extended {
 943                 for name, id := range cn.PeerExtensionIDs {
 944                         if id != msg.ExtendedID {
 945                                 continue
 946                         }
 947                         torrent.Add(fmt.Sprintf("Extended messages written for protocol %q", name), 1)
 948                 }
 949         }
 950         cn.allStats(func(cs *ConnStats) { cs.wroteMsg(msg) })
 951 }
 952
 953 func (cn *PeerConn) readMsg(msg *pp.Message) {
 954         cn.allStats(func(cs *ConnStats) { cs.readMsg(msg) })
 955 }
 956
 957 // After handshake, we know what Torrent and Client stats to include for a
 958 // connection.
 959 func (cn *Peer) postHandshakeStats(f func(*ConnStats)) {
 960         t := cn.t
 961         f(&t.stats)
 962         f(&t.cl.stats)
 963 }
 964
 965 // All ConnStats that include this connection. Some objects are not known
 966 // until the handshake is complete, after which it's expected to reconcile the
 967 // differences.
 968 func (cn *Peer) allStats(f func(*ConnStats)) {
 969         f(&cn._stats)
 970         if cn.reconciledHandshakeStats {
 971                 cn.postHandshakeStats(f)
 972         }
 973 }
 974
 975 func (cn *PeerConn) wroteBytes(n int64) {
 976         cn.allStats(add(n, func(cs *ConnStats) *Count { return &cs.BytesWritten }))
 977 }
 978
 979 func (cn *PeerConn) readBytes(n int64) {
 980         cn.allStats(add(n, func(cs *ConnStats) *Count { return &cs.BytesRead }))
 981 }
 982
 983 // Returns whether the connection could be useful to us. We're seeding and
 984 // they want data, we don't have metainfo and they can provide it, etc.
 985 func (c *Peer) useful() bool {
 986         t := c.t
 987         if c.closed.IsSet() {
 988                 return false
 989         }
 990         if !t.haveInfo() {
 991                 return c.supportsExtension("ut_metadata")
 992         }
 993         if t.seeding() && c.peerInterested {
 994                 return true
 995         }
 996         if c.peerHasWantedPieces() {
 997                 return true
 998         }
 999         return false
1000 }
1001
1002 func (c *Peer) lastHelpful() (ret time.Time) {
1003         ret = c.lastUsefulChunkReceived
1004         if c.t.seeding() && c.lastChunkSent.After(ret) {
1005                 ret = c.lastChunkSent
1006         }
1007         return
1008 }
1009
1010 func (c *PeerConn) fastEnabled() bool {
1011         return c.PeerExtensionBytes.SupportsFast() && c.t.cl.config.Extensions.SupportsFast()
1012 }
1013
1014 func (c *PeerConn) reject(r Request) {
1015         if !c.fastEnabled() {
1016                 panic("fast not enabled")
1017         }
1018         c.post(r.ToMsg(pp.Reject))
1019         delete(c.peerRequests, r)
1020 }
1021
1022 func (c *PeerConn) onReadRequest(r Request) error {
1023         requestedChunkLengths.Add(strconv.FormatUint(r.Length.Uint64(), 10), 1)
1024         if _, ok := c.peerRequests[r]; ok {
1025                 torrent.Add("duplicate requests received", 1)
1026                 return nil
1027         }
1028         if c.choking {
1029                 torrent.Add("requests received while choking", 1)
1030                 if c.fastEnabled() {
1031                         torrent.Add("requests rejected while choking", 1)
1032                         c.reject(r)
1033                 }
1034                 return nil
1035         }
1036         if len(c.peerRequests) >= maxRequests {
1037                 torrent.Add("requests received while queue full", 1)
1038                 if c.fastEnabled() {
1039                         c.reject(r)
1040                 }
1041                 // BEP 6 says we may close here if we choose.
1042                 return nil
1043         }
1044         if !c.t.havePiece(pieceIndex(r.Index)) {
1045                 // This isn't necessarily them screwing up. We can drop pieces
1046                 // from our storage, and can't communicate this to peers
1047                 // except by reconnecting.
1048                 requestsReceivedForMissingPieces.Add(1)
1049                 return fmt.Errorf("peer requested piece we don't have: %v", r.Index.Int())
1050         }
1051         // Check this after we know we have the piece, so that the piece length will be known.
1052         if r.Begin+r.Length > c.t.pieceLength(pieceIndex(r.Index)) {
1053                 torrent.Add("bad requests received", 1)
1054                 return errors.New("bad Request")
1055         }
1056         if c.peerRequests == nil {
1057                 c.peerRequests = make(map[Request]*peerRequestState, maxRequests)
1058         }
1059         value := &peerRequestState{}
1060         c.peerRequests[r] = value
1061         go c.peerRequestDataReader(r, value)
1062         //c.tickleWriter()
1063         return nil
1064 }
1065
1066 func (c *PeerConn) peerRequestDataReader(r Request, prs *peerRequestState) {
1067         b, err := readPeerRequestData(r, c)
1068         c.locker().Lock()
1069         defer c.locker().Unlock()
1070         if err != nil {
1071                 c.peerRequestDataReadFailed(err, r)
1072         } else {
1073                 if b == nil {
1074                         panic("data must be non-nil to trigger send")
1075                 }
1076                 prs.data = b
1077                 c.tickleWriter()
1078         }
1079 }
1080
1081 // If this is maintained correctly, we might be able to support optional synchronous reading for
1082 // chunk sending, the way it used to work.
1083 func (c *PeerConn) peerRequestDataReadFailed(err error, r Request) {
1084         c.logger.WithDefaultLevel(log.Warning).Printf("error reading chunk for peer Request %v: %v", r, err)
1085         i := pieceIndex(r.Index)
1086         if c.t.pieceComplete(i) {
1087                 // There used to be more code here that just duplicated the following break. Piece
1088                 // completions are currently cached, so I'm not sure how helpful this update is, except to
1089                 // pull any completion changes pushed to the storage backend in failed reads that got us
1090                 // here.
1091                 c.t.updatePieceCompletion(i)
1092         }
1093         // If we failed to send a chunk, choke the peer to ensure they flush all their requests. We've
1094         // probably dropped a piece from storage, but there's no way to communicate this to the peer. If
1095         // they ask for it again, we'll kick them to allow us to send them an updated bitfield on the
1096         // next connect. TODO: Support rejecting here too.
1097         if c.choking {
1098                 c.logger.WithDefaultLevel(log.Warning).Printf("already choking peer, requests might not be rejected correctly")
1099         }
1100         c.choke(c.post)
1101 }
1102
1103 func readPeerRequestData(r Request, c *PeerConn) ([]byte, error) {
1104         b := make([]byte, r.Length)
1105         p := c.t.info.Piece(int(r.Index))
1106         n, err := c.t.readAt(b, p.Offset()+int64(r.Begin))
1107         if n == len(b) {
1108                 if err == io.EOF {
1109                         err = nil
1110                 }
1111         } else {
1112                 if err == nil {
1113                         panic("expected error")
1114                 }
1115         }
1116         return b, err
1117 }
1118
1119 func runSafeExtraneous(f func()) {
1120         if true {
1121                 go f()
1122         } else {
1123                 f()
1124         }
1125 }
1126
1127 // Processes incoming BitTorrent wire-protocol messages. The client lock is held upon entry and
1128 // exit. Returning will end the connection.
1129 func (c *PeerConn) mainReadLoop() (err error) {
1130         defer func() {
1131                 if err != nil {
1132                         torrent.Add("connection.mainReadLoop returned with error", 1)
1133                 } else {
1134                         torrent.Add("connection.mainReadLoop returned with no error", 1)
1135                 }
1136         }()
1137         t := c.t
1138         cl := t.cl
1139
1140         decoder := pp.Decoder{
1141                 R:         bufio.NewReaderSize(c.r, 1<<17),
1142                 MaxLength: 256 * 1024,
1143                 Pool:      t.chunkPool,
1144         }
1145         for {
1146                 var msg pp.Message
1147                 func() {
1148                         cl.unlock()
1149                         defer cl.lock()
1150                         err = decoder.Decode(&msg)
1151                 }()
1152                 if cb := c.callbacks.ReadMessage; cb != nil && err == nil {
1153                         cb(c, &msg)
1154                 }
1155                 if t.closed.IsSet() || c.closed.IsSet() {
1156                         return nil
1157                 }
1158                 if err != nil {
1159                         return err
1160                 }
1161                 c.readMsg(&msg)
1162                 c.lastMessageReceived = time.Now()
1163                 if msg.Keepalive {
1164                         receivedKeepalives.Add(1)
1165                         continue
1166                 }
1167                 messageTypesReceived.Add(msg.Type.String(), 1)
1168                 if msg.Type.FastExtension() && !c.fastEnabled() {
1169                         runSafeExtraneous(func() { torrent.Add("fast messages received when extension is disabled", 1) })
1170                         return fmt.Errorf("received fast extension message (type=%v) but extension is disabled", msg.Type)
1171                 }
1172                 switch msg.Type {
1173                 case pp.Choke:
1174                         c.peerChoking = true
1175                         if !c.fastEnabled() {
1176                                 c.deleteAllRequests()
1177                         }
1178                         // We can then reset our interest.
1179                         c.updateRequests()
1180                         c.updateExpectingChunks()
1181                 case pp.Unchoke:
1182                         c.peerChoking = false
1183                         c.tickleWriter()
1184                         c.updateExpectingChunks()
1185                 case pp.Interested:
1186                         c.peerInterested = true
1187                         c.tickleWriter()
1188                 case pp.NotInterested:
1189                         c.peerInterested = false
1190                         // We don't clear their requests since it isn't clear in the spec.
1191                         // We'll probably choke them for this, which will clear them if
1192                         // appropriate, and is clearly specified.
1193                 case pp.Have:
1194                         err = c.peerSentHave(pieceIndex(msg.Index))
1195                 case pp.Bitfield:
1196                         err = c.peerSentBitfield(msg.Bitfield)
1197                 case pp.Request:
1198                         r := newRequestFromMessage(&msg)
1199                         err = c.onReadRequest(r)
1200                 case pp.Piece:
1201                         err = c.receiveChunk(&msg)
1202                         if len(msg.Piece) == int(t.chunkSize) {
1203                                 t.chunkPool.Put(&msg.Piece)
1204                         }
1205                         if err != nil {
1206                                 err = fmt.Errorf("receiving chunk: %s", err)
1207                         }
1208                 case pp.Cancel:
1209                         req := newRequestFromMessage(&msg)
1210                         c.onPeerSentCancel(req)
1211                 case pp.Port:
1212                         ipa, ok := tryIpPortFromNetAddr(c.RemoteAddr)
1213                         if !ok {
1214                                 break
1215                         }
1216                         pingAddr := net.UDPAddr{
1217                                 IP:   ipa.IP,
1218                                 Port: ipa.Port,
1219                         }
1220                         if msg.Port != 0 {
1221                                 pingAddr.Port = int(msg.Port)
1222                         }
1223                         cl.eachDhtServer(func(s DhtServer) {
1224                                 go s.Ping(&pingAddr)
1225                         })
1226                 case pp.Suggest:
1227                         torrent.Add("suggests received", 1)
1228                         log.Fmsg("peer suggested piece %d", msg.Index).AddValues(c, msg.Index).SetLevel(log.Debug).Log(c.t.logger)
1229                         c.updateRequests()
1230                 case pp.HaveAll:
1231                         err = c.onPeerSentHaveAll()
1232                 case pp.HaveNone:
1233                         err = c.peerSentHaveNone()
1234                 case pp.Reject:
1235                         c.remoteRejectedRequest(newRequestFromMessage(&msg))
1236                 case pp.AllowedFast:
1237                         torrent.Add("allowed fasts received", 1)
1238                         log.Fmsg("peer allowed fast: %d", msg.Index).AddValues(c).SetLevel(log.Debug).Log(c.t.logger)
1239                         c.peerAllowedFast.Add(int(msg.Index))
1240                         c.updateRequests()
1241                 case pp.Extended:
1242                         err = c.onReadExtendedMsg(msg.ExtendedID, msg.ExtendedPayload)
1243                 default:
1244                         err = fmt.Errorf("received unknown message type: %#v", msg.Type)
1245                 }
1246                 if err != nil {
1247                         return err
1248                 }
1249         }
1250 }
1251
1252 func (c *Peer) remoteRejectedRequest(r Request) {
1253         if c.deleteRequest(r) {
1254                 c.decExpectedChunkReceive(r)
1255         }
1256 }
1257
1258 func (c *Peer) decExpectedChunkReceive(r Request) {
1259         count := c.validReceiveChunks[r]
1260         if count == 1 {
1261                 delete(c.validReceiveChunks, r)
1262         } else if count > 1 {
1263                 c.validReceiveChunks[r] = count - 1
1264         } else {
1265                 panic(r)
1266         }
1267 }
1268
1269 func (c *PeerConn) onReadExtendedMsg(id pp.ExtensionNumber, payload []byte) (err error) {
1270         defer func() {
1271                 // TODO: Should we still do this?
1272                 if err != nil {
1273                         // These clients use their own extension IDs for outgoing message
1274                         // types, which is incorrect.
1275                         if bytes.HasPrefix(c.PeerID[:], []byte("-SD0100-")) || strings.HasPrefix(string(c.PeerID[:]), "-XL0012-") {
1276                                 err = nil
1277                         }
1278                 }
1279         }()
1280         t := c.t
1281         cl := t.cl
1282         switch id {
1283         case pp.HandshakeExtendedID:
1284                 var d pp.ExtendedHandshakeMessage
1285                 if err := bencode.Unmarshal(payload, &d); err != nil {
1286                         c.logger.Printf("error parsing extended handshake message %q: %s", payload, err)
1287                         return errors.Wrap(err, "unmarshalling extended handshake payload")
1288                 }
1289                 if cb := c.callbacks.ReadExtendedHandshake; cb != nil {
1290                         cb(c, &d)
1291                 }
1292                 //c.logger.WithDefaultLevel(log.Debug).Printf("received extended handshake message:\n%s", spew.Sdump(d))
1293                 if d.Reqq != 0 {
1294                         c.PeerMaxRequests = d.Reqq
1295                 }
1296                 c.PeerClientName = d.V
1297                 if c.PeerExtensionIDs == nil {
1298                         c.PeerExtensionIDs = make(map[pp.ExtensionName]pp.ExtensionNumber, len(d.M))
1299                 }
1300                 c.PeerListenPort = d.Port
1301                 c.PeerPrefersEncryption = d.Encryption
1302                 for name, id := range d.M {
1303                         if _, ok := c.PeerExtensionIDs[name]; !ok {
1304                                 torrent.Add(fmt.Sprintf("peers supporting extension %q", name), 1)
1305                         }
1306                         c.PeerExtensionIDs[name] = id
1307                 }
1308                 if d.MetadataSize != 0 {
1309                         if err = t.setMetadataSize(d.MetadataSize); err != nil {
1310                                 return errors.Wrapf(err, "setting metadata size to %d", d.MetadataSize)
1311                         }
1312                 }
1313                 c.requestPendingMetadata()
1314                 if !t.cl.config.DisablePEX {
1315                         t.pex.Add(c) // we learnt enough now
1316                         c.pex.Init(c)
1317                 }
1318                 return nil
1319         case metadataExtendedId:
1320                 err := cl.gotMetadataExtensionMsg(payload, t, c)
1321                 if err != nil {
1322                         return fmt.Errorf("handling metadata extension message: %w", err)
1323                 }
1324                 return nil
1325         case pexExtendedId:
1326                 if !c.pex.IsEnabled() {
1327                         return nil // or hang-up maybe?
1328                 }
1329                 return c.pex.Recv(payload)
1330         default:
1331                 return fmt.Errorf("unexpected extended message ID: %v", id)
1332         }
1333 }
1334
1335 // Set both the Reader and Writer for the connection from a single ReadWriter.
1336 func (cn *PeerConn) setRW(rw io.ReadWriter) {
1337         cn.r = rw
1338         cn.w = rw
1339 }
1340
1341 // Returns the Reader and Writer as a combined ReadWriter.
1342 func (cn *PeerConn) rw() io.ReadWriter {
1343         return struct {
1344                 io.Reader
1345                 io.Writer
1346         }{cn.r, cn.w}
1347 }
1348
1349 // Handle a received chunk from a peer.
1350 func (c *Peer) receiveChunk(msg *pp.Message) error {
1351         t := c.t
1352         cl := t.cl
1353         torrent.Add("chunks received", 1)
1354
1355         req := newRequestFromMessage(msg)
1356
1357         if c.peerChoking {
1358                 torrent.Add("chunks received while choking", 1)
1359         }
1360
1361         if c.validReceiveChunks[req] <= 0 {
1362                 torrent.Add("chunks received unexpected", 1)
1363                 return errors.New("received unexpected chunk")
1364         }
1365         c.decExpectedChunkReceive(req)
1366
1367         if c.peerChoking && c.peerAllowedFast.Get(int(req.Index)) {
1368                 torrent.Add("chunks received due to allowed fast", 1)
1369         }
1370
1371         // TODO: This needs to happen immediately, to prevent cancels occurring asynchronously when have
1372         // actually already received the piece, while we have the Client unlocked to write the data out.
1373         {
1374                 if _, ok := c.requests[req]; ok {
1375                         for _, f := range c.callbacks.ReceivedRequested {
1376                                 f(PeerMessageEvent{c, msg})
1377                         }
1378                 }
1379                 // Request has been satisfied.
1380                 if c.deleteRequest(req) {
1381                         if c.expectingChunks() {
1382                                 c._chunksReceivedWhileExpecting++
1383                         }
1384                 } else {
1385                         torrent.Add("chunks received unwanted", 1)
1386                 }
1387         }
1388
1389         // Do we actually want this chunk?
1390         if t.haveChunk(req) {
1391                 torrent.Add("chunks received wasted", 1)
1392                 c.allStats(add(1, func(cs *ConnStats) *Count { return &cs.ChunksReadWasted }))
1393                 return nil
1394         }
1395
1396         piece := &t.pieces[req.Index]
1397
1398         c.allStats(add(1, func(cs *ConnStats) *Count { return &cs.ChunksReadUseful }))
1399         c.allStats(add(int64(len(msg.Piece)), func(cs *ConnStats) *Count { return &cs.BytesReadUsefulData }))
1400         for _, f := range c.t.cl.config.Callbacks.ReceivedUsefulData {
1401                 f(ReceivedUsefulDataEvent{c, msg})
1402         }
1403         c.lastUsefulChunkReceived = time.Now()
1404         // if t.fastestPeer != c {
1405         // log.Printf("setting fastest connection %p", c)
1406         // }
1407         t.fastestPeer = c
1408
1409         // Need to record that it hasn't been written yet, before we attempt to do
1410         // anything with it.
1411         piece.incrementPendingWrites()
1412         // Record that we have the chunk, so we aren't trying to download it while
1413         // waiting for it to be written to storage.
1414         piece.unpendChunkIndex(chunkIndex(req.ChunkSpec, t.chunkSize))
1415
1416         // Cancel pending requests for this chunk from *other* peers.
1417         t.iterPeers(func(p *Peer) {
1418                 if p == c {
1419                         return
1420                 }
1421                 p.postCancel(req)
1422         })
1423
1424         err := func() error {
1425                 cl.unlock()
1426                 defer cl.lock()
1427                 concurrentChunkWrites.Add(1)
1428                 defer concurrentChunkWrites.Add(-1)
1429                 // Write the chunk out. Note that the upper bound on chunk writing concurrency will be the
1430                 // number of connections. We write inline with receiving the chunk (with this lock dance),
1431                 // because we want to handle errors synchronously and I haven't thought of a nice way to
1432                 // defer any concurrency to the storage and have that notify the client of errors. TODO: Do
1433                 // that instead.
1434                 return t.writeChunk(int(msg.Index), int64(msg.Begin), msg.Piece)
1435         }()
1436
1437         piece.decrementPendingWrites()
1438
1439         if err != nil {
1440                 c.logger.WithDefaultLevel(log.Error).Printf("writing received chunk %v: %v", req, err)
1441                 t.pendRequest(req)
1442                 //t.updatePieceCompletion(pieceIndex(msg.Index))
1443                 t.onWriteChunkErr(err)
1444                 return nil
1445         }
1446
1447         c.onDirtiedPiece(pieceIndex(req.Index))
1448
1449         // We need to ensure the piece is only queued once, so only the last chunk writer gets this job.
1450         if t.pieceAllDirty(pieceIndex(req.Index)) && piece.pendingWrites == 0 {
1451                 t.queuePieceCheck(pieceIndex(req.Index))
1452                 // We don't pend all chunks here anymore because we don't want code dependent on the dirty
1453                 // chunk status (such as the haveChunk call above) to have to check all the various other
1454                 // piece states like queued for hash, hashing etc. This does mean that we need to be sure
1455                 // that chunk pieces are pended at an appropriate time later however.
1456         }
1457
1458         cl.event.Broadcast()
1459         // We do this because we've written a chunk, and may change PieceState.Partial.
1460         t.publishPieceChange(pieceIndex(req.Index))
1461
1462         return nil
1463 }
1464
1465 func (c *Peer) onDirtiedPiece(piece pieceIndex) {
1466         if c.peerTouchedPieces == nil {
1467                 c.peerTouchedPieces = make(map[pieceIndex]struct{})
1468         }
1469         c.peerTouchedPieces[piece] = struct{}{}
1470         ds := &c.t.pieces[piece].dirtiers
1471         if *ds == nil {
1472                 *ds = make(map[*Peer]struct{})
1473         }
1474         (*ds)[c] = struct{}{}
1475 }
1476
1477 func (c *PeerConn) uploadAllowed() bool {
1478         if c.t.cl.config.NoUpload {
1479                 return false
1480         }
1481         if c.t.dataUploadDisallowed {
1482                 return false
1483         }
1484         if c.t.seeding() {
1485                 return true
1486         }
1487         if !c.peerHasWantedPieces() {
1488                 return false
1489         }
1490         // Don't upload more than 100 KiB more than we download.
1491         if c._stats.BytesWrittenData.Int64() >= c._stats.BytesReadData.Int64()+100<<10 {
1492                 return false
1493         }
1494         return true
1495 }
1496
1497 func (c *PeerConn) setRetryUploadTimer(delay time.Duration) {
1498         if c.uploadTimer == nil {
1499                 c.uploadTimer = time.AfterFunc(delay, c.writerCond.Broadcast)
1500         } else {
1501                 c.uploadTimer.Reset(delay)
1502         }
1503 }
1504
1505 // Also handles choking and unchoking of the remote peer.
1506 func (c *PeerConn) upload(msg func(pp.Message) bool) bool {
1507         // Breaking or completing this loop means we don't want to upload to the
1508         // peer anymore, and we choke them.
1509 another:
1510         for c.uploadAllowed() {
1511                 // We want to upload to the peer.
1512                 if !c.unchoke(msg) {
1513                         return false
1514                 }
1515                 for r, state := range c.peerRequests {
1516                         if state.data == nil {
1517                                 continue
1518                         }
1519                         res := c.t.cl.config.UploadRateLimiter.ReserveN(time.Now(), int(r.Length))
1520                         if !res.OK() {
1521                                 panic(fmt.Sprintf("upload rate limiter burst size < %d", r.Length))
1522                         }
1523                         delay := res.Delay()
1524                         if delay > 0 {
1525                                 res.Cancel()
1526                                 c.setRetryUploadTimer(delay)
1527                                 // Hard to say what to return here.
1528                                 return true
1529                         }
1530                         more := c.sendChunk(r, msg, state)
1531                         delete(c.peerRequests, r)
1532                         if !more {
1533                                 return false
1534                         }
1535                         goto another
1536                 }
1537                 return true
1538         }
1539         return c.choke(msg)
1540 }
1541
1542 func (cn *PeerConn) drop() {
1543         cn.t.dropConnection(cn)
1544 }
1545
1546 func (cn *Peer) netGoodPiecesDirtied() int64 {
1547         return cn._stats.PiecesDirtiedGood.Int64() - cn._stats.PiecesDirtiedBad.Int64()
1548 }
1549
1550 func (c *Peer) peerHasWantedPieces() bool {
1551         return !c._pieceRequestOrder.IsEmpty()
1552 }
1553
1554 func (c *Peer) numLocalRequests() int {
1555         return len(c.requests)
1556 }
1557
1558 func (c *Peer) deleteRequest(r Request) bool {
1559         if _, ok := c.requests[r]; !ok {
1560                 return false
1561         }
1562         delete(c.requests, r)
1563         for _, f := range c.callbacks.DeletedRequest {
1564                 f(PeerRequestEvent{c, r})
1565         }
1566         c.updateExpectingChunks()
1567         c.t.requestStrategy.hooks().deletedRequest(r)
1568         pr := c.t.pendingRequests
1569         pr[r]--
1570         n := pr[r]
1571         if n == 0 {
1572                 delete(pr, r)
1573         }
1574         if n < 0 {
1575                 panic(n)
1576         }
1577         // If a request fails, updating the requests for the current peer first may miss the opportunity
1578         // to try other peers for that request instead, depending on the request strategy. This might
1579         // only affect webseed peers though, since they synchronously issue new requests: PeerConns do
1580         // it in the writer routine.
1581         const updateCurrentConnRequestsFirst = false
1582         if updateCurrentConnRequestsFirst {
1583                 c.updateRequests()
1584         }
1585         // Give other conns a chance to pick up the request.
1586         c.t.iterPeers(func(_c *Peer) {
1587                 // We previously checked that the peer wasn't interested to to only wake connections that
1588                 // were unable to issue requests due to starvation by the request strategy. There could be
1589                 // performance ramifications.
1590                 if _c != c && c.peerHasPiece(pieceIndex(r.Index)) {
1591                         _c.updateRequests()
1592                 }
1593         })
1594         if !updateCurrentConnRequestsFirst {
1595                 c.updateRequests()
1596         }
1597         return true
1598 }
1599
1600 func (c *Peer) deleteAllRequests() {
1601         for r := range c.requests {
1602                 c.deleteRequest(r)
1603         }
1604         if len(c.requests) != 0 {
1605                 panic(len(c.requests))
1606         }
1607         // for c := range c.t.conns {
1608         //      c.tickleWriter()
1609         // }
1610 }
1611
1612 // This is called when something has changed that should wake the writer, such as putting stuff into
1613 // the writeBuffer, or changing some state that the writer can act on.
1614 func (c *PeerConn) tickleWriter() {
1615         c.writerCond.Broadcast()
1616 }
1617
1618 func (c *Peer) postCancel(r Request) bool {
1619         if !c.deleteRequest(r) {
1620                 return false
1621         }
1622         c.peerImpl._postCancel(r)
1623         return true
1624 }
1625
1626 func (c *PeerConn) _postCancel(r Request) {
1627         c.post(makeCancelMessage(r))
1628 }
1629
1630 func (c *PeerConn) sendChunk(r Request, msg func(pp.Message) bool, state *peerRequestState) (more bool) {
1631         c.lastChunkSent = time.Now()
1632         return msg(pp.Message{
1633                 Type:  pp.Piece,
1634                 Index: r.Index,
1635                 Begin: r.Begin,
1636                 Piece: state.data,
1637         })
1638 }
1639
1640 func (c *PeerConn) setTorrent(t *Torrent) {
1641         if c.t != nil {
1642                 panic("connection already associated with a torrent")
1643         }
1644         c.t = t
1645         c.logger.WithDefaultLevel(log.Debug).Printf("set torrent=%v", t)
1646         t.reconcileHandshakeStats(c)
1647 }
1648
1649 func (c *Peer) peerPriority() (peerPriority, error) {
1650         return bep40Priority(c.remoteIpPort(), c.t.cl.publicAddr(c.remoteIp()))
1651 }
1652
1653 func (c *Peer) remoteIp() net.IP {
1654         host, _, _ := net.SplitHostPort(c.RemoteAddr.String())
1655         return net.ParseIP(host)
1656 }
1657
1658 func (c *Peer) remoteIpPort() IpPort {
1659         ipa, _ := tryIpPortFromNetAddr(c.RemoteAddr)
1660         return IpPort{ipa.IP, uint16(ipa.Port)}
1661 }
1662
1663 func (c *PeerConn) pexPeerFlags() pp.PexPeerFlags {
1664         f := pp.PexPeerFlags(0)
1665         if c.PeerPrefersEncryption {
1666                 f |= pp.PexPrefersEncryption
1667         }
1668         if c.outgoing {
1669                 f |= pp.PexOutgoingConn
1670         }
1671         if c.utp() {
1672                 f |= pp.PexSupportsUtp
1673         }
1674         return f
1675 }
1676
1677 // This returns the address to use if we want to dial the peer again. It incorporates the peer's
1678 // advertised listen port.
1679 func (c *PeerConn) dialAddr() PeerRemoteAddr {
1680         if !c.outgoing && c.PeerListenPort != 0 {
1681                 switch addr := c.RemoteAddr.(type) {
1682                 case *net.TCPAddr:
1683                         dialAddr := *addr
1684                         dialAddr.Port = c.PeerListenPort
1685                         return &dialAddr
1686                 case *net.UDPAddr:
1687                         dialAddr := *addr
1688                         dialAddr.Port = c.PeerListenPort
1689                         return &dialAddr
1690                 }
1691         }
1692         return c.RemoteAddr
1693 }
1694
1695 func (c *PeerConn) pexEvent(t pexEventType) pexEvent {
1696         f := c.pexPeerFlags()
1697         addr := c.dialAddr()
1698         return pexEvent{t, addr, f}
1699 }
1700
1701 func (c *PeerConn) String() string {
1702         return fmt.Sprintf("connection %p", c)
1703 }
1704
1705 func (c *Peer) trust() connectionTrust {
1706         return connectionTrust{c.trusted, c.netGoodPiecesDirtied()}
1707 }
1708
1709 type connectionTrust struct {
1710         Implicit            bool
1711         NetGoodPiecesDirted int64
1712 }
1713
1714 func (l connectionTrust) Less(r connectionTrust) bool {
1715         return multiless.New().Bool(l.Implicit, r.Implicit).Int64(l.NetGoodPiecesDirted, r.NetGoodPiecesDirted).Less()
1716 }
1717
1718 func (cn *Peer) requestStrategyConnection() requestStrategyConnection {
1719         return cn
1720 }
1721
1722 func (cn *Peer) chunksReceivedWhileExpecting() int64 {
1723         return cn._chunksReceivedWhileExpecting
1724 }
1725
1726 func (cn *Peer) fastest() bool {
1727         return cn == cn.t.fastestPeer
1728 }
1729
1730 func (cn *Peer) peerMaxRequests() int {
1731         return cn.PeerMaxRequests
1732 }
1733
1734 // Returns the pieces the peer has claimed to have.
1735 func (cn *PeerConn) PeerPieces() bitmap.Bitmap {
1736         cn.locker().RLock()
1737         defer cn.locker().RUnlock()
1738         return cn.peerPieces()
1739 }
1740
1741 func (cn *Peer) peerPieces() bitmap.Bitmap {
1742         ret := cn._peerPieces.Copy()
1743         if cn.peerSentHaveAll {
1744                 ret.AddRange(0, cn.t.numPieces())
1745         }
1746         return ret
1747 }
1748
1749 func (cn *Peer) pieceRequestOrder() *prioritybitmap.PriorityBitmap {
1750         return &cn._pieceRequestOrder
1751 }
1752
1753 func (cn *Peer) stats() *ConnStats {
1754         return &cn._stats
1755 }
1756
1757 func (cn *Peer) torrent() requestStrategyTorrent {
1758         return cn.t.requestStrategyTorrent()
1759 }
1760
1761 func (p *Peer) TryAsPeerConn() (*PeerConn, bool) {
1762         pc, ok := p.peerImpl.(*PeerConn)
1763         return pc, ok
1764 }