peer.go

   1 package torrent
   2
   3 import (
   4         "errors"
   5         "fmt"
   6         "io"
   7         "net"
   8         "strings"
   9         "sync"
  10         "time"
  11
  12         "github.com/RoaringBitmap/roaring"
  13         "github.com/anacrolix/chansync"
  14         . "github.com/anacrolix/generics"
  15         "github.com/anacrolix/log"
  16         "github.com/anacrolix/missinggo/iter"
  17         "github.com/anacrolix/missinggo/v2/bitmap"
  18         "github.com/anacrolix/multiless"
  19
  20         "github.com/anacrolix/torrent/internal/alloclim"
  21         "github.com/anacrolix/torrent/mse"
  22         pp "github.com/anacrolix/torrent/peer_protocol"
  23         request_strategy "github.com/anacrolix/torrent/request-strategy"
  24         typedRoaring "github.com/anacrolix/torrent/typed-roaring"
  25 )
  26
  27 type (
  28         Peer struct {
  29                 // First to ensure 64-bit alignment for atomics. See #262.
  30                 _stats ConnStats
  31
  32                 t *Torrent
  33
  34                 peerImpl
  35                 callbacks *Callbacks
  36
  37                 outgoing   bool
  38                 Network    string
  39                 RemoteAddr PeerRemoteAddr
  40                 // The local address as observed by the remote peer. WebRTC seems to get this right without needing hints from the
  41                 // config.
  42                 localPublicAddr peerLocalPublicAddr
  43                 bannableAddr    Option[bannableAddr]
  44                 // True if the connection is operating over MSE obfuscation.
  45                 headerEncrypted bool
  46                 cryptoMethod    mse.CryptoMethod
  47                 Discovery       PeerSource
  48                 trusted         bool
  49                 closed          chansync.SetOnce
  50                 // Set true after we've added our ConnStats generated during handshake to
  51                 // other ConnStat instances as determined when the *Torrent became known.
  52                 reconciledHandshakeStats bool
  53
  54                 lastMessageReceived     time.Time
  55                 completedHandshake      time.Time
  56                 lastUsefulChunkReceived time.Time
  57                 lastChunkSent           time.Time
  58
  59                 // Stuff controlled by the local peer.
  60                 needRequestUpdate    string
  61                 requestState         request_strategy.PeerRequestState
  62                 updateRequestsTimer  *time.Timer
  63                 lastRequestUpdate    time.Time
  64                 peakRequests         maxRequests
  65                 lastBecameInterested time.Time
  66                 priorInterest        time.Duration
  67
  68                 lastStartedExpectingToReceiveChunks time.Time
  69                 cumulativeExpectedToReceiveChunks   time.Duration
  70                 _chunksReceivedWhileExpecting       int64
  71
  72                 choking                                bool
  73                 piecesReceivedSinceLastRequestUpdate   maxRequests
  74                 maxPiecesReceivedBetweenRequestUpdates maxRequests
  75                 // Chunks that we might reasonably expect to receive from the peer. Due to latency, buffering,
  76                 // and implementation differences, we may receive chunks that are no longer in the set of
  77                 // requests actually want. This could use a roaring.BSI if the memory use becomes noticeable.
  78                 validReceiveChunks map[RequestIndex]int
  79                 // Indexed by metadata piece, set to true if posted and pending a
  80                 // response.
  81                 metadataRequests []bool
  82                 sentHaves        bitmap.Bitmap
  83
  84                 // Stuff controlled by the remote peer.
  85                 peerInterested        bool
  86                 peerChoking           bool
  87                 peerRequests          map[Request]*peerRequestState
  88                 PeerPrefersEncryption bool // as indicated by 'e' field in extension handshake
  89                 // The highest possible number of pieces the torrent could have based on
  90                 // communication with the peer. Generally only useful until we have the
  91                 // torrent info.
  92                 peerMinPieces pieceIndex
  93                 // Pieces we've accepted chunks for from the peer.
  94                 peerTouchedPieces map[pieceIndex]struct{}
  95                 peerAllowedFast   typedRoaring.Bitmap[pieceIndex]
  96
  97                 PeerMaxRequests maxRequests // Maximum pending requests the peer allows.
  98
  99                 logger         log.Logger
 100                 protocolLogger log.Logger
 101         }
 102
 103         PeerSource string
 104
 105         peerRequestState struct {
 106                 data             []byte
 107                 allocReservation *alloclim.Reservation
 108         }
 109
 110         PeerRemoteAddr interface {
 111                 String() string
 112         }
 113
 114         peerRequests = orderedBitmap[RequestIndex]
 115 )
 116
 117 const (
 118         PeerSourceUtHolepunch     = "C"
 119         PeerSourceTracker         = "Tr"
 120         PeerSourceIncoming        = "I"
 121         PeerSourceDhtGetPeers     = "Hg" // Peers we found by searching a DHT.
 122         PeerSourceDhtAnnouncePeer = "Ha" // Peers that were announced to us by a DHT.
 123         PeerSourcePex             = "X"
 124         // The peer was given directly, such as through a magnet link.
 125         PeerSourceDirect = "M"
 126 )
 127
 128 // Returns the Torrent a Peer belongs to. Shouldn't change for the lifetime of the Peer. May be nil
 129 // if we are the receiving end of a connection and the handshake hasn't been received or accepted
 130 // yet.
 131 func (p *Peer) Torrent() *Torrent {
 132         return p.t
 133 }
 134
 135 func (p *Peer) initRequestState() {
 136         p.requestState.Requests = &peerRequests{}
 137 }
 138
 139 func (cn *Peer) updateExpectingChunks() {
 140         if cn.expectingChunks() {
 141                 if cn.lastStartedExpectingToReceiveChunks.IsZero() {
 142                         cn.lastStartedExpectingToReceiveChunks = time.Now()
 143                 }
 144         } else {
 145                 if !cn.lastStartedExpectingToReceiveChunks.IsZero() {
 146                         cn.cumulativeExpectedToReceiveChunks += time.Since(cn.lastStartedExpectingToReceiveChunks)
 147                         cn.lastStartedExpectingToReceiveChunks = time.Time{}
 148                 }
 149         }
 150 }
 151
 152 func (cn *Peer) expectingChunks() bool {
 153         if cn.requestState.Requests.IsEmpty() {
 154                 return false
 155         }
 156         if !cn.requestState.Interested {
 157                 return false
 158         }
 159         if !cn.peerChoking {
 160                 return true
 161         }
 162         haveAllowedFastRequests := false
 163         cn.peerAllowedFast.Iterate(func(i pieceIndex) bool {
 164                 haveAllowedFastRequests = roaringBitmapRangeCardinality[RequestIndex](
 165                         cn.requestState.Requests,
 166                         cn.t.pieceRequestIndexOffset(i),
 167                         cn.t.pieceRequestIndexOffset(i+1),
 168                 ) == 0
 169                 return !haveAllowedFastRequests
 170         })
 171         return haveAllowedFastRequests
 172 }
 173
 174 func (cn *Peer) remoteChokingPiece(piece pieceIndex) bool {
 175         return cn.peerChoking && !cn.peerAllowedFast.Contains(piece)
 176 }
 177
 178 func (cn *Peer) cumInterest() time.Duration {
 179         ret := cn.priorInterest
 180         if cn.requestState.Interested {
 181                 ret += time.Since(cn.lastBecameInterested)
 182         }
 183         return ret
 184 }
 185
 186 func (cn *Peer) locker() *lockWithDeferreds {
 187         return cn.t.cl.locker()
 188 }
 189
 190 func (cn *PeerConn) supportsExtension(ext pp.ExtensionName) bool {
 191         _, ok := cn.PeerExtensionIDs[ext]
 192         return ok
 193 }
 194
 195 // The best guess at number of pieces in the torrent for this peer.
 196 func (cn *Peer) bestPeerNumPieces() pieceIndex {
 197         if cn.t.haveInfo() {
 198                 return cn.t.numPieces()
 199         }
 200         return cn.peerMinPieces
 201 }
 202
 203 func (cn *Peer) completedString() string {
 204         have := pieceIndex(cn.peerPieces().GetCardinality())
 205         if all, _ := cn.peerHasAllPieces(); all {
 206                 have = cn.bestPeerNumPieces()
 207         }
 208         return fmt.Sprintf("%d/%d", have, cn.bestPeerNumPieces())
 209 }
 210
 211 func eventAgeString(t time.Time) string {
 212         if t.IsZero() {
 213                 return "never"
 214         }
 215         return fmt.Sprintf("%.2fs ago", time.Since(t).Seconds())
 216 }
 217
 218 // Inspired by https://github.com/transmission/transmission/wiki/Peer-Status-Text.
 219 func (cn *Peer) statusFlags() (ret string) {
 220         c := func(b byte) {
 221                 ret += string([]byte{b})
 222         }
 223         if cn.requestState.Interested {
 224                 c('i')
 225         }
 226         if cn.choking {
 227                 c('c')
 228         }
 229         c(':')
 230         ret += cn.connectionFlags()
 231         c(':')
 232         if cn.peerInterested {
 233                 c('i')
 234         }
 235         if cn.peerChoking {
 236                 c('c')
 237         }
 238         return
 239 }
 240
 241 func (cn *Peer) downloadRate() float64 {
 242         num := cn._stats.BytesReadUsefulData.Int64()
 243         if num == 0 {
 244                 return 0
 245         }
 246         return float64(num) / cn.totalExpectingTime().Seconds()
 247 }
 248
 249 func (p *Peer) DownloadRate() float64 {
 250         p.locker().RLock()
 251         defer p.locker().RUnlock()
 252
 253         return p.downloadRate()
 254 }
 255
 256 func (cn *Peer) iterContiguousPieceRequests(f func(piece pieceIndex, count int)) {
 257         var last Option[pieceIndex]
 258         var count int
 259         next := func(item Option[pieceIndex]) {
 260                 if item == last {
 261                         count++
 262                 } else {
 263                         if count != 0 {
 264                                 f(last.Value, count)
 265                         }
 266                         last = item
 267                         count = 1
 268                 }
 269         }
 270         cn.requestState.Requests.Iterate(func(requestIndex request_strategy.RequestIndex) bool {
 271                 next(Some(cn.t.pieceIndexOfRequestIndex(requestIndex)))
 272                 return true
 273         })
 274         next(None[pieceIndex]())
 275 }
 276
 277 func (cn *Peer) writeStatus(w io.Writer) {
 278         // \t isn't preserved in <pre> blocks?
 279         if cn.closed.IsSet() {
 280                 fmt.Fprint(w, "CLOSED: ")
 281         }
 282         fmt.Fprintln(w, strings.Join(cn.peerImplStatusLines(), "\n"))
 283         prio, err := cn.peerPriority()
 284         prioStr := fmt.Sprintf("%08x", prio)
 285         if err != nil {
 286                 prioStr += ": " + err.Error()
 287         }
 288         fmt.Fprintf(w, "bep40-prio: %v\n", prioStr)
 289         fmt.Fprintf(w, "last msg: %s, connected: %s, last helpful: %s, itime: %s, etime: %s\n",
 290                 eventAgeString(cn.lastMessageReceived),
 291                 eventAgeString(cn.completedHandshake),
 292                 eventAgeString(cn.lastHelpful()),
 293                 cn.cumInterest(),
 294                 cn.totalExpectingTime(),
 295         )
 296         fmt.Fprintf(w,
 297                 "%s completed, %d pieces touched, good chunks: %v/%v:%v reqq: %d+%v/(%d/%d):%d/%d, flags: %s, dr: %.1f KiB/s\n",
 298                 cn.completedString(),
 299                 len(cn.peerTouchedPieces),
 300                 &cn._stats.ChunksReadUseful,
 301                 &cn._stats.ChunksRead,
 302                 &cn._stats.ChunksWritten,
 303                 cn.requestState.Requests.GetCardinality(),
 304                 cn.requestState.Cancelled.GetCardinality(),
 305                 cn.nominalMaxRequests(),
 306                 cn.PeerMaxRequests,
 307                 len(cn.peerRequests),
 308                 localClientReqq,
 309                 cn.statusFlags(),
 310                 cn.downloadRate()/(1<<10),
 311         )
 312         fmt.Fprintf(w, "requested pieces:")
 313         cn.iterContiguousPieceRequests(func(piece pieceIndex, count int) {
 314                 fmt.Fprintf(w, " %v(%v)", piece, count)
 315         })
 316         fmt.Fprintf(w, "\n")
 317 }
 318
 319 func (p *Peer) close() {
 320         if !p.closed.Set() {
 321                 return
 322         }
 323         if p.updateRequestsTimer != nil {
 324                 p.updateRequestsTimer.Stop()
 325         }
 326         for _, prs := range p.peerRequests {
 327                 prs.allocReservation.Drop()
 328         }
 329         p.peerImpl.onClose()
 330         if p.t != nil {
 331                 p.t.decPeerPieceAvailability(p)
 332         }
 333         for _, f := range p.callbacks.PeerClosed {
 334                 f(p)
 335         }
 336 }
 337
 338 func (p *Peer) Close() error {
 339         p.locker().Lock()
 340         defer p.locker().Unlock()
 341         p.close()
 342         return nil
 343 }
 344
 345 // Peer definitely has a piece, for purposes of requesting. So it's not sufficient that we think
 346 // they do (known=true).
 347 func (cn *Peer) peerHasPiece(piece pieceIndex) bool {
 348         if all, known := cn.peerHasAllPieces(); all && known {
 349                 return true
 350         }
 351         return cn.peerPieces().ContainsInt(piece)
 352 }
 353
 354 // 64KiB, but temporarily less to work around an issue with WebRTC. TODO: Update when
 355 // https://github.com/pion/datachannel/issues/59 is fixed.
 356 const (
 357         writeBufferHighWaterLen = 1 << 15
 358         writeBufferLowWaterLen  = writeBufferHighWaterLen / 2
 359 )
 360
 361 var (
 362         interestedMsgLen = len(pp.Message{Type: pp.Interested}.MustMarshalBinary())
 363         requestMsgLen    = len(pp.Message{Type: pp.Request}.MustMarshalBinary())
 364         // This is the maximum request count that could fit in the write buffer if it's at or below the
 365         // low water mark when we run maybeUpdateActualRequestState.
 366         maxLocalToRemoteRequests = (writeBufferHighWaterLen - writeBufferLowWaterLen - interestedMsgLen) / requestMsgLen
 367 )
 368
 369 // The actual value to use as the maximum outbound requests.
 370 func (cn *Peer) nominalMaxRequests() maxRequests {
 371         return maxInt(1, minInt(cn.PeerMaxRequests, cn.peakRequests*2, maxLocalToRemoteRequests))
 372 }
 373
 374 func (cn *Peer) totalExpectingTime() (ret time.Duration) {
 375         ret = cn.cumulativeExpectedToReceiveChunks
 376         if !cn.lastStartedExpectingToReceiveChunks.IsZero() {
 377                 ret += time.Since(cn.lastStartedExpectingToReceiveChunks)
 378         }
 379         return
 380 }
 381
 382 func (cn *Peer) setInterested(interested bool) bool {
 383         if cn.requestState.Interested == interested {
 384                 return true
 385         }
 386         cn.requestState.Interested = interested
 387         if interested {
 388                 cn.lastBecameInterested = time.Now()
 389         } else if !cn.lastBecameInterested.IsZero() {
 390                 cn.priorInterest += time.Since(cn.lastBecameInterested)
 391         }
 392         cn.updateExpectingChunks()
 393         // log.Printf("%p: setting interest: %v", cn, interested)
 394         return cn.writeInterested(interested)
 395 }
 396
 397 // The function takes a message to be sent, and returns true if more messages
 398 // are okay.
 399 type messageWriter func(pp.Message) bool
 400
 401 // This function seems to only used by Peer.request. It's all logic checks, so maybe we can no-op it
 402 // when we want to go fast.
 403 func (cn *Peer) shouldRequest(r RequestIndex) error {
 404         err := cn.t.checkValidReceiveChunk(cn.t.requestIndexToRequest(r))
 405         if err != nil {
 406                 return err
 407         }
 408         pi := cn.t.pieceIndexOfRequestIndex(r)
 409         if cn.requestState.Cancelled.Contains(r) {
 410                 return errors.New("request is cancelled and waiting acknowledgement")
 411         }
 412         if !cn.peerHasPiece(pi) {
 413                 return errors.New("requesting piece peer doesn't have")
 414         }
 415         if !cn.t.peerIsActive(cn) {
 416                 panic("requesting but not in active conns")
 417         }
 418         if cn.closed.IsSet() {
 419                 panic("requesting when connection is closed")
 420         }
 421         if cn.t.hashingPiece(pi) {
 422                 panic("piece is being hashed")
 423         }
 424         if cn.t.pieceQueuedForHash(pi) {
 425                 panic("piece is queued for hash")
 426         }
 427         if cn.peerChoking && !cn.peerAllowedFast.Contains(pi) {
 428                 // This could occur if we made a request with the fast extension, and then got choked and
 429                 // haven't had the request rejected yet.
 430                 if !cn.requestState.Requests.Contains(r) {
 431                         panic("peer choking and piece not allowed fast")
 432                 }
 433         }
 434         return nil
 435 }
 436
 437 func (cn *Peer) mustRequest(r RequestIndex) bool {
 438         more, err := cn.request(r)
 439         if err != nil {
 440                 panic(err)
 441         }
 442         return more
 443 }
 444
 445 func (cn *Peer) request(r RequestIndex) (more bool, err error) {
 446         if err := cn.shouldRequest(r); err != nil {
 447                 panic(err)
 448         }
 449         if cn.requestState.Requests.Contains(r) {
 450                 return true, nil
 451         }
 452         if maxRequests(cn.requestState.Requests.GetCardinality()) >= cn.nominalMaxRequests() {
 453                 return true, errors.New("too many outstanding requests")
 454         }
 455         cn.requestState.Requests.Add(r)
 456         if cn.validReceiveChunks == nil {
 457                 cn.validReceiveChunks = make(map[RequestIndex]int)
 458         }
 459         cn.validReceiveChunks[r]++
 460         cn.t.requestState[r] = requestState{
 461                 peer: cn,
 462                 when: time.Now(),
 463         }
 464         cn.updateExpectingChunks()
 465         ppReq := cn.t.requestIndexToRequest(r)
 466         for _, f := range cn.callbacks.SentRequest {
 467                 f(PeerRequestEvent{cn, ppReq})
 468         }
 469         return cn.peerImpl._request(ppReq), nil
 470 }
 471
 472 func (me *Peer) cancel(r RequestIndex) {
 473         if !me.deleteRequest(r) {
 474                 panic("request not existing should have been guarded")
 475         }
 476         if me._cancel(r) {
 477                 // Record that we expect to get a cancel ack.
 478                 if !me.requestState.Cancelled.CheckedAdd(r) {
 479                         panic("request already cancelled")
 480                 }
 481         }
 482         me.decPeakRequests()
 483         if me.isLowOnRequests() {
 484                 me.updateRequests("Peer.cancel")
 485         }
 486 }
 487
 488 // Sets a reason to update requests, and if there wasn't already one, handle it.
 489 func (cn *Peer) updateRequests(reason string) {
 490         if cn.needRequestUpdate != "" {
 491                 return
 492         }
 493         cn.needRequestUpdate = reason
 494         cn.handleUpdateRequests()
 495 }
 496
 497 // Emits the indices in the Bitmaps bms in order, never repeating any index.
 498 // skip is mutated during execution, and its initial values will never be
 499 // emitted.
 500 func iterBitmapsDistinct(skip *bitmap.Bitmap, bms ...bitmap.Bitmap) iter.Func {
 501         return func(cb iter.Callback) {
 502                 for _, bm := range bms {
 503                         if !iter.All(
 504                                 func(_i interface{}) bool {
 505                                         i := _i.(int)
 506                                         if skip.Contains(bitmap.BitIndex(i)) {
 507                                                 return true
 508                                         }
 509                                         skip.Add(bitmap.BitIndex(i))
 510                                         return cb(i)
 511                                 },
 512                                 bm.Iter,
 513                         ) {
 514                                 return
 515                         }
 516                 }
 517         }
 518 }
 519
 520 // After handshake, we know what Torrent and Client stats to include for a
 521 // connection.
 522 func (cn *Peer) postHandshakeStats(f func(*ConnStats)) {
 523         t := cn.t
 524         f(&t.stats)
 525         f(&t.cl.connStats)
 526 }
 527
 528 // All ConnStats that include this connection. Some objects are not known
 529 // until the handshake is complete, after which it's expected to reconcile the
 530 // differences.
 531 func (cn *Peer) allStats(f func(*ConnStats)) {
 532         f(&cn._stats)
 533         if cn.reconciledHandshakeStats {
 534                 cn.postHandshakeStats(f)
 535         }
 536 }
 537
 538 func (cn *Peer) readBytes(n int64) {
 539         cn.allStats(add(n, func(cs *ConnStats) *Count { return &cs.BytesRead }))
 540 }
 541
 542 func (c *Peer) lastHelpful() (ret time.Time) {
 543         ret = c.lastUsefulChunkReceived
 544         if c.t.seeding() && c.lastChunkSent.After(ret) {
 545                 ret = c.lastChunkSent
 546         }
 547         return
 548 }
 549
 550 // Returns whether any part of the chunk would lie outside a piece of the given length.
 551 func chunkOverflowsPiece(cs ChunkSpec, pieceLength pp.Integer) bool {
 552         switch {
 553         default:
 554                 return false
 555         case cs.Begin+cs.Length > pieceLength:
 556         // Check for integer overflow
 557         case cs.Begin > pp.IntegerMax-cs.Length:
 558         }
 559         return true
 560 }
 561
 562 func runSafeExtraneous(f func()) {
 563         if true {
 564                 go f()
 565         } else {
 566                 f()
 567         }
 568 }
 569
 570 // Returns true if it was valid to reject the request.
 571 func (c *Peer) remoteRejectedRequest(r RequestIndex) bool {
 572         if c.deleteRequest(r) {
 573                 c.decPeakRequests()
 574         } else if !c.requestState.Cancelled.CheckedRemove(r) {
 575                 return false
 576         }
 577         if c.isLowOnRequests() {
 578                 c.updateRequests("Peer.remoteRejectedRequest")
 579         }
 580         c.decExpectedChunkReceive(r)
 581         return true
 582 }
 583
 584 func (c *Peer) decExpectedChunkReceive(r RequestIndex) {
 585         count := c.validReceiveChunks[r]
 586         if count == 1 {
 587                 delete(c.validReceiveChunks, r)
 588         } else if count > 1 {
 589                 c.validReceiveChunks[r] = count - 1
 590         } else {
 591                 panic(r)
 592         }
 593 }
 594
 595 func (c *Peer) doChunkReadStats(size int64) {
 596         c.allStats(func(cs *ConnStats) { cs.receivedChunk(size) })
 597 }
 598
 599 // Handle a received chunk from a peer.
 600 func (c *Peer) receiveChunk(msg *pp.Message) error {
 601         chunksReceived.Add("total", 1)
 602
 603         ppReq := newRequestFromMessage(msg)
 604         t := c.t
 605         err := t.checkValidReceiveChunk(ppReq)
 606         if err != nil {
 607                 err = log.WithLevel(log.Warning, err)
 608                 return err
 609         }
 610         req := c.t.requestIndexFromRequest(ppReq)
 611
 612         recordBlockForSmartBan := sync.OnceFunc(func() {
 613                 c.recordBlockForSmartBan(req, msg.Piece)
 614         })
 615         // This needs to occur before we return, but we try to do it when the client is unlocked. It
 616         // can't be done before checking if chunks are valid because they won't be deallocated by piece
 617         // hashing if they're out of bounds.
 618         defer recordBlockForSmartBan()
 619
 620         if c.peerChoking {
 621                 chunksReceived.Add("while choked", 1)
 622         }
 623
 624         if c.validReceiveChunks[req] <= 0 {
 625                 chunksReceived.Add("unexpected", 1)
 626                 return errors.New("received unexpected chunk")
 627         }
 628         c.decExpectedChunkReceive(req)
 629
 630         if c.peerChoking && c.peerAllowedFast.Contains(pieceIndex(ppReq.Index)) {
 631                 chunksReceived.Add("due to allowed fast", 1)
 632         }
 633
 634         // The request needs to be deleted immediately to prevent cancels occurring asynchronously when
 635         // have actually already received the piece, while we have the Client unlocked to write the data
 636         // out.
 637         intended := false
 638         {
 639                 if c.requestState.Requests.Contains(req) {
 640                         for _, f := range c.callbacks.ReceivedRequested {
 641                                 f(PeerMessageEvent{c, msg})
 642                         }
 643                 }
 644                 // Request has been satisfied.
 645                 if c.deleteRequest(req) || c.requestState.Cancelled.CheckedRemove(req) {
 646                         intended = true
 647                         if !c.peerChoking {
 648                                 c._chunksReceivedWhileExpecting++
 649                         }
 650                         if c.isLowOnRequests() {
 651                                 c.updateRequests("Peer.receiveChunk deleted request")
 652                         }
 653                 } else {
 654                         chunksReceived.Add("unintended", 1)
 655                 }
 656         }
 657
 658         cl := t.cl
 659
 660         // Do we actually want this chunk?
 661         if t.haveChunk(ppReq) {
 662                 // panic(fmt.Sprintf("%+v", ppReq))
 663                 chunksReceived.Add("redundant", 1)
 664                 c.allStats(add(1, func(cs *ConnStats) *Count { return &cs.ChunksReadWasted }))
 665                 return nil
 666         }
 667
 668         piece := &t.pieces[ppReq.Index]
 669
 670         c.allStats(add(1, func(cs *ConnStats) *Count { return &cs.ChunksReadUseful }))
 671         c.allStats(add(int64(len(msg.Piece)), func(cs *ConnStats) *Count { return &cs.BytesReadUsefulData }))
 672         if intended {
 673                 c.piecesReceivedSinceLastRequestUpdate++
 674                 c.allStats(add(int64(len(msg.Piece)), func(cs *ConnStats) *Count { return &cs.BytesReadUsefulIntendedData }))
 675         }
 676         for _, f := range c.t.cl.config.Callbacks.ReceivedUsefulData {
 677                 f(ReceivedUsefulDataEvent{c, msg})
 678         }
 679         c.lastUsefulChunkReceived = time.Now()
 680
 681         // Need to record that it hasn't been written yet, before we attempt to do
 682         // anything with it.
 683         piece.incrementPendingWrites()
 684         // Record that we have the chunk, so we aren't trying to download it while
 685         // waiting for it to be written to storage.
 686         piece.unpendChunkIndex(chunkIndexFromChunkSpec(ppReq.ChunkSpec, t.chunkSize))
 687
 688         // Cancel pending requests for this chunk from *other* peers.
 689         if p := t.requestingPeer(req); p != nil {
 690                 if p == c {
 691                         panic("should not be pending request from conn that just received it")
 692                 }
 693                 p.cancel(req)
 694         }
 695
 696         err = func() error {
 697                 cl.unlock()
 698                 defer cl.lock()
 699                 // Opportunistically do this here while we aren't holding the client lock.
 700                 recordBlockForSmartBan()
 701                 concurrentChunkWrites.Add(1)
 702                 defer concurrentChunkWrites.Add(-1)
 703                 // Write the chunk out. Note that the upper bound on chunk writing concurrency will be the
 704                 // number of connections. We write inline with receiving the chunk (with this lock dance),
 705                 // because we want to handle errors synchronously and I haven't thought of a nice way to
 706                 // defer any concurrency to the storage and have that notify the client of errors. TODO: Do
 707                 // that instead.
 708                 return t.writeChunk(int(msg.Index), int64(msg.Begin), msg.Piece)
 709         }()
 710
 711         piece.decrementPendingWrites()
 712
 713         if err != nil {
 714                 c.logger.WithDefaultLevel(log.Error).Printf("writing received chunk %v: %v", req, err)
 715                 t.pendRequest(req)
 716                 // Necessary to pass TestReceiveChunkStorageFailureSeederFastExtensionDisabled. I think a
 717                 // request update runs while we're writing the chunk that just failed. Then we never do a
 718                 // fresh update after pending the failed request.
 719                 c.updateRequests("Peer.receiveChunk error writing chunk")
 720                 t.onWriteChunkErr(err)
 721                 return nil
 722         }
 723
 724         c.onDirtiedPiece(pieceIndex(ppReq.Index))
 725
 726         // We need to ensure the piece is only queued once, so only the last chunk writer gets this job.
 727         if t.pieceAllDirty(pieceIndex(ppReq.Index)) && piece.pendingWrites == 0 {
 728                 t.queuePieceCheck(pieceIndex(ppReq.Index))
 729                 // We don't pend all chunks here anymore because we don't want code dependent on the dirty
 730                 // chunk status (such as the haveChunk call above) to have to check all the various other
 731                 // piece states like queued for hash, hashing etc. This does mean that we need to be sure
 732                 // that chunk pieces are pended at an appropriate time later however.
 733         }
 734
 735         cl.event.Broadcast()
 736         // We do this because we've written a chunk, and may change PieceState.Partial.
 737         t.publishPieceStateChange(pieceIndex(ppReq.Index))
 738
 739         return nil
 740 }
 741
 742 func (c *Peer) onDirtiedPiece(piece pieceIndex) {
 743         if c.peerTouchedPieces == nil {
 744                 c.peerTouchedPieces = make(map[pieceIndex]struct{})
 745         }
 746         c.peerTouchedPieces[piece] = struct{}{}
 747         ds := &c.t.pieces[piece].dirtiers
 748         if *ds == nil {
 749                 *ds = make(map[*Peer]struct{})
 750         }
 751         (*ds)[c] = struct{}{}
 752 }
 753
 754 func (cn *Peer) netGoodPiecesDirtied() int64 {
 755         return cn._stats.PiecesDirtiedGood.Int64() - cn._stats.PiecesDirtiedBad.Int64()
 756 }
 757
 758 func (c *Peer) peerHasWantedPieces() bool {
 759         if all, _ := c.peerHasAllPieces(); all {
 760                 return !c.t.haveAllPieces() && !c.t._pendingPieces.IsEmpty()
 761         }
 762         if !c.t.haveInfo() {
 763                 return !c.peerPieces().IsEmpty()
 764         }
 765         return c.peerPieces().Intersects(&c.t._pendingPieces)
 766 }
 767
 768 // Returns true if an outstanding request is removed. Cancelled requests should be handled
 769 // separately.
 770 func (c *Peer) deleteRequest(r RequestIndex) bool {
 771         if !c.requestState.Requests.CheckedRemove(r) {
 772                 return false
 773         }
 774         for _, f := range c.callbacks.DeletedRequest {
 775                 f(PeerRequestEvent{c, c.t.requestIndexToRequest(r)})
 776         }
 777         c.updateExpectingChunks()
 778         if c.t.requestingPeer(r) != c {
 779                 panic("only one peer should have a given request at a time")
 780         }
 781         delete(c.t.requestState, r)
 782         // c.t.iterPeers(func(p *Peer) {
 783         //      if p.isLowOnRequests() {
 784         //              p.updateRequests("Peer.deleteRequest")
 785         //      }
 786         // })
 787         return true
 788 }
 789
 790 func (c *Peer) deleteAllRequests(reason string) {
 791         if c.requestState.Requests.IsEmpty() {
 792                 return
 793         }
 794         c.requestState.Requests.IterateSnapshot(func(x RequestIndex) bool {
 795                 if !c.deleteRequest(x) {
 796                         panic("request should exist")
 797                 }
 798                 return true
 799         })
 800         c.assertNoRequests()
 801         c.t.iterPeers(func(p *Peer) {
 802                 if p.isLowOnRequests() {
 803                         p.updateRequests(reason)
 804                 }
 805         })
 806         return
 807 }
 808
 809 func (c *Peer) assertNoRequests() {
 810         if !c.requestState.Requests.IsEmpty() {
 811                 panic(c.requestState.Requests.GetCardinality())
 812         }
 813 }
 814
 815 func (c *Peer) cancelAllRequests() {
 816         c.requestState.Requests.IterateSnapshot(func(x RequestIndex) bool {
 817                 c.cancel(x)
 818                 return true
 819         })
 820         c.assertNoRequests()
 821         return
 822 }
 823
 824 func (c *Peer) peerPriority() (peerPriority, error) {
 825         return bep40Priority(c.remoteIpPort(), c.localPublicAddr)
 826 }
 827
 828 func (c *Peer) remoteIp() net.IP {
 829         host, _, _ := net.SplitHostPort(c.RemoteAddr.String())
 830         return net.ParseIP(host)
 831 }
 832
 833 func (c *Peer) remoteIpPort() IpPort {
 834         ipa, _ := tryIpPortFromNetAddr(c.RemoteAddr)
 835         return IpPort{ipa.IP, uint16(ipa.Port)}
 836 }
 837
 838 func (c *Peer) trust() connectionTrust {
 839         return connectionTrust{c.trusted, c.netGoodPiecesDirtied()}
 840 }
 841
 842 type connectionTrust struct {
 843         Implicit            bool
 844         NetGoodPiecesDirted int64
 845 }
 846
 847 func (l connectionTrust) Less(r connectionTrust) bool {
 848         return multiless.New().Bool(l.Implicit, r.Implicit).Int64(l.NetGoodPiecesDirted, r.NetGoodPiecesDirted).Less()
 849 }
 850
 851 // Returns a new Bitmap that includes bits for all pieces the peer could have based on their claims.
 852 func (cn *Peer) newPeerPieces() *roaring.Bitmap {
 853         // TODO: Can we use copy on write?
 854         ret := cn.peerPieces().Clone()
 855         if all, _ := cn.peerHasAllPieces(); all {
 856                 if cn.t.haveInfo() {
 857                         ret.AddRange(0, bitmap.BitRange(cn.t.numPieces()))
 858                 } else {
 859                         ret.AddRange(0, bitmap.ToEnd)
 860                 }
 861         }
 862         return ret
 863 }
 864
 865 func (cn *Peer) stats() *ConnStats {
 866         return &cn._stats
 867 }
 868
 869 func (p *Peer) TryAsPeerConn() (*PeerConn, bool) {
 870         pc, ok := p.peerImpl.(*PeerConn)
 871         return pc, ok
 872 }
 873
 874 func (p *Peer) uncancelledRequests() uint64 {
 875         return p.requestState.Requests.GetCardinality()
 876 }
 877
 878 type peerLocalPublicAddr = IpPort
 879
 880 func (p *Peer) isLowOnRequests() bool {
 881         return p.requestState.Requests.IsEmpty() && p.requestState.Cancelled.IsEmpty()
 882 }
 883
 884 func (p *Peer) decPeakRequests() {
 885         // // This can occur when peak requests are altered by the update request timer to be lower than
 886         // // the actual number of outstanding requests. Let's let it go negative and see what happens. I
 887         // // wonder what happens if maxRequests is not signed.
 888         // if p.peakRequests < 1 {
 889         //      panic(p.peakRequests)
 890         // }
 891         p.peakRequests--
 892 }
 893
 894 func (p *Peer) recordBlockForSmartBan(req RequestIndex, blockData []byte) {
 895         if p.bannableAddr.Ok {
 896                 p.t.smartBanCache.RecordBlock(p.bannableAddr.Value, req, blockData)
 897         }
 898 }