peerconn.go

   1 package torrent
   2
   3 import (
   4         "bufio"
   5         "bytes"
   6         "context"
   7         "errors"
   8         "fmt"
   9         "io"
  10         "math/rand"
  11         "net"
  12         "strconv"
  13         "strings"
  14         "time"
  15
  16         "github.com/RoaringBitmap/roaring"
  17         . "github.com/anacrolix/generics"
  18         "github.com/anacrolix/log"
  19         "github.com/anacrolix/missinggo/v2/bitmap"
  20         "github.com/anacrolix/multiless"
  21         "golang.org/x/time/rate"
  22
  23         "github.com/anacrolix/torrent/bencode"
  24         "github.com/anacrolix/torrent/internal/alloclim"
  25         "github.com/anacrolix/torrent/metainfo"
  26         "github.com/anacrolix/torrent/mse"
  27         pp "github.com/anacrolix/torrent/peer_protocol"
  28 )
  29
  30 // Maintains the state of a BitTorrent-protocol based connection with a peer.
  31 type PeerConn struct {
  32         Peer
  33
  34         // A string that should identify the PeerConn's net.Conn endpoints. The net.Conn could
  35         // be wrapping WebRTC, uTP, or TCP etc. Used in writing the conn status for peers.
  36         connString string
  37
  38         // See BEP 3 etc.
  39         PeerID             PeerID
  40         PeerExtensionBytes pp.PeerExtensionBits
  41
  42         // The actual Conn, used for closing, and setting socket options. Do not use methods on this
  43         // while holding any mutexes.
  44         conn net.Conn
  45         // The Reader and Writer for this Conn, with hooks installed for stats,
  46         // limiting, deadlines etc.
  47         w io.Writer
  48         r io.Reader
  49
  50         messageWriter peerConnMsgWriter
  51
  52         uploadTimer *time.Timer
  53         pex         pexConnState
  54
  55         // The pieces the peer has claimed to have.
  56         _peerPieces roaring.Bitmap
  57         // The peer has everything. This can occur due to a special message, when
  58         // we may not even know the number of pieces in the torrent yet.
  59         peerSentHaveAll bool
  60
  61         peerRequestDataAllocLimiter alloclim.Limiter
  62 }
  63
  64 func (cn *PeerConn) peerImplStatusLines() []string {
  65         return []string{fmt.Sprintf("%+-55q %s %s", cn.PeerID, cn.PeerExtensionBytes, cn.connString)}
  66 }
  67
  68 // Returns true if the connection is over IPv6.
  69 func (cn *PeerConn) ipv6() bool {
  70         ip := cn.remoteIp()
  71         if ip.To4() != nil {
  72                 return false
  73         }
  74         return len(ip) == net.IPv6len
  75 }
  76
  77 // Returns true the if the dialer/initiator has the lower client peer ID. TODO: Find the
  78 // specification for this.
  79 func (cn *PeerConn) isPreferredDirection() bool {
  80         return bytes.Compare(cn.t.cl.peerID[:], cn.PeerID[:]) < 0 == cn.outgoing
  81 }
  82
  83 // Returns whether the left connection should be preferred over the right one,
  84 // considering only their networking properties. If ok is false, we can't
  85 // decide.
  86 func (l *PeerConn) hasPreferredNetworkOver(r *PeerConn) bool {
  87         var ml multiless.Computation
  88         ml = ml.Bool(r.isPreferredDirection(), l.isPreferredDirection())
  89         ml = ml.Bool(l.utp(), r.utp())
  90         ml = ml.Bool(r.ipv6(), l.ipv6())
  91         return ml.Less()
  92 }
  93
  94 func (cn *PeerConn) peerHasAllPieces() (all, known bool) {
  95         if cn.peerSentHaveAll {
  96                 return true, true
  97         }
  98         if !cn.t.haveInfo() {
  99                 return false, false
 100         }
 101         return cn._peerPieces.GetCardinality() == uint64(cn.t.numPieces()), true
 102 }
 103
 104 func (cn *PeerConn) onGotInfo(info *metainfo.Info) {
 105         cn.setNumPieces(info.NumPieces())
 106 }
 107
 108 // Correct the PeerPieces slice length. Return false if the existing slice is invalid, such as by
 109 // receiving badly sized BITFIELD, or invalid HAVE messages.
 110 func (cn *PeerConn) setNumPieces(num pieceIndex) {
 111         cn._peerPieces.RemoveRange(bitmap.BitRange(num), bitmap.ToEnd)
 112         cn.peerPiecesChanged()
 113 }
 114
 115 func (cn *PeerConn) peerPieces() *roaring.Bitmap {
 116         return &cn._peerPieces
 117 }
 118
 119 func (cn *PeerConn) connectionFlags() (ret string) {
 120         c := func(b byte) {
 121                 ret += string([]byte{b})
 122         }
 123         if cn.cryptoMethod == mse.CryptoMethodRC4 {
 124                 c('E')
 125         } else if cn.headerEncrypted {
 126                 c('e')
 127         }
 128         ret += string(cn.Discovery)
 129         if cn.utp() {
 130                 c('U')
 131         }
 132         return
 133 }
 134
 135 func (cn *PeerConn) utp() bool {
 136         return parseNetworkString(cn.Network).Udp
 137 }
 138
 139 func (cn *PeerConn) onClose() {
 140         if cn.pex.IsEnabled() {
 141                 cn.pex.Close()
 142         }
 143         cn.tickleWriter()
 144         if cn.conn != nil {
 145                 go cn.conn.Close()
 146         }
 147         if cb := cn.callbacks.PeerConnClosed; cb != nil {
 148                 cb(cn)
 149         }
 150 }
 151
 152 // Writes a message into the write buffer. Returns whether it's okay to keep writing. Writing is
 153 // done asynchronously, so it may be that we're not able to honour backpressure from this method.
 154 func (cn *PeerConn) write(msg pp.Message) bool {
 155         torrent.Add(fmt.Sprintf("messages written of type %s", msg.Type.String()), 1)
 156         // We don't need to track bytes here because the connection's Writer has that behaviour injected
 157         // (although there's some delay between us buffering the message, and the connection writer
 158         // flushing it out.).
 159         notFull := cn.messageWriter.write(msg)
 160         // Last I checked only Piece messages affect stats, and we don't write those.
 161         cn.wroteMsg(&msg)
 162         cn.tickleWriter()
 163         return notFull
 164 }
 165
 166 func (cn *PeerConn) requestMetadataPiece(index int) {
 167         eID := cn.PeerExtensionIDs[pp.ExtensionNameMetadata]
 168         if eID == pp.ExtensionDeleteNumber {
 169                 return
 170         }
 171         if index < len(cn.metadataRequests) && cn.metadataRequests[index] {
 172                 return
 173         }
 174         cn.logger.WithDefaultLevel(log.Debug).Printf("requesting metadata piece %d", index)
 175         cn.write(pp.MetadataExtensionRequestMsg(eID, index))
 176         for index >= len(cn.metadataRequests) {
 177                 cn.metadataRequests = append(cn.metadataRequests, false)
 178         }
 179         cn.metadataRequests[index] = true
 180 }
 181
 182 func (cn *PeerConn) requestedMetadataPiece(index int) bool {
 183         return index < len(cn.metadataRequests) && cn.metadataRequests[index]
 184 }
 185
 186 func (cn *PeerConn) onPeerSentCancel(r Request) {
 187         if _, ok := cn.peerRequests[r]; !ok {
 188                 torrent.Add("unexpected cancels received", 1)
 189                 return
 190         }
 191         if cn.fastEnabled() {
 192                 cn.reject(r)
 193         } else {
 194                 delete(cn.peerRequests, r)
 195         }
 196 }
 197
 198 func (cn *PeerConn) choke(msg messageWriter) (more bool) {
 199         if cn.choking {
 200                 return true
 201         }
 202         cn.choking = true
 203         more = msg(pp.Message{
 204                 Type: pp.Choke,
 205         })
 206         if !cn.fastEnabled() {
 207                 cn.deleteAllPeerRequests()
 208         }
 209         return
 210 }
 211
 212 func (cn *PeerConn) deleteAllPeerRequests() {
 213         for _, state := range cn.peerRequests {
 214                 state.allocReservation.Drop()
 215         }
 216         cn.peerRequests = nil
 217 }
 218
 219 func (cn *PeerConn) unchoke(msg func(pp.Message) bool) bool {
 220         if !cn.choking {
 221                 return true
 222         }
 223         cn.choking = false
 224         return msg(pp.Message{
 225                 Type: pp.Unchoke,
 226         })
 227 }
 228
 229 func (pc *PeerConn) writeInterested(interested bool) bool {
 230         return pc.write(pp.Message{
 231                 Type: func() pp.MessageType {
 232                         if interested {
 233                                 return pp.Interested
 234                         } else {
 235                                 return pp.NotInterested
 236                         }
 237                 }(),
 238         })
 239 }
 240
 241 func (me *PeerConn) _request(r Request) bool {
 242         return me.write(pp.Message{
 243                 Type:   pp.Request,
 244                 Index:  r.Index,
 245                 Begin:  r.Begin,
 246                 Length: r.Length,
 247         })
 248 }
 249
 250 func (me *PeerConn) _cancel(r RequestIndex) bool {
 251         me.write(makeCancelMessage(me.t.requestIndexToRequest(r)))
 252         // Transmission does not send rejects for received cancels. See
 253         // https://github.com/transmission/transmission/pull/2275.
 254         return me.fastEnabled() && !me.remoteIsTransmission()
 255 }
 256
 257 func (cn *PeerConn) fillWriteBuffer() {
 258         if cn.messageWriter.writeBuffer.Len() > writeBufferLowWaterLen {
 259                 // Fully committing to our max requests requires sufficient space (see
 260                 // maxLocalToRemoteRequests). Flush what we have instead. We also prefer always to make
 261                 // requests than to do PEX or upload, so we short-circuit before handling those. Any update
 262                 // request reason will not be cleared, so we'll come right back here when there's space. We
 263                 // can't do this in maybeUpdateActualRequestState because it's a method on Peer and has no
 264                 // knowledge of write buffers.
 265         }
 266         cn.maybeUpdateActualRequestState()
 267         if cn.pex.IsEnabled() {
 268                 if flow := cn.pex.Share(cn.write); !flow {
 269                         return
 270                 }
 271         }
 272         cn.upload(cn.write)
 273 }
 274
 275 func (cn *PeerConn) have(piece pieceIndex) {
 276         if cn.sentHaves.Get(bitmap.BitIndex(piece)) {
 277                 return
 278         }
 279         cn.write(pp.Message{
 280                 Type:  pp.Have,
 281                 Index: pp.Integer(piece),
 282         })
 283         cn.sentHaves.Add(bitmap.BitIndex(piece))
 284 }
 285
 286 func (cn *PeerConn) postBitfield() {
 287         if cn.sentHaves.Len() != 0 {
 288                 panic("bitfield must be first have-related message sent")
 289         }
 290         if !cn.t.haveAnyPieces() {
 291                 return
 292         }
 293         cn.write(pp.Message{
 294                 Type:     pp.Bitfield,
 295                 Bitfield: cn.t.bitfield(),
 296         })
 297         cn.sentHaves = bitmap.Bitmap{cn.t._completedPieces.Clone()}
 298 }
 299
 300 func (cn *PeerConn) handleUpdateRequests() {
 301         // The writer determines the request state as needed when it can write.
 302         cn.tickleWriter()
 303 }
 304
 305 func (cn *PeerConn) raisePeerMinPieces(newMin pieceIndex) {
 306         if newMin > cn.peerMinPieces {
 307                 cn.peerMinPieces = newMin
 308         }
 309 }
 310
 311 func (cn *PeerConn) peerSentHave(piece pieceIndex) error {
 312         if cn.t.haveInfo() && piece >= cn.t.numPieces() || piece < 0 {
 313                 return errors.New("invalid piece")
 314         }
 315         if cn.peerHasPiece(piece) {
 316                 return nil
 317         }
 318         cn.raisePeerMinPieces(piece + 1)
 319         if !cn.peerHasPiece(piece) {
 320                 cn.t.incPieceAvailability(piece)
 321         }
 322         cn._peerPieces.Add(uint32(piece))
 323         if cn.t.wantPieceIndex(piece) {
 324                 cn.updateRequests("have")
 325         }
 326         cn.peerPiecesChanged()
 327         return nil
 328 }
 329
 330 func (cn *PeerConn) peerSentBitfield(bf []bool) error {
 331         if len(bf)%8 != 0 {
 332                 panic("expected bitfield length divisible by 8")
 333         }
 334         // We know that the last byte means that at most the last 7 bits are wasted.
 335         cn.raisePeerMinPieces(pieceIndex(len(bf) - 7))
 336         if cn.t.haveInfo() && len(bf) > int(cn.t.numPieces()) {
 337                 // Ignore known excess pieces.
 338                 bf = bf[:cn.t.numPieces()]
 339         }
 340         bm := boolSliceToBitmap(bf)
 341         if cn.t.haveInfo() && pieceIndex(bm.GetCardinality()) == cn.t.numPieces() {
 342                 cn.onPeerHasAllPieces()
 343                 return nil
 344         }
 345         if !bm.IsEmpty() {
 346                 cn.raisePeerMinPieces(pieceIndex(bm.Maximum()) + 1)
 347         }
 348         shouldUpdateRequests := false
 349         if cn.peerSentHaveAll {
 350                 if !cn.t.deleteConnWithAllPieces(&cn.Peer) {
 351                         panic(cn)
 352                 }
 353                 cn.peerSentHaveAll = false
 354                 if !cn._peerPieces.IsEmpty() {
 355                         panic("if peer has all, we expect no individual peer pieces to be set")
 356                 }
 357         } else {
 358                 bm.Xor(&cn._peerPieces)
 359         }
 360         cn.peerSentHaveAll = false
 361         // bm is now 'on' for pieces that are changing
 362         bm.Iterate(func(x uint32) bool {
 363                 pi := pieceIndex(x)
 364                 if cn._peerPieces.Contains(x) {
 365                         // Then we must be losing this piece
 366                         cn.t.decPieceAvailability(pi)
 367                 } else {
 368                         if !shouldUpdateRequests && cn.t.wantPieceIndex(pieceIndex(x)) {
 369                                 shouldUpdateRequests = true
 370                         }
 371                         // We must be gaining this piece
 372                         cn.t.incPieceAvailability(pieceIndex(x))
 373                 }
 374                 return true
 375         })
 376         // Apply the changes. If we had everything previously, this should be empty, so xor is the same
 377         // as or.
 378         cn._peerPieces.Xor(&bm)
 379         if shouldUpdateRequests {
 380                 cn.updateRequests("bitfield")
 381         }
 382         // We didn't guard this before, I see no reason to do it now.
 383         cn.peerPiecesChanged()
 384         return nil
 385 }
 386
 387 func (cn *PeerConn) onPeerHasAllPieces() {
 388         t := cn.t
 389         if t.haveInfo() {
 390                 cn._peerPieces.Iterate(func(x uint32) bool {
 391                         t.decPieceAvailability(pieceIndex(x))
 392                         return true
 393                 })
 394         }
 395         t.addConnWithAllPieces(&cn.Peer)
 396         cn.peerSentHaveAll = true
 397         cn._peerPieces.Clear()
 398         if !cn.t._pendingPieces.IsEmpty() {
 399                 cn.updateRequests("Peer.onPeerHasAllPieces")
 400         }
 401         cn.peerPiecesChanged()
 402 }
 403
 404 func (cn *PeerConn) onPeerSentHaveAll() error {
 405         cn.onPeerHasAllPieces()
 406         return nil
 407 }
 408
 409 func (cn *PeerConn) peerSentHaveNone() error {
 410         if !cn.peerSentHaveAll {
 411                 cn.t.decPeerPieceAvailability(&cn.Peer)
 412         }
 413         cn._peerPieces.Clear()
 414         cn.peerSentHaveAll = false
 415         cn.peerPiecesChanged()
 416         return nil
 417 }
 418
 419 func (c *PeerConn) requestPendingMetadata() {
 420         if c.t.haveInfo() {
 421                 return
 422         }
 423         if c.PeerExtensionIDs[pp.ExtensionNameMetadata] == 0 {
 424                 // Peer doesn't support this.
 425                 return
 426         }
 427         // Request metadata pieces that we don't have in a random order.
 428         var pending []int
 429         for index := 0; index < c.t.metadataPieceCount(); index++ {
 430                 if !c.t.haveMetadataPiece(index) && !c.requestedMetadataPiece(index) {
 431                         pending = append(pending, index)
 432                 }
 433         }
 434         rand.Shuffle(len(pending), func(i, j int) { pending[i], pending[j] = pending[j], pending[i] })
 435         for _, i := range pending {
 436                 c.requestMetadataPiece(i)
 437         }
 438 }
 439
 440 func (cn *PeerConn) wroteMsg(msg *pp.Message) {
 441         torrent.Add(fmt.Sprintf("messages written of type %s", msg.Type.String()), 1)
 442         if msg.Type == pp.Extended {
 443                 for name, id := range cn.PeerExtensionIDs {
 444                         if id != msg.ExtendedID {
 445                                 continue
 446                         }
 447                         torrent.Add(fmt.Sprintf("Extended messages written for protocol %q", name), 1)
 448                 }
 449         }
 450         cn.allStats(func(cs *ConnStats) { cs.wroteMsg(msg) })
 451 }
 452
 453 func (cn *PeerConn) wroteBytes(n int64) {
 454         cn.allStats(add(n, func(cs *ConnStats) *Count { return &cs.BytesWritten }))
 455 }
 456
 457 func (c *PeerConn) fastEnabled() bool {
 458         return c.PeerExtensionBytes.SupportsFast() && c.t.cl.config.Extensions.SupportsFast()
 459 }
 460
 461 func (c *PeerConn) reject(r Request) {
 462         if !c.fastEnabled() {
 463                 panic("fast not enabled")
 464         }
 465         c.write(r.ToMsg(pp.Reject))
 466         // It is possible to reject a request before it is added to peer requests due to being invalid.
 467         if state, ok := c.peerRequests[r]; ok {
 468                 state.allocReservation.Drop()
 469                 delete(c.peerRequests, r)
 470         }
 471 }
 472
 473 func (c *PeerConn) maximumPeerRequestChunkLength() (_ Option[int]) {
 474         uploadRateLimiter := c.t.cl.config.UploadRateLimiter
 475         if uploadRateLimiter.Limit() == rate.Inf {
 476                 return
 477         }
 478         return Some(uploadRateLimiter.Burst())
 479 }
 480
 481 // startFetch is for testing purposes currently.
 482 func (c *PeerConn) onReadRequest(r Request, startFetch bool) error {
 483         requestedChunkLengths.Add(strconv.FormatUint(r.Length.Uint64(), 10), 1)
 484         if _, ok := c.peerRequests[r]; ok {
 485                 torrent.Add("duplicate requests received", 1)
 486                 if c.fastEnabled() {
 487                         return errors.New("received duplicate request with fast enabled")
 488                 }
 489                 return nil
 490         }
 491         if c.choking {
 492                 torrent.Add("requests received while choking", 1)
 493                 if c.fastEnabled() {
 494                         torrent.Add("requests rejected while choking", 1)
 495                         c.reject(r)
 496                 }
 497                 return nil
 498         }
 499         // TODO: What if they've already requested this?
 500         if len(c.peerRequests) >= localClientReqq {
 501                 torrent.Add("requests received while queue full", 1)
 502                 if c.fastEnabled() {
 503                         c.reject(r)
 504                 }
 505                 // BEP 6 says we may close here if we choose.
 506                 return nil
 507         }
 508         if opt := c.maximumPeerRequestChunkLength(); opt.Ok && int(r.Length) > opt.Value {
 509                 err := fmt.Errorf("peer requested chunk too long (%v)", r.Length)
 510                 c.logger.Levelf(log.Warning, err.Error())
 511                 if c.fastEnabled() {
 512                         c.reject(r)
 513                         return nil
 514                 } else {
 515                         return err
 516                 }
 517         }
 518         if !c.t.havePiece(pieceIndex(r.Index)) {
 519                 // TODO: Tell the peer we don't have the piece, and reject this request.
 520                 requestsReceivedForMissingPieces.Add(1)
 521                 return fmt.Errorf("peer requested piece we don't have: %v", r.Index.Int())
 522         }
 523         pieceLength := c.t.pieceLength(pieceIndex(r.Index))
 524         // Check this after we know we have the piece, so that the piece length will be known.
 525         if chunkOverflowsPiece(r.ChunkSpec, pieceLength) {
 526                 torrent.Add("bad requests received", 1)
 527                 return errors.New("chunk overflows piece")
 528         }
 529         if c.peerRequests == nil {
 530                 c.peerRequests = make(map[Request]*peerRequestState, localClientReqq)
 531         }
 532         value := &peerRequestState{
 533                 allocReservation: c.peerRequestDataAllocLimiter.Reserve(int64(r.Length)),
 534         }
 535         c.peerRequests[r] = value
 536         if startFetch {
 537                 // TODO: Limit peer request data read concurrency.
 538                 go c.peerRequestDataReader(r, value)
 539         }
 540         return nil
 541 }
 542
 543 func (c *PeerConn) peerRequestDataReader(r Request, prs *peerRequestState) {
 544         b, err := c.readPeerRequestData(r, prs)
 545         c.locker().Lock()
 546         defer c.locker().Unlock()
 547         if err != nil {
 548                 c.peerRequestDataReadFailed(err, r)
 549         } else {
 550                 if b == nil {
 551                         panic("data must be non-nil to trigger send")
 552                 }
 553                 torrent.Add("peer request data read successes", 1)
 554                 prs.data = b
 555                 // This might be required for the error case too (#752 and #753).
 556                 c.tickleWriter()
 557         }
 558 }
 559
 560 // If this is maintained correctly, we might be able to support optional synchronous reading for
 561 // chunk sending, the way it used to work.
 562 func (c *PeerConn) peerRequestDataReadFailed(err error, r Request) {
 563         torrent.Add("peer request data read failures", 1)
 564         logLevel := log.Warning
 565         if c.t.hasStorageCap() {
 566                 // It's expected that pieces might drop. See
 567                 // https://github.com/anacrolix/torrent/issues/702#issuecomment-1000953313.
 568                 logLevel = log.Debug
 569         }
 570         c.logger.WithDefaultLevel(logLevel).Printf("error reading chunk for peer Request %v: %v", r, err)
 571         if c.t.closed.IsSet() {
 572                 return
 573         }
 574         i := pieceIndex(r.Index)
 575         if c.t.pieceComplete(i) {
 576                 // There used to be more code here that just duplicated the following break. Piece
 577                 // completions are currently cached, so I'm not sure how helpful this update is, except to
 578                 // pull any completion changes pushed to the storage backend in failed reads that got us
 579                 // here.
 580                 c.t.updatePieceCompletion(i)
 581         }
 582         // We've probably dropped a piece from storage, but there's no way to communicate this to the
 583         // peer. If they ask for it again, we kick them allowing us to send them updated piece states if
 584         // we reconnect. TODO: Instead, we could just try to update them with Bitfield or HaveNone and
 585         // if they kick us for breaking protocol, on reconnect we will be compliant again (at least
 586         // initially).
 587         if c.fastEnabled() {
 588                 c.reject(r)
 589         } else {
 590                 if c.choking {
 591                         // If fast isn't enabled, I think we would have wiped all peer requests when we last
 592                         // choked, and requests while we're choking would be ignored. It could be possible that
 593                         // a peer request data read completed concurrently to it being deleted elsewhere.
 594                         c.logger.WithDefaultLevel(log.Warning).Printf("already choking peer, requests might not be rejected correctly")
 595                 }
 596                 // Choking a non-fast peer should cause them to flush all their requests.
 597                 c.choke(c.write)
 598         }
 599 }
 600
 601 func (c *PeerConn) readPeerRequestData(r Request, prs *peerRequestState) ([]byte, error) {
 602         // Should we depend on Torrent closure here? I think it's okay to get cancelled from elsewhere,
 603         // or fail to read and then cleanup.
 604         ctx := context.Background()
 605         err := prs.allocReservation.Wait(ctx)
 606         if err != nil {
 607                 if ctx.Err() == nil {
 608                         // The error is from the reservation itself. Something is very broken, or we're not
 609                         // guarding against excessively large requests.
 610                         err = log.WithLevel(log.Critical, err)
 611                 }
 612                 err = fmt.Errorf("waiting for alloc limit reservation: %w", err)
 613                 return nil, err
 614         }
 615         b := make([]byte, r.Length)
 616         p := c.t.info.Piece(int(r.Index))
 617         n, err := c.t.readAt(b, p.Offset()+int64(r.Begin))
 618         if n == len(b) {
 619                 if err == io.EOF {
 620                         err = nil
 621                 }
 622         } else {
 623                 if err == nil {
 624                         panic("expected error")
 625                 }
 626         }
 627         return b, err
 628 }
 629
 630 func (c *PeerConn) logProtocolBehaviour(level log.Level, format string, arg ...interface{}) {
 631         c.logger.WithContextText(fmt.Sprintf(
 632                 "peer id %q, ext v %q", c.PeerID, c.PeerClientName.Load(),
 633         )).SkipCallers(1).Levelf(level, format, arg...)
 634 }
 635
 636 // Processes incoming BitTorrent wire-protocol messages. The client lock is held upon entry and
 637 // exit. Returning will end the connection.
 638 func (c *PeerConn) mainReadLoop() (err error) {
 639         defer func() {
 640                 if err != nil {
 641                         torrent.Add("connection.mainReadLoop returned with error", 1)
 642                 } else {
 643                         torrent.Add("connection.mainReadLoop returned with no error", 1)
 644                 }
 645         }()
 646         t := c.t
 647         cl := t.cl
 648
 649         decoder := pp.Decoder{
 650                 R:         bufio.NewReaderSize(c.r, 1<<17),
 651                 MaxLength: 4 * pp.Integer(max(int64(t.chunkSize), defaultChunkSize)),
 652                 Pool:      &t.chunkPool,
 653         }
 654         for {
 655                 var msg pp.Message
 656                 func() {
 657                         cl.unlock()
 658                         defer cl.lock()
 659                         err = decoder.Decode(&msg)
 660                 }()
 661                 if cb := c.callbacks.ReadMessage; cb != nil && err == nil {
 662                         cb(c, &msg)
 663                 }
 664                 if t.closed.IsSet() || c.closed.IsSet() {
 665                         return nil
 666                 }
 667                 if err != nil {
 668                         return err
 669                 }
 670                 c.lastMessageReceived = time.Now()
 671                 if msg.Keepalive {
 672                         receivedKeepalives.Add(1)
 673                         continue
 674                 }
 675                 messageTypesReceived.Add(msg.Type.String(), 1)
 676                 if msg.Type.FastExtension() && !c.fastEnabled() {
 677                         runSafeExtraneous(func() { torrent.Add("fast messages received when extension is disabled", 1) })
 678                         return fmt.Errorf("received fast extension message (type=%v) but extension is disabled", msg.Type)
 679                 }
 680                 switch msg.Type {
 681                 case pp.Choke:
 682                         if c.peerChoking {
 683                                 break
 684                         }
 685                         if !c.fastEnabled() {
 686                                 c.deleteAllRequests("choked by non-fast PeerConn")
 687                         } else {
 688                                 // We don't decrement pending requests here, let's wait for the peer to either
 689                                 // reject or satisfy the outstanding requests. Additionally, some peers may unchoke
 690                                 // us and resume where they left off, we don't want to have piled on to those chunks
 691                                 // in the meanwhile. I think a peer's ability to abuse this should be limited: they
 692                                 // could let us request a lot of stuff, then choke us and never reject, but they're
 693                                 // only a single peer, our chunk balancing should smooth over this abuse.
 694                         }
 695                         c.peerChoking = true
 696                         c.updateExpectingChunks()
 697                 case pp.Unchoke:
 698                         if !c.peerChoking {
 699                                 // Some clients do this for some reason. Transmission doesn't error on this, so we
 700                                 // won't for consistency.
 701                                 c.logProtocolBehaviour(log.Debug, "received unchoke when already unchoked")
 702                                 break
 703                         }
 704                         c.peerChoking = false
 705                         preservedCount := 0
 706                         c.requestState.Requests.Iterate(func(x RequestIndex) bool {
 707                                 if !c.peerAllowedFast.Contains(c.t.pieceIndexOfRequestIndex(x)) {
 708                                         preservedCount++
 709                                 }
 710                                 return true
 711                         })
 712                         if preservedCount != 0 {
 713                                 // TODO: Yes this is a debug log but I'm not happy with the state of the logging lib
 714                                 // right now.
 715                                 c.logger.Levelf(log.Debug,
 716                                         "%v requests were preserved while being choked (fast=%v)",
 717                                         preservedCount,
 718                                         c.fastEnabled())
 719
 720                                 torrent.Add("requestsPreservedThroughChoking", int64(preservedCount))
 721                         }
 722                         if !c.t._pendingPieces.IsEmpty() {
 723                                 c.updateRequests("unchoked")
 724                         }
 725                         c.updateExpectingChunks()
 726                 case pp.Interested:
 727                         c.peerInterested = true
 728                         c.tickleWriter()
 729                 case pp.NotInterested:
 730                         c.peerInterested = false
 731                         // We don't clear their requests since it isn't clear in the spec.
 732                         // We'll probably choke them for this, which will clear them if
 733                         // appropriate, and is clearly specified.
 734                 case pp.Have:
 735                         err = c.peerSentHave(pieceIndex(msg.Index))
 736                 case pp.Bitfield:
 737                         err = c.peerSentBitfield(msg.Bitfield)
 738                 case pp.Request:
 739                         r := newRequestFromMessage(&msg)
 740                         err = c.onReadRequest(r, true)
 741                         if err != nil {
 742                                 err = fmt.Errorf("on reading request %v: %w", r, err)
 743                         }
 744                 case pp.Piece:
 745                         c.doChunkReadStats(int64(len(msg.Piece)))
 746                         err = c.receiveChunk(&msg)
 747                         if len(msg.Piece) == int(t.chunkSize) {
 748                                 t.chunkPool.Put(&msg.Piece)
 749                         }
 750                         if err != nil {
 751                                 err = fmt.Errorf("receiving chunk: %w", err)
 752                         }
 753                 case pp.Cancel:
 754                         req := newRequestFromMessage(&msg)
 755                         c.onPeerSentCancel(req)
 756                 case pp.Port:
 757                         ipa, ok := tryIpPortFromNetAddr(c.RemoteAddr)
 758                         if !ok {
 759                                 break
 760                         }
 761                         pingAddr := net.UDPAddr{
 762                                 IP:   ipa.IP,
 763                                 Port: ipa.Port,
 764                         }
 765                         if msg.Port != 0 {
 766                                 pingAddr.Port = int(msg.Port)
 767                         }
 768                         cl.eachDhtServer(func(s DhtServer) {
 769                                 go s.Ping(&pingAddr)
 770                         })
 771                 case pp.Suggest:
 772                         torrent.Add("suggests received", 1)
 773                         log.Fmsg("peer suggested piece %d", msg.Index).AddValues(c, msg.Index).LogLevel(log.Debug, c.t.logger)
 774                         c.updateRequests("suggested")
 775                 case pp.HaveAll:
 776                         err = c.onPeerSentHaveAll()
 777                 case pp.HaveNone:
 778                         err = c.peerSentHaveNone()
 779                 case pp.Reject:
 780                         req := newRequestFromMessage(&msg)
 781                         if !c.remoteRejectedRequest(c.t.requestIndexFromRequest(req)) {
 782                                 c.logger.Printf("received invalid reject [request=%v, peer=%v]", req, c)
 783                                 err = fmt.Errorf("received invalid reject [request=%v]", req)
 784                         }
 785                 case pp.AllowedFast:
 786                         torrent.Add("allowed fasts received", 1)
 787                         log.Fmsg("peer allowed fast: %d", msg.Index).AddValues(c).LogLevel(log.Debug, c.t.logger)
 788                         c.updateRequests("PeerConn.mainReadLoop allowed fast")
 789                 case pp.Extended:
 790                         err = c.onReadExtendedMsg(msg.ExtendedID, msg.ExtendedPayload)
 791                 default:
 792                         err = fmt.Errorf("received unknown message type: %#v", msg.Type)
 793                 }
 794                 if err != nil {
 795                         return err
 796                 }
 797         }
 798 }
 799
 800 func (c *PeerConn) onReadExtendedMsg(id pp.ExtensionNumber, payload []byte) (err error) {
 801         defer func() {
 802                 // TODO: Should we still do this?
 803                 if err != nil {
 804                         // These clients use their own extension IDs for outgoing message
 805                         // types, which is incorrect.
 806                         if bytes.HasPrefix(c.PeerID[:], []byte("-SD0100-")) || strings.HasPrefix(string(c.PeerID[:]), "-XL0012-") {
 807                                 err = nil
 808                         }
 809                 }
 810         }()
 811         t := c.t
 812         cl := t.cl
 813         switch id {
 814         case pp.HandshakeExtendedID:
 815                 var d pp.ExtendedHandshakeMessage
 816                 if err := bencode.Unmarshal(payload, &d); err != nil {
 817                         c.logger.Printf("error parsing extended handshake message %q: %s", payload, err)
 818                         return fmt.Errorf("unmarshalling extended handshake payload: %w", err)
 819                 }
 820                 if cb := c.callbacks.ReadExtendedHandshake; cb != nil {
 821                         cb(c, &d)
 822                 }
 823                 // c.logger.WithDefaultLevel(log.Debug).Printf("received extended handshake message:\n%s", spew.Sdump(d))
 824                 if d.Reqq != 0 {
 825                         c.PeerMaxRequests = d.Reqq
 826                 }
 827                 c.PeerClientName.Store(d.V)
 828                 if c.PeerExtensionIDs == nil {
 829                         c.PeerExtensionIDs = make(map[pp.ExtensionName]pp.ExtensionNumber, len(d.M))
 830                 }
 831                 c.PeerListenPort = d.Port
 832                 c.PeerPrefersEncryption = d.Encryption
 833                 for name, id := range d.M {
 834                         if _, ok := c.PeerExtensionIDs[name]; !ok {
 835                                 peersSupportingExtension.Add(
 836                                         // expvar.Var.String must produce valid JSON. "ut_payme\xeet_address" was being
 837                                         // entered here which caused problems later when unmarshalling.
 838                                         strconv.Quote(string(name)),
 839                                         1)
 840                         }
 841                         c.PeerExtensionIDs[name] = id
 842                 }
 843                 if d.MetadataSize != 0 {
 844                         if err = t.setMetadataSize(d.MetadataSize); err != nil {
 845                                 return fmt.Errorf("setting metadata size to %d: %w", d.MetadataSize, err)
 846                         }
 847                 }
 848                 c.requestPendingMetadata()
 849                 if !t.cl.config.DisablePEX {
 850                         t.pex.Add(c) // we learnt enough now
 851                         c.pex.Init(c)
 852                 }
 853                 return nil
 854         case metadataExtendedId:
 855                 err := cl.gotMetadataExtensionMsg(payload, t, c)
 856                 if err != nil {
 857                         return fmt.Errorf("handling metadata extension message: %w", err)
 858                 }
 859                 return nil
 860         case pexExtendedId:
 861                 if !c.pex.IsEnabled() {
 862                         return nil // or hang-up maybe?
 863                 }
 864                 return c.pex.Recv(payload)
 865         default:
 866                 return fmt.Errorf("unexpected extended message ID: %v", id)
 867         }
 868 }
 869
 870 // Set both the Reader and Writer for the connection from a single ReadWriter.
 871 func (cn *PeerConn) setRW(rw io.ReadWriter) {
 872         cn.r = rw
 873         cn.w = rw
 874 }
 875
 876 // Returns the Reader and Writer as a combined ReadWriter.
 877 func (cn *PeerConn) rw() io.ReadWriter {
 878         return struct {
 879                 io.Reader
 880                 io.Writer
 881         }{cn.r, cn.w}
 882 }
 883
 884 func (c *PeerConn) uploadAllowed() bool {
 885         if c.t.cl.config.NoUpload {
 886                 return false
 887         }
 888         if c.t.dataUploadDisallowed {
 889                 return false
 890         }
 891         if c.t.seeding() {
 892                 return true
 893         }
 894         if !c.peerHasWantedPieces() {
 895                 return false
 896         }
 897         // Don't upload more than 100 KiB more than we download.
 898         if c._stats.BytesWrittenData.Int64() >= c._stats.BytesReadData.Int64()+100<<10 {
 899                 return false
 900         }
 901         return true
 902 }
 903
 904 func (c *PeerConn) setRetryUploadTimer(delay time.Duration) {
 905         if c.uploadTimer == nil {
 906                 c.uploadTimer = time.AfterFunc(delay, c.tickleWriter)
 907         } else {
 908                 c.uploadTimer.Reset(delay)
 909         }
 910 }
 911
 912 // Also handles choking and unchoking of the remote peer.
 913 func (c *PeerConn) upload(msg func(pp.Message) bool) bool {
 914         // Breaking or completing this loop means we don't want to upload to the
 915         // peer anymore, and we choke them.
 916 another:
 917         for c.uploadAllowed() {
 918                 // We want to upload to the peer.
 919                 if !c.unchoke(msg) {
 920                         return false
 921                 }
 922                 for r, state := range c.peerRequests {
 923                         if state.data == nil {
 924                                 continue
 925                         }
 926                         res := c.t.cl.config.UploadRateLimiter.ReserveN(time.Now(), int(r.Length))
 927                         if !res.OK() {
 928                                 panic(fmt.Sprintf("upload rate limiter burst size < %d", r.Length))
 929                         }
 930                         delay := res.Delay()
 931                         if delay > 0 {
 932                                 res.Cancel()
 933                                 c.setRetryUploadTimer(delay)
 934                                 // Hard to say what to return here.
 935                                 return true
 936                         }
 937                         more := c.sendChunk(r, msg, state)
 938                         delete(c.peerRequests, r)
 939                         if !more {
 940                                 return false
 941                         }
 942                         goto another
 943                 }
 944                 return true
 945         }
 946         return c.choke(msg)
 947 }
 948
 949 func (cn *PeerConn) drop() {
 950         cn.t.dropConnection(cn)
 951 }
 952
 953 func (cn *PeerConn) ban() {
 954         cn.t.cl.banPeerIP(cn.remoteIp())
 955 }
 956
 957 // This is called when something has changed that should wake the writer, such as putting stuff into
 958 // the writeBuffer, or changing some state that the writer can act on.
 959 func (c *PeerConn) tickleWriter() {
 960         c.messageWriter.writeCond.Broadcast()
 961 }
 962
 963 func (c *PeerConn) sendChunk(r Request, msg func(pp.Message) bool, state *peerRequestState) (more bool) {
 964         c.lastChunkSent = time.Now()
 965         state.allocReservation.Release()
 966         return msg(pp.Message{
 967                 Type:  pp.Piece,
 968                 Index: r.Index,
 969                 Begin: r.Begin,
 970                 Piece: state.data,
 971         })
 972 }
 973
 974 func (c *PeerConn) setTorrent(t *Torrent) {
 975         if c.t != nil {
 976                 panic("connection already associated with a torrent")
 977         }
 978         c.t = t
 979         c.logger.WithDefaultLevel(log.Debug).Printf("set torrent=%v", t)
 980         t.reconcileHandshakeStats(c)
 981 }
 982
 983 func (c *PeerConn) pexPeerFlags() pp.PexPeerFlags {
 984         f := pp.PexPeerFlags(0)
 985         if c.PeerPrefersEncryption {
 986                 f |= pp.PexPrefersEncryption
 987         }
 988         if c.outgoing {
 989                 f |= pp.PexOutgoingConn
 990         }
 991         if c.utp() {
 992                 f |= pp.PexSupportsUtp
 993         }
 994         return f
 995 }
 996
 997 // This returns the address to use if we want to dial the peer again. It incorporates the peer's
 998 // advertised listen port.
 999 func (c *PeerConn) dialAddr() PeerRemoteAddr {
1000         if !c.outgoing && c.PeerListenPort != 0 {
1001                 switch addr := c.RemoteAddr.(type) {
1002                 case *net.TCPAddr:
1003                         dialAddr := *addr
1004                         dialAddr.Port = c.PeerListenPort
1005                         return &dialAddr
1006                 case *net.UDPAddr:
1007                         dialAddr := *addr
1008                         dialAddr.Port = c.PeerListenPort
1009                         return &dialAddr
1010                 }
1011         }
1012         return c.RemoteAddr
1013 }
1014
1015 func (c *PeerConn) pexEvent(t pexEventType) pexEvent {
1016         f := c.pexPeerFlags()
1017         addr := c.dialAddr()
1018         return pexEvent{t, addr, f, nil}
1019 }
1020
1021 func (c *PeerConn) String() string {
1022         return fmt.Sprintf("%T %p [id=%q, exts=%v, v=%q]", c, c, c.PeerID, c.PeerExtensionBytes, c.PeerClientName.Load())
1023 }
1024
1025 // Returns the pieces the peer could have based on their claims. If we don't know how many pieces
1026 // are in the torrent, it could be a very large range the peer has sent HaveAll.
1027 func (cn *PeerConn) PeerPieces() *roaring.Bitmap {
1028         cn.locker().RLock()
1029         defer cn.locker().RUnlock()
1030         return cn.newPeerPieces()
1031 }
1032
1033 func (pc *PeerConn) remoteIsTransmission() bool {
1034         return bytes.HasPrefix(pc.PeerID[:], []byte("-TR")) && pc.PeerID[7] == '-'
1035 }