peer.go

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