Peer.go | 4 ++-- Peers.go | 2 +- client.go | 52 ++++++++++++++++++++++++++-------------------------- client_test.go | 2 +- conn_stats.go | 2 +- connection.go => peerconn.go | 318 +++++++++++++++++++++++++++--------------------------- connection_test.go => peerconn_test.go | 4 ++-- misc.go | 2 +- piece.go | 2 +- t.go | 11 +++++++++-- torrent.go | 57 +++++++++++++++++++++++++++-------------------------- worst_conns.go | 6 +++--- diff --git a/Peer.go b/Peer.go index c71aff211bd79431cf519845b1a2b71b0ccad05f..784e38c34a1709031582cdb772f5c53eb48e1109 100644 --- a/Peer.go +++ b/Peer.go @@ -12,7 +12,7 @@ // Peer connection info, handed about publicly. type Peer struct { Id [20]byte Addr net.Addr - Source peerSource + Source PeerSource // Peer is known to support encryption. SupportsEncryption bool peer_protocol.PexPeerFlags @@ -23,7 +23,7 @@ // FromPex generate Peer from peer exchange func (me *Peer) FromPex(na krpc.NodeAddr, fs peer_protocol.PexPeerFlags) { me.Addr = ipPortAddr{append([]byte(nil), na.IP...), na.Port} - me.Source = peerSourcePex + me.Source = PeerSourcePex // If they prefer encryption, they must support it. if fs.Get(peer_protocol.PexPrefersEncryption) { me.SupportsEncryption = true diff --git a/Peers.go b/Peers.go index 0c2d726f70bb05f9cb5b4bb13fbd9083e3f8c59c..a49247e2c61b8b04efeb55a63cbadf483f582e69 100644 --- a/Peers.go +++ b/Peers.go @@ -25,7 +25,7 @@ func (ret Peers) AppendFromTracker(ps []tracker.Peer) Peers { for _, p := range ps { _p := Peer{ Addr: ipPortAddr{p.IP, p.Port}, - Source: peerSourceTracker, + Source: PeerSourceTracker, } copy(_p.Id[:], p.ID) ret = append(ret, _p) diff --git a/client.go b/client.go index f26d94bdaeac56658d21692bb965f4478baa8e64..9945133bdc17948b904d4676576dd1e58d8a153e 100644 --- a/client.go +++ b/client.go @@ -463,7 +463,7 @@ if tc, ok := nc.(*net.TCPConn); ok { tc.SetLinger(0) } c := cl.newConnection(nc, false, nc.RemoteAddr(), nc.RemoteAddr().Network()) - c.Discovery = peerSourceIncoming + c.Discovery = PeerSourceIncoming cl.runReceivedConn(c) } @@ -619,7 +619,7 @@ } // Performs initiator handshakes and returns a connection. Returns nil // *connection if no connection for valid reasons. -func (cl *Client) handshakesConnection(ctx context.Context, nc net.Conn, t *Torrent, encryptHeader bool, remoteAddr net.Addr, network string) (c *connection, err error) { +func (cl *Client) handshakesConnection(ctx context.Context, nc net.Conn, t *Torrent, encryptHeader bool, remoteAddr net.Addr, network string) (c *PeerConn, err error) { c = cl.newConnection(nc, true, remoteAddr, network) c.headerEncrypted = encryptHeader ctx, cancel := context.WithTimeout(ctx, cl.config.HandshakesTimeout) @@ -638,7 +638,7 @@ } // Returns nil connection and nil error if no connection could be established // for valid reasons. -func (cl *Client) establishOutgoingConnEx(t *Torrent, addr net.Addr, obfuscatedHeader bool) (*connection, error) { +func (cl *Client) establishOutgoingConnEx(t *Torrent, addr net.Addr, obfuscatedHeader bool) (*PeerConn, error) { dialCtx, cancel := context.WithTimeout(context.Background(), func() time.Duration { cl.rLock() defer cl.rUnlock() @@ -662,7 +662,7 @@ } // Returns nil connection and nil error if no connection could be established // for valid reasons. -func (cl *Client) establishOutgoingConn(t *Torrent, addr net.Addr) (c *connection, err error) { +func (cl *Client) establishOutgoingConn(t *Torrent, addr net.Addr) (c *PeerConn, err error) { torrent.Add("establish outgoing connection", 1) obfuscatedHeaderFirst := cl.config.HeaderObfuscationPolicy.Preferred c, err = cl.establishOutgoingConnEx(t, addr, obfuscatedHeaderFirst) @@ -687,7 +687,7 @@ } // Called to dial out and run a connection. The addr we're given is already // considered half-open. -func (cl *Client) outgoingConnection(t *Torrent, addr net.Addr, ps peerSource, trusted bool) { +func (cl *Client) outgoingConnection(t *Torrent, addr net.Addr, ps PeerSource, trusted bool) { cl.dialRateLimiter.Wait(context.Background()) c, err := cl.establishOutgoingConn(t, addr) cl.lock() @@ -701,7 +701,7 @@ cl.logger.Printf("error establishing outgoing connection to %v: %v", addr, err) } return } - defer c.Close() + defer c.close() c.Discovery = ps c.trusted = trusted cl.runHandshookConn(c, t) @@ -712,7 +712,7 @@ func (cl *Client) incomingPeerPort() int { return cl.LocalPort() } -func (cl *Client) initiateHandshakes(c *connection, t *Torrent) error { +func (cl *Client) initiateHandshakes(c *PeerConn, t *Torrent) error { if c.headerEncrypted { var rw io.ReadWriter var err error @@ -765,7 +765,7 @@ } } // Do encryption and bittorrent handshakes as receiver. -func (cl *Client) receiveHandshakes(c *connection) (t *Torrent, err error) { +func (cl *Client) receiveHandshakes(c *PeerConn) (t *Torrent, err error) { defer perf.ScopeTimerErr(&err)() var rw io.ReadWriter rw, c.headerEncrypted, c.cryptoMethod, err = handleEncryption(c.rw(), cl.forSkeys, cl.config.HeaderObfuscationPolicy, cl.config.CryptoSelector) @@ -800,7 +800,7 @@ cl.unlock() return } -func (cl *Client) connBtHandshake(c *connection, ih *metainfo.Hash) (ret metainfo.Hash, err error) { +func (cl *Client) connBtHandshake(c *PeerConn, ih *metainfo.Hash) (ret metainfo.Hash, err error) { res, err := pp.Handshake(c.rw(), ih, cl.peerID, cl.extensionBytes) if err != nil { return @@ -812,7 +812,7 @@ c.completedHandshake = time.Now() return } -func (cl *Client) runReceivedConn(c *connection) { +func (cl *Client) runReceivedConn(c *PeerConn) { err := c.conn.SetDeadline(time.Now().Add(cl.config.HandshakesTimeout)) if err != nil { panic(err) @@ -846,7 +846,7 @@ defer cl.unlock() cl.runHandshookConn(c, t) } -func (cl *Client) runHandshookConn(c *connection, t *Torrent) { +func (cl *Client) runHandshookConn(c *PeerConn, t *Torrent) { c.setTorrent(t) if c.PeerID == cl.peerID { if c.outgoing { @@ -880,9 +880,9 @@ } } // See the order given in Transmission's tr_peerMsgsNew. -func (cl *Client) sendInitialMessages(conn *connection, torrent *Torrent) { +func (cl *Client) sendInitialMessages(conn *PeerConn, torrent *Torrent) { if conn.PeerExtensionBytes.SupportsExtended() && cl.extensionBytes.SupportsExtended() { - conn.Post(pp.Message{ + conn.post(pp.Message{ Type: pp.Extended, ExtendedID: pp.HandshakeExtendedID, ExtendedPayload: func() []byte { @@ -911,19 +911,19 @@ } func() { if conn.fastEnabled() { if torrent.haveAllPieces() { - conn.Post(pp.Message{Type: pp.HaveAll}) + conn.post(pp.Message{Type: pp.HaveAll}) conn.sentHaves.AddRange(0, bitmap.BitIndex(conn.t.NumPieces())) return } else if !torrent.haveAnyPieces() { - conn.Post(pp.Message{Type: pp.HaveNone}) + conn.post(pp.Message{Type: pp.HaveNone}) conn.sentHaves.Clear() return } } - conn.PostBitfield() + conn.postBitfield() }() if conn.PeerExtensionBytes.SupportsDHT() && cl.extensionBytes.SupportsDHT() && cl.haveDhtServer() { - conn.Post(pp.Message{ + conn.post(pp.Message{ Type: pp.Port, Port: cl.dhtPort(), }) @@ -945,7 +945,7 @@ return } // Process incoming ut_metadata message. -func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *Torrent, c *connection) error { +func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *Torrent, c *PeerConn) error { var d map[string]int err := bencode.Unmarshal(payload, &d) if _, ok := err.(bencode.ErrUnusedTrailingBytes); ok { @@ -973,12 +973,12 @@ c.lastUsefulChunkReceived = time.Now() return t.maybeCompleteMetadata() case pp.RequestMetadataExtensionMsgType: if !t.haveMetadataPiece(piece) { - c.Post(t.newMetadataExtensionMessage(c, pp.RejectMetadataExtensionMsgType, d["piece"], nil)) + c.post(t.newMetadataExtensionMessage(c, pp.RejectMetadataExtensionMsgType, d["piece"], nil)) return nil } start := (1 << 14) * piece c.logger.Printf("sending metadata piece %d", piece) - c.Post(t.newMetadataExtensionMessage(c, pp.DataMetadataExtensionMsgType, piece, t.metadataBytes[start:start+t.metadataPieceSize(piece)])) + c.post(t.newMetadataExtensionMessage(c, pp.DataMetadataExtensionMsgType, piece, t.metadataBytes[start:start+t.metadataPieceSize(piece)])) return nil case pp.RejectMetadataExtensionMsgType: return nil @@ -1027,7 +1027,7 @@ getPrio: func(p Peer) peerPriority { return bep40PriorityIgnoreError(cl.publicAddr(addrIpOrNil(p.Addr)), p.addr()) }, }, - conns: make(map[*connection]struct{}, 2*cl.config.EstablishedConnsPerTorrent), + conns: make(map[*PeerConn]struct{}, 2*cl.config.EstablishedConnsPerTorrent), halfOpen: make(map[string]Peer), pieceStateChanges: pubsub.NewPubSub(), @@ -1222,12 +1222,12 @@ } cl.badPeerIPs[ip.String()] = struct{}{} } -func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr net.Addr, network string) (c *connection) { - c = &connection{ +func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr net.Addr, network string) (c *PeerConn) { + c = &PeerConn{ conn: nc, outgoing: outgoing, - Choked: true, - PeerChoked: true, + choking: true, + peerChoking: true, PeerMaxRequests: 250, writeBuffer: new(bytes.Buffer), remoteAddr: remoteAddr, @@ -1257,7 +1257,7 @@ return } t.addPeers([]Peer{{ Addr: ipPortAddr{ip, port}, - Source: peerSourceDhtAnnouncePeer, + Source: PeerSourceDhtAnnouncePeer, }}) } diff --git a/client_test.go b/client_test.go index 6748bd30bc66291f0d422a8fddd225d9693fd6bf..5c454633fd8f3183b5ef472a3c694241f5ead931 100644 --- a/client_test.go +++ b/client_test.go @@ -547,7 +547,7 @@ }) require.NoError(t, err) assert.True(t, _new) defer tt.Drop() - cn := &connection{ + cn := &PeerConn{ t: tt, } assert.NoError(t, cn.peerSentHave(0)) diff --git a/conn_stats.go b/conn_stats.go index bb2bc26bc90b38f120ba8cd301894d311b2d3d89..9b0865a08051ea4a1bf61739df806642d8d4810d 100644 --- a/conn_stats.go +++ b/conn_stats.go @@ -105,7 +105,7 @@ } type connStatsReadWriter struct { rw io.ReadWriter - c *connection + c *PeerConn } func (me connStatsReadWriter) Write(b []byte) (n int, err error) { diff --git a/connection.go b/peerconn.go rename from connection.go rename to peerconn.go index edddebd8d54465095841e96246c732d571426473..67ce044152a875dd958f182cb6bcc48296cec0de 100644 --- a/connection.go +++ b/peerconn.go @@ -25,18 +25,18 @@ "github.com/anacrolix/torrent/mse" pp "github.com/anacrolix/torrent/peer_protocol" ) -type peerSource string +type PeerSource string const ( - peerSourceTracker = "Tr" - peerSourceIncoming = "I" - peerSourceDhtGetPeers = "Hg" // Peers we found by searching a DHT. - peerSourceDhtAnnouncePeer = "Ha" // Peers that were announced to us by a DHT. - peerSourcePex = "X" + PeerSourceTracker = "Tr" + PeerSourceIncoming = "I" + PeerSourceDhtGetPeers = "Hg" // Peers we found by searching a DHT. + PeerSourceDhtAnnouncePeer = "Ha" // Peers that were announced to us by a DHT. + PeerSourcePex = "X" ) // Maintains the state of a connection with a peer. -type connection struct { +type PeerConn struct { // First to ensure 64-bit alignment for atomics. See #262. _stats ConnStats @@ -53,7 +53,7 @@ r io.Reader // True if the connection is operating over MSE obfuscation. headerEncrypted bool cryptoMethod mse.CryptoMethod - Discovery peerSource + Discovery PeerSource trusted bool closed missinggo.Event // Set true after we've added our ConnStats generated during handshake to @@ -66,7 +66,7 @@ lastUsefulChunkReceived time.Time lastChunkSent time.Time // Stuff controlled by the local peer. - Interested bool + interested bool lastBecameInterested time.Time priorInterest time.Duration @@ -74,7 +74,7 @@ lastStartedExpectingToReceiveChunks time.Time cumulativeExpectedToReceiveChunks time.Duration _chunksReceivedWhileExpecting int64 - Choked bool + choking bool requests map[request]struct{} requestsLowWater int // Chunks that we might reasonably expect to receive from the peer. Due to @@ -88,9 +88,9 @@ sentHaves bitmap.Bitmap // Stuff controlled by the remote peer. PeerID PeerID - PeerInterested bool - PeerChoked bool - PeerRequests map[request]struct{} + peerInterested bool + peerChoking bool + peerRequests map[request]struct{} PeerExtensionBytes pp.PeerExtensionBits // The pieces the peer has claimed to have. _peerPieces bitmap.Bitmap @@ -119,7 +119,7 @@ logger log.Logger } -func (cn *connection) updateExpectingChunks() { +func (cn *PeerConn) updateExpectingChunks() { if cn.expectingChunks() { if cn.lastStartedExpectingToReceiveChunks.IsZero() { cn.lastStartedExpectingToReceiveChunks = time.Now() @@ -132,12 +132,12 @@ } } } -func (cn *connection) expectingChunks() bool { - return cn.Interested && !cn.PeerChoked +func (cn *PeerConn) expectingChunks() bool { + return cn.interested && !cn.peerChoking } // Returns true if the connection is over IPv6. -func (cn *connection) ipv6() bool { +func (cn *PeerConn) ipv6() bool { ip := addrIpOrNil(cn.remoteAddr) if ip.To4() != nil { return false @@ -147,14 +147,14 @@ } // Returns true the if the dialer/initiator has the lower client peer ID. TODO: Find the // specification for this. -func (cn *connection) isPreferredDirection() bool { +func (cn *PeerConn) isPreferredDirection() bool { return bytes.Compare(cn.t.cl.peerID[:], cn.PeerID[:]) < 0 == cn.outgoing } // Returns whether the left connection should be preferred over the right one, // considering only their networking properties. If ok is false, we can't // decide. -func (l *connection) hasPreferredNetworkOver(r *connection) (left, ok bool) { +func (l *PeerConn) hasPreferredNetworkOver(r *PeerConn) (left, ok bool) { var ml multiLess ml.NextBool(l.isPreferredDirection(), r.isPreferredDirection()) ml.NextBool(!l.utp(), !r.utp()) @@ -162,15 +162,15 @@ ml.NextBool(l.ipv6(), r.ipv6()) return ml.FinalOk() } -func (cn *connection) cumInterest() time.Duration { +func (cn *PeerConn) cumInterest() time.Duration { ret := cn.priorInterest - if cn.Interested { + if cn.interested { ret += time.Since(cn.lastBecameInterested) } return ret } -func (cn *connection) peerHasAllPieces() (all bool, known bool) { +func (cn *PeerConn) peerHasAllPieces() (all bool, known bool) { if cn.peerSentHaveAll { return true, true } @@ -180,28 +180,28 @@ } return bitmap.Flip(cn._peerPieces, 0, bitmap.BitIndex(cn.t.numPieces())).IsEmpty(), true } -func (cn *connection) mu() sync.Locker { +func (cn *PeerConn) mu() sync.Locker { return cn.t.cl.locker() } -func (cn *connection) localAddr() net.Addr { +func (cn *PeerConn) localAddr() net.Addr { return cn.conn.LocalAddr() } -func (cn *connection) supportsExtension(ext pp.ExtensionName) bool { +func (cn *PeerConn) supportsExtension(ext pp.ExtensionName) bool { _, ok := cn.PeerExtensionIDs[ext] return ok } // The best guess at number of pieces in the torrent for this peer. -func (cn *connection) bestPeerNumPieces() pieceIndex { +func (cn *PeerConn) bestPeerNumPieces() pieceIndex { if cn.t.haveInfo() { return cn.t.numPieces() } return cn.peerMinPieces } -func (cn *connection) completedString() string { +func (cn *PeerConn) completedString() string { have := pieceIndex(cn._peerPieces.Len()) if cn.peerSentHaveAll { have = cn.bestPeerNumPieces() @@ -212,7 +212,7 @@ // Correct the PeerPieces slice length. Return false if the existing slice is // invalid, such as by receiving badly sized BITFIELD, or invalid HAVE // messages. -func (cn *connection) setNumPieces(num pieceIndex) error { +func (cn *PeerConn) setNumPieces(num pieceIndex) error { cn._peerPieces.RemoveRange(bitmap.BitIndex(num), bitmap.ToEnd) cn.peerPiecesChanged() return nil @@ -225,7 +225,7 @@ } return fmt.Sprintf("%.2fs ago", time.Since(t).Seconds()) } -func (cn *connection) connectionFlags() (ret string) { +func (cn *PeerConn) connectionFlags() (ret string) { c := func(b byte) { ret += string([]byte{b}) } @@ -241,28 +241,28 @@ } return } -func (cn *connection) utp() bool { +func (cn *PeerConn) utp() bool { return parseNetworkString(cn.network).Udp } // Inspired by https://github.com/transmission/transmission/wiki/Peer-Status-Text. -func (cn *connection) statusFlags() (ret string) { +func (cn *PeerConn) statusFlags() (ret string) { c := func(b byte) { ret += string([]byte{b}) } - if cn.Interested { + if cn.interested { c('i') } - if cn.Choked { + if cn.choking { c('c') } c('-') ret += cn.connectionFlags() c('-') - if cn.PeerInterested { + if cn.peerInterested { c('i') } - if cn.PeerChoked { + if cn.peerChoking { c('c') } return @@ -270,15 +270,15 @@ } // func (cn *connection) String() string { // var buf bytes.Buffer -// cn.WriteStatus(&buf, nil) +// cn.writeStatus(&buf, nil) // return buf.String() // } -func (cn *connection) downloadRate() float64 { +func (cn *PeerConn) downloadRate() float64 { return float64(cn._stats.BytesReadUsefulData.Int64()) / cn.cumInterest().Seconds() } -func (cn *connection) WriteStatus(w io.Writer, t *Torrent) { +func (cn *PeerConn) writeStatus(w io.Writer, t *Torrent) { // \t isn't preserved in
blocks?
fmt.Fprintf(w, "%+-55q %s %s-%s\n", cn.PeerID, cn.PeerExtensionBytes, cn.localAddr(), cn.remoteAddr)
fmt.Fprintf(w, " last msg: %s, connected: %s, last helpful: %s, itime: %s, etime: %s\n",
@@ -298,7 +298,7 @@ &cn._stats.ChunksWritten,
cn.requestsLowWater,
cn.numLocalRequests(),
cn.nominalMaxRequests(),
- len(cn.PeerRequests),
+ len(cn.peerRequests),
cn.statusFlags(),
cn.downloadRate()/(1<<10),
)
@@ -314,7 +314,7 @@ }(),
)
}
-func (cn *connection) Close() {
+func (cn *PeerConn) close() {
if !cn.closed.Set() {
return
}
@@ -326,12 +326,12 @@ go cn.conn.Close()
}
}
-func (cn *connection) PeerHasPiece(piece pieceIndex) bool {
+func (cn *PeerConn) peerHasPiece(piece pieceIndex) bool {
return cn.peerSentHaveAll || cn._peerPieces.Contains(bitmap.BitIndex(piece))
}
// Writes a message into the write buffer.
-func (cn *connection) Post(msg pp.Message) {
+func (cn *PeerConn) post(msg pp.Message) {
torrent.Add(fmt.Sprintf("messages posted of type %s", msg.Type.String()), 1)
// We don't need to track bytes here because a connection.w Writer wrapper
// takes care of that (although there's some delay between us recording
@@ -343,7 +343,7 @@ cn.wroteMsg(&msg)
cn.tickleWriter()
}
-func (cn *connection) requestMetadataPiece(index int) {
+func (cn *PeerConn) requestMetadataPiece(index int) {
eID := cn.PeerExtensionIDs[pp.ExtensionNameMetadata]
if eID == 0 {
return
@@ -352,7 +352,7 @@ if index < len(cn.metadataRequests) && cn.metadataRequests[index] {
return
}
cn.logger.Printf("requesting metadata piece %d", index)
- cn.Post(pp.Message{
+ cn.post(pp.Message{
Type: pp.Extended,
ExtendedID: eID,
ExtendedPayload: func() []byte {
@@ -372,12 +372,12 @@ }
cn.metadataRequests[index] = true
}
-func (cn *connection) requestedMetadataPiece(index int) bool {
+func (cn *PeerConn) requestedMetadataPiece(index int) bool {
return index < len(cn.metadataRequests) && cn.metadataRequests[index]
}
// The actual value to use as the maximum outbound requests.
-func (cn *connection) nominalMaxRequests() (ret int) {
+func (cn *PeerConn) nominalMaxRequests() (ret int) {
return int(clamp(
1,
int64(cn.PeerMaxRequests),
@@ -385,7 +385,7 @@ int64(cn.t.requestStrategy.nominalMaxRequests(cn.requestStrategyConnection())),
))
}
-func (cn *connection) totalExpectingTime() (ret time.Duration) {
+func (cn *PeerConn) totalExpectingTime() (ret time.Duration) {
ret = cn.cumulativeExpectedToReceiveChunks
if !cn.lastStartedExpectingToReceiveChunks.IsZero() {
ret += time.Since(cn.lastStartedExpectingToReceiveChunks)
@@ -394,52 +394,52 @@ return
}
-func (cn *connection) onPeerSentCancel(r request) {
- if _, ok := cn.PeerRequests[r]; !ok {
+func (cn *PeerConn) onPeerSentCancel(r request) {
+ if _, ok := cn.peerRequests[r]; !ok {
torrent.Add("unexpected cancels received", 1)
return
}
if cn.fastEnabled() {
cn.reject(r)
} else {
- delete(cn.PeerRequests, r)
+ delete(cn.peerRequests, r)
}
}
-func (cn *connection) Choke(msg messageWriter) (more bool) {
- if cn.Choked {
+func (cn *PeerConn) choke(msg messageWriter) (more bool) {
+ if cn.choking {
return true
}
- cn.Choked = true
+ cn.choking = true
more = msg(pp.Message{
Type: pp.Choke,
})
if cn.fastEnabled() {
- for r := range cn.PeerRequests {
+ for r := range cn.peerRequests {
// TODO: Don't reject pieces in allowed fast set.
cn.reject(r)
}
} else {
- cn.PeerRequests = nil
+ cn.peerRequests = nil
}
return
}
-func (cn *connection) Unchoke(msg func(pp.Message) bool) bool {
- if !cn.Choked {
+func (cn *PeerConn) unchoke(msg func(pp.Message) bool) bool {
+ if !cn.choking {
return true
}
- cn.Choked = false
+ cn.choking = false
return msg(pp.Message{
Type: pp.Unchoke,
})
}
-func (cn *connection) SetInterested(interested bool, msg func(pp.Message) bool) bool {
- if cn.Interested == interested {
+func (cn *PeerConn) setInterested(interested bool, msg func(pp.Message) bool) bool {
+ if cn.interested == interested {
return true
}
- cn.Interested = interested
+ cn.interested = interested
if interested {
cn.lastBecameInterested = time.Now()
} else if !cn.lastBecameInterested.IsZero() {
@@ -463,11 +463,11 @@ // are okay.
type messageWriter func(pp.Message) bool
// Proxies the messageWriter's response.
-func (cn *connection) request(r request, mw messageWriter) 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)) {
+ if !cn.peerHasPiece(pieceIndex(r.Index)) {
panic("requesting piece peer doesn't have")
}
if _, ok := cn.t.conns[cn]; !ok {
@@ -476,11 +476,11 @@ }
if cn.closed.IsSet() {
panic("requesting when connection is closed")
}
- if cn.PeerChoked {
+ if cn.peerChoking {
if cn.peerAllowedFast.Get(int(r.Index)) {
torrent.Add("allowed fast requests sent", 1)
} else {
- panic("requesting while choked and not allowed fast")
+ panic("requesting while choking and not allowed fast")
}
}
if cn.t.hashingPiece(pieceIndex(r.Index)) {
@@ -508,9 +508,9 @@ Length: r.Length,
})
}
-func (cn *connection) fillWriteBuffer(msg func(pp.Message) bool) {
+func (cn *PeerConn) fillWriteBuffer(msg func(pp.Message) bool) {
if !cn.t.networkingEnabled {
- if !cn.SetInterested(false, msg) {
+ if !cn.setInterested(false, msg) {
return
}
if len(cn.requests) != 0 {
@@ -527,7 +527,7 @@ if len(cn.requests) <= cn.requestsLowWater {
filledBuffer := false
cn.iterPendingPieces(func(pieceIndex pieceIndex) bool {
cn.iterPendingRequests(pieceIndex, func(r request) bool {
- if !cn.SetInterested(true, msg) {
+ if !cn.setInterested(true, msg) {
filledBuffer = true
return false
}
@@ -536,7 +536,7 @@ return false
}
// Choking is looked at here because our interest is dependent
// on whether we'd make requests in its absence.
- if cn.PeerChoked {
+ if cn.peerChoking {
if !cn.peerAllowedFast.Get(bitmap.BitIndex(r.Index)) {
return false
}
@@ -564,7 +564,7 @@
// Routine that writes to the peer. Some of what to write is buffered by
// activity elsewhere in the Client, and some is determined locally when the
// connection is writable.
-func (cn *connection) writer(keepAliveTimeout time.Duration) {
+func (cn *PeerConn) writer(keepAliveTimeout time.Duration) {
var (
lastWrite time.Time = time.Now()
keepAliveTimer *time.Timer
@@ -579,7 +579,7 @@ keepAliveTimer.Reset(keepAliveTimeout)
})
cn.mu().Lock()
defer cn.mu().Unlock()
- defer cn.Close()
+ defer cn.close()
defer keepAliveTimer.Stop()
frontBuf := new(bytes.Buffer)
for {
@@ -622,32 +622,32 @@ frontBuf.Reset()
}
}
-func (cn *connection) Have(piece pieceIndex) {
+func (cn *PeerConn) have(piece pieceIndex) {
if cn.sentHaves.Get(bitmap.BitIndex(piece)) {
return
}
- cn.Post(pp.Message{
+ cn.post(pp.Message{
Type: pp.Have,
Index: pp.Integer(piece),
})
cn.sentHaves.Add(bitmap.BitIndex(piece))
}
-func (cn *connection) PostBitfield() {
+func (cn *PeerConn) postBitfield() {
if cn.sentHaves.Len() != 0 {
panic("bitfield must be first have-related message sent")
}
if !cn.t.haveAnyPieces() {
return
}
- cn.Post(pp.Message{
+ cn.post(pp.Message{
Type: pp.Bitfield,
Bitfield: cn.t.bitfield(),
})
cn.sentHaves = cn.t._completedPieces.Copy()
}
-func (cn *connection) updateRequests() {
+func (cn *PeerConn) updateRequests() {
// log.Print("update requests")
cn.tickleWriter()
}
@@ -701,21 +701,21 @@ // fastest connection, and we have active readers that signal an ordering preference. It's
// 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
// significant wasted bandwidth.
-func (cn *connection) shouldRequestWithoutBias() bool {
+func (cn *PeerConn) shouldRequestWithoutBias() bool {
return cn.t.requestStrategy.shouldRequestWithoutBias(cn.requestStrategyConnection())
}
-func (cn *connection) iterPendingPieces(f func(pieceIndex) bool) bool {
+func (cn *PeerConn) iterPendingPieces(f func(pieceIndex) bool) bool {
if !cn.t.haveInfo() {
return false
}
return cn.t.requestStrategy.iterPendingPieces(cn, f)
}
-func (cn *connection) iterPendingPiecesUntyped(f iter.Callback) {
+func (cn *PeerConn) iterPendingPiecesUntyped(f iter.Callback) {
cn.iterPendingPieces(func(i pieceIndex) bool { return f(i) })
}
-func (cn *connection) iterPendingRequests(piece pieceIndex, f func(request) bool) bool {
+func (cn *PeerConn) iterPendingRequests(piece pieceIndex, f func(request) bool) bool {
return cn.t.requestStrategy.iterUndirtiedChunks(
cn.t.piece(piece).requestStrategyPiece(),
func(cs chunkSpec) bool {
@@ -725,7 +725,7 @@ )
}
// check callers updaterequests
-func (cn *connection) stopRequestingPiece(piece pieceIndex) bool {
+func (cn *PeerConn) stopRequestingPiece(piece pieceIndex) bool {
return cn._pieceRequestOrder.Remove(bitmap.BitIndex(piece))
}
@@ -733,9 +733,9 @@ // This is distinct from Torrent piece priority, which is the user's
// preference. Connection piece priority is specific to a connection and is
// used to pseudorandomly avoid connections always requesting the same pieces
// and thus wasting effort.
-func (cn *connection) updatePiecePriority(piece pieceIndex) bool {
+func (cn *PeerConn) updatePiecePriority(piece pieceIndex) bool {
tpp := cn.t.piecePriority(piece)
- if !cn.PeerHasPiece(piece) {
+ if !cn.peerHasPiece(piece) {
tpp = PiecePriorityNone
}
if tpp == PiecePriorityNone {
@@ -746,14 +746,14 @@ prio = cn.t.requestStrategy.piecePriority(cn, piece, tpp, prio)
return cn._pieceRequestOrder.Set(bitmap.BitIndex(piece), prio) || cn.shouldRequestWithoutBias()
}
-func (cn *connection) getPieceInclination() []int {
+func (cn *PeerConn) getPieceInclination() []int {
if cn.pieceInclination == nil {
cn.pieceInclination = cn.t.getConnPieceInclination()
}
return cn.pieceInclination
}
-func (cn *connection) discardPieceInclination() {
+func (cn *PeerConn) discardPieceInclination() {
if cn.pieceInclination == nil {
return
}
@@ -761,7 +761,7 @@ cn.t.putPieceInclination(cn.pieceInclination)
cn.pieceInclination = nil
}
-func (cn *connection) peerPiecesChanged() {
+func (cn *PeerConn) peerPiecesChanged() {
if cn.t.haveInfo() {
prioritiesChanged := false
for i := pieceIndex(0); i < cn.t.numPieces(); i++ {
@@ -775,17 +775,17 @@ }
}
}
-func (cn *connection) raisePeerMinPieces(newMin pieceIndex) {
+func (cn *PeerConn) raisePeerMinPieces(newMin pieceIndex) {
if newMin > cn.peerMinPieces {
cn.peerMinPieces = newMin
}
}
-func (cn *connection) peerSentHave(piece pieceIndex) error {
+func (cn *PeerConn) peerSentHave(piece pieceIndex) error {
if cn.t.haveInfo() && piece >= cn.t.numPieces() || piece < 0 {
return errors.New("invalid piece")
}
- if cn.PeerHasPiece(piece) {
+ if cn.peerHasPiece(piece) {
return nil
}
cn.raisePeerMinPieces(piece + 1)
@@ -796,7 +796,7 @@ }
return nil
}
-func (cn *connection) peerSentBitfield(bf []bool) error {
+func (cn *PeerConn) peerSentBitfield(bf []bool) error {
cn.peerSentHaveAll = false
if len(bf)%8 != 0 {
panic("expected bitfield length divisible by 8")
@@ -818,21 +818,21 @@ cn.peerPiecesChanged()
return nil
}
-func (cn *connection) onPeerSentHaveAll() error {
+func (cn *PeerConn) onPeerSentHaveAll() error {
cn.peerSentHaveAll = true
cn._peerPieces.Clear()
cn.peerPiecesChanged()
return nil
}
-func (cn *connection) peerSentHaveNone() error {
+func (cn *PeerConn) peerSentHaveNone() error {
cn._peerPieces.Clear()
cn.peerSentHaveAll = false
cn.peerPiecesChanged()
return nil
}
-func (c *connection) requestPendingMetadata() {
+func (c *PeerConn) requestPendingMetadata() {
if c.t.haveInfo() {
return
}
@@ -853,18 +853,18 @@ c.requestMetadataPiece(i)
}
}
-func (cn *connection) wroteMsg(msg *pp.Message) {
+func (cn *PeerConn) wroteMsg(msg *pp.Message) {
torrent.Add(fmt.Sprintf("messages written of type %s", msg.Type.String()), 1)
cn.allStats(func(cs *ConnStats) { cs.wroteMsg(msg) })
}
-func (cn *connection) readMsg(msg *pp.Message) {
+func (cn *PeerConn) readMsg(msg *pp.Message) {
cn.allStats(func(cs *ConnStats) { cs.readMsg(msg) })
}
// After handshake, we know what Torrent and Client stats to include for a
// connection.
-func (cn *connection) postHandshakeStats(f func(*ConnStats)) {
+func (cn *PeerConn) postHandshakeStats(f func(*ConnStats)) {
t := cn.t
f(&t.stats)
f(&t.cl.stats)
@@ -873,24 +873,24 @@
// 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 *connection) allStats(f func(*ConnStats)) {
+func (cn *PeerConn) allStats(f func(*ConnStats)) {
f(&cn._stats)
if cn.reconciledHandshakeStats {
cn.postHandshakeStats(f)
}
}
-func (cn *connection) wroteBytes(n int64) {
+func (cn *PeerConn) wroteBytes(n int64) {
cn.allStats(add(n, func(cs *ConnStats) *Count { return &cs.BytesWritten }))
}
-func (cn *connection) readBytes(n int64) {
+func (cn *PeerConn) readBytes(n int64) {
cn.allStats(add(n, func(cs *ConnStats) *Count { return &cs.BytesRead }))
}
// 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.
-func (c *connection) useful() bool {
+func (c *PeerConn) useful() bool {
t := c.t
if c.closed.IsSet() {
return false
@@ -898,7 +898,7 @@ }
if !t.haveInfo() {
return c.supportsExtension("ut_metadata")
}
- if t.seeding() && c.PeerInterested {
+ if t.seeding() && c.peerInterested {
return true
}
if c.peerHasWantedPieces() {
@@ -907,7 +907,7 @@ }
return false
}
-func (c *connection) lastHelpful() (ret time.Time) {
+func (c *PeerConn) lastHelpful() (ret time.Time) {
ret = c.lastUsefulChunkReceived
if c.t.seeding() && c.lastChunkSent.After(ret) {
ret = c.lastChunkSent
@@ -915,25 +915,25 @@ }
return
}
-func (c *connection) fastEnabled() bool {
+func (c *PeerConn) fastEnabled() bool {
return c.PeerExtensionBytes.SupportsFast() && c.t.cl.extensionBytes.SupportsFast()
}
-func (c *connection) reject(r request) {
+func (c *PeerConn) reject(r request) {
if !c.fastEnabled() {
panic("fast not enabled")
}
- c.Post(r.ToMsg(pp.Reject))
- delete(c.PeerRequests, r)
+ c.post(r.ToMsg(pp.Reject))
+ delete(c.peerRequests, r)
}
-func (c *connection) onReadRequest(r request) error {
+func (c *PeerConn) onReadRequest(r request) error {
requestedChunkLengths.Add(strconv.FormatUint(r.Length.Uint64(), 10), 1)
- if _, ok := c.PeerRequests[r]; ok {
+ if _, ok := c.peerRequests[r]; ok {
torrent.Add("duplicate requests received", 1)
return nil
}
- if c.Choked {
+ if c.choking {
torrent.Add("requests received while choking", 1)
if c.fastEnabled() {
torrent.Add("requests rejected while choking", 1)
@@ -941,7 +941,7 @@ c.reject(r)
}
return nil
}
- if len(c.PeerRequests) >= maxRequests {
+ if len(c.peerRequests) >= maxRequests {
torrent.Add("requests received while queue full", 1)
if c.fastEnabled() {
c.reject(r)
@@ -961,17 +961,17 @@ if r.Begin+r.Length > c.t.pieceLength(pieceIndex(r.Index)) {
torrent.Add("bad requests received", 1)
return errors.New("bad request")
}
- if c.PeerRequests == nil {
- c.PeerRequests = make(map[request]struct{}, maxRequests)
+ if c.peerRequests == nil {
+ c.peerRequests = make(map[request]struct{}, maxRequests)
}
- c.PeerRequests[r] = struct{}{}
+ c.peerRequests[r] = struct{}{}
c.tickleWriter()
return nil
}
// Processes incoming BitTorrent wire-protocol messages. The client lock is held upon entry and
// exit. Returning will end the connection.
-func (c *connection) mainReadLoop() (err error) {
+func (c *PeerConn) mainReadLoop() (err error) {
defer func() {
if err != nil {
torrent.Add("connection.mainReadLoop returned with error", 1)
@@ -1012,20 +1012,20 @@ return fmt.Errorf("received fast extension message (type=%v) but extension is disabled", msg.Type)
}
switch msg.Type {
case pp.Choke:
- c.PeerChoked = true
+ c.peerChoking = true
c.deleteAllRequests()
// We can then reset our interest.
c.updateRequests()
c.updateExpectingChunks()
case pp.Unchoke:
- c.PeerChoked = false
+ c.peerChoking = false
c.tickleWriter()
c.updateExpectingChunks()
case pp.Interested:
- c.PeerInterested = true
+ c.peerInterested = true
c.tickleWriter()
case pp.NotInterested:
- c.PeerInterested = false
+ c.peerInterested = false
// We don't clear their requests since it isn't clear in the spec.
// We'll probably choke them for this, which will clear them if
// appropriate, and is clearly specified.
@@ -1089,7 +1089,7 @@ }
}
}
-func (c *connection) onReadExtendedMsg(id pp.ExtensionNumber, payload []byte) (err error) {
+func (c *PeerConn) onReadExtendedMsg(id pp.ExtensionNumber, payload []byte) (err error) {
defer func() {
// TODO: Should we still do this?
if err != nil {
@@ -1158,13 +1158,13 @@ }
}
// Set both the Reader and Writer for the connection from a single ReadWriter.
-func (cn *connection) setRW(rw io.ReadWriter) {
+func (cn *PeerConn) setRW(rw io.ReadWriter) {
cn.r = rw
cn.w = rw
}
// Returns the Reader and Writer as a combined ReadWriter.
-func (cn *connection) rw() io.ReadWriter {
+func (cn *PeerConn) rw() io.ReadWriter {
return struct {
io.Reader
io.Writer
@@ -1172,15 +1172,15 @@ }{cn.r, cn.w}
}
// Handle a received chunk from a peer.
-func (c *connection) receiveChunk(msg *pp.Message) error {
+func (c *PeerConn) receiveChunk(msg *pp.Message) error {
t := c.t
cl := t.cl
torrent.Add("chunks received", 1)
req := newRequestFromMessage(msg)
- if c.PeerChoked {
- torrent.Add("chunks received while choked", 1)
+ if c.peerChoking {
+ torrent.Add("chunks received while choking", 1)
}
if _, ok := c.validReceiveChunks[req]; !ok {
@@ -1189,7 +1189,7 @@ return errors.New("received unexpected chunk")
}
delete(c.validReceiveChunks, req)
- if c.PeerChoked && c.peerAllowedFast.Get(int(req.Index)) {
+ if c.peerChoking && c.peerAllowedFast.Get(int(req.Index)) {
torrent.Add("chunks received due to allowed fast", 1)
}
@@ -1271,19 +1271,19 @@
return nil
}
-func (c *connection) onDirtiedPiece(piece pieceIndex) {
+func (c *PeerConn) 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[*connection]struct{})
+ *ds = make(map[*PeerConn]struct{})
}
(*ds)[c] = struct{}{}
}
-func (c *connection) uploadAllowed() bool {
+func (c *PeerConn) uploadAllowed() bool {
if c.t.cl.config.NoUpload {
return false
}
@@ -1300,7 +1300,7 @@ }
return true
}
-func (c *connection) setRetryUploadTimer(delay time.Duration) {
+func (c *PeerConn) setRetryUploadTimer(delay time.Duration) {
if c.uploadTimer == nil {
c.uploadTimer = time.AfterFunc(delay, c.writerCond.Broadcast)
} else {
@@ -1309,16 +1309,16 @@ }
}
// Also handles choking and unchoking of the remote peer.
-func (c *connection) upload(msg func(pp.Message) bool) bool {
+func (c *PeerConn) upload(msg func(pp.Message) bool) bool {
// Breaking or completing this loop means we don't want to upload to the
// peer anymore, and we choke them.
another:
for c.uploadAllowed() {
// We want to upload to the peer.
- if !c.Unchoke(msg) {
+ if !c.unchoke(msg) {
return false
}
- for r := range c.PeerRequests {
+ for r := range c.peerRequests {
res := c.t.cl.config.UploadRateLimiter.ReserveN(time.Now(), int(r.Length))
if !res.OK() {
panic(fmt.Sprintf("upload rate limiter burst size < %d", r.Length))
@@ -1348,7 +1348,7 @@ // ask for it again, we'll kick them to allow us to send them
// an updated bitfield.
break another
}
- delete(c.PeerRequests, r)
+ delete(c.peerRequests, r)
if !more {
return false
}
@@ -1356,26 +1356,26 @@ goto another
}
return true
}
- return c.Choke(msg)
+ return c.choke(msg)
}
-func (cn *connection) Drop() {
+func (cn *PeerConn) drop() {
cn.t.dropConnection(cn)
}
-func (cn *connection) netGoodPiecesDirtied() int64 {
+func (cn *PeerConn) netGoodPiecesDirtied() int64 {
return cn._stats.PiecesDirtiedGood.Int64() - cn._stats.PiecesDirtiedBad.Int64()
}
-func (c *connection) peerHasWantedPieces() bool {
+func (c *PeerConn) peerHasWantedPieces() bool {
return !c._pieceRequestOrder.IsEmpty()
}
-func (c *connection) numLocalRequests() int {
+func (c *PeerConn) numLocalRequests() int {
return len(c.requests)
}
-func (c *connection) deleteRequest(r request) bool {
+func (c *PeerConn) deleteRequest(r request) bool {
if _, ok := c.requests[r]; !ok {
return false
}
@@ -1393,14 +1393,14 @@ panic(n)
}
c.updateRequests()
for _c := range c.t.conns {
- if !_c.Interested && _c != c && c.PeerHasPiece(pieceIndex(r.Index)) {
+ if !_c.interested && _c != c && c.peerHasPiece(pieceIndex(r.Index)) {
_c.updateRequests()
}
}
return true
}
-func (c *connection) deleteAllRequests() {
+func (c *PeerConn) deleteAllRequests() {
for r := range c.requests {
c.deleteRequest(r)
}
@@ -1412,19 +1412,19 @@ // c.tickleWriter()
// }
}
-func (c *connection) tickleWriter() {
+func (c *PeerConn) tickleWriter() {
c.writerCond.Broadcast()
}
-func (c *connection) postCancel(r request) bool {
+func (c *PeerConn) postCancel(r request) bool {
if !c.deleteRequest(r) {
return false
}
- c.Post(makeCancelMessage(r))
+ c.post(makeCancelMessage(r))
return true
}
-func (c *connection) sendChunk(r request, msg func(pp.Message) bool) (more bool, err error) {
+func (c *PeerConn) sendChunk(r request, msg func(pp.Message) bool) (more bool, err error) {
// Count the chunk being sent, even if it isn't.
b := make([]byte, r.Length)
p := c.t.info.Piece(int(r.Index))
@@ -1447,7 +1447,7 @@ c.lastChunkSent = time.Now()
return
}
-func (c *connection) setTorrent(t *Torrent) {
+func (c *PeerConn) setTorrent(t *Torrent) {
if c.t != nil {
panic("connection already associated with a torrent")
}
@@ -1456,24 +1456,24 @@ c.logger.Printf("torrent=%v", t)
t.reconcileHandshakeStats(c)
}
-func (c *connection) peerPriority() peerPriority {
+func (c *PeerConn) peerPriority() peerPriority {
return bep40PriorityIgnoreError(c.remoteIpPort(), c.t.cl.publicAddr(c.remoteIp()))
}
-func (c *connection) remoteIp() net.IP {
+func (c *PeerConn) remoteIp() net.IP {
return addrIpOrNil(c.remoteAddr)
}
-func (c *connection) remoteIpPort() IpPort {
+func (c *PeerConn) remoteIpPort() IpPort {
ipa, _ := tryIpPortFromNetAddr(c.remoteAddr)
return IpPort{ipa.IP, uint16(ipa.Port)}
}
-func (c *connection) String() string {
+func (c *PeerConn) String() string {
return fmt.Sprintf("connection %p", c)
}
-func (c *connection) trust() connectionTrust {
+func (c *PeerConn) trust() connectionTrust {
return connectionTrust{c.trusted, c.netGoodPiecesDirtied()}
}
@@ -1486,23 +1486,23 @@ func (l connectionTrust) Less(r connectionTrust) bool {
return multiless.New().Bool(l.Implicit, r.Implicit).Int64(l.NetGoodPiecesDirted, r.NetGoodPiecesDirted).Less()
}
-func (cn *connection) requestStrategyConnection() requestStrategyConnection {
+func (cn *PeerConn) requestStrategyConnection() requestStrategyConnection {
return cn
}
-func (cn *connection) chunksReceivedWhileExpecting() int64 {
+func (cn *PeerConn) chunksReceivedWhileExpecting() int64 {
return cn._chunksReceivedWhileExpecting
}
-func (cn *connection) fastest() bool {
+func (cn *PeerConn) fastest() bool {
return cn == cn.t.fastestConn
}
-func (cn *connection) peerMaxRequests() int {
+func (cn *PeerConn) peerMaxRequests() int {
return cn.PeerMaxRequests
}
-func (cn *connection) peerPieces() bitmap.Bitmap {
+func (cn *PeerConn) peerPieces() bitmap.Bitmap {
ret := cn._peerPieces.Copy()
if cn.peerSentHaveAll {
ret.AddRange(0, cn.t.numPieces())
@@ -1510,14 +1510,14 @@ }
return ret
}
-func (cn *connection) pieceRequestOrder() *prioritybitmap.PriorityBitmap {
+func (cn *PeerConn) pieceRequestOrder() *prioritybitmap.PriorityBitmap {
return &cn._pieceRequestOrder
}
-func (cn *connection) stats() *ConnStats {
+func (cn *PeerConn) stats() *ConnStats {
return &cn._stats
}
-func (cn *connection) torrent() requestStrategyTorrent {
+func (cn *PeerConn) torrent() requestStrategyTorrent {
return cn.t.requestStrategyTorrent()
}
diff --git a/connection_test.go b/peerconn_test.go
rename from connection_test.go
rename to peerconn_test.go
index 2f11fc2b7e5745367e452a1abf8e5601d0a2122c..002a5fca399de2858da971b632394f961dc2c721 100644
--- a/connection_test.go
+++ b/peerconn_test.go
@@ -34,10 +34,10 @@ c.w = w
go c.writer(time.Minute)
c.mu().Lock()
c.t._completedPieces.Add(1)
- c.PostBitfield( /*[]bool{false, true, false}*/ )
+ c.postBitfield( /*[]bool{false, true, false}*/ )
c.mu().Unlock()
c.mu().Lock()
- c.Have(2)
+ c.have(2)
c.mu().Unlock()
b := make([]byte, 15)
n, err := io.ReadFull(r, b)
diff --git a/misc.go b/misc.go
index d1f179a3fe0a158b308620d93da72aeea1c24215..b49ff7f431a8b7d48a8f5f123d652bdc45d9f92d 100644
--- a/misc.go
+++ b/misc.go
@@ -106,7 +106,7 @@ }
return ret
}
-func connLessTrusted(l, r *connection) bool {
+func connLessTrusted(l, r *PeerConn) bool {
return l.trust().Less(r.trust())
}
diff --git a/piece.go b/piece.go
index 28b87747ae57f06b718a4676694d77f9e5ed8932..97b6087ca3a63925f4bd6590aee210010a1a2322 100644
--- a/piece.go
+++ b/piece.go
@@ -61,7 +61,7 @@ noPendingWrites sync.Cond
// Connections that have written data to this piece since its last check.
// This can include connections that have closed.
- dirtiers map[*connection]struct{}
+ dirtiers map[*PeerConn]struct{}
}
func (p *Piece) String() string {
diff --git a/t.go b/t.go
index 95b69fe7528b90b72598e3c170d7f3eb2e21e999..82c18bfc841faf99a75fde8ed9b70e27c1424d3c 100644
--- a/t.go
+++ b/t.go
@@ -9,8 +9,7 @@
"github.com/anacrolix/torrent/metainfo"
)
-// The torrent's infohash. This is fixed and cannot change. It uniquely
-// identifies a torrent.
+// The Torrent's infohash. This is fixed and cannot change. It uniquely identifies a torrent.
func (t *Torrent) InfoHash() metainfo.Hash {
return t.infoHash
}
@@ -244,3 +243,11 @@
func (t *Torrent) Piece(i pieceIndex) *Piece {
return t.piece(i)
}
+
+func (t *Torrent) PeerConns() []*PeerConn {
+ ret := make([]*PeerConn, 0, len(t.conns))
+ for c := range t.conns {
+ ret = append(ret, c)
+ }
+ return ret
+}
diff --git a/torrent.go b/torrent.go
index 4cd6cd4b9e3821323624e6544a84e90c047fa34e..2c18b73df67e8c43c091058c76edbcdce8192ab2 100644
--- a/torrent.go
+++ b/torrent.go
@@ -74,12 +74,12 @@ files *[]*File
// Active peer connections, running message stream loops. TODO: Make this
// open (not-closed) connections only.
- conns map[*connection]struct{}
+ conns map[*PeerConn]struct{}
maxEstablishedConns int
// Set of addrs to which we're attempting to connect. Connections are
// half-open until all handshakes are completed.
halfOpen map[string]Peer
- fastestConn *connection
+ fastestConn *PeerConn
// 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
@@ -236,8 +236,8 @@ }
return false
}
-func (t *Torrent) unclosedConnsAsSlice() (ret []*connection) {
- ret = make([]*connection, 0, len(t.conns))
+func (t *Torrent) unclosedConnsAsSlice() (ret []*PeerConn) {
+ ret = make([]*PeerConn, 0, len(t.conns))
for c := range t.conns {
if !c.closed.IsSet() {
ret = append(ret, c)
@@ -385,7 +385,7 @@ func (t *Torrent) onSetInfo() {
for conn := range t.conns {
if err := conn.setNumPieces(t.numPieces()); err != nil {
t.logger.Printf("closing connection: %s", err)
- conn.Close()
+ conn.close()
}
}
for i := range t.pieces {
@@ -485,7 +485,7 @@ func (t *Torrent) metadataPieceSize(piece int) int {
return metadataPieceSize(len(t.metadataBytes), piece)
}
-func (t *Torrent) newMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
+func (t *Torrent) newMetadataExtensionMessage(c *PeerConn, msgType int, piece int, data []byte) pp.Message {
d := map[string]int{
"msg_type": msgType,
"piece": piece,
@@ -617,7 +617,7 @@ conns := t.connsAsSlice()
slices.Sort(conns, worseConn)
for i, c := range conns {
fmt.Fprintf(w, "%2d. ", i+1)
- c.WriteStatus(w, t)
+ c.writeStatus(w, t)
}
}
@@ -702,7 +702,7 @@ t.storage.Close()
t.storageLock.Unlock()
}
for conn := range t.conns {
- conn.Close()
+ conn.close()
}
t.cl.event.Broadcast()
t.pieceStateChanges.Close()
@@ -835,11 +835,11 @@ // The worst connection is one that hasn't been sent, or sent anything useful
// for the longest. A bad connection is one that usually sends us unwanted
// pieces, or has been in worser half of the established connections for more
// than a minute.
-func (t *Torrent) worstBadConn() *connection {
+func (t *Torrent) worstBadConn() *PeerConn {
wcs := worseConnSlice{t.unclosedConnsAsSlice()}
heap.Init(&wcs)
for wcs.Len() != 0 {
- c := heap.Pop(&wcs).(*connection)
+ c := heap.Pop(&wcs).(*PeerConn)
if c._stats.ChunksReadWasted.Int64() >= 6 && c._stats.ChunksReadWasted.Int64() > c._stats.ChunksReadUseful.Int64() {
return c
}
@@ -1033,7 +1033,7 @@ }
func (t *Torrent) numReceivedConns() (ret int) {
for c := range t.conns {
- if c.Discovery == peerSourceIncoming {
+ if c.Discovery == PeerSourceIncoming {
ret++
}
}
@@ -1194,7 +1194,7 @@ return t.setInfoBytes(b)
}
// Returns true if connection is removed from torrent.Conns.
-func (t *Torrent) deleteConnection(c *connection) (ret bool) {
+func (t *Torrent) deleteConnection(c *PeerConn) (ret bool) {
if !c.closed.IsSet() {
panic("connection is not closed")
// There are behaviours prevented by the closed state that will fail
@@ -1219,9 +1219,9 @@ // panic(t.lastRequested)
//}
}
-func (t *Torrent) dropConnection(c *connection) {
+func (t *Torrent) dropConnection(c *PeerConn) {
t.cl.event.Broadcast()
- c.Close()
+ c.close()
if t.deleteConnection(c) {
t.openNewConns()
}
@@ -1352,7 +1352,7 @@ continue
}
t.addPeer(Peer{
Addr: ipPortAddr{cp.IP, cp.Port},
- Source: peerSourceDhtGetPeers,
+ Source: PeerSourceDhtGetPeers,
})
}
cl.unlock()
@@ -1440,7 +1440,7 @@ }
// Reconcile bytes transferred before connection was associated with a
// torrent.
-func (t *Torrent) reconcileHandshakeStats(c *connection) {
+func (t *Torrent) reconcileHandshakeStats(c *PeerConn) {
if c._stats != (ConnStats{
// Handshakes should only increment these fields:
BytesWritten: c._stats.BytesWritten,
@@ -1456,7 +1456,7 @@ c.reconciledHandshakeStats = true
}
// Returns true if the connection is added.
-func (t *Torrent) addConnection(c *connection) (err error) {
+func (t *Torrent) addConnection(c *PeerConn) (err error) {
defer func() {
if err == nil {
torrent.Add("added connections", 1)
@@ -1473,7 +1473,7 @@ if !t.cl.config.dropDuplicatePeerIds {
continue
}
if left, ok := c.hasPreferredNetworkOver(c0); ok && left {
- c0.Close()
+ c0.close()
t.deleteConnection(c0)
} else {
return errors.New("existing connection preferred")
@@ -1484,7 +1484,7 @@ c := t.worstBadConn()
if c == nil {
return errors.New("don't want conns")
}
- c.Close()
+ c.close()
t.deleteConnection(c)
}
if len(t.conns) >= t.maxEstablishedConns {
@@ -1517,7 +1517,7 @@ oldMax = t.maxEstablishedConns
t.maxEstablishedConns = max
wcs := slices.HeapInterface(slices.FromMapKeys(t.conns), worseConn)
for len(t.conns) > t.maxEstablishedConns && wcs.Len() > 0 {
- t.dropConnection(wcs.Pop().(*connection))
+ t.dropConnection(wcs.Pop().(*PeerConn))
}
t.openNewConns()
return oldMax
@@ -1568,7 +1568,7 @@ // Y u do dis peer?!
c.stats().incrementPiecesDirtiedBad()
}
- bannableTouchers := make([]*connection, 0, len(p.dirtiers))
+ bannableTouchers := make([]*PeerConn, 0, len(p.dirtiers))
for c := range p.dirtiers {
if !c.trusted {
bannableTouchers = append(bannableTouchers, c)
@@ -1593,7 +1593,7 @@
if len(bannableTouchers) >= 1 {
c := bannableTouchers[0]
t.cl.banPeerIP(c.remoteIp())
- c.Drop()
+ c.drop()
}
}
t.onIncompletePiece(piece)
@@ -1613,7 +1613,7 @@ func (t *Torrent) onPieceCompleted(piece pieceIndex) {
t.pendAllChunkSpecs(piece)
t.cancelRequestsForPiece(piece)
for conn := range t.conns {
- conn.Have(piece)
+ conn.have(piece)
}
}
@@ -1634,11 +1634,11 @@ // favourably anyway.
// for c := range t.conns {
// if c.sentHave(piece) {
- // c.Drop()
+ // c.drop()
// }
// }
for conn := range t.conns {
- if conn.PeerHasPiece(piece) {
+ if conn.peerHasPiece(piece) {
conn.updateRequests()
}
}
@@ -1707,7 +1707,7 @@ delete(p.dirtiers, c)
}
}
-func (t *Torrent) connsAsSlice() (ret []*connection) {
+func (t *Torrent) connsAsSlice() (ret []*PeerConn) {
for c := range t.conns {
ret = append(ret, c)
}
@@ -1750,7 +1750,8 @@ t.halfOpen[addr.String()] = peer
go t.cl.outgoingConnection(t, addr, peer.Source, peer.Trusted)
}
-// Adds each a trusted, pending peer for each of the Client's addresses.
+// Adds a trusted, pending peer for each of the given Client's addresses. Typically used in tests to
+// quickly make one Client visible to the Torrent of another Client.
func (t *Torrent) AddClientPeer(cl *Client) {
t.AddPeers(func() (ps []Peer) {
for _, la := range cl.ListenAddrs() {
@@ -1799,7 +1800,7 @@ torrent.Add("request timeouts", 1)
cb.t.cl.lock()
defer cb.t.cl.unlock()
for cn := range cb.t.conns {
- if cn.PeerHasPiece(pieceIndex(r.Index)) {
+ if cn.peerHasPiece(pieceIndex(r.Index)) {
cn.updateRequests()
}
}
diff --git a/worst_conns.go b/worst_conns.go
index c43652bdc3941dca8b36644115f426c40899c8b3..3d89c7bbd218771bf53afeb8b3721a9c14f8e57b 100644
--- a/worst_conns.go
+++ b/worst_conns.go
@@ -8,7 +8,7 @@
"github.com/anacrolix/multiless"
)
-func worseConn(l, r *connection) bool {
+func worseConn(l, r *PeerConn) bool {
less, ok := multiless.New().Bool(
l.useful(), r.useful()).CmpInt64(
l.lastHelpful().Sub(r.lastHelpful()).Nanoseconds()).CmpInt64(
@@ -22,7 +22,7 @@ return less
}
type worseConnSlice struct {
- conns []*connection
+ conns []*PeerConn
}
var _ heap.Interface = &worseConnSlice{}
@@ -43,7 +43,7 @@ return ret
}
func (me *worseConnSlice) Push(x interface{}) {
- me.conns = append(me.conns, x.(*connection))
+ me.conns = append(me.conns, x.(*PeerConn))
}
func (me worseConnSlice) Swap(i, j int) {