PeerSourcePex = "X"
)
+type peerImpl interface {
+ updateRequests()
+ writeInterested(interested bool) bool
+ cancel(request) bool
+ request(request) bool
+ connectionFlags() string
+ drop()
+}
+
type peer struct {
- t *Torrent
+ // First to ensure 64-bit alignment for atomics. See #262.
+ _stats ConnStats
+
+ t *Torrent
+
+ peerImpl
+
connString string
outgoing bool
network string
// Maintains the state of a connection with a peer.
type PeerConn struct {
- // First to ensure 64-bit alignment for atomics. See #262.
- _stats ConnStats
-
peer
// The actual Conn, used for closing, and setting socket options.
writerCond sync.Cond
}
-func (cn *PeerConn) updateExpectingChunks() {
+func (cn *peer) updateExpectingChunks() {
if cn.expectingChunks() {
if cn.lastStartedExpectingToReceiveChunks.IsZero() {
cn.lastStartedExpectingToReceiveChunks = time.Now()
}
}
-func (cn *PeerConn) expectingChunks() bool {
+func (cn *peer) expectingChunks() bool {
return cn.interested && !cn.peerChoking
}
return ml.FinalOk()
}
-func (cn *PeerConn) cumInterest() time.Duration {
+func (cn *peer) cumInterest() time.Duration {
ret := cn.priorInterest
if cn.interested {
ret += time.Since(cn.lastBecameInterested)
}
// The best guess at number of pieces in the torrent for this peer.
-func (cn *PeerConn) bestPeerNumPieces() pieceIndex {
+func (cn *peer) bestPeerNumPieces() pieceIndex {
if cn.t.haveInfo() {
return cn.t.numPieces()
}
return cn.peerMinPieces
}
-func (cn *PeerConn) completedString() string {
+func (cn *peer) completedString() string {
have := pieceIndex(cn._peerPieces.Len())
if cn.peerSentHaveAll {
have = cn.bestPeerNumPieces()
}
// Inspired by https://github.com/transmission/transmission/wiki/Peer-Status-Text.
-func (cn *PeerConn) statusFlags() (ret string) {
+func (cn *peer) statusFlags() (ret string) {
c := func(b byte) {
ret += string([]byte{b})
}
// return buf.String()
// }
-func (cn *PeerConn) downloadRate() float64 {
+func (cn *peer) downloadRate() float64 {
return float64(cn._stats.BytesReadUsefulData.Int64()) / cn.cumInterest().Seconds()
}
-func (cn *PeerConn) writeStatus(w io.Writer, t *Torrent) {
+func (cn *peer) writeStatus(w io.Writer, t *Torrent) {
// \t isn't preserved in <pre> blocks?
fmt.Fprintf(w, "%+-55q %s %s\n", cn.PeerID, cn.PeerExtensionBytes, cn.connString)
fmt.Fprintf(w, " last msg: %s, connected: %s, last helpful: %s, itime: %s, etime: %s\n",
}
}
-func (cn *PeerConn) peerHasPiece(piece pieceIndex) bool {
+func (cn *peer) peerHasPiece(piece pieceIndex) bool {
return cn.peerSentHaveAll || cn._peerPieces.Contains(bitmap.BitIndex(piece))
}
}
// The actual value to use as the maximum outbound requests.
-func (cn *PeerConn) nominalMaxRequests() (ret int) {
+func (cn *peer) nominalMaxRequests() (ret int) {
return int(clamp(
1,
int64(cn.PeerMaxRequests),
))
}
-func (cn *PeerConn) totalExpectingTime() (ret time.Duration) {
+func (cn *peer) totalExpectingTime() (ret time.Duration) {
ret = cn.cumulativeExpectedToReceiveChunks
if !cn.lastStartedExpectingToReceiveChunks.IsZero() {
ret += time.Since(cn.lastStartedExpectingToReceiveChunks)
})
}
-func (cn *PeerConn) setInterested(interested bool, msg func(pp.Message) bool) bool {
+func (cn *peer) setInterested(interested bool) bool {
if cn.interested == interested {
return true
}
}
cn.updateExpectingChunks()
// log.Printf("%p: setting interest: %v", cn, interested)
- return msg(pp.Message{
+ return cn.writeInterested(interested)
+}
+
+func (pc *PeerConn) writeInterested(interested bool) bool {
+ return pc.write(pp.Message{
Type: func() pp.MessageType {
if interested {
return pp.Interested
// are okay.
type messageWriter func(pp.Message) bool
-// Proxies the messageWriter's response.
-func (cn *PeerConn) request(r request, mw messageWriter) bool {
+func (cn *peer) request(r request) bool {
if _, ok := cn.requests[r]; ok {
panic("chunk already requested")
}
if !cn.peerHasPiece(pieceIndex(r.Index)) {
panic("requesting piece peer doesn't have")
}
- if _, ok := cn.t.conns[cn]; !ok {
+ if !cn.t.peerIsActive(cn) {
panic("requesting but not in active conns")
}
if cn.closed.IsSet() {
cn.t.pendingRequests[r]++
cn.t.requestStrategy.hooks().sentRequest(r)
cn.updateExpectingChunks()
- return mw(pp.Message{
+ return cn.peerImpl.request(r)
+}
+
+func (me *PeerConn) request(r request) bool {
+ return me.write(pp.Message{
Type: pp.Request,
Index: r.Index,
Begin: r.Begin,
})
}
-func (cn *PeerConn) fillWriteBuffer(msg func(pp.Message) bool) {
+func (me *PeerConn) cancel(r request) bool {
+ return me.write(makeCancelMessage(r))
+}
+
+func (cn *peer) doRequestState() bool {
if !cn.t.networkingEnabled || cn.t.dataDownloadDisallowed {
- if !cn.setInterested(false, msg) {
- return
+ if !cn.setInterested(false) {
+ return false
}
if len(cn.requests) != 0 {
for r := range cn.requests {
cn.deleteRequest(r)
// log.Printf("%p: cancelling request: %v", cn, r)
- if !msg(makeCancelMessage(r)) {
- return
+ if !cn.peerImpl.cancel(r) {
+ return false
}
}
}
filledBuffer := false
cn.iterPendingPieces(func(pieceIndex pieceIndex) bool {
cn.iterPendingRequests(pieceIndex, func(r request) bool {
- if !cn.setInterested(true, msg) {
+ if !cn.setInterested(true) {
filledBuffer = true
return false
}
if _, ok := cn.requests[r]; ok {
return true
}
- filledBuffer = !cn.request(r, msg)
+ filledBuffer = !cn.request(r)
return !filledBuffer
})
return !filledBuffer
// If we didn't completely top up the requests, we shouldn't mark
// the low water, since we'll want to top up the requests as soon
// as we have more write buffer space.
- return
+ return false
}
cn.requestsLowWater = len(cn.requests) / 2
}
+ return true
+}
+
+func (cn *PeerConn) fillWriteBuffer() {
+ if !cn.doRequestState() {
+ return
+ }
if cn.pex.IsEnabled() {
- if flow := cn.pex.Share(msg); !flow {
+ if flow := cn.pex.Share(cn.write); !flow {
return
}
}
- cn.upload(msg)
+ cn.upload(cn.write)
+}
+
+func (cn *PeerConn) write(msg pp.Message) bool {
+ cn.wroteMsg(&msg)
+ cn.writeBuffer.Write(msg.MustMarshalBinary())
+ torrent.Add(fmt.Sprintf("messages filled of type %s", msg.Type.String()), 1)
+ // 64KiB, but temporarily less to work around an issue with WebRTC. TODO: Update
+ // when https://github.com/pion/datachannel/issues/59 is fixed.
+ return cn.writeBuffer.Len() < 1<<15
}
// Routine that writes to the peer. Some of what to write is buffered by
return
}
if cn.writeBuffer.Len() == 0 {
- cn.fillWriteBuffer(func(msg pp.Message) bool {
- cn.wroteMsg(&msg)
- cn.writeBuffer.Write(msg.MustMarshalBinary())
- torrent.Add(fmt.Sprintf("messages filled of type %s", msg.Type.String()), 1)
- // 64KiB, but temporarily less to work around an issue with WebRTC. TODO: Update
- // when https://github.com/pion/datachannel/issues/59 is fixed.
- return cn.writeBuffer.Len() < 1<<15
- })
+ cn.fillWriteBuffer()
}
if cn.writeBuffer.Len() == 0 && time.Since(lastWrite) >= keepAliveTimeout {
cn.writeBuffer.Write(pp.Message{Keepalive: true}.MustMarshalBinary())
return cn.t.requestStrategy.shouldRequestWithoutBias(cn.requestStrategyConnection())
}
-func (cn *PeerConn) iterPendingPieces(f func(pieceIndex) bool) bool {
+func (cn *peer) iterPendingPieces(f func(pieceIndex) bool) bool {
if !cn.t.haveInfo() {
return false
}
return cn.t.requestStrategy.iterPendingPieces(cn, f)
}
-func (cn *PeerConn) iterPendingPiecesUntyped(f iter.Callback) {
+func (cn *peer) iterPendingPiecesUntyped(f iter.Callback) {
cn.iterPendingPieces(func(i pieceIndex) bool { return f(i) })
}
-func (cn *PeerConn) iterPendingRequests(piece pieceIndex, f func(request) bool) bool {
+func (cn *peer) iterPendingRequests(piece pieceIndex, f func(request) bool) bool {
return cn.t.requestStrategy.iterUndirtiedChunks(
cn.t.piece(piece).requestStrategyPiece(),
func(cs chunkSpec) bool {
// After handshake, we know what Torrent and Client stats to include for a
// connection.
-func (cn *PeerConn) postHandshakeStats(f func(*ConnStats)) {
+func (cn *peer) postHandshakeStats(f func(*ConnStats)) {
t := cn.t
f(&t.stats)
f(&t.cl.stats)
// All ConnStats that include this connection. Some objects are not known
// until the handshake is complete, after which it's expected to reconcile the
// differences.
-func (cn *PeerConn) allStats(f func(*ConnStats)) {
+func (cn *peer) allStats(f func(*ConnStats)) {
f(&cn._stats)
if cn.reconciledHandshakeStats {
cn.postHandshakeStats(f)
return false
}
-func (c *PeerConn) lastHelpful() (ret time.Time) {
+func (c *peer) lastHelpful() (ret time.Time) {
ret = c.lastUsefulChunkReceived
if c.t.seeding() && c.lastChunkSent.After(ret) {
ret = c.lastChunkSent
}
}
-func (c *PeerConn) decExpectedChunkReceive(r request) {
+func (c *peer) decExpectedChunkReceive(r request) {
count := c.validReceiveChunks[r]
if count == 1 {
delete(c.validReceiveChunks, r)
}
// Handle a received chunk from a peer.
-func (c *PeerConn) receiveChunk(msg *pp.Message) error {
+func (c *peer) receiveChunk(msg *pp.Message) error {
t := c.t
cl := t.cl
torrent.Add("chunks received", 1)
return nil
}
-func (c *PeerConn) onDirtiedPiece(piece pieceIndex) {
+func (c *peer) onDirtiedPiece(piece pieceIndex) {
if c.peerTouchedPieces == nil {
c.peerTouchedPieces = make(map[pieceIndex]struct{})
}
c.peerTouchedPieces[piece] = struct{}{}
ds := &c.t.pieces[piece].dirtiers
if *ds == nil {
- *ds = make(map[*PeerConn]struct{})
+ *ds = make(map[*peer]struct{})
}
(*ds)[c] = struct{}{}
}
cn.t.dropConnection(cn)
}
-func (cn *PeerConn) netGoodPiecesDirtied() int64 {
+func (cn *peer) netGoodPiecesDirtied() int64 {
return cn._stats.PiecesDirtiedGood.Int64() - cn._stats.PiecesDirtiedBad.Int64()
}
return !c._pieceRequestOrder.IsEmpty()
}
-func (c *PeerConn) numLocalRequests() int {
+func (c *peer) numLocalRequests() int {
return len(c.requests)
}
-func (c *PeerConn) deleteRequest(r request) bool {
+func (c *peer) deleteRequest(r request) bool {
if _, ok := c.requests[r]; !ok {
return false
}
panic(n)
}
c.updateRequests()
- for _c := range c.t.conns {
+ c.t.iterPeers(func(_c *peer) {
if !_c.interested && _c != c && c.peerHasPiece(pieceIndex(r.Index)) {
_c.updateRequests()
}
- }
+ })
return true
}
return bep40Priority(c.remoteIpPort(), c.t.cl.publicAddr(c.remoteIp()))
}
-func (c *PeerConn) remoteIp() net.IP {
+func (c *peer) remoteIp() net.IP {
return addrIpOrNil(c.remoteAddr)
}
return fmt.Sprintf("connection %p", c)
}
-func (c *PeerConn) trust() connectionTrust {
+func (c *peer) trust() connectionTrust {
return connectionTrust{c.trusted, c.netGoodPiecesDirtied()}
}
return multiless.New().Bool(l.Implicit, r.Implicit).Int64(l.NetGoodPiecesDirted, r.NetGoodPiecesDirted).Less()
}
-func (cn *PeerConn) requestStrategyConnection() requestStrategyConnection {
+func (cn *peer) requestStrategyConnection() requestStrategyConnection {
return cn
}
-func (cn *PeerConn) chunksReceivedWhileExpecting() int64 {
+func (cn *peer) chunksReceivedWhileExpecting() int64 {
return cn._chunksReceivedWhileExpecting
}
-func (cn *PeerConn) fastest() bool {
+func (cn *peer) fastest() bool {
return cn == cn.t.fastestConn
}
-func (cn *PeerConn) peerMaxRequests() int {
+func (cn *peer) peerMaxRequests() int {
return cn.PeerMaxRequests
}
return cn.peerPieces()
}
-func (cn *PeerConn) peerPieces() bitmap.Bitmap {
+func (cn *peer) peerPieces() bitmap.Bitmap {
ret := cn._peerPieces.Copy()
if cn.peerSentHaveAll {
ret.AddRange(0, cn.t.numPieces())
return ret
}
-func (cn *PeerConn) pieceRequestOrder() *prioritybitmap.PriorityBitmap {
+func (cn *peer) pieceRequestOrder() *prioritybitmap.PriorityBitmap {
return &cn._pieceRequestOrder
}
-func (cn *PeerConn) stats() *ConnStats {
+func (cn *peer) stats() *ConnStats {
return &cn._stats
}
-func (cn *PeerConn) torrent() requestStrategyTorrent {
+func (cn *peer) torrent() requestStrategyTorrent {
return cn.t.requestStrategyTorrent()
}