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

Export Peer

This commit is contained in:
Matt Joiner 2021-01-20 13:10:32 +11:00
parent 03b1abafb9
commit 27108bd2f7
11 changed files with 106 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
client.go
View File

@ -1114,7 +1114,7 @@ func (cl *Client) newTorrent(ih metainfo.Hash, specStorage storage.ClientImpl) (
metadataChanged: sync.Cond{ metadataChanged: sync.Cond{
L: cl.locker(), L: cl.locker(),
}, },
webSeeds: make(map[string]*peer), webSeeds: make(map[string]*Peer),
} }
t._pendingPieces.NewSet = priorityBitmapStableNewSet t._pendingPieces.NewSet = priorityBitmapStableNewSet
t.requestStrategy = cl.config.DefaultRequestStrategy(t.requestStrategyCallbacks(), &cl._mu) t.requestStrategy = cl.config.DefaultRequestStrategy(t.requestStrategyCallbacks(), &cl._mu)
@ -1368,7 +1368,7 @@ func (cl *Client) banPeerIP(ip net.IP) {
func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr net.Addr, network, connString string) (c *PeerConn) { func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr net.Addr, network, connString string) (c *PeerConn) {
c = &PeerConn{ c = &PeerConn{
peer: peer{ Peer: Peer{
outgoing: outgoing, outgoing: outgoing,
choking: true, choking: true,
peerChoking: true, peerChoking: true,

2
client_test.go
View File

@ -547,7 +547,7 @@ func TestPeerInvalidHave(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
assert.True(t, _new) assert.True(t, _new)
defer tt.Drop() defer tt.Drop()
cn := &PeerConn{peer: peer{ cn := &PeerConn{Peer: Peer{
t: tt, t: tt,
}} }}
cn.peerImpl = cn cn.peerImpl = cn

2
misc.go
View File

@ -106,7 +106,7 @@ func chunkIndexSpec(index pp.Integer, pieceLength, chunkSize pp.Integer) chunkSp
return ret return ret
} }
func connLessTrusted(l, r *peer) bool { func connLessTrusted(l, r *Peer) bool {
return l.trust().Less(r.trust()) return l.trust().Less(r.trust())
} }

114
peerconn.go
View File

@ -42,7 +42,7 @@ type peerRequestState struct {
data []byte data []byte
} }
type peer struct { type Peer struct {
// First to ensure 64-bit alignment for atomics. See #262. // First to ensure 64-bit alignment for atomics. See #262.
_stats ConnStats _stats ConnStats
@ -120,7 +120,7 @@ type peer struct {
// Maintains the state of a BitTorrent-protocol based connection with a peer. // Maintains the state of a BitTorrent-protocol based connection with a peer.
type PeerConn struct { type PeerConn struct {
peer Peer
// A string that should identify the PeerConn's net.Conn endpoints. The net.Conn could // A string that should identify the PeerConn's net.Conn endpoints. The net.Conn could
// be wrapping WebRTC, uTP, or TCP etc. Used in writing the conn status for peers. // be wrapping WebRTC, uTP, or TCP etc. Used in writing the conn status for peers.
@ -150,7 +150,7 @@ func (cn *PeerConn) connStatusString() string {
return fmt.Sprintf("%+-55q %s %s", cn.PeerID, cn.PeerExtensionBytes, cn.connString) return fmt.Sprintf("%+-55q %s %s", cn.PeerID, cn.PeerExtensionBytes, cn.connString)
} }
func (cn *peer) updateExpectingChunks() { func (cn *Peer) updateExpectingChunks() {
if cn.expectingChunks() { if cn.expectingChunks() {
if cn.lastStartedExpectingToReceiveChunks.IsZero() { if cn.lastStartedExpectingToReceiveChunks.IsZero() {
cn.lastStartedExpectingToReceiveChunks = time.Now() cn.lastStartedExpectingToReceiveChunks = time.Now()
@ -163,7 +163,7 @@ func (cn *peer) updateExpectingChunks() {
} }
} }
func (cn *peer) expectingChunks() bool { func (cn *Peer) expectingChunks() bool {
return cn.interested && !cn.peerChoking return cn.interested && !cn.peerChoking
} }
@ -193,7 +193,7 @@ func (l *PeerConn) hasPreferredNetworkOver(r *PeerConn) (left, ok bool) {
return ml.FinalOk() return ml.FinalOk()
} }
func (cn *peer) cumInterest() time.Duration { func (cn *Peer) cumInterest() time.Duration {
ret := cn.priorInterest ret := cn.priorInterest
if cn.interested { if cn.interested {
ret += time.Since(cn.lastBecameInterested) ret += time.Since(cn.lastBecameInterested)
@ -201,7 +201,7 @@ func (cn *peer) cumInterest() time.Duration {
return ret return ret
} }
func (cn *peer) peerHasAllPieces() (all bool, known bool) { func (cn *Peer) peerHasAllPieces() (all bool, known bool) {
if cn.peerSentHaveAll { if cn.peerSentHaveAll {
return true, true return true, true
} }
@ -215,20 +215,20 @@ func (cn *PeerConn) locker() *lockWithDeferreds {
return cn.t.cl.locker() return cn.t.cl.locker()
} }
func (cn *peer) supportsExtension(ext pp.ExtensionName) bool { func (cn *Peer) supportsExtension(ext pp.ExtensionName) bool {
_, ok := cn.PeerExtensionIDs[ext] _, ok := cn.PeerExtensionIDs[ext]
return ok return ok
} }
// The best guess at number of pieces in the torrent for this peer. // The best guess at number of pieces in the torrent for this peer.
func (cn *peer) bestPeerNumPieces() pieceIndex { func (cn *Peer) bestPeerNumPieces() pieceIndex {
if cn.t.haveInfo() { if cn.t.haveInfo() {
return cn.t.numPieces() return cn.t.numPieces()
} }
return cn.peerMinPieces return cn.peerMinPieces
} }
func (cn *peer) completedString() string { func (cn *Peer) completedString() string {
have := pieceIndex(cn._peerPieces.Len()) have := pieceIndex(cn._peerPieces.Len())
if cn.peerSentHaveAll { if cn.peerSentHaveAll {
have = cn.bestPeerNumPieces() have = cn.bestPeerNumPieces()
@ -275,7 +275,7 @@ func (cn *PeerConn) utp() bool {
} }
// Inspired by https://github.com/transmission/transmission/wiki/Peer-Status-Text. // Inspired by https://github.com/transmission/transmission/wiki/Peer-Status-Text.
func (cn *peer) statusFlags() (ret string) { func (cn *Peer) statusFlags() (ret string) {
c := func(b byte) { c := func(b byte) {
ret += string([]byte{b}) ret += string([]byte{b})
} }
@ -303,11 +303,11 @@ func (cn *peer) statusFlags() (ret string) {
// return buf.String() // return buf.String()
// } // }
func (cn *peer) downloadRate() float64 { func (cn *Peer) downloadRate() float64 {
return float64(cn._stats.BytesReadUsefulData.Int64()) / cn.cumInterest().Seconds() return float64(cn._stats.BytesReadUsefulData.Int64()) / cn.cumInterest().Seconds()
} }
func (cn *peer) writeStatus(w io.Writer, t *Torrent) { func (cn *Peer) writeStatus(w io.Writer, t *Torrent) {
// \t isn't preserved in <pre> blocks? // \t isn't preserved in <pre> blocks?
fmt.Fprintln(w, cn.connStatusString()) fmt.Fprintln(w, cn.connStatusString())
fmt.Fprintf(w, " last msg: %s, connected: %s, last helpful: %s, itime: %s, etime: %s\n", fmt.Fprintf(w, " last msg: %s, connected: %s, last helpful: %s, itime: %s, etime: %s\n",
@ -343,7 +343,7 @@ func (cn *peer) writeStatus(w io.Writer, t *Torrent) {
) )
} }
func (cn *peer) close() { func (cn *Peer) close() {
if !cn.closed.Set() { if !cn.closed.Set() {
return return
} }
@ -365,7 +365,7 @@ func (cn *PeerConn) onClose() {
} }
} }
func (cn *peer) peerHasPiece(piece pieceIndex) bool { func (cn *Peer) peerHasPiece(piece pieceIndex) bool {
return cn.peerSentHaveAll || cn._peerPieces.Contains(bitmap.BitIndex(piece)) return cn.peerSentHaveAll || cn._peerPieces.Contains(bitmap.BitIndex(piece))
} }
@ -430,7 +430,7 @@ func (cn *PeerConn) requestedMetadataPiece(index int) bool {
} }
// The actual value to use as the maximum outbound requests. // The actual value to use as the maximum outbound requests.
func (cn *peer) nominalMaxRequests() (ret int) { func (cn *Peer) nominalMaxRequests() (ret int) {
return int(clamp( return int(clamp(
1, 1,
int64(cn.PeerMaxRequests), int64(cn.PeerMaxRequests),
@ -438,7 +438,7 @@ func (cn *peer) nominalMaxRequests() (ret int) {
)) ))
} }
func (cn *peer) totalExpectingTime() (ret time.Duration) { func (cn *Peer) totalExpectingTime() (ret time.Duration) {
ret = cn.cumulativeExpectedToReceiveChunks ret = cn.cumulativeExpectedToReceiveChunks
if !cn.lastStartedExpectingToReceiveChunks.IsZero() { if !cn.lastStartedExpectingToReceiveChunks.IsZero() {
ret += time.Since(cn.lastStartedExpectingToReceiveChunks) ret += time.Since(cn.lastStartedExpectingToReceiveChunks)
@ -488,7 +488,7 @@ func (cn *PeerConn) unchoke(msg func(pp.Message) bool) bool {
}) })
} }
func (cn *peer) setInterested(interested bool) bool { func (cn *Peer) setInterested(interested bool) bool {
if cn.interested == interested { if cn.interested == interested {
return true return true
} }
@ -519,7 +519,7 @@ func (pc *PeerConn) writeInterested(interested bool) bool {
// are okay. // are okay.
type messageWriter func(pp.Message) bool type messageWriter func(pp.Message) bool
func (cn *peer) request(r request) bool { func (cn *Peer) request(r request) bool {
if _, ok := cn.requests[r]; ok { if _, ok := cn.requests[r]; ok {
panic("chunk already requested") panic("chunk already requested")
} }
@ -572,7 +572,7 @@ func (me *PeerConn) cancel(r request) bool {
return me.write(makeCancelMessage(r)) return me.write(makeCancelMessage(r))
} }
func (cn *peer) doRequestState() bool { func (cn *Peer) doRequestState() bool {
if !cn.t.networkingEnabled || cn.t.dataDownloadDisallowed { if !cn.t.networkingEnabled || cn.t.dataDownloadDisallowed {
if !cn.setInterested(false) { if !cn.setInterested(false) {
return false return false
@ -771,21 +771,21 @@ func iterUnbiasedPieceRequestOrder(cn requestStrategyConnection, f func(piece pi
// conceivable that the best connection should do this, since it's least likely to waste our time if // conceivable that the best connection should do this, since it's least likely to waste our time if
// assigned to the highest priority pieces, and assigning more than one this role would cause // assigned to the highest priority pieces, and assigning more than one this role would cause
// significant wasted bandwidth. // significant wasted bandwidth.
func (cn *peer) shouldRequestWithoutBias() bool { func (cn *Peer) shouldRequestWithoutBias() bool {
return cn.t.requestStrategy.shouldRequestWithoutBias(cn.requestStrategyConnection()) return cn.t.requestStrategy.shouldRequestWithoutBias(cn.requestStrategyConnection())
} }
func (cn *peer) iterPendingPieces(f func(pieceIndex) bool) bool { func (cn *Peer) iterPendingPieces(f func(pieceIndex) bool) bool {
if !cn.t.haveInfo() { if !cn.t.haveInfo() {
return false return false
} }
return cn.t.requestStrategy.iterPendingPieces(cn, f) return cn.t.requestStrategy.iterPendingPieces(cn, f)
} }
func (cn *peer) iterPendingPiecesUntyped(f iter.Callback) { func (cn *Peer) iterPendingPiecesUntyped(f iter.Callback) {
cn.iterPendingPieces(func(i pieceIndex) bool { return f(i) }) cn.iterPendingPieces(func(i pieceIndex) bool { return f(i) })
} }
func (cn *peer) iterPendingRequests(piece pieceIndex, f func(request) bool) bool { func (cn *Peer) iterPendingRequests(piece pieceIndex, f func(request) bool) bool {
return cn.t.requestStrategy.iterUndirtiedChunks( return cn.t.requestStrategy.iterUndirtiedChunks(
cn.t.piece(piece).requestStrategyPiece(), cn.t.piece(piece).requestStrategyPiece(),
func(cs chunkSpec) bool { func(cs chunkSpec) bool {
@ -795,7 +795,7 @@ func (cn *peer) iterPendingRequests(piece pieceIndex, f func(request) bool) bool
} }
// check callers updaterequests // check callers updaterequests
func (cn *peer) stopRequestingPiece(piece pieceIndex) bool { func (cn *Peer) stopRequestingPiece(piece pieceIndex) bool {
return cn._pieceRequestOrder.Remove(bitmap.BitIndex(piece)) return cn._pieceRequestOrder.Remove(bitmap.BitIndex(piece))
} }
@ -803,7 +803,7 @@ func (cn *peer) stopRequestingPiece(piece pieceIndex) bool {
// preference. Connection piece priority is specific to a connection and is // preference. Connection piece priority is specific to a connection and is
// used to pseudorandomly avoid connections always requesting the same pieces // used to pseudorandomly avoid connections always requesting the same pieces
// and thus wasting effort. // and thus wasting effort.
func (cn *peer) updatePiecePriority(piece pieceIndex) bool { func (cn *Peer) updatePiecePriority(piece pieceIndex) bool {
tpp := cn.t.piecePriority(piece) tpp := cn.t.piecePriority(piece)
if !cn.peerHasPiece(piece) { if !cn.peerHasPiece(piece) {
tpp = PiecePriorityNone tpp = PiecePriorityNone
@ -816,14 +816,14 @@ func (cn *peer) 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 (cn *peer) getPieceInclination() []int { func (cn *Peer) getPieceInclination() []int {
if cn.pieceInclination == nil { if cn.pieceInclination == nil {
cn.pieceInclination = cn.t.getConnPieceInclination() cn.pieceInclination = cn.t.getConnPieceInclination()
} }
return cn.pieceInclination return cn.pieceInclination
} }
func (cn *peer) discardPieceInclination() { func (cn *Peer) discardPieceInclination() {
if cn.pieceInclination == nil { if cn.pieceInclination == nil {
return return
} }
@ -843,7 +843,7 @@ func (cn *PeerConn) peerPiecesChanged() {
cn.updateRequests() cn.updateRequests()
} }
} }
cn.t.maybeDropMutuallyCompletePeer(&cn.peer) cn.t.maybeDropMutuallyCompletePeer(&cn.Peer)
} }
func (cn *PeerConn) raisePeerMinPieces(newMin pieceIndex) { func (cn *PeerConn) raisePeerMinPieces(newMin pieceIndex) {
@ -861,7 +861,7 @@ func (cn *PeerConn) peerSentHave(piece pieceIndex) error {
} }
cn.raisePeerMinPieces(piece + 1) cn.raisePeerMinPieces(piece + 1)
cn._peerPieces.Set(bitmap.BitIndex(piece), true) cn._peerPieces.Set(bitmap.BitIndex(piece), true)
cn.t.maybeDropMutuallyCompletePeer(&cn.peer) cn.t.maybeDropMutuallyCompletePeer(&cn.Peer)
if cn.updatePiecePriority(piece) { if cn.updatePiecePriority(piece) {
cn.updateRequests() cn.updateRequests()
} }
@ -944,7 +944,7 @@ func (cn *PeerConn) readMsg(msg *pp.Message) {
// After handshake, we know what Torrent and Client stats to include for a // After handshake, we know what Torrent and Client stats to include for a
// connection. // connection.
func (cn *peer) postHandshakeStats(f func(*ConnStats)) { func (cn *Peer) postHandshakeStats(f func(*ConnStats)) {
t := cn.t t := cn.t
f(&t.stats) f(&t.stats)
f(&t.cl.stats) f(&t.cl.stats)
@ -953,7 +953,7 @@ func (cn *peer) postHandshakeStats(f func(*ConnStats)) {
// All ConnStats that include this connection. Some objects are not known // All ConnStats that include this connection. Some objects are not known
// until the handshake is complete, after which it's expected to reconcile the // until the handshake is complete, after which it's expected to reconcile the
// differences. // differences.
func (cn *peer) allStats(f func(*ConnStats)) { func (cn *Peer) allStats(f func(*ConnStats)) {
f(&cn._stats) f(&cn._stats)
if cn.reconciledHandshakeStats { if cn.reconciledHandshakeStats {
cn.postHandshakeStats(f) cn.postHandshakeStats(f)
@ -970,7 +970,7 @@ func (cn *PeerConn) readBytes(n int64) {
// Returns whether the connection could be useful to us. We're seeding and // Returns whether the connection could be useful to us. We're seeding and
// they want data, we don't have metainfo and they can provide it, etc. // they want data, we don't have metainfo and they can provide it, etc.
func (c *peer) useful() bool { func (c *Peer) useful() bool {
t := c.t t := c.t
if c.closed.IsSet() { if c.closed.IsSet() {
return false return false
@ -987,7 +987,7 @@ func (c *peer) useful() bool {
return false return false
} }
func (c *peer) lastHelpful() (ret time.Time) { func (c *Peer) lastHelpful() (ret time.Time) {
ret = c.lastUsefulChunkReceived ret = c.lastUsefulChunkReceived
if c.t.seeding() && c.lastChunkSent.After(ret) { if c.t.seeding() && c.lastChunkSent.After(ret) {
ret = c.lastChunkSent ret = c.lastChunkSent
@ -1237,13 +1237,13 @@ func (c *PeerConn) mainReadLoop() (err error) {
} }
} }
func (c *peer) remoteRejectedRequest(r request) { func (c *Peer) remoteRejectedRequest(r request) {
if c.deleteRequest(r) { if c.deleteRequest(r) {
c.decExpectedChunkReceive(r) c.decExpectedChunkReceive(r)
} }
} }
func (c *peer) decExpectedChunkReceive(r request) { func (c *Peer) decExpectedChunkReceive(r request) {
count := c.validReceiveChunks[r] count := c.validReceiveChunks[r]
if count == 1 { if count == 1 {
delete(c.validReceiveChunks, r) delete(c.validReceiveChunks, r)
@ -1335,7 +1335,7 @@ func (cn *PeerConn) rw() io.ReadWriter {
} }
// Handle a received chunk from a peer. // Handle a received chunk from a peer.
func (c *peer) receiveChunk(msg *pp.Message) error { func (c *Peer) receiveChunk(msg *pp.Message) error {
t := c.t t := c.t
cl := t.cl cl := t.cl
torrent.Add("chunks received", 1) torrent.Add("chunks received", 1)
@ -1437,14 +1437,14 @@ func (c *peer) receiveChunk(msg *pp.Message) error {
return nil return nil
} }
func (c *peer) onDirtiedPiece(piece pieceIndex) { func (c *Peer) onDirtiedPiece(piece pieceIndex) {
if c.peerTouchedPieces == nil { if c.peerTouchedPieces == nil {
c.peerTouchedPieces = make(map[pieceIndex]struct{}) c.peerTouchedPieces = make(map[pieceIndex]struct{})
} }
c.peerTouchedPieces[piece] = struct{}{} c.peerTouchedPieces[piece] = struct{}{}
ds := &c.t.pieces[piece].dirtiers ds := &c.t.pieces[piece].dirtiers
if *ds == nil { if *ds == nil {
*ds = make(map[*peer]struct{}) *ds = make(map[*Peer]struct{})
} }
(*ds)[c] = struct{}{} (*ds)[c] = struct{}{}
} }
@ -1518,19 +1518,19 @@ func (cn *PeerConn) drop() {
cn.t.dropConnection(cn) cn.t.dropConnection(cn)
} }
func (cn *peer) netGoodPiecesDirtied() int64 { func (cn *Peer) netGoodPiecesDirtied() int64 {
return cn._stats.PiecesDirtiedGood.Int64() - cn._stats.PiecesDirtiedBad.Int64() return cn._stats.PiecesDirtiedGood.Int64() - cn._stats.PiecesDirtiedBad.Int64()
} }
func (c *peer) peerHasWantedPieces() bool { func (c *Peer) peerHasWantedPieces() bool {
return !c._pieceRequestOrder.IsEmpty() return !c._pieceRequestOrder.IsEmpty()
} }
func (c *peer) numLocalRequests() int { func (c *Peer) numLocalRequests() int {
return len(c.requests) return len(c.requests)
} }
func (c *peer) deleteRequest(r request) bool { func (c *Peer) deleteRequest(r request) bool {
if _, ok := c.requests[r]; !ok { if _, ok := c.requests[r]; !ok {
return false return false
} }
@ -1555,7 +1555,7 @@ func (c *peer) deleteRequest(r request) bool {
c.updateRequests() c.updateRequests()
} }
// Give other conns a chance to pick up the request. // Give other conns a chance to pick up the request.
c.t.iterPeers(func(_c *peer) { c.t.iterPeers(func(_c *Peer) {
// We previously checked that the peer wasn't interested to to only wake connections that // We previously checked that the peer wasn't interested to to only wake connections that
// were unable to issue requests due to starvation by the request strategy. There could be // were unable to issue requests due to starvation by the request strategy. There could be
// performance ramifications. // performance ramifications.
@ -1569,7 +1569,7 @@ func (c *peer) deleteRequest(r request) bool {
return true return true
} }
func (c *peer) deleteAllRequests() { func (c *Peer) deleteAllRequests() {
for r := range c.requests { for r := range c.requests {
c.deleteRequest(r) c.deleteRequest(r)
} }
@ -1587,7 +1587,7 @@ func (c *PeerConn) tickleWriter() {
c.writerCond.Broadcast() c.writerCond.Broadcast()
} }
func (c *peer) postCancel(r request) bool { func (c *Peer) postCancel(r request) bool {
if !c.deleteRequest(r) { if !c.deleteRequest(r) {
return false return false
} }
@ -1618,15 +1618,15 @@ func (c *PeerConn) setTorrent(t *Torrent) {
t.reconcileHandshakeStats(c) t.reconcileHandshakeStats(c)
} }
func (c *peer) peerPriority() (peerPriority, error) { func (c *Peer) peerPriority() (peerPriority, error) {
return bep40Priority(c.remoteIpPort(), c.t.cl.publicAddr(c.remoteIp())) return bep40Priority(c.remoteIpPort(), c.t.cl.publicAddr(c.remoteIp()))
} }
func (c *peer) remoteIp() net.IP { func (c *Peer) remoteIp() net.IP {
return addrIpOrNil(c.RemoteAddr) return addrIpOrNil(c.RemoteAddr)
} }
func (c *peer) remoteIpPort() IpPort { func (c *Peer) remoteIpPort() IpPort {
ipa, _ := tryIpPortFromNetAddr(c.RemoteAddr) ipa, _ := tryIpPortFromNetAddr(c.RemoteAddr)
return IpPort{ipa.IP, uint16(ipa.Port)} return IpPort{ipa.IP, uint16(ipa.Port)}
} }
@ -1673,7 +1673,7 @@ func (c *PeerConn) String() string {
return fmt.Sprintf("connection %p", c) return fmt.Sprintf("connection %p", c)
} }
func (c *peer) trust() connectionTrust { func (c *Peer) trust() connectionTrust {
return connectionTrust{c.trusted, c.netGoodPiecesDirtied()} return connectionTrust{c.trusted, c.netGoodPiecesDirtied()}
} }
@ -1686,19 +1686,19 @@ func (l connectionTrust) Less(r connectionTrust) bool {
return multiless.New().Bool(l.Implicit, r.Implicit).Int64(l.NetGoodPiecesDirted, r.NetGoodPiecesDirted).Less() return multiless.New().Bool(l.Implicit, r.Implicit).Int64(l.NetGoodPiecesDirted, r.NetGoodPiecesDirted).Less()
} }
func (cn *peer) requestStrategyConnection() requestStrategyConnection { func (cn *Peer) requestStrategyConnection() requestStrategyConnection {
return cn return cn
} }
func (cn *peer) chunksReceivedWhileExpecting() int64 { func (cn *Peer) chunksReceivedWhileExpecting() int64 {
return cn._chunksReceivedWhileExpecting return cn._chunksReceivedWhileExpecting
} }
func (cn *peer) fastest() bool { func (cn *Peer) fastest() bool {
return cn == cn.t.fastestPeer return cn == cn.t.fastestPeer
} }
func (cn *peer) peerMaxRequests() int { func (cn *Peer) peerMaxRequests() int {
return cn.PeerMaxRequests return cn.PeerMaxRequests
} }
@ -1709,7 +1709,7 @@ func (cn *PeerConn) PeerPieces() bitmap.Bitmap {
return cn.peerPieces() return cn.peerPieces()
} }
func (cn *peer) peerPieces() bitmap.Bitmap { func (cn *Peer) peerPieces() bitmap.Bitmap {
ret := cn._peerPieces.Copy() ret := cn._peerPieces.Copy()
if cn.peerSentHaveAll { if cn.peerSentHaveAll {
ret.AddRange(0, cn.t.numPieces()) ret.AddRange(0, cn.t.numPieces())
@ -1717,14 +1717,14 @@ func (cn *peer) peerPieces() bitmap.Bitmap {
return ret return ret
} }
func (cn *peer) pieceRequestOrder() *prioritybitmap.PriorityBitmap { func (cn *Peer) pieceRequestOrder() *prioritybitmap.PriorityBitmap {
return &cn._pieceRequestOrder return &cn._pieceRequestOrder
} }
func (cn *peer) stats() *ConnStats { func (cn *Peer) stats() *ConnStats {
return &cn._stats return &cn._stats
} }
func (cn *peer) torrent() requestStrategyTorrent { func (cn *Peer) torrent() requestStrategyTorrent {
return cn.t.requestStrategyTorrent() return cn.t.requestStrategyTorrent()
} }

24
peerconn_test.go
View File

@ -155,14 +155,14 @@ func TestConnPexPeerFlags(t *testing.T) {
conn *PeerConn conn *PeerConn
f pp.PexPeerFlags f pp.PexPeerFlags
}{ }{
{&PeerConn{peer: peer{outgoing: false, PeerPrefersEncryption: false}}, 0}, {&PeerConn{Peer: Peer{outgoing: false, PeerPrefersEncryption: false}}, 0},
{&PeerConn{peer: peer{outgoing: false, PeerPrefersEncryption: true}}, pp.PexPrefersEncryption}, {&PeerConn{Peer: Peer{outgoing: false, PeerPrefersEncryption: true}}, pp.PexPrefersEncryption},
{&PeerConn{peer: peer{outgoing: true, PeerPrefersEncryption: false}}, pp.PexOutgoingConn}, {&PeerConn{Peer: Peer{outgoing: true, PeerPrefersEncryption: false}}, pp.PexOutgoingConn},
{&PeerConn{peer: peer{outgoing: true, PeerPrefersEncryption: true}}, pp.PexOutgoingConn | pp.PexPrefersEncryption}, {&PeerConn{Peer: Peer{outgoing: true, PeerPrefersEncryption: true}}, pp.PexOutgoingConn | pp.PexPrefersEncryption},
{&PeerConn{peer: peer{RemoteAddr: udpAddr}}, pp.PexSupportsUtp}, {&PeerConn{Peer: Peer{RemoteAddr: udpAddr}}, pp.PexSupportsUtp},
{&PeerConn{peer: peer{RemoteAddr: udpAddr, outgoing: true}}, pp.PexOutgoingConn | pp.PexSupportsUtp}, {&PeerConn{Peer: Peer{RemoteAddr: udpAddr, outgoing: true}}, pp.PexOutgoingConn | pp.PexSupportsUtp},
{&PeerConn{peer: peer{RemoteAddr: tcpAddr, outgoing: true}}, pp.PexOutgoingConn}, {&PeerConn{Peer: Peer{RemoteAddr: tcpAddr, outgoing: true}}, pp.PexOutgoingConn},
{&PeerConn{peer: peer{RemoteAddr: tcpAddr}}, 0}, {&PeerConn{Peer: Peer{RemoteAddr: tcpAddr}}, 0},
} }
for i, tc := range testcases { for i, tc := range testcases {
f := tc.conn.pexPeerFlags() f := tc.conn.pexPeerFlags()
@ -184,22 +184,22 @@ func TestConnPexEvent(t *testing.T) {
}{ }{
{ {
pexAdd, pexAdd,
&PeerConn{peer: peer{RemoteAddr: udpAddr}}, &PeerConn{Peer: Peer{RemoteAddr: udpAddr}},
pexEvent{pexAdd, udpAddr, pp.PexSupportsUtp}, pexEvent{pexAdd, udpAddr, pp.PexSupportsUtp},
}, },
{ {
pexDrop, pexDrop,
&PeerConn{peer: peer{RemoteAddr: tcpAddr, outgoing: true, PeerListenPort: dialTcpAddr.Port}}, &PeerConn{Peer: Peer{RemoteAddr: tcpAddr, outgoing: true, PeerListenPort: dialTcpAddr.Port}},
pexEvent{pexDrop, tcpAddr, pp.PexOutgoingConn}, pexEvent{pexDrop, tcpAddr, pp.PexOutgoingConn},
}, },
{ {
pexAdd, pexAdd,
&PeerConn{peer: peer{RemoteAddr: tcpAddr, PeerListenPort: dialTcpAddr.Port}}, &PeerConn{Peer: Peer{RemoteAddr: tcpAddr, PeerListenPort: dialTcpAddr.Port}},
pexEvent{pexAdd, dialTcpAddr, 0}, pexEvent{pexAdd, dialTcpAddr, 0},
}, },
{ {
pexDrop, pexDrop,
&PeerConn{peer: peer{RemoteAddr: udpAddr, PeerListenPort: dialUdpAddr.Port}}, &PeerConn{Peer: Peer{RemoteAddr: udpAddr, PeerListenPort: dialUdpAddr.Port}},
pexEvent{pexDrop, dialUdpAddr, pp.PexSupportsUtp}, pexEvent{pexDrop, dialUdpAddr, pp.PexSupportsUtp},
}, },
} }

16
pex_test.go
View File

@ -36,7 +36,7 @@ var (
func TestPexAdded(t *testing.T) { func TestPexAdded(t *testing.T) {
t.Run("noHold", func(t *testing.T) { t.Run("noHold", func(t *testing.T) {
s := new(pexState) s := new(pexState)
s.Add(&PeerConn{peer: peer{RemoteAddr: addrs[0], outgoing: true}}) s.Add(&PeerConn{Peer: Peer{RemoteAddr: addrs[0], outgoing: true}})
targ := &pexState{ targ := &pexState{
ev: []pexEvent{ ev: []pexEvent{
pexEvent{pexAdd, addrs[0], pp.PexOutgoingConn}, pexEvent{pexAdd, addrs[0], pp.PexOutgoingConn},
@ -52,7 +52,7 @@ func TestPexAdded(t *testing.T) {
}, },
nc: 0, nc: 0,
} }
s.Add(&PeerConn{peer: peer{RemoteAddr: addrs[0]}}) s.Add(&PeerConn{Peer: Peer{RemoteAddr: addrs[0]}})
targ := &pexState{ targ := &pexState{
hold: []pexEvent{ hold: []pexEvent{
pexEvent{pexDrop, addrs[1], 0}, pexEvent{pexDrop, addrs[1], 0},
@ -72,7 +72,7 @@ func TestPexAdded(t *testing.T) {
}, },
nc: pexTargAdded, nc: pexTargAdded,
} }
s.Add(&PeerConn{peer: peer{RemoteAddr: addrs[0]}}) s.Add(&PeerConn{Peer: Peer{RemoteAddr: addrs[0]}})
targ := &pexState{ targ := &pexState{
hold: []pexEvent{}, hold: []pexEvent{},
ev: []pexEvent{ ev: []pexEvent{
@ -88,7 +88,7 @@ func TestPexAdded(t *testing.T) {
func TestPexDropped(t *testing.T) { func TestPexDropped(t *testing.T) {
t.Run("belowTarg", func(t *testing.T) { t.Run("belowTarg", func(t *testing.T) {
s := &pexState{nc: 1} s := &pexState{nc: 1}
s.Drop(&PeerConn{peer: peer{RemoteAddr: addrs[0]}, pex: pexConnState{Listed: true}}) s.Drop(&PeerConn{Peer: Peer{RemoteAddr: addrs[0]}, pex: pexConnState{Listed: true}})
targ := &pexState{ targ := &pexState{
hold: []pexEvent{pexEvent{pexDrop, addrs[0], 0}}, hold: []pexEvent{pexEvent{pexDrop, addrs[0], 0}},
nc: 0, nc: 0,
@ -97,7 +97,7 @@ func TestPexDropped(t *testing.T) {
}) })
t.Run("aboveTarg", func(t *testing.T) { t.Run("aboveTarg", func(t *testing.T) {
s := &pexState{nc: pexTargAdded + 1} s := &pexState{nc: pexTargAdded + 1}
s.Drop(&PeerConn{peer: peer{RemoteAddr: addrs[0]}, pex: pexConnState{Listed: true}}) s.Drop(&PeerConn{Peer: Peer{RemoteAddr: addrs[0]}, pex: pexConnState{Listed: true}})
targ := &pexState{ targ := &pexState{
ev: []pexEvent{pexEvent{pexDrop, addrs[0], 0}}, ev: []pexEvent{pexEvent{pexDrop, addrs[0], 0}},
nc: pexTargAdded, nc: pexTargAdded,
@ -106,7 +106,7 @@ func TestPexDropped(t *testing.T) {
}) })
t.Run("aboveTargNotListed", func(t *testing.T) { t.Run("aboveTargNotListed", func(t *testing.T) {
s := &pexState{nc: pexTargAdded + 1} s := &pexState{nc: pexTargAdded + 1}
s.Drop(&PeerConn{peer: peer{RemoteAddr: addrs[0]}, pex: pexConnState{Listed: false}}) s.Drop(&PeerConn{Peer: Peer{RemoteAddr: addrs[0]}, pex: pexConnState{Listed: false}})
targ := &pexState{nc: pexTargAdded + 1} targ := &pexState{nc: pexTargAdded + 1}
require.EqualValues(t, targ, s) require.EqualValues(t, targ, s)
}) })
@ -322,7 +322,7 @@ func TestPexInitialNoCutoff(t *testing.T) {
c := addrgen(n) c := addrgen(n)
for addr := range c { for addr := range c {
s.Add(&PeerConn{peer: peer{RemoteAddr: addr}}) s.Add(&PeerConn{Peer: Peer{RemoteAddr: addr}})
} }
m, seq := s.Genmsg(0) m, seq := s.Genmsg(0)
@ -340,7 +340,7 @@ func benchmarkPexInitialN(b *testing.B, npeers int) {
var s pexState var s pexState
c := addrgen(npeers) c := addrgen(npeers)
for addr := range c { for addr := range c {
s.Add(&PeerConn{peer: peer{RemoteAddr: addr}}) s.Add(&PeerConn{Peer: Peer{RemoteAddr: addr}})
s.Genmsg(0) s.Genmsg(0)
} }
} }

2
piece.go
View File

@ -63,7 +63,7 @@ type Piece struct {
// Connections that have written data to this piece since its last check. // Connections that have written data to this piece since its last check.
// This can include connections that have closed. // This can include connections that have closed.
dirtiers map[*peer]struct{} dirtiers map[*Peer]struct{}
} }
func (p *Piece) String() string { func (p *Piece) String() string {

2
struct_test.go
View File

@ -8,5 +8,5 @@ import (
func TestStructSizes(t *testing.T) { func TestStructSizes(t *testing.T) {
t.Log("[]*File", unsafe.Sizeof([]*File(nil))) t.Log("[]*File", unsafe.Sizeof([]*File(nil)))
t.Log("Piece", unsafe.Sizeof(Piece{})) t.Log("Piece", unsafe.Sizeof(Piece{}))
t.Log("map[*peer]struct{}", unsafe.Sizeof(map[*peer]struct{}(nil))) t.Log("map[*peer]struct{}", unsafe.Sizeof(map[*Peer]struct{}(nil)))
} }

40
torrent.go
View File

@ -85,7 +85,7 @@ type Torrent struct {
fileIndex segments.Index fileIndex segments.Index
files *[]*File files *[]*File
webSeeds map[string]*peer webSeeds map[string]*Peer
// Active peer connections, running message stream loops. TODO: Make this // Active peer connections, running message stream loops. TODO: Make this
// open (not-closed) connections only. // open (not-closed) connections only.
@ -94,7 +94,7 @@ type Torrent struct {
// Set of addrs to which we're attempting to connect. Connections are // Set of addrs to which we're attempting to connect. Connections are
// half-open until all handshakes are completed. // half-open until all handshakes are completed.
halfOpen map[string]PeerInfo halfOpen map[string]PeerInfo
fastestPeer *peer fastestPeer *Peer
// Reserve of peers to connect to. A peer can be both here and in the // Reserve of peers to connect to. A peer can be both here and in the
// active connections if were told about the peer after connecting with // active connections if were told about the peer after connecting with
@ -410,7 +410,7 @@ func (t *Torrent) setInfo(info *metainfo.Info) error {
// This seems to be all the follow-up tasks after info is set, that can't fail. // This seems to be all the follow-up tasks after info is set, that can't fail.
func (t *Torrent) onSetInfo() { func (t *Torrent) onSetInfo() {
t.iterPeers(func(p *peer) { t.iterPeers(func(p *Peer) {
p.onGotInfo(t.info) p.onGotInfo(t.info)
}) })
for i := range t.pieces { for i := range t.pieces {
@ -828,7 +828,7 @@ func (t *Torrent) havePiece(index pieceIndex) bool {
func (t *Torrent) maybeDropMutuallyCompletePeer( func (t *Torrent) maybeDropMutuallyCompletePeer(
// I'm not sure about taking peer here, not all peer implementations actually drop. Maybe that's okay? // I'm not sure about taking peer here, not all peer implementations actually drop. Maybe that's okay?
p *peer, p *Peer,
) { ) {
if !t.cl.config.DropMutuallyCompletePeers { if !t.cl.config.DropMutuallyCompletePeers {
return return
@ -984,7 +984,7 @@ func (t *Torrent) maybeNewConns() {
func (t *Torrent) piecePriorityChanged(piece pieceIndex) { func (t *Torrent) piecePriorityChanged(piece pieceIndex) {
// t.logger.Printf("piece %d priority changed", piece) // t.logger.Printf("piece %d priority changed", piece)
t.iterPeers(func(c *peer) { t.iterPeers(func(c *Peer) {
if c.updatePiecePriority(piece) { if c.updatePiecePriority(piece) {
// log.Print("conn piece priority changed") // log.Print("conn piece priority changed")
c.updateRequests() c.updateRequests()
@ -1294,7 +1294,7 @@ func (t *Torrent) deleteConnection(c *PeerConn) (ret bool) {
} }
func (t *Torrent) numActivePeers() (num int) { func (t *Torrent) numActivePeers() (num int) {
t.iterPeers(func(*peer) { t.iterPeers(func(*Peer) {
num++ num++
}) })
return return
@ -1715,7 +1715,7 @@ func (t *Torrent) SetMaxEstablishedConns(max int) (oldMax int) {
oldMax = t.maxEstablishedConns oldMax = t.maxEstablishedConns
t.maxEstablishedConns = max t.maxEstablishedConns = max
wcs := slices.HeapInterface(slices.FromMapKeys(t.conns), func(l, r *PeerConn) bool { wcs := slices.HeapInterface(slices.FromMapKeys(t.conns), func(l, r *PeerConn) bool {
return worseConn(&l.peer, &r.peer) return worseConn(&l.Peer, &r.Peer)
}) })
for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 { for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 {
t.dropConnection(wcs.Pop().(*PeerConn)) t.dropConnection(wcs.Pop().(*PeerConn))
@ -1782,7 +1782,7 @@ func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
c.stats().incrementPiecesDirtiedBad() c.stats().incrementPiecesDirtiedBad()
} }
bannableTouchers := make([]*peer, 0, len(p.dirtiers)) bannableTouchers := make([]*Peer, 0, len(p.dirtiers))
for c := range p.dirtiers { for c := range p.dirtiers {
if !c.trusted { if !c.trusted {
bannableTouchers = append(bannableTouchers, c) bannableTouchers = append(bannableTouchers, c)
@ -1828,7 +1828,7 @@ func (t *Torrent) onPieceCompleted(piece pieceIndex) {
t.cancelRequestsForPiece(piece) t.cancelRequestsForPiece(piece)
for conn := range t.conns { for conn := range t.conns {
conn.have(piece) conn.have(piece)
t.maybeDropMutuallyCompletePeer(&conn.peer) t.maybeDropMutuallyCompletePeer(&conn.Peer)
} }
} }
@ -1852,7 +1852,7 @@ func (t *Torrent) onIncompletePiece(piece pieceIndex) {
// c.drop() // c.drop()
// } // }
// } // }
t.iterPeers(func(conn *peer) { t.iterPeers(func(conn *Peer) {
if conn.peerHasPiece(piece) { if conn.peerHasPiece(piece) {
conn.updateRequests() conn.updateRequests()
} }
@ -1924,8 +1924,8 @@ func (t *Torrent) clearPieceTouchers(pi pieceIndex) {
} }
} }
func (t *Torrent) peersAsSlice() (ret []*peer) { func (t *Torrent) peersAsSlice() (ret []*Peer) {
t.iterPeers(func(p *peer) { t.iterPeers(func(p *Peer) {
ret = append(ret, p) ret = append(ret, p)
}) })
return return
@ -2016,7 +2016,7 @@ func (cb torrentRequestStrategyCallbacks) requestTimedOut(r request) {
torrent.Add("request timeouts", 1) torrent.Add("request timeouts", 1)
cb.t.cl.lock() cb.t.cl.lock()
defer cb.t.cl.unlock() defer cb.t.cl.unlock()
cb.t.iterPeers(func(cn *peer) { cb.t.iterPeers(func(cn *Peer) {
if cn.peerHasPiece(pieceIndex(r.Index)) { if cn.peerHasPiece(pieceIndex(r.Index)) {
cn.updateRequests() cn.updateRequests()
} }
@ -2045,7 +2045,7 @@ func (t *Torrent) DisallowDataDownload() {
func (t *Torrent) disallowDataDownloadLocked() { func (t *Torrent) disallowDataDownloadLocked() {
t.dataDownloadDisallowed = true t.dataDownloadDisallowed = true
t.iterPeers(func(c *peer) { t.iterPeers(func(c *Peer) {
c.updateRequests() c.updateRequests()
}) })
t.tickleReaders() t.tickleReaders()
@ -2056,7 +2056,7 @@ func (t *Torrent) AllowDataDownload() {
defer t.cl.unlock() defer t.cl.unlock()
t.dataDownloadDisallowed = false t.dataDownloadDisallowed = false
t.tickleReaders() t.tickleReaders()
t.iterPeers(func(c *peer) { t.iterPeers(func(c *Peer) {
c.updateRequests() c.updateRequests()
}) })
} }
@ -2089,9 +2089,9 @@ func (t *Torrent) SetOnWriteChunkError(f func(error)) {
t.userOnWriteChunkErr = f t.userOnWriteChunkErr = f
} }
func (t *Torrent) iterPeers(f func(*peer)) { func (t *Torrent) iterPeers(f func(*Peer)) {
for pc := range t.conns { for pc := range t.conns {
f(&pc.peer) f(&pc.Peer)
} }
for _, ws := range t.webSeeds { for _, ws := range t.webSeeds {
f(ws) f(ws)
@ -2107,7 +2107,7 @@ func (t *Torrent) addWebSeed(url string) {
} }
const maxRequests = 10 const maxRequests = 10
ws := webseedPeer{ ws := webseedPeer{
peer: peer{ peer: Peer{
t: t, t: t,
outgoing: true, outgoing: true,
network: "http", network: "http",
@ -2129,8 +2129,8 @@ func (t *Torrent) addWebSeed(url string) {
t.webSeeds[url] = &ws.peer t.webSeeds[url] = &ws.peer
} }
func (t *Torrent) peerIsActive(p *peer) (active bool) { func (t *Torrent) peerIsActive(p *Peer) (active bool) {
t.iterPeers(func(p1 *peer) { t.iterPeers(func(p1 *Peer) {
if p1 == p { if p1 == p {
active = true active = true
} }

2
webseed-peer.go
View File

@ -15,7 +15,7 @@ import (
type webseedPeer struct { type webseedPeer struct {
client webseed.Client client webseed.Client
requests map[request]webseed.Request requests map[request]webseed.Request
peer peer peer Peer
} }
var _ peerImpl = (*webseedPeer)(nil) var _ peerImpl = (*webseedPeer)(nil)

4
worst_conns.go
View File

@ -8,7 +8,7 @@ import (
"github.com/anacrolix/multiless" "github.com/anacrolix/multiless"
) )
func worseConn(l, r *peer) bool { func worseConn(l, r *Peer) bool {
less, ok := multiless.New().Bool( less, ok := multiless.New().Bool(
l.useful(), r.useful()).CmpInt64( l.useful(), r.useful()).CmpInt64(
l.lastHelpful().Sub(r.lastHelpful()).Nanoseconds()).CmpInt64( l.lastHelpful().Sub(r.lastHelpful()).Nanoseconds()).CmpInt64(
@ -45,7 +45,7 @@ func (me worseConnSlice) Len() int {
} }
func (me worseConnSlice) Less(i, j int) bool { func (me worseConnSlice) Less(i, j int) bool {
return worseConn(&me.conns[i].peer, &me.conns[j].peer) return worseConn(&me.conns[i].Peer, &me.conns[j].Peer)
} }
func (me *worseConnSlice) Pop() interface{} { func (me *worseConnSlice) Pop() interface{} {