Seperate peer and PeerConn

client.go

@@ -1290,6 +1290,7 @@ func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr net.Addr,
 		conn:        nc,
 		writeBuffer: new(bytes.Buffer),
 	}
+	c.peerImpl = c
 	c.logger = cl.logger.WithValues(c).WithDefaultLevel(log.Debug).WithText(func(m log.Msg) string {
 		return fmt.Sprintf("%v: %s", c, m.Text())
 	})

peerconn.go

@@ -35,8 +35,23 @@ const (
 	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
@@ -111,9 +126,6 @@ type peer struct {
 
 // 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.
@@ -128,7 +140,7 @@ type PeerConn struct {
 	writerCond  sync.Cond
 }
 
-func (cn *PeerConn) updateExpectingChunks() {
+func (cn *peer) updateExpectingChunks() {
 	if cn.expectingChunks() {
 		if cn.lastStartedExpectingToReceiveChunks.IsZero() {
 			cn.lastStartedExpectingToReceiveChunks = time.Now()
@@ -141,7 +153,7 @@ func (cn *PeerConn) updateExpectingChunks() {
 	}
 }
 
-func (cn *PeerConn) expectingChunks() bool {
+func (cn *peer) expectingChunks() bool {
 	return cn.interested && !cn.peerChoking
 }
 
@@ -171,7 +183,7 @@ func (l *PeerConn) hasPreferredNetworkOver(r *PeerConn) (left, ok bool) {
 	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)
@@ -203,14 +215,14 @@ func (cn *PeerConn) supportsExtension(ext pp.ExtensionName) bool {
 }
 
 // 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()
@@ -255,7 +267,7 @@ func (cn *PeerConn) utp() bool {
 }
 
 // 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})
 	}
@@ -283,11 +295,11 @@ func (cn *PeerConn) statusFlags() (ret string) {
 // 	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",
@@ -338,7 +350,7 @@ func (cn *PeerConn) close() {
 	}
 }
 
-func (cn *PeerConn) peerHasPiece(piece pieceIndex) bool {
+func (cn *peer) peerHasPiece(piece pieceIndex) bool {
 	return cn.peerSentHaveAll || cn._peerPieces.Contains(bitmap.BitIndex(piece))
 }
 
@@ -389,7 +401,7 @@ func (cn *PeerConn) requestedMetadataPiece(index int) bool {
 }
 
 // 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),
@@ -397,7 +409,7 @@ func (cn *PeerConn) nominalMaxRequests() (ret int) {
 	))
 }
 
-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)
@@ -447,7 +459,7 @@ func (cn *PeerConn) unchoke(msg func(pp.Message) bool) bool {
 	})
 }
 
-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
 	}
@@ -459,7 +471,11 @@ func (cn *PeerConn) setInterested(interested bool, msg func(pp.Message) bool) bo
 	}
 	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
@@ -474,15 +490,14 @@ func (cn *PeerConn) setInterested(interested bool, msg func(pp.Message) bool) bo
 // are okay.
 type messageWriter func(pp.Message) bool
 
-// Proxies the messageWriter's response.
+func (cn *peer) request(r request) bool {
-func (cn *PeerConn) request(r request, mw messageWriter) 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() {
@@ -512,7 +527,11 @@ func (cn *PeerConn) request(r request, mw messageWriter) bool {
 	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,
@@ -520,17 +539,21 @@ func (cn *PeerConn) request(r request, mw messageWriter) bool {
 	})
 }
 
-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) {
+		if !cn.setInterested(false) {
-			return
+			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)) {
+				if !cn.peerImpl.cancel(r) {
-					return
+					return false
 				}
 			}
 		}
@@ -538,7 +561,7 @@ func (cn *PeerConn) fillWriteBuffer(msg func(pp.Message) bool) {
 		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
 				}
@@ -555,7 +578,7 @@ func (cn *PeerConn) fillWriteBuffer(msg func(pp.Message) bool) {
 				if _, ok := cn.requests[r]; ok {
 					return true
 				}
-				filledBuffer = !cn.request(r, msg)
+				filledBuffer = !cn.request(r)
 				return !filledBuffer
 			})
 			return !filledBuffer
