Drop support for go 1.20

[btrtrc.git] / torrent.go

diff --git a/torrent.go b/torrent.go
index 49ae06054d4b362c87f892652f15e9c1e0a7bed4..6385a3fc026f9a2474ddc029908ef236ab30713a 100644 (file)
--- a/torrent.go
+++ b/torrent.go
@@ -830,7 +830,10 @@ func (t *Torrent) newMetaInfo() metainfo.MetaInfo {
        }
 }
 
-// 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()
@@ -1057,7 +1060,7 @@ func (t *Torrent) havePiece(index pieceIndex) bool {
 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
@@ -1071,7 +1074,7 @@ func (t *Torrent) maybeDropMutuallyCompletePeer(
        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()
 }
 
@@ -1715,10 +1718,9 @@ func (t *Torrent) startScrapingTracker(_url string) {
        }
        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
        }
@@ -2141,7 +2143,7 @@ func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
                }
                err := p.Storage().MarkComplete()
                if err != nil {
-                       t.logger.Printf("%T: error marking piece complete %d: %s", t.storage, piece, err)
+                       t.logger.Levelf(log.Warning, "%T: error marking piece complete %d: %s", t.storage, piece, err)
                }
                t.cl.lock()
 
@@ -2186,7 +2188,7 @@ func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
                        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
@@ -2217,7 +2219,7 @@ func (t *Torrent) onPieceCompleted(piece pieceIndex) {
        t.piece(piece).readerCond.Broadcast()
        for conn := range t.conns {
                conn.have(piece)
-               t.maybeDropMutuallyCompletePeer(&conn.Peer)
+               t.maybeDropMutuallyCompletePeer(conn)
        }
 }
 
@@ -2249,7 +2251,7 @@ func (t *Torrent) onIncompletePiece(piece pieceIndex) {
 }
 
 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() {
        }
 }
 
@@ -2739,18 +2741,11 @@ func (t *Torrent) peerConnsWithDialAddrPort(target netip.AddrPort) (ret []*PeerC
        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)
        }
@@ -2767,10 +2762,38 @@ func sendUtHolepunchMsg(
        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)
@@ -2801,6 +2824,14 @@ func (t *Torrent) handleReceivedUtHolepunchMsg(msg utHolepunch.Msg, sender *Peer
                }
                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,
@@ -2818,6 +2849,7 @@ func (t *Torrent) handleReceivedUtHolepunchMsg(msg utHolepunch.Msg, sender *Peer
                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:
@@ -2825,6 +2857,18 @@ func (t *Torrent) handleReceivedUtHolepunchMsg(msg utHolepunch.Msg, sender *Peer
        }
 }
 
+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 {