connString: connString,
conn: nc,
}
+ c.initRequestState()
// TODO: Need to be much more explicit about this, including allowing non-IP bannable addresses.
if remoteAddr != nil {
netipAddrPort, err := netip.ParseAddrPort(remoteAddr.String())
"github.com/anacrolix/torrent/mse"
pp "github.com/anacrolix/torrent/peer_protocol"
request_strategy "github.com/anacrolix/torrent/request-strategy"
+ "github.com/anacrolix/torrent/typed-roaring"
)
type PeerSource string
peerMinPieces pieceIndex
// Pieces we've accepted chunks for from the peer.
peerTouchedPieces map[pieceIndex]struct{}
- peerAllowedFast roaring.Bitmap
+ peerAllowedFast typedRoaring.Bitmap[pieceIndex]
PeerMaxRequests maxRequests // Maximum pending requests the peer allows.
PeerExtensionIDs map[pp.ExtensionName]pp.ExtensionNumber
logger log.Logger
}
+type peerRequests struct {
+ typedRoaring.Bitmap[RequestIndex]
+}
+
+func (p *peerRequests) IterateSnapshot(f func(request_strategy.RequestIndex) bool) {
+ p.Clone().Iterate(f)
+}
+
+func (p *Peer) initRequestState() {
+ p.requestState.Requests = &peerRequests{}
+}
+
// Maintains the state of a BitTorrent-protocol based connection with a peer.
type PeerConn struct {
Peer
return true
}
haveAllowedFastRequests := false
- cn.peerAllowedFast.Iterate(func(i uint32) bool {
- haveAllowedFastRequests = roaringBitmapRangeCardinality(
- &cn.requestState.Requests,
- cn.t.pieceRequestIndexOffset(pieceIndex(i)),
- cn.t.pieceRequestIndexOffset(pieceIndex(i+1)),
+ cn.peerAllowedFast.Iterate(func(i pieceIndex) bool {
+ haveAllowedFastRequests = roaringBitmapRangeCardinality[RequestIndex](
+ cn.requestState.Requests,
+ cn.t.pieceRequestIndexOffset(i),
+ cn.t.pieceRequestIndexOffset(i+1),
) == 0
return !haveAllowedFastRequests
})
}
func (cn *Peer) remoteChokingPiece(piece pieceIndex) bool {
- return cn.peerChoking && !cn.peerAllowedFast.Contains(bitmap.BitIndex(piece))
+ return cn.peerChoking && !cn.peerAllowedFast.Contains(piece)
}
// Returns true if the connection is over IPv6.
func (cn *Peer) numRequestsByPiece() (ret map[pieceIndex]int) {
ret = make(map[pieceIndex]int)
- cn.requestState.Requests.Iterate(func(x uint32) bool {
- ret[pieceIndex(x/cn.t.chunksPerRegularPiece())]++
+ cn.requestState.Requests.Iterate(func(x RequestIndex) bool {
+ ret[cn.t.pieceIndexOfRequestIndex(x)]++
return true
})
return
if cn.t.pieceQueuedForHash(pi) {
panic("piece is queued for hash")
}
- if cn.peerChoking && !cn.peerAllowedFast.Contains(bitmap.BitIndex(pi)) {
+ if cn.peerChoking && !cn.peerAllowedFast.Contains(pi) {
// This could occur if we made a request with the fast extension, and then got choked and
// haven't had the request rejected yet.
if !cn.requestState.Requests.Contains(r) {
break
}
if !c.fastEnabled() {
- if !c.deleteAllRequests().IsEmpty() {
- c.t.iterPeers(func(p *Peer) {
- if p.isLowOnRequests() {
- p.updateRequests("choked by non-fast PeerConn")
- }
- })
- }
+ c.deleteAllRequests("choked by non-fast PeerConn")
} else {
// We don't decrement pending requests here, let's wait for the peer to either
// reject or satisfy the outstanding requests. Additionally, some peers may unchoke
}
c.peerChoking = false
preservedCount := 0
- c.requestState.Requests.Iterate(func(x uint32) bool {
- if !c.peerAllowedFast.Contains(x / c.t.chunksPerRegularPiece()) {
+ c.requestState.Requests.Iterate(func(x RequestIndex) bool {
+ if !c.peerAllowedFast.Contains(c.t.pieceIndexOfRequestIndex(x)) {
preservedCount++
}
return true
}
c.decExpectedChunkReceive(req)
- if c.peerChoking && c.peerAllowedFast.Contains(bitmap.BitIndex(ppReq.Index)) {
+ if c.peerChoking && c.peerAllowedFast.Contains(pieceIndex(ppReq.Index)) {
chunksReceived.Add("due to allowed fast", 1)
}
return true
}
-func (c *Peer) deleteAllRequests() (deleted *roaring.Bitmap) {
- deleted = c.requestState.Requests.Clone()
- deleted.Iterate(func(x uint32) bool {
+func (c *Peer) deleteAllRequests(reason string) {
+ if c.requestState.Requests.IsEmpty() {
+ return
+ }
+ c.requestState.Requests.IterateSnapshot(func(x RequestIndex) bool {
if !c.deleteRequest(x) {
panic("request should exist")
}
return true
})
c.assertNoRequests()
+ c.t.iterPeers(func(p *Peer) {
+ if p.isLowOnRequests() {
+ p.updateRequests(reason)
+ }
+ })
return
}
}
}
-func (c *Peer) cancelAllRequests() (cancelled *roaring.Bitmap) {
- cancelled = c.requestState.Requests.Clone()
- cancelled.Iterate(func(x uint32) bool {
+func (c *Peer) cancelAllRequests() {
+ c.requestState.Requests.IterateSnapshot(func(x RequestIndex) bool {
c.cancel(x)
return true
})
pc := PeerConn{
Peer: Peer{t: tt},
}
+ pc.initRequestState()
pc.peerImpl = &pc
tt.conns[&pc] = struct{}{}
c.Assert(pc.onPeerSentHaveAll(), qt.IsNil)
"fmt"
"sync"
- "github.com/RoaringBitmap/roaring"
"github.com/anacrolix/chansync"
"github.com/anacrolix/missinggo/v2/bitmap"
"github.com/anacrolix/torrent/metainfo"
pp "github.com/anacrolix/torrent/peer_protocol"
"github.com/anacrolix/torrent/storage"
+ "github.com/anacrolix/torrent/typed-roaring"
)
type Piece struct {
}
func (p *Piece) numDirtyChunks() chunkIndexType {
- return chunkIndexType(roaringBitmapRangeCardinality(
+ return chunkIndexType(roaringBitmapRangeCardinality[RequestIndex](
&p.t.dirtyChunks,
p.requestIndexOffset(),
p.t.pieceRequestIndexOffset(p.index+1)))
// Use an iterator to jump between dirty bits.
type undirtiedChunksIter struct {
- TorrentDirtyChunks *roaring.Bitmap
+ TorrentDirtyChunks *typedRoaring.Bitmap[RequestIndex]
StartRequestIndex RequestIndex
EndRequestIndex RequestIndex
}
return false
}
-func (r requestStrategyTorrent) ChunksPerPiece() uint32 {
- return r.t.chunksPerRegularPiece()
-}
-
func (r requestStrategyTorrent) PieceLength() int64 {
return r.t.info.PieceLength
}
)
type (
- RequestIndex = uint32
- ChunkIndex = uint32
+ RequestIndex uint32
+ ChunkIndex = RequestIndex
Request = types.Request
pieceIndex = types.PieceIndex
piecePriority = types.PiecePriority
package request_strategy
import (
- "github.com/RoaringBitmap/roaring"
+ "github.com/anacrolix/torrent/typed-roaring"
)
type PeerRequestState struct {
Interested bool
// Expecting. TODO: This should be ordered so webseed requesters initiate in the same order they
// were assigned.
- Requests roaring.Bitmap
+ Requests PeerRequests
// Cancelled and waiting response
- Cancelled roaring.Bitmap
+ Cancelled typedRoaring.Bitmap[RequestIndex]
+}
+
+// A set of request indices iterable by order added.
+type PeerRequests interface {
+ // Can be more efficient than GetCardinality.
+ IsEmpty() bool
+ // See roaring.Bitmap.GetCardinality.
+ GetCardinality() uint64
+ Contains(RequestIndex) bool
+ // Should not adjust iteration order if item already exists, although I don't think that usage
+ // exists.
+ Add(RequestIndex)
+ // See roaring.Bitmap.Rank.
+ Rank(RequestIndex) uint64
+ // Must yield in order items were added.
+ Iterate(func(RequestIndex) bool)
+ // See roaring.Bitmap.CheckedRemove.
+ CheckedRemove(RequestIndex) bool
+ // Iterate a snapshot of the values. It is safe to mutate the underlying data structure.
+ IterateSnapshot(func(RequestIndex) bool)
}
type Torrent interface {
IgnorePiece(int) bool
- ChunksPerPiece() uint32
PieceLength() int64
}
chunkIndexType = request_strategy.ChunkIndex
)
-type peerRequests struct {
+type desiredPeerRequests struct {
requestIndexes []RequestIndex
peer *Peer
}
-func (p *peerRequests) Len() int {
+func (p *desiredPeerRequests) Len() int {
return len(p.requestIndexes)
}
-func (p *peerRequests) Less(i, j int) bool {
+func (p *desiredPeerRequests) Less(i, j int) bool {
leftRequest := p.requestIndexes[i]
rightRequest := p.requestIndexes[j]
t := p.peer.t
- leftPieceIndex := leftRequest / t.chunksPerRegularPiece()
- rightPieceIndex := rightRequest / t.chunksPerRegularPiece()
+ leftPieceIndex := t.pieceIndexOfRequestIndex(leftRequest)
+ rightPieceIndex := t.pieceIndexOfRequestIndex(rightRequest)
ml := multiless.New()
// Push requests that can't be served right now to the end. But we don't throw them away unless
// there's a better alternative. This is for when we're using the fast extension and get choked
!p.peer.peerAllowedFast.Contains(rightPieceIndex),
)
}
- leftPiece := t.piece(int(leftPieceIndex))
- rightPiece := t.piece(int(rightPieceIndex))
+ leftPiece := t.piece(leftPieceIndex)
+ rightPiece := t.piece(rightPieceIndex)
// Putting this first means we can steal requests from lesser-performing peers for our first few
// new requests.
ml = ml.Int(
return ml.Less()
}
-func (p *peerRequests) Swap(i, j int) {
+func (p *desiredPeerRequests) Swap(i, j int) {
p.requestIndexes[i], p.requestIndexes[j] = p.requestIndexes[j], p.requestIndexes[i]
}
-func (p *peerRequests) Push(x interface{}) {
+func (p *desiredPeerRequests) Push(x interface{}) {
p.requestIndexes = append(p.requestIndexes, x.(RequestIndex))
}
-func (p *peerRequests) Pop() interface{} {
+func (p *desiredPeerRequests) Pop() interface{} {
last := len(p.requestIndexes) - 1
x := p.requestIndexes[last]
p.requestIndexes = p.requestIndexes[:last]
}
type desiredRequestState struct {
- Requests peerRequests
+ Requests desiredPeerRequests
Interested bool
}
return
}
input := p.t.getRequestStrategyInput()
- requestHeap := peerRequests{
+ requestHeap := desiredPeerRequests{
peer: p,
}
request_strategy.GetRequestablePieces(
if !p.peerHasPiece(pieceIndex) {
return
}
- allowedFast := p.peerAllowedFast.ContainsInt(pieceIndex)
+ allowedFast := p.peerAllowedFast.Contains(pieceIndex)
p.t.piece(pieceIndex).undirtiedChunksIter.Iter(func(ci request_strategy.ChunkIndex) {
r := p.t.pieceRequestIndexOffset(pieceIndex) + ci
if !allowedFast {
package torrent
import (
- "github.com/RoaringBitmap/roaring"
+ "github.com/anacrolix/torrent/typed-roaring"
)
// Return the number of bits set in the range. To do this we need the rank of the item before the
// first, and the rank of the last item. An off-by-one minefield. Hopefully I haven't missed
// something in roaring's API that provides this.
-func roaringBitmapRangeCardinality(bm *roaring.Bitmap, start, end uint32) (card uint64) {
+func roaringBitmapRangeCardinality[T typedRoaring.BitConstraint](bm interface{ Rank(T) uint64 }, start, end T) (card uint64) {
card = bm.Rank(end - 1)
if start != 0 {
card -= bm.Rank(start - 1)
"github.com/anacrolix/multiless"
"github.com/anacrolix/sync"
request_strategy "github.com/anacrolix/torrent/request-strategy"
+ typedRoaring "github.com/anacrolix/torrent/typed-roaring"
"github.com/davecgh/go-spew/spew"
"github.com/pion/datachannel"
pendingRequests map[RequestIndex]*Peer
lastRequested map[RequestIndex]time.Time
// Chunks we've written to since the corresponding piece was last checked.
- dirtyChunks roaring.Bitmap
+ dirtyChunks typedRoaring.Bitmap[RequestIndex]
pex pexState
return chunkIndexType((t.pieceLength(piece) + t.chunkSize - 1) / t.chunkSize)
}
-func (t *Torrent) chunksPerRegularPiece() uint32 {
- return uint32((pp.Integer(t.usualPieceSize()) + t.chunkSize - 1) / t.chunkSize)
+func (t *Torrent) chunksPerRegularPiece() chunkIndexType {
+ return chunkIndexType((pp.Integer(t.usualPieceSize()) + t.chunkSize - 1) / t.chunkSize)
}
func (t *Torrent) numRequests() RequestIndex {
if t.numPieces() == 0 {
return 0
}
- return uint32(t.numPieces()-1)*t.chunksPerRegularPiece() + t.pieceNumChunks(t.numPieces()-1)
+ return RequestIndex(t.numPieces()-1)*t.chunksPerRegularPiece() + t.pieceNumChunks(t.numPieces()-1)
}
func (t *Torrent) pendAllChunkSpecs(pieceIndex pieceIndex) {
if !c.peerHasPiece(piece) {
return
}
- if c.requestState.Interested && c.peerChoking && !c.peerAllowedFast.Contains(uint32(piece)) {
+ if c.requestState.Interested && c.peerChoking && !c.peerAllowedFast.Contains(piece) {
return
}
c.updateRequests(reason)
}
}
torrent.Add("deleted connections", 1)
- if !c.deleteAllRequests().IsEmpty() {
- t.iterPeers(func(p *Peer) {
- if p.isLowOnRequests() {
- p.updateRequests("Torrent.deletePeerConn")
- }
- })
- }
+ c.deleteAllRequests("Torrent.deletePeerConn")
t.assertPendingRequests()
if t.numActivePeers() == 0 && len(t.connsWithAllPieces) != 0 {
panic(t.connsWithAllPieces)
activeRequests: make(map[Request]webseed.Request, maxRequests),
maxRequests: maxRequests,
}
+ ws.peer.initRequestState()
for _, opt := range opts {
opt(&ws.client)
}
}
func (t *Torrent) requestIndexFromRequest(r Request) RequestIndex {
- return t.pieceRequestIndexOffset(pieceIndex(r.Index)) + uint32(r.Begin/t.chunkSize)
+ return t.pieceRequestIndexOffset(pieceIndex(r.Index)) + RequestIndex(r.Begin/t.chunkSize)
}
func (t *Torrent) pieceRequestIndexOffset(piece pieceIndex) RequestIndex {
--- /dev/null
+package typedRoaring
+
+import (
+ "github.com/RoaringBitmap/roaring"
+)
+
+type Bitmap[T BitConstraint] struct {
+ roaring.Bitmap
+}
+
+func (me *Bitmap[T]) Contains(x T) bool {
+ return me.Bitmap.Contains(uint32(x))
+}
+
+func (me Bitmap[T]) Iterate(f func(x T) bool) {
+ me.Bitmap.Iterate(func(x uint32) bool {
+ return f(T(x))
+ })
+}
+
+func (me *Bitmap[T]) Add(x T) {
+ me.Bitmap.Add(uint32(x))
+}
+
+func (me *Bitmap[T]) Rank(x T) uint64 {
+ return me.Bitmap.Rank(uint32(x))
+}
+
+func (me *Bitmap[T]) CheckedRemove(x T) bool {
+ return me.Bitmap.CheckedRemove(uint32(x))
+}
+
+func (me *Bitmap[T]) Clone() Bitmap[T] {
+ return Bitmap[T]{*me.Bitmap.Clone()}
+}
+
+func (me *Bitmap[T]) CheckedAdd(x T) bool {
+ return me.Bitmap.CheckedAdd(uint32(x))
+}
+
+func (me *Bitmap[T]) Remove(x T) {
+ me.Bitmap.Remove(uint32(x))
+}
+
+func (me *Bitmap[T]) Iterator() Iterator[T] {
+ return Iterator[T]{me.Bitmap.Iterator()}
+}
--- /dev/null
+package typedRoaring
+
+type BitConstraint interface {
+ ~int | ~uint32
+}
--- /dev/null
+package typedRoaring
+
+import (
+ "github.com/RoaringBitmap/roaring"
+)
+
+type Iterator[T BitConstraint] struct {
+ roaring.IntPeekable
+}
+
+func (t Iterator[T]) Next() T {
+ return T(t.IntPeekable.Next())
+}
+
+func (t Iterator[T]) AdvanceIfNeeded(minVal T) {
+ t.IntPeekable.AdvanceIfNeeded(uint32(minVal))
+}
import (
"testing"
- "github.com/RoaringBitmap/roaring"
+ typedRoaring "github.com/anacrolix/torrent/typed-roaring"
)
func BenchmarkUndirtiedChunksIter(b *testing.B) {
- var bitmap roaring.Bitmap
+ var bitmap typedRoaring.Bitmap[RequestIndex]
a := undirtiedChunksIter{
TorrentDirtyChunks: &bitmap,
StartRequestIndex: 69,
start:
for !ws.peer.closed.IsSet() {
restart := false
- ws.peer.requestState.Requests.Iterate(func(x uint32) bool {
+ ws.peer.requestState.Requests.Iterate(func(x RequestIndex) bool {
r := ws.peer.t.requestIndexToRequest(x)
if _, ok := ws.activeRequests[r]; ok {
return true