FedP2P/request_strategy.go

package torrent

import (
	"math"
	"sync"
	"time"

	"github.com/anacrolix/missinggo/iter"
	"github.com/anacrolix/missinggo/v2/bitmap"
	"github.com/anacrolix/missinggo/v2/prioritybitmap"

	pp "github.com/anacrolix/torrent/peer_protocol"
)

type requestStrategyPiece interface {
	numChunks() pp.Integer
	dirtyChunks() bitmap.Bitmap
	chunkIndexRequest(i pp.Integer) request
}

type requestStrategyTorrent interface {
	numConns() int
	numReaders() int
	numPieces() int
	readerPiecePriorities() (now, readahead bitmap.Bitmap)
	ignorePieces() bitmap.Bitmap
	pendingPieces() *prioritybitmap.PriorityBitmap
}

type requestStrategyConnection interface {
	torrent() requestStrategyTorrent
	peerPieces() bitmap.Bitmap
	pieceRequestOrder() *prioritybitmap.PriorityBitmap
	fastest() bool
	stats() *ConnStats
	totalExpectingTime() time.Duration
	peerMaxRequests() int
	chunksReceivedWhileExpecting() int64
}

type requestStrategyDefaults struct{}

func (requestStrategyDefaults) hooks() requestStrategyHooks {
	return requestStrategyHooks{
		sentRequest:    func(request) {},
		deletedRequest: func(request) {},
	}
}

type requestStrategy interface {
	iterPendingPieces(requestStrategyConnection, func(pieceIndex) bool) bool
	iterUndirtiedChunks(requestStrategyPiece, func(chunkSpec) bool) bool
	nominalMaxRequests(requestStrategyConnection) int
	shouldRequestWithoutBias(requestStrategyConnection) bool
	piecePriority(requestStrategyConnection, pieceIndex, piecePriority, int) int
	hooks() requestStrategyHooks
}

type requestStrategyHooks struct {
	sentRequest    func(request)
	deletedRequest func(request)
}

type requestStrategyCallbacks interface {
	requestTimedOut(request)
}

// Favour higher priority pieces with some fuzzing to reduce overlaps and wastage across
// connections.
type requestStrategyFuzzing struct {
	requestStrategyDefaults
}

// The fastest connection downloads strictly in order of priority, while all others adhere to their
// piece inclinations.
type requestStrategyFastest struct {
	requestStrategyDefaults
}

func newRequestStrategyMaker(rs requestStrategy) RequestStrategyMaker {
	return func(requestStrategyCallbacks, sync.Locker) requestStrategy {
		return rs
	}
}

func RequestStrategyFastest() RequestStrategyMaker {
	return newRequestStrategyMaker(requestStrategyFastest{})
}

func RequestStrategyFuzzing() RequestStrategyMaker {
	return newRequestStrategyMaker(requestStrategyFuzzing{})
}

func (requestStrategyFastest) shouldRequestWithoutBias(cn requestStrategyConnection) bool {
	if cn.torrent().numReaders() == 0 {
		return false
	}
	if cn.torrent().numConns() == 1 {
		return true
	}
	if cn.fastest() {
		return true
	}
	return false
}

// Requests are strictly by piece priority, and not duplicated until duplicateRequestTimeout is
// reached.
type requestStrategyDuplicateRequestTimeout struct {
	requestStrategyDefaults
	// How long to avoid duplicating a pending request.
	duplicateRequestTimeout time.Duration

	callbacks requestStrategyCallbacks

	// The last time we requested a chunk. Deleting the request from any connection will clear this
	// value.
	lastRequested map[request]*time.Timer
	// The lock to take when running a request timeout handler.
	timeoutLocker sync.Locker
}

type RequestStrategyMaker func(callbacks requestStrategyCallbacks, clientLocker sync.Locker) requestStrategy

func RequestStrategyDuplicateRequestTimeout(duplicateRequestTimeout time.Duration) RequestStrategyMaker {
	return func(callbacks requestStrategyCallbacks, clientLocker sync.Locker) requestStrategy {
		return requestStrategyDuplicateRequestTimeout{
			duplicateRequestTimeout: duplicateRequestTimeout,
			callbacks:               callbacks,
			lastRequested:           make(map[request]*time.Timer),
			timeoutLocker:           clientLocker,
		}
	}
}

func (rs requestStrategyDuplicateRequestTimeout) hooks() requestStrategyHooks {
	return requestStrategyHooks{
		deletedRequest: func(r request) {
			if t, ok := rs.lastRequested[r]; ok {
				t.Stop()
				delete(rs.lastRequested, r)
			}
		},
		sentRequest: rs.onSentRequest,
	}
}

