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()
}