tc.SetLinger(0)
}
c := cl.newConnection(nc, false, nc.RemoteAddr(), nc.RemoteAddr().Network())
- c.Discovery = peerSourceIncoming
+ c.Discovery = PeerSourceIncoming
cl.runReceivedConn(c)
}
// 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)
// 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()
// 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)
// 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()
}
return
}
- defer c.Close()
+ defer c.close()
c.Discovery = ps
c.trusted = trusted
cl.runHandshookConn(c, t)
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
}
// 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)
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
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)
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 {
}
// 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 {
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(),
})
}
// 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 {
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
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(),
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,
}
t.addPeers([]Peer{{
Addr: ipPortAddr{ip, port},
- Source: peerSourceDhtAnnouncePeer,
+ Source: PeerSourceDhtAnnouncePeer,
}})
}
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
// 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
lastChunkSent time.Time
// Stuff controlled by the local peer.
- Interested bool
+ interested bool
lastBecameInterested time.Time
priorInterest time.Duration
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
// 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
logger log.Logger
}
-func (cn *connection) updateExpectingChunks() {
+func (cn *PeerConn) updateExpectingChunks() {
if cn.expectingChunks() {
if cn.lastStartedExpectingToReceiveChunks.IsZero() {
cn.lastStartedExpectingToReceiveChunks = time.Now()
}
}
-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
// 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())
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
}
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()
// 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
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})
}
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
// 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 <pre> 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",
cn.requestsLowWater,
cn.numLocalRequests(),
cn.nominalMaxRequests(),
- len(cn.PeerRequests),
+ len(cn.peerRequests),
cn.statusFlags(),
cn.downloadRate()/(1<<10),
)
)
}
-func (cn *connection) Close() {
+func (cn *PeerConn) close() {
if !cn.closed.Set() {
return
}
}
}
-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
cn.tickleWriter()
}
-func (cn *connection) requestMetadataPiece(index int) {
+func (cn *PeerConn) requestMetadataPiece(index int) {
eID := cn.PeerExtensionIDs[pp.ExtensionNameMetadata]
if eID == 0 {
return
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 {
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),
))
}
-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)
}
-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() {
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 {
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)) {
})
}
-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 {
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
}
}
// 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
}
// 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
})
cn.mu().Lock()
defer cn.mu().Unlock()
- defer cn.Close()
+ defer cn.close()
defer keepAliveTimer.Stop()
frontBuf := new(bytes.Buffer)
for {
}
}
-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()
}
// 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 {
}
// check callers updaterequests
-func (cn *connection) stopRequestingPiece(piece pieceIndex) bool {
+func (cn *PeerConn) stopRequestingPiece(piece pieceIndex) bool {
return cn._pieceRequestOrder.Remove(bitmap.BitIndex(piece))
}
// 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 {
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
}
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++ {
}
}
-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)
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")
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
}
}
}
-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)
// 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
if !t.haveInfo() {
return c.supportsExtension("ut_metadata")
}
- if t.seeding() && c.PeerInterested {
+ if t.seeding() && c.peerInterested {
return true
}
if c.peerHasWantedPieces() {
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
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)
}
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)
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)
}
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.
}
}
-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 {
}
// 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
}
// 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 {
}
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)
}
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
}
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 {
}
// 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))
// an updated bitfield.
break another
}
- delete(c.PeerRequests, r)
+ delete(c.peerRequests, r)
if !more {
return false
}
}
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
}
}
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)
}
// }
}
-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))
return
}
-func (c *connection) setTorrent(t *Torrent) {
+func (c *PeerConn) setTorrent(t *Torrent) {
if c.t != nil {
panic("connection already associated with a torrent")
}
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()}
}
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())
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()
}
// 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
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)
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 {
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,
slices.Sort(conns, worseConn)
for i, c := range conns {
fmt.Fprintf(w, "%2d. ", i+1)
- c.WriteStatus(w, t)
+ c.writeStatus(w, t)
}
}
t.storageLock.Unlock()
}
for conn := range t.conns {
- conn.Close()
+ conn.close()
}
t.cl.event.Broadcast()
t.pieceStateChanges.Close()
// 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
}
func (t *Torrent) numReceivedConns() (ret int) {
for c := range t.conns {
- if c.Discovery == peerSourceIncoming {
+ if c.Discovery == PeerSourceIncoming {
ret++
}
}
}
// 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
//}
}
-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()
}
}
t.addPeer(Peer{
Addr: ipPortAddr{cp.IP, cp.Port},
- Source: peerSourceDhtGetPeers,
+ Source: PeerSourceDhtGetPeers,
})
}
cl.unlock()
// 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,
}
// 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)
continue
}
if left, ok := c.hasPreferredNetworkOver(c0); ok && left {
- c0.Close()
+ c0.close()
t.deleteConnection(c0)
} else {
return errors.New("existing connection preferred")
if c == nil {
return errors.New("don't want conns")
}
- c.Close()
+ c.close()
t.deleteConnection(c)
}
if len(t.conns) >= 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
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)
if len(bannableTouchers) >= 1 {
c := bannableTouchers[0]
t.cl.banPeerIP(c.remoteIp())
- c.Drop()
+ c.drop()
}
}
t.onIncompletePiece(piece)
t.pendAllChunkSpecs(piece)
t.cancelRequestsForPiece(piece)
for conn := range t.conns {
- conn.Have(piece)
+ conn.have(piece)
}
}
// 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()
}
}
}
}
-func (t *Torrent) connsAsSlice() (ret []*connection) {
+func (t *Torrent) connsAsSlice() (ret []*PeerConn) {
for c := range t.conns {
ret = append(ret, c)
}
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() {
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()
}
}