}
}
-// Get bytes left
+// Returns a count of bytes that are not complete in storage, and not pending being written to
+// storage. This value is from the perspective of the download manager, and may not agree with the
+// actual state in storage. If you want read data synchronously you should use a Reader. See
+// https://github.com/anacrolix/torrent/issues/828.
func (t *Torrent) BytesMissing() (n int64) {
t.cl.rLock()
n = t.bytesMissingLocked()
func (t *Torrent) maybeDropMutuallyCompletePeer(
// I'm not sure about taking peer here, not all peer implementations actually drop. Maybe that's
// okay?
- p *Peer,
+ p *PeerConn,
) {
if !t.cl.config.DropMutuallyCompletePeers {
return
if p.useful() {
return
}
- t.logger.WithDefaultLevel(log.Debug).Printf("dropping %v, which is mutually complete", p)
+ p.logger.Levelf(log.Debug, "is mutually complete; dropping")
p.drop()
}
}
u, err := url.Parse(_url)
if err != nil {
- // URLs with a leading '*' appear to be a uTorrent convention to
- // disable trackers.
+ // URLs with a leading '*' appear to be a uTorrent convention to disable trackers.
if _url[0] != '*' {
- log.Str("error parsing tracker url").AddValues("url", _url).Log(t.logger)
+ t.logger.Levelf(log.Warning, "error parsing tracker url: %v", err)
}
return
}
if len(bannableTouchers) >= 1 {
c := bannableTouchers[0]
if len(bannableTouchers) != 1 {
- t.logger.Levelf(log.Warning, "would have banned %v for touching piece %v after failed piece check", c.remoteIp(), piece)
+ t.logger.Levelf(log.Debug, "would have banned %v for touching piece %v after failed piece check", c.remoteIp(), piece)
} else {
// Turns out it's still useful to ban peers like this because if there's only a
// single peer for a piece, and we never progress that piece to completion, we
t.piece(piece).readerCond.Broadcast()
for conn := range t.conns {
conn.have(piece)
- t.maybeDropMutuallyCompletePeer(&conn.Peer)
+ t.maybeDropMutuallyCompletePeer(conn)
}
}
}
func (t *Torrent) tryCreateMorePieceHashers() {
- for !t.closed.IsSet() && t.activePieceHashes < 2 && t.tryCreatePieceHasher() {
+ for !t.closed.IsSet() && t.activePieceHashes < t.cl.config.PieceHashersPerTorrent && t.tryCreatePieceHasher() {
}
}
return
}
-func makeUtHolepunchMsgForPeerConn(
+func wrapUtHolepunchMsgForPeerConn(
recipient *PeerConn,
- msgType utHolepunch.MsgType,
- addrPort netip.AddrPort,
- errCode utHolepunch.ErrCode,
+ msg utHolepunch.Msg,
) pp.Message {
- utHolepunchMsg := utHolepunch.Msg{
- MsgType: msgType,
- AddrPort: addrPort,
- ErrCode: errCode,
- }
- extendedPayload, err := utHolepunchMsg.MarshalBinary()
+ extendedPayload, err := msg.MarshalBinary()
if err != nil {
panic(err)
}
addrPort netip.AddrPort,
errCode utHolepunch.ErrCode,
) {
- pc.write(makeUtHolepunchMsgForPeerConn(pc, msgType, addrPort, errCode))
+ holepunchMsg := utHolepunch.Msg{
+ MsgType: msgType,
+ AddrPort: addrPort,
+ ErrCode: errCode,
+ }
+ incHolepunchMessagesSent(holepunchMsg)
+ ppMsg := wrapUtHolepunchMsgForPeerConn(pc, holepunchMsg)
+ pc.write(ppMsg)
+}
+
+func incHolepunchMessages(msg utHolepunch.Msg, verb string) {
+ torrent.Add(
+ fmt.Sprintf(
+ "holepunch %v %v messages %v",
+ msg.MsgType,
+ addrPortProtocolStr(msg.AddrPort),
+ verb,
+ ),
+ 1,
+ )
+}
+
+func incHolepunchMessagesReceived(msg utHolepunch.Msg) {
+ incHolepunchMessages(msg, "received")
+}
+
+func incHolepunchMessagesSent(msg utHolepunch.Msg) {
+ incHolepunchMessages(msg, "sent")
}
func (t *Torrent) handleReceivedUtHolepunchMsg(msg utHolepunch.Msg, sender *PeerConn) error {
+ incHolepunchMessagesReceived(msg)
switch msg.MsgType {
case utHolepunch.Rendezvous:
t.logger.Printf("got holepunch rendezvous request for %v from %p", msg.AddrPort, sender)
}
return nil
case utHolepunch.Connect:
+ holepunchAddr := msg.AddrPort
+ t.logger.Printf("got holepunch connect request for %v from %p", holepunchAddr, sender)
+ if g.MapContains(t.cl.undialableWithoutHolepunch, holepunchAddr) {
+ setAdd(&t.cl.undialableWithoutHolepunchDialedAfterHolepunchConnect, holepunchAddr)
+ if g.MapContains(t.cl.accepted, holepunchAddr) {
+ setAdd(&t.cl.probablyOnlyConnectedDueToHolepunch, holepunchAddr)
+ }
+ }
opts := outgoingConnOpts{
peerInfo: PeerInfo{
Addr: msg.AddrPort,
initiateConn(opts, true)
return nil
case utHolepunch.Error:
+ torrent.Add("holepunch error messages received", 1)
t.logger.Levelf(log.Debug, "received ut_holepunch error message from %v: %v", sender, msg.ErrCode)
return nil
default:
}
}
-func (t *Torrent) startHolepunchRendezvous(addrPort netip.AddrPort) error {
+func addrPortProtocolStr(addrPort netip.AddrPort) string {
+ addr := addrPort.Addr()
+ switch {
+ case addr.Is4():
+ return "ipv4"
+ case addr.Is6():
+ return "ipv6"
+ default:
+ panic(addrPort)
+ }
+}
+
+func (t *Torrent) trySendHolepunchRendezvous(addrPort netip.AddrPort) error {
rzsSent := 0
for pc := range t.conns {
if !pc.supportsExtension(utHolepunch.ExtensionName) {
defer cl.rUnlock()
return t.dialTimeout()
}
-
-func (t *Torrent) startHolepunchRendezvousForPeerRemoteAddr(addr PeerRemoteAddr) error {
- addrPort, err := addrPortFromPeerRemoteAddr(addr)
- if err != nil {
- return err
- }
- return t.startHolepunchRendezvous(addrPort)
-}