func (requestStrategyDefaults) piecePriority(cn requestStrategyConnection, piece pieceIndex, tpp piecePriority, prio int) int {
	return prio
}

func (rs requestStrategyDuplicateRequestTimeout) iterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool) bool {
	for i := pp.Integer(0); i < pp.Integer(p.numChunks()); i++ {
		if p.dirtyChunks().Get(bitmap.BitIndex(i)) {
			continue
		}
		r := p.chunkIndexRequest(i)
		if rs.wouldDuplicateRecent(r) {
			continue
		}
		if !f(r.chunkSpec) {
			return false
		}
	}
	return true
}

func (requestStrategyDefaults) iterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool) bool {
	chunkIndices := p.dirtyChunks().Copy()
	chunkIndices.FlipRange(0, bitmap.BitIndex(p.numChunks()))
	return iter.ForPerm(chunkIndices.Len(), func(i int) bool {
		ci, err := chunkIndices.RB.Select(uint32(i))
		if err != nil {
			panic(err)
		}
		return f(p.chunkIndexRequest(pp.Integer(ci)).chunkSpec)
	})
}

func (requestStrategyFuzzing) piecePriority(cn requestStrategyConnection, piece pieceIndex, tpp piecePriority, prio int) int {
	switch tpp {
	case PiecePriorityNormal:
	case PiecePriorityReadahead:
		prio -= int(cn.torrent().numPieces())
	case PiecePriorityNext, PiecePriorityNow:
		prio -= 2 * int(cn.torrent().numPieces())
	default:
		panic(tpp)
	}
	prio += int(piece / 3)
	return prio
}

func (requestStrategyDuplicateRequestTimeout) iterPendingPieces(cn requestStrategyConnection, f func(pieceIndex) bool) bool {
	return iterUnbiasedPieceRequestOrder(cn, f)
}
func defaultIterPendingPieces(rs requestStrategy, cn requestStrategyConnection, f func(pieceIndex) bool) bool {
	if rs.shouldRequestWithoutBias(cn) {
		return iterUnbiasedPieceRequestOrder(cn, f)
	} else {
		return cn.pieceRequestOrder().IterTyped(func(i int) bool {
			return f(pieceIndex(i))
		})
	}
}
func (rs requestStrategyFuzzing) iterPendingPieces(cn requestStrategyConnection, cb func(pieceIndex) bool) bool {
	return defaultIterPendingPieces(rs, cn, cb)
}
func (rs requestStrategyFastest) iterPendingPieces(cn requestStrategyConnection, cb func(pieceIndex) bool) bool {
	return defaultIterPendingPieces(rs, cn, cb)
}

func (requestStrategyDefaults) shouldRequestWithoutBias(cn requestStrategyConnection) bool {
	return false
}

func (rs requestStrategyDuplicateRequestTimeout) onSentRequest(r request) {
	rs.lastRequested[r] = time.AfterFunc(rs.duplicateRequestTimeout, func() {
		rs.timeoutLocker.Lock()
		delete(rs.lastRequested, r)
		rs.timeoutLocker.Unlock()
		rs.callbacks.requestTimedOut(r)
	})
}

// The actual value to use as the maximum outbound requests.
func (rs requestStrategyDuplicateRequestTimeout) nominalMaxRequests(cn requestStrategyConnection) (ret int) {
	expectingTime := int64(cn.totalExpectingTime())
	if expectingTime == 0 {
		expectingTime = math.MaxInt64
	} else {
		expectingTime *= 2
	}
	return int(clamp(
		1,
		int64(cn.peerMaxRequests()),
		max(
			// It makes sense to always pipeline at least one connection, since latency must be
			// non-zero.
			2,
			// Request only as many as we expect to receive in the duplicateRequestTimeout
			// window. We are trying to avoid having to duplicate requests.
			cn.chunksReceivedWhileExpecting()*int64(rs.duplicateRequestTimeout)/expectingTime,
		),
	))
}
func (requestStrategyDefaults) nominalMaxRequests(cn requestStrategyConnection) int {
	return int(
		max(64,
			cn.stats().ChunksReadUseful.Int64()-(cn.stats().ChunksRead.Int64()-cn.stats().ChunksReadUseful.Int64())))
}

func (rs requestStrategyDuplicateRequestTimeout) wouldDuplicateRecent(r request) bool {
	// This piece has been requested on another connection, and the duplicate request timer is still
	// running.
	_, ok := rs.lastRequested[r]
	return ok
}