requesting.go

   1 package torrent
   2
   3 import (
   4         "time"
   5         "unsafe"
   6
   7         "github.com/anacrolix/missinggo/v2/bitmap"
   8
   9         "github.com/anacrolix/chansync"
  10         request_strategy "github.com/anacrolix/torrent/request-strategy"
  11 )
  12
  13 func (cl *Client) requester() {
  14         for {
  15                 update := func() chansync.Signaled {
  16                         cl.lock()
  17                         defer cl.unlock()
  18                         cl.doRequests()
  19                         return cl.updateRequests.Signaled()
  20                 }()
  21                 minWait := time.After(100 * time.Millisecond)
  22                 maxWait := time.After(1000 * time.Millisecond)
  23                 select {
  24                 case <-cl.closed.Done():
  25                         return
  26                 case <-minWait:
  27                 case <-maxWait:
  28                 }
  29                 select {
  30                 case <-cl.closed.Done():
  31                         return
  32                 case <-update:
  33                 case <-maxWait:
  34                 }
  35         }
  36 }
  37
  38 func (cl *Client) tickleRequester() {
  39         cl.updateRequests.Broadcast()
  40 }
  41
  42 func (cl *Client) getRequestStrategyInput() request_strategy.Input {
  43         ts := make([]request_strategy.Torrent, 0, len(cl.torrents))
  44         for _, t := range cl.torrents {
  45                 if !t.haveInfo() {
  46                         // This would be removed if metadata is handled here.
  47                         continue
  48                 }
  49                 rst := request_strategy.Torrent{
  50                         InfoHash:       t.infoHash,
  51                         ChunksPerPiece: (t.usualPieceSize() + int(t.chunkSize) - 1) / int(t.chunkSize),
  52                 }
  53                 if t.storage != nil {
  54                         rst.Capacity = t.storage.Capacity
  55                 }
  56                 rst.Pieces = make([]request_strategy.Piece, 0, len(t.pieces))
  57                 for i := range t.pieces {
  58                         p := &t.pieces[i]
  59                         rst.Pieces = append(rst.Pieces, request_strategy.Piece{
  60                                 Request:           !t.ignorePieceForRequests(i),
  61                                 Priority:          p.purePriority(),
  62                                 Partial:           t.piecePartiallyDownloaded(i),
  63                                 Availability:      p.availability,
  64                                 Length:            int64(p.length()),
  65                                 NumPendingChunks:  int(t.pieceNumPendingChunks(i)),
  66                                 IterPendingChunks: p.iterUndirtiedChunks,
  67                         })
  68                 }
  69                 t.iterPeers(func(p *Peer) {
  70                         if p.closed.IsSet() {
  71                                 return
  72                         }
  73                         if p.piecesReceivedSinceLastRequestUpdate > p.maxPiecesReceivedBetweenRequestUpdates {
  74                                 p.maxPiecesReceivedBetweenRequestUpdates = p.piecesReceivedSinceLastRequestUpdate
  75                         }
  76                         p.piecesReceivedSinceLastRequestUpdate = 0
  77                         rst.Peers = append(rst.Peers, request_strategy.Peer{
  78                                 HasPiece:    p.peerHasPiece,
  79                                 MaxRequests: p.nominalMaxRequests(),
  80                                 HasExistingRequest: func(r RequestIndex) bool {
  81                                         return p.actualRequestState.Requests.Contains(r)
  82                                 },
  83                                 Choking: p.peerChoking,
  84                                 PieceAllowedFast: func(i pieceIndex) bool {
  85                                         return p.peerAllowedFast.Contains(bitmap.BitIndex(i))
  86                                 },
  87                                 DownloadRate: p.downloadRate(),
  88                                 Age:          time.Since(p.completedHandshake),
  89                                 Id: peerId{
  90                                         Peer: p,
  91                                         ptr:  uintptr(unsafe.Pointer(p)),
  92                                 },
  93                         })
  94                 })
  95                 ts = append(ts, rst)
  96         }
  97         return request_strategy.Input{
  98                 Torrents:           ts,
  99                 MaxUnverifiedBytes: cl.config.MaxUnverifiedBytes,
 100         }
 101 }
 102
 103 func (cl *Client) doRequests() {
 104         nextPeerStates := request_strategy.Run(cl.getRequestStrategyInput())
 105         for p, state := range nextPeerStates {
 106                 setPeerNextRequestState(p, state)
 107         }
 108 }
 109
 110 type peerId struct {
 111         *Peer
 112         ptr uintptr
 113 }
 114
 115 func (p peerId) Uintptr() uintptr {
 116         return p.ptr
 117 }
 118
 119 func setPeerNextRequestState(_p request_strategy.PeerId, rp request_strategy.PeerNextRequestState) {
 120         p := _p.(peerId).Peer
 121         p.nextRequestState = rp
 122         p.onNextRequestStateChanged()
 123 }
 124
 125 type RequestIndex = request_strategy.RequestIndex
 126 type chunkIndexType = request_strategy.ChunkIndex
 127
 128 func (p *Peer) applyNextRequestState() bool {
 129         if p.actualRequestState.Requests.GetCardinality() > uint64(p.nominalMaxRequests()/2) {
 130                 return true
 131         }
 132         type piece struct {
 133                 index   int
 134                 endGame bool
 135         }
 136         var pieceOrder []piece
 137         request_strategy.GetRequestablePieces(
 138                 p.t.cl.getRequestStrategyInput(),
 139                 func(t *request_strategy.Torrent, rsp *request_strategy.Piece, pieceIndex int) {
 140                         if t.InfoHash != p.t.infoHash {
 141                                 return
 142                         }
 143                         if !p.peerHasPiece(pieceIndex) {
 144                                 return
 145                         }
 146                         pieceOrder = append(pieceOrder, piece{
 147                                 index:   pieceIndex,
 148                                 endGame: rsp.Priority == PiecePriorityNow,
 149                         })
 150                 },
 151         )
 152         more := true
 153         interested := false
 154         for _, endGameIter := range []bool{false, true} {
 155                 for _, piece := range pieceOrder {
 156                         tp := p.t.piece(piece.index)
 157                         tp.iterUndirtiedChunks(func(cs chunkIndexType) {
 158                                 req := cs + tp.requestIndexOffset()
 159                                 if !piece.endGame && !endGameIter && p.t.pendingRequests[req] > 0 {
 160                                         return
 161                                 }
 162                                 interested = true
 163                                 more = p.setInterested(true)
 164                                 if !more {
 165                                         return
 166                                 }
 167                                 if maxRequests(p.actualRequestState.Requests.GetCardinality()) >= p.nominalMaxRequests() {
 168                                         return
 169                                 }
 170                                 if p.peerChoking && !p.peerAllowedFast.Contains(bitmap.BitIndex(piece.index)) {
 171                                         return
 172                                 }
 173                                 var err error
 174                                 more, err = p.request(req)
 175                                 if err != nil {
 176                                         panic(err)
 177                                 }
 178                         })
 179                         if interested && maxRequests(p.actualRequestState.Requests.GetCardinality()) >= p.nominalMaxRequests() {
 180                                 break
 181                         }
 182                         if !more {
 183                                 break
 184                         }
 185                 }
 186                 if !more {
 187                         break
 188                 }
 189         }
 190         if !more {
 191                 return false
 192         }
 193         if !interested {
 194                 p.setInterested(false)
 195         }
 196         return more
 197 }