peer.go

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