@@ -564,16 +587,32 @@ func (cn *PeerConn) fillWriteBuffer(msg func(pp.Message) bool) {
 			// 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
@@ -602,14 +641,7 @@ func (cn *PeerConn) writer(keepAliveTimeout time.Duration) {
 			return
 		}
 		if cn.writeBuffer.Len() == 0 {
-			cn.fillWriteBuffer(func(msg pp.Message) bool {
+			cn.fillWriteBuffer()
-				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
-			})
 		}
 		if cn.writeBuffer.Len() == 0 && time.Since(lastWrite) >= keepAliveTimeout {
 			cn.writeBuffer.Write(pp.Message{Keepalive: true}.MustMarshalBinary())
@@ -723,17 +755,17 @@ func (cn *PeerConn) shouldRequestWithoutBias() bool {
 	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 {
@@ -890,7 +922,7 @@ func (cn *PeerConn) readMsg(msg *pp.Message) {
 
 // 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)
@@ -899,7 +931,7 @@ func (cn *PeerConn) postHandshakeStats(f func(*ConnStats)) {
 // 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)
@@ -933,7 +965,7 @@ func (c *PeerConn) useful() bool {
 	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
@@ -1126,7 +1158,7 @@ func (c *PeerConn) mainReadLoop() (err error) {
 	}
 }
 
-func (c *PeerConn) decExpectedChunkReceive(r request) {
+func (c *peer) decExpectedChunkReceive(r request) {
 	count := c.validReceiveChunks[r]
 	if count == 1 {
 		delete(c.validReceiveChunks, r)
@@ -1215,7 +1247,7 @@ func (cn *PeerConn) rw() io.ReadWriter {
 }
 
 // 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)
@@ -1314,14 +1346,14 @@ func (c *PeerConn) receiveChunk(msg *pp.Message) error {
 	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{}{}
 }
@@ -1406,7 +1438,7 @@ func (cn *PeerConn) drop() {
 	cn.t.dropConnection(cn)
 }
 
-func (cn *PeerConn) netGoodPiecesDirtied() int64 {
+func (cn *peer) netGoodPiecesDirtied() int64 {
 	return cn._stats.PiecesDirtiedGood.Int64() - cn._stats.PiecesDirtiedBad.Int64()
 }
 
@@ -1414,11 +1446,11 @@ func (c *PeerConn) peerHasWantedPieces() bool {
 	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
 	}
@@ -1435,11 +1467,11 @@ func (c *PeerConn) deleteRequest(r request) bool {
 		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
 }
 
@@ -1503,7 +1535,7 @@ func (c *PeerConn) peerPriority() (peerPriority, error) {
 	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)
 }
 
@@ -1552,7 +1584,7 @@ func (c *PeerConn) String() string {
 	return fmt.Sprintf("connection %p", c)
 }
 
-func (c *PeerConn) trust() connectionTrust {
+func (c *peer) trust() connectionTrust {
 	return connectionTrust{c.trusted, c.netGoodPiecesDirtied()}
 }
 
@@ -1565,19 +1597,19 @@ func (l connectionTrust) Less(r connectionTrust) bool {
 	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
 }
 
@@ -1588,7 +1620,7 @@ func (cn *PeerConn) PeerPieces() bitmap.Bitmap {
 	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())
@@ -1596,14 +1628,14 @@ func (cn *PeerConn) peerPieces() bitmap.Bitmap {
 	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()
 }

piece.go

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

torrent.go

@@ -84,7 +84,7 @@ type Torrent struct {
 	// Set of addrs to which we're attempting to connect. Connections are
 	// half-open until all handshakes are completed.
 	halfOpen    map[string]PeerInfo
-	fastestConn *PeerConn
+	fastestConn *peer
 
 	// 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
@@ -1691,7 +1691,7 @@ func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
 				c.stats().incrementPiecesDirtiedBad()
 			}
 
-			bannableTouchers := make([]*PeerConn, 0, len(p.dirtiers))
+			bannableTouchers := make([]*peer, 0, len(p.dirtiers))
 			for c := range p.dirtiers {
 				if !c.trusted {
 					bannableTouchers = append(bannableTouchers, c)
@@ -1974,3 +1974,18 @@ func (t *Torrent) SetOnWriteChunkError(f func(error)) {
 	defer t.cl.unlock()
 	t.userOnWriteChunkErr = f
 }
+
+func (t *Torrent) iterPeers(f func(*peer)) {
+	for pc := range t.conns {
+		f(&pc.peer)
+	}
+}
+
+func (t *Torrent) peerIsActive(p *peer) (active bool) {
+	t.iterPeers(func(p1 *peer) {
+		if p1 == p {
+			active = true
+		}
+	})
+	return
+}