requesting.go

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