dlss
/
FedP2P
4
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Expose request strategies

This commit is contained in:
Matt Joiner 2020-01-24 17:30:57 +11:00
parent 2044457959
commit 48eb7ff3f2
4 changed files with 41 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
config.go
View File

@ -138,7 +138,7 @@ type ClientConfig struct {
// OnQuery hook func // OnQuery hook func
DHTOnQuery func(query *krpc.Msg, source net.Addr) (propagate bool) DHTOnQuery func(query *krpc.Msg, source net.Addr) (propagate bool)
DefaultRequestStrategy requestStrategyMaker DefaultRequestStrategy RequestStrategyMaker
} }
func (cfg *ClientConfig) SetListenAddr(addr string) *ClientConfig { func (cfg *ClientConfig) SetListenAddr(addr string) *ClientConfig {
@ -177,7 +177,7 @@ func NewDefaultClientConfig() *ClientConfig {
ListenPort: 42069, ListenPort: 42069,
Logger: log.Default, Logger: log.Default,
DefaultRequestStrategy: requestStrategyThreeMaker(5 * time.Second), DefaultRequestStrategy: requestStrategyDuplicateRequestTimeout(5 * time.Second),
} }
//cc.ConnTracker.SetNoMaxEntries() //cc.ConnTracker.SetNoMaxEntries()
//cc.ConnTracker.Timeout = func(conntrack.Entry) time.Duration { return 0 } //cc.ConnTracker.Timeout = func(conntrack.Entry) time.Duration { return 0 }

32
connection.go
View File

@ -388,7 +388,7 @@ func (cn *connection) nominalMaxRequests() (ret int) {
} }
// The actual value to use as the maximum outbound requests. // The actual value to use as the maximum outbound requests.
func (rs requestStrategyThree) nominalMaxRequests(cn requestStrategyConnection) (ret int) { func (rs requestStrategyDuplicateRequestTimeout) nominalMaxRequests(cn requestStrategyConnection) (ret int) {
expectingTime := int64(cn.totalExpectingTime()) expectingTime := int64(cn.totalExpectingTime())
if expectingTime == 0 { if expectingTime == 0 {
expectingTime = math.MaxInt64 expectingTime = math.MaxInt64
@ -413,10 +413,10 @@ func defaultNominalMaxRequests(cn requestStrategyConnection) int {
max(64, max(64,
cn.stats().ChunksReadUseful.Int64()-(cn.stats().ChunksRead.Int64()-cn.stats().ChunksReadUseful.Int64()))) cn.stats().ChunksReadUseful.Int64()-(cn.stats().ChunksRead.Int64()-cn.stats().ChunksReadUseful.Int64())))
} }
func (rs requestStrategyOne) nominalMaxRequests(cn requestStrategyConnection) int { func (rs requestStrategyFuzzing) nominalMaxRequests(cn requestStrategyConnection) int {
return defaultNominalMaxRequests(cn) return defaultNominalMaxRequests(cn)
} }
func (rs requestStrategyTwo) nominalMaxRequests(cn requestStrategyConnection) int { func (rs requestStrategyFastest) nominalMaxRequests(cn requestStrategyConnection) int {
return defaultNominalMaxRequests(cn) return defaultNominalMaxRequests(cn)
} }
@ -543,7 +543,7 @@ func (cn *connection) request(r request, mw messageWriter) bool {
}) })
} }
func (rs requestStrategyThree) onSentRequest(r request) { func (rs requestStrategyDuplicateRequestTimeout) onSentRequest(r request) {
rs.lastRequested[r] = time.AfterFunc(rs.duplicateRequestTimeout, func() { rs.lastRequested[r] = time.AfterFunc(rs.duplicateRequestTimeout, func() {
rs.timeoutLocker.Lock() rs.timeoutLocker.Lock()
delete(rs.lastRequested, r) delete(rs.lastRequested, r)
@ -753,7 +753,7 @@ func defaultShouldRequestWithoutBias(cn requestStrategyConnection) bool {
return false return false
} }
func (requestStrategyTwo) shouldRequestWithoutBias(cn requestStrategyConnection) bool { func (requestStrategyFastest) shouldRequestWithoutBias(cn requestStrategyConnection) bool {
if cn.torrent().numReaders() == 0 { if cn.torrent().numReaders() == 0 {
return false return false
} }
@ -766,11 +766,11 @@ func (requestStrategyTwo) shouldRequestWithoutBias(cn requestStrategyConnection)
return false return false
} }
func (requestStrategyOne) shouldRequestWithoutBias(cn requestStrategyConnection) bool { func (requestStrategyFuzzing) shouldRequestWithoutBias(cn requestStrategyConnection) bool {
return defaultShouldRequestWithoutBias(cn) return defaultShouldRequestWithoutBias(cn)
} }
func (requestStrategyThree) shouldRequestWithoutBias(cn requestStrategyConnection) bool { func (requestStrategyDuplicateRequestTimeout) shouldRequestWithoutBias(cn requestStrategyConnection) bool {
return defaultShouldRequestWithoutBias(cn) return defaultShouldRequestWithoutBias(cn)
} }
@ -780,7 +780,7 @@ func (cn *connection) iterPendingPieces(f func(pieceIndex) bool) bool {
} }
return cn.t.requestStrategy.iterPendingPieces(cn, f) return cn.t.requestStrategy.iterPendingPieces(cn, f)
} }
func (requestStrategyThree) iterPendingPieces(cn requestStrategyConnection, f func(pieceIndex) bool) bool { func (requestStrategyDuplicateRequestTimeout) iterPendingPieces(cn requestStrategyConnection, f func(pieceIndex) bool) bool {
return iterUnbiasedPieceRequestOrder(cn, f) return iterUnbiasedPieceRequestOrder(cn, f)
} }
func defaultIterPendingPieces(rs requestStrategy, cn requestStrategyConnection, f func(pieceIndex) bool) bool { func defaultIterPendingPieces(rs requestStrategy, cn requestStrategyConnection, f func(pieceIndex) bool) bool {
@ -792,10 +792,10 @@ func defaultIterPendingPieces(rs requestStrategy, cn requestStrategyConnection,
}) })
} }
} }
func (rs requestStrategyOne) iterPendingPieces(cn requestStrategyConnection, cb func(pieceIndex) bool) bool { func (rs requestStrategyFuzzing) iterPendingPieces(cn requestStrategyConnection, cb func(pieceIndex) bool) bool {
return defaultIterPendingPieces(rs, cn, cb) return defaultIterPendingPieces(rs, cn, cb)
} }
func (rs requestStrategyTwo) iterPendingPieces(cn requestStrategyConnection, cb func(pieceIndex) bool) bool { func (rs requestStrategyFastest) iterPendingPieces(cn requestStrategyConnection, cb func(pieceIndex) bool) bool {
return defaultIterPendingPieces(rs, cn, cb) return defaultIterPendingPieces(rs, cn, cb)
} }
@ -812,7 +812,7 @@ func (cn *connection) iterPendingRequests(piece pieceIndex, f func(request) bool
) )
} }
func (rs requestStrategyThree) iterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool) bool { func (rs requestStrategyDuplicateRequestTimeout) iterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool) bool {
for i := pp.Integer(0); i < pp.Integer(p.numChunks()); i++ { for i := pp.Integer(0); i < pp.Integer(p.numChunks()); i++ {
if p.dirtyChunks().Get(bitmap.BitIndex(i)) { if p.dirtyChunks().Get(bitmap.BitIndex(i)) {
continue continue
@ -840,11 +840,11 @@ func defaultIterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool)
}) })
} }
func (rs requestStrategyOne) iterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool) bool { func (rs requestStrategyFuzzing) iterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool) bool {
return defaultIterUndirtiedChunks(p, f) return defaultIterUndirtiedChunks(p, f)
} }
func (rs requestStrategyTwo) iterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool) bool { func (rs requestStrategyFastest) iterUndirtiedChunks(p requestStrategyPiece, f func(chunkSpec) bool) bool {
return defaultIterUndirtiedChunks(p, f) return defaultIterUndirtiedChunks(p, f)
} }
@ -870,7 +870,7 @@ func (cn *connection) updatePiecePriority(piece pieceIndex) bool {
return cn._pieceRequestOrder.Set(bitmap.BitIndex(piece), prio) || cn.shouldRequestWithoutBias() return cn._pieceRequestOrder.Set(bitmap.BitIndex(piece), prio) || cn.shouldRequestWithoutBias()
} }
func (requestStrategyOne) piecePriority(cn requestStrategyConnection, piece pieceIndex, tpp piecePriority, prio int) int { func (requestStrategyFuzzing) piecePriority(cn requestStrategyConnection, piece pieceIndex, tpp piecePriority, prio int) int {
switch tpp { switch tpp {
case PiecePriorityNormal: case PiecePriorityNormal:
case PiecePriorityReadahead: case PiecePriorityReadahead:
@ -888,11 +888,11 @@ func defaultPiecePriority(cn requestStrategyConnection, piece pieceIndex, tpp pi
return prio return prio
} }
func (requestStrategyTwo) piecePriority(cn requestStrategyConnection, piece pieceIndex, tpp piecePriority, prio int) int { func (requestStrategyFastest) piecePriority(cn requestStrategyConnection, piece pieceIndex, tpp piecePriority, prio int) int {
return defaultPiecePriority(cn, piece, tpp, prio) return defaultPiecePriority(cn, piece, tpp, prio)
} }
func (requestStrategyThree) piecePriority(cn requestStrategyConnection, piece pieceIndex, tpp piecePriority, prio int) int { func (requestStrategyDuplicateRequestTimeout) piecePriority(cn requestStrategyConnection, piece pieceIndex, tpp piecePriority, prio int) int {
return defaultPiecePriority(cn, piece, tpp, prio) return defaultPiecePriority(cn, piece, tpp, prio)
} }

2
piece.go
View File

@ -263,7 +263,7 @@ func (p *Piece) dirtyChunks() bitmap.Bitmap {
return p._dirtyChunks return p._dirtyChunks
} }
func (rs requestStrategyThree) wouldDuplicateRecent(r request) bool { func (rs requestStrategyDuplicateRequestTimeout) wouldDuplicateRecent(r request) bool {
// This piece has been requested on another connection, and the duplicate request timer is still // This piece has been requested on another connection, and the duplicate request timer is still
// running. // running.
_, ok := rs.lastRequested[r] _, ok := rs.lastRequested[r]

31
request_strategy.go
View File

@ -65,17 +65,31 @@ type requestStrategyCallbacks interface {
// Favour higher priority pieces with some fuzzing to reduce overlaps and wastage across // Favour higher priority pieces with some fuzzing to reduce overlaps and wastage across
// connections. // connections.
type requestStrategyOne struct { type requestStrategyFuzzing struct {
requestStrategyDefaults requestStrategyDefaults
} }
// The fastest connection downloads strictly in order of priority, while all others adhere to their // The fastest connection downloads strictly in order of priority, while all others adhere to their
// piece inclinations. // piece inclinations.
type requestStrategyTwo struct { type requestStrategyFastest struct {
requestStrategyDefaults requestStrategyDefaults
} }
func (requestStrategyTwo) ShouldRequestWithoutBias(cn requestStrategyConnection) bool { 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 { if cn.torrent().numReaders() == 0 {
return false return false
} }
@ -90,7 +104,7 @@ func (requestStrategyTwo) ShouldRequestWithoutBias(cn requestStrategyConnection)
// Requests are strictly by piece priority, and not duplicated until duplicateRequestTimeout is // Requests are strictly by piece priority, and not duplicated until duplicateRequestTimeout is
// reached. // reached.
type requestStrategyThree struct { type requestStrategyDuplicateRequestTimeout struct {
// How long to avoid duplicating a pending request. // How long to avoid duplicating a pending request.
duplicateRequestTimeout time.Duration duplicateRequestTimeout time.Duration
@ -103,11 +117,11 @@ type requestStrategyThree struct {
timeoutLocker sync.Locker timeoutLocker sync.Locker
} }
type requestStrategyMaker func(callbacks requestStrategyCallbacks, clientLocker sync.Locker) requestStrategy type RequestStrategyMaker func(callbacks requestStrategyCallbacks, clientLocker sync.Locker) requestStrategy
func requestStrategyThreeMaker(duplicateRequestTimeout time.Duration) requestStrategyMaker { func RequestStrategyDuplicateRequestTimeout(duplicateRequestTimeout time.Duration) RequestStrategyMaker {
return func(callbacks requestStrategyCallbacks, clientLocker sync.Locker) requestStrategy { return func(callbacks requestStrategyCallbacks, clientLocker sync.Locker) requestStrategy {
return requestStrategyThree{ return requestStrategyDuplicateRequestTimeout{
duplicateRequestTimeout: duplicateRequestTimeout, duplicateRequestTimeout: duplicateRequestTimeout,
callbacks: callbacks, callbacks: callbacks,
lastRequested: make(map[request]*time.Timer), lastRequested: make(map[request]*time.Timer),
@ -116,7 +130,7 @@ func requestStrategyThreeMaker(duplicateRequestTimeout time.Duration) requestStr
} }
} }
func (rs requestStrategyThree) hooks() requestStrategyHooks { func (rs requestStrategyDuplicateRequestTimeout) hooks() requestStrategyHooks {
return requestStrategyHooks{ return requestStrategyHooks{
deletedRequest: func(r request) { deletedRequest: func(r request) {
if t, ok := rs.lastRequested[r]; ok { if t, ok := rs.lastRequested[r]; ok {
@ -126,5 +140,4 @@ func (rs requestStrategyThree) hooks() requestStrategyHooks {
}, },
sentRequest: rs.onSentRequest, sentRequest: rs.onSentRequest,
} }
} }