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         "github.com/anacrolix/torrent/types"
  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                 select {
  23                 case <-cl.closed.Done():
  24                         return
  25                 case <-time.After(100 * time.Millisecond):
  26                 }
  27                 select {
  28                 case <-cl.closed.Done():
  29                         return
  30                 case <-update:
  31                 case <-time.After(time.Second):
  32                 }
  33         }
  34 }
  35
  36 func (cl *Client) tickleRequester() {
  37         cl.updateRequests.Broadcast()
  38 }
  39
  40 func (cl *Client) doRequests() {
  41         ts := make([]request_strategy.Torrent, 0, len(cl.torrents))
  42         for _, t := range cl.torrents {
  43                 rst := request_strategy.Torrent{
  44                         StableId: uintptr(unsafe.Pointer(t)),
  45                 }
  46                 if t.storage != nil {
  47                         rst.Capacity = t.storage.Capacity
  48                 }
  49                 rst.Pieces = make([]request_strategy.Piece, 0, len(t.pieces))
  50                 for i := range t.pieces {
  51                         p := &t.pieces[i]
  52                         rst.Pieces = append(rst.Pieces, request_strategy.Piece{
  53                                 Request:          !t.ignorePieceForRequests(i),
  54                                 Priority:         p.purePriority(),
  55                                 Partial:          t.piecePartiallyDownloaded(i),
  56                                 Availability:     p.availability,
  57                                 Length:           int64(p.length()),
  58                                 NumPendingChunks: int(t.pieceNumPendingChunks(i)),
  59                                 IterPendingChunks: func(f func(types.ChunkSpec)) {
  60                                         p.iterUndirtiedChunks(func(cs ChunkSpec) bool {
  61                                                 f(cs)
  62                                                 return true
  63                                         })
  64                                 },
  65                         })
  66                 }
  67                 t.iterPeers(func(p *Peer) {
  68                         if p.closed.IsSet() {
  69                                 return
  70                         }
  71                         if p.piecesReceivedSinceLastRequestUpdate > p.maxPiecesReceivedBetweenRequestUpdates {
  72                                 p.maxPiecesReceivedBetweenRequestUpdates = p.piecesReceivedSinceLastRequestUpdate
  73                         }
  74                         p.piecesReceivedSinceLastRequestUpdate = 0
  75                         rst.Peers = append(rst.Peers, request_strategy.Peer{
  76                                 HasPiece:    p.peerHasPiece,
  77                                 MaxRequests: p.nominalMaxRequests(),
  78                                 HasExistingRequest: func(r request_strategy.Request) bool {
  79                                         _, ok := p.actualRequestState.Requests[r]
  80                                         return ok
  81                                 },
  82                                 Choking: p.peerChoking,
  83                                 PieceAllowedFast: func(i pieceIndex) bool {
  84                                         return p.peerAllowedFast.Contains(bitmap.BitIndex(i))
  85                                 },
  86                                 DownloadRate: p.downloadRate(),
  87                                 Age:          time.Since(p.completedHandshake),
  88                                 Id: peerId{
  89                                         Peer: p,
  90                                         ptr:  uintptr(unsafe.Pointer(p)),
  91                                 },
  92                         })
  93                 })
  94                 ts = append(ts, rst)
  95         }
  96         nextPeerStates := request_strategy.Run(request_strategy.Input{
  97                 Torrents:           ts,
  98                 MaxUnverifiedBytes: cl.config.MaxUnverifiedBytes,
  99         })
 100         for p, state := range nextPeerStates {
 101                 setPeerNextRequestState(p, state)
 102         }
 103 }
 104
 105 type peerId struct {
 106         *Peer
 107         ptr uintptr
 108 }
 109
 110 func (p peerId) Uintptr() uintptr {
 111         return p.ptr
 112 }
 113
 114 func setPeerNextRequestState(_p request_strategy.PeerId, rp request_strategy.PeerNextRequestState) {
 115         p := _p.(peerId).Peer
 116         p.nextRequestState = rp
 117         p.onNextRequestStateChanged()
 118 }
 119
 120 func (p *Peer) applyNextRequestState() bool {
 121         next := p.nextRequestState
 122         current := p.actualRequestState
 123         if !p.setInterested(next.Interested) {
 124                 return false
 125         }
 126         for req := range current.Requests {
 127                 if _, ok := next.Requests[req]; !ok {
 128                         if !p.cancel(req) {
 129                                 return false
 130                         }
 131                 }
 132         }
 133         for req := range next.Requests {
 134                 more, err := p.request(req)
 135                 if err != nil {
 136                         panic(err)
 137                 } /* else {
 138                         log.Print(req)
 139                 } */
 140                 if !more {
 141                         return false
 142                 }
 143         }
 144         return true
 145 }