import (
"bytes"
"container/heap"
- "context"
"crypto/sha1"
"errors"
"fmt"
"io"
+ "math/rand"
"net/netip"
"net/url"
"sort"
"github.com/anacrolix/chansync"
"github.com/anacrolix/chansync/events"
"github.com/anacrolix/dht/v2"
+ . "github.com/anacrolix/generics"
"github.com/anacrolix/log"
"github.com/anacrolix/missinggo/perf"
- "github.com/anacrolix/missinggo/pubsub"
"github.com/anacrolix/missinggo/slices"
"github.com/anacrolix/missinggo/v2"
"github.com/anacrolix/missinggo/v2/bitmap"
+ "github.com/anacrolix/missinggo/v2/pubsub"
"github.com/anacrolix/multiless"
"github.com/anacrolix/sync"
- "github.com/anacrolix/torrent/option"
- request_strategy "github.com/anacrolix/torrent/request-strategy"
"github.com/davecgh/go-spew/spew"
- "github.com/pion/datachannel"
"github.com/anacrolix/torrent/bencode"
"github.com/anacrolix/torrent/common"
"github.com/anacrolix/torrent/metainfo"
pp "github.com/anacrolix/torrent/peer_protocol"
+ request_strategy "github.com/anacrolix/torrent/request-strategy"
"github.com/anacrolix/torrent/segments"
"github.com/anacrolix/torrent/storage"
"github.com/anacrolix/torrent/tracker"
+ typedRoaring "github.com/anacrolix/torrent/typed-roaring"
"github.com/anacrolix/torrent/webseed"
- "github.com/anacrolix/torrent/webtorrent"
)
// Maintains state of torrent within a Client. Many methods should not be called before the info is
userOnWriteChunkErr func(error)
closed chansync.SetOnce
+ onClose []func()
infoHash metainfo.Hash
pieces []Piece
+
+ // The order pieces are requested if there's no stronger reason like availability or priority.
+ pieceRequestOrder []int
// Values are the piece indices that changed.
- pieceStateChanges *pubsub.PubSub
+ pieceStateChanges pubsub.PubSub[PieceStateChange]
// The size of chunks to request from peers over the wire. This is
// normally 16KiB by convention these days.
chunkSize pp.Integer
chunkPool sync.Pool
// Total length of the torrent in bytes. Stored because it's not O(1) to
// get this from the info dict.
- length *int64
+ _length Option[int64]
// The storage to open when the info dict becomes available.
storageOpener *storage.Client
fileIndex segments.Index
files *[]*File
+ _chunksPerRegularPiece chunkIndexType
+
webSeeds map[string]*Peer
// Active peer connections, running message stream loops. TODO: Make this
// open (not-closed) connections only.
initialPieceCheckDisabled bool
connsWithAllPieces map[*Peer]struct{}
- // Count of each request across active connections.
- pendingRequests map[RequestIndex]*Peer
- lastRequested map[RequestIndex]time.Time
+
+ requestState map[RequestIndex]requestState
// Chunks we've written to since the corresponding piece was last checked.
- dirtyChunks roaring.Bitmap
+ dirtyChunks typedRoaring.Bitmap[RequestIndex]
pex pexState
// Is On when all pieces are complete.
Complete chansync.Flag
+ // Torrent sources in use keyed by the source string.
+ activeSources sync.Map
+ sourcesLogger log.Logger
+
smartBanCache smartBanCache
+
+ // Large allocations reused between request state updates.
+ requestPieceStates []request_strategy.PieceRequestOrderState
+ requestIndexes []RequestIndex
+}
+
+func (t *Torrent) length() int64 {
+ return t._length.Value
}
func (t *Torrent) selectivePieceAvailabilityFromPeers(i pieceIndex) (count int) {
beginFile := pieceFirstFileIndex(piece.torrentBeginOffset(), files)
endFile := pieceEndFileIndex(piece.torrentEndOffset(), files)
piece.files = files[beginFile:endFile]
- piece.undirtiedChunksIter = undirtiedChunksIter{
- TorrentDirtyChunks: &t.dirtyChunks,
- StartRequestIndex: piece.requestIndexOffset(),
- EndRequestIndex: piece.requestIndexOffset() + piece.numChunks(),
- }
}
}
for _, f := range t.info.UpvertedFiles() {
l += f.Length
}
- t.length = &l
+ t._length = Some(l)
}
// TODO: This shouldn't fail for storage reasons. Instead we should handle storage failure
t.nameMu.Lock()
t.info = info
t.nameMu.Unlock()
+ t._chunksPerRegularPiece = chunkIndexType((pp.Integer(t.usualPieceSize()) + t.chunkSize - 1) / t.chunkSize)
t.updateComplete()
t.fileIndex = segments.NewIndex(common.LengthIterFromUpvertedFiles(info.UpvertedFiles()))
t.displayName = "" // Save a few bytes lol.
// This seems to be all the follow-up tasks after info is set, that can't fail.
func (t *Torrent) onSetInfo() {
+ t.pieceRequestOrder = rand.Perm(t.numPieces())
t.initPieceRequestOrder()
+ MakeSliceWithLength(&t.requestPieceStates, t.numPieces())
for i := range t.pieces {
p := &t.pieces[i]
// Need to add relativeAvailability before updating piece completion, as that may result in conns
}
p.relativeAvailability = t.selectivePieceAvailabilityFromPeers(i)
t.addRequestOrderPiece(i)
- t.updatePieceCompletion(pieceIndex(i))
+ t.updatePieceCompletion(i)
if !t.initialPieceCheckDisabled && !p.storageCompletionOk {
// t.logger.Printf("piece %s completion unknown, queueing check", p)
- t.queuePieceCheck(pieceIndex(i))
+ t.queuePieceCheck(i)
}
}
t.cl.event.Broadcast()
close(t.gotMetainfoC)
t.updateWantPeersEvent()
- t.pendingRequests = make(map[RequestIndex]*Peer)
- t.lastRequested = make(map[RequestIndex]time.Time)
+ t.requestState = make(map[RequestIndex]requestState)
t.tryCreateMorePieceHashers()
t.iterPeers(func(p *Peer) {
p.onGotInfo(t.info)
return
}
if uint32(size) > maxMetadataSize {
- return errors.New("bad size")
+ return log.WithLevel(log.Warning, errors.New("bad size"))
}
if len(t.metadataBytes) == size {
return
t.nameMu.RLock()
defer t.nameMu.RUnlock()
if t.haveInfo() {
- return t.info.Name
+ return t.info.BestName()
}
if t.displayName != "" {
return t.displayName
}
return worseConn(i, j)
})
+ var buf bytes.Buffer
for i, c := range peers {
fmt.Fprintf(w, "%2d. ", i+1)
- c.writeStatus(w, t)
+ buf.Reset()
+ c.writeStatus(&buf, t)
+ w.Write(bytes.TrimRight(
+ bytes.ReplaceAll(buf.Bytes(), []byte("\n"), []byte("\n ")),
+ " "))
}
}
}
func (t *Torrent) numPieces() pieceIndex {
- return pieceIndex(t.info.NumPieces())
+ return t.info.NumPieces()
}
func (t *Torrent) numPiecesCompleted() (num pieceIndex) {
}
func (t *Torrent) close(wg *sync.WaitGroup) (err error) {
- t.closed.Set()
+ if !t.closed.Set() {
+ err = errors.New("already closed")
+ return
+ }
+ for _, f := range t.onClose {
+ f()
+ }
if t.storage != nil {
wg.Add(1)
go func() {
if t.storage != nil {
t.deletePieceRequestOrder()
}
+ t.assertAllPiecesRelativeAvailabilityZero()
+ t.pex.Reset()
+ t.cl.event.Broadcast()
+ t.pieceStateChanges.Close()
+ t.updateWantPeersEvent()
+ return
+}
+
+func (t *Torrent) assertAllPiecesRelativeAvailabilityZero() {
for i := range t.pieces {
p := t.piece(i)
if p.relativeAvailability != 0 {
panic(fmt.Sprintf("piece %v has relative availability %v", i, p.relativeAvailability))
}
}
- t.pex.Reset()
- t.cl.event.Broadcast()
- t.pieceStateChanges.Close()
- t.updateWantPeersEvent()
- return
}
func (t *Torrent) requestOffset(r Request) int64 {
- return torrentRequestOffset(*t.length, int64(t.usualPieceSize()), r)
+ return torrentRequestOffset(t.length(), int64(t.usualPieceSize()), r)
}
// Return the request that would include the given offset into the torrent data. Returns !ok if
// there is no such request.
func (t *Torrent) offsetRequest(off int64) (req Request, ok bool) {
- return torrentOffsetRequest(*t.length, t.info.PieceLength, int64(t.chunkSize), off)
+ return torrentOffsetRequest(t.length(), t.info.PieceLength, int64(t.chunkSize), off)
}
func (t *Torrent) writeChunk(piece int, begin int64, data []byte) (err error) {
return chunkIndexType((t.pieceLength(piece) + t.chunkSize - 1) / t.chunkSize)
}
-func (t *Torrent) chunksPerRegularPiece() uint32 {
- return uint32((pp.Integer(t.usualPieceSize()) + t.chunkSize - 1) / t.chunkSize)
+func (t *Torrent) chunksPerRegularPiece() chunkIndexType {
+ return t._chunksPerRegularPiece
}
-func (t *Torrent) numRequests() RequestIndex {
+func (t *Torrent) numChunks() RequestIndex {
if t.numPieces() == 0 {
return 0
}
- return uint32(t.numPieces()-1)*t.chunksPerRegularPiece() + t.pieceNumChunks(t.numPieces()-1)
+ return RequestIndex(t.numPieces()-1)*t.chunksPerRegularPiece() + t.pieceNumChunks(t.numPieces()-1)
}
func (t *Torrent) pendAllChunkSpecs(pieceIndex pieceIndex) {
return 0
}
if piece == t.numPieces()-1 {
- ret := pp.Integer(*t.length % t.info.PieceLength)
+ ret := pp.Integer(t.length() % t.info.PieceLength)
if ret != 0 {
return ret
}
}
_, err = storagePiece.WriteTo(io.MultiWriter(writers...))
if logPieceContents {
- log.Printf("hashed %q with copy err %v", examineBuf.Bytes(), err)
+ t.logger.WithDefaultLevel(log.Debug).Printf("hashed %q with copy err %v", examineBuf.Bytes(), err)
}
smartBanWriter.Flush()
differingPeers = smartBanWriter.badPeers
if !c.peerHasPiece(piece) {
return
}
- if c.requestState.Interested && c.peerChoking && !c.peerAllowedFast.Contains(uint32(piece)) {
+ if c.requestState.Interested && c.peerChoking && !c.peerAllowedFast.Contains(piece) {
return
}
c.updateRequests(reason)
// Returns the range of pieces [begin, end) that contains the extent of bytes.
func (t *Torrent) byteRegionPieces(off, size int64) (begin, end pieceIndex) {
- if off >= *t.length {
+ if off >= t.length() {
return
}
if off < 0 {
}
func (t *Torrent) pendRequest(req RequestIndex) {
- t.piece(int(req / t.chunksPerRegularPiece())).pendChunkIndex(req % t.chunksPerRegularPiece())
+ t.piece(t.pieceIndexOfRequestIndex(req)).pendChunkIndex(req % t.chunksPerRegularPiece())
}
func (t *Torrent) pieceCompletionChanged(piece pieceIndex, reason string) {
t.logger.Printf("marked piece %v complete but still has dirtiers", piece)
}
if changed {
- log.Fstr("piece %d completion changed: %+v -> %+v", piece, cached, uncached).SetLevel(log.Debug).Log(t.logger)
+ log.Fstr("piece %d completion changed: %+v -> %+v", piece, cached, uncached).LogLevel(log.Debug, t.logger)
t.pieceCompletionChanged(piece, "Torrent.updatePieceCompletion")
}
return changed
if !t.haveInfo() {
return 0
}
- return *t.length - t.bytesLeft()
+ return t.length() - t.bytesLeft()
}
func (t *Torrent) SetInfoBytes(b []byte) (err error) {
}
}
torrent.Add("deleted connections", 1)
- if !c.deleteAllRequests().IsEmpty() {
- t.iterPeers(func(p *Peer) {
- if p.isLowOnRequests() {
- p.updateRequests("Torrent.deletePeerConn")
- }
- })
- }
+ c.deleteAllRequests("Torrent.deletePeerConn")
t.assertPendingRequests()
if t.numActivePeers() == 0 && len(t.connsWithAllPieces) != 0 {
panic(t.connsWithAllPieces)
}
// var actual pendingRequests
// if t.haveInfo() {
- // actual.m = make([]int, t.numRequests())
+ // actual.m = make([]int, t.numChunks())
// }
// t.iterPeers(func(p *Peer) {
// p.requestState.Requests.Iterate(func(x uint32) bool {
return true
}
-func (t *Torrent) onWebRtcConn(
- c datachannel.ReadWriteCloser,
- dcc webtorrent.DataChannelContext,
-) {
- defer c.Close()
- netConn := webrtcNetConn{
- ReadWriteCloser: c,
- DataChannelContext: dcc,
- }
- peerRemoteAddr := netConn.RemoteAddr()
- if t.cl.badPeerAddr(peerRemoteAddr) {
- return
- }
- pc, err := t.cl.initiateProtocolHandshakes(
- context.Background(),
- netConn,
- t,
- dcc.LocalOffered,
- false,
- netConn.RemoteAddr(),
- webrtcNetwork,
- fmt.Sprintf("webrtc offer_id %x", dcc.OfferId),
- )
- if err != nil {
- t.logger.WithDefaultLevel(log.Error).Printf("error in handshaking webrtc connection: %v", err)
- return
- }
- if dcc.LocalOffered {
- pc.Discovery = PeerSourceTracker
- } else {
- pc.Discovery = PeerSourceIncoming
- }
- pc.conn.SetWriteDeadline(time.Time{})
- t.cl.lock()
- defer t.cl.unlock()
- err = t.cl.runHandshookConn(pc, t)
- if err != nil {
- t.logger.WithDefaultLevel(log.Critical).Printf("error running handshook webrtc conn: %v", err)
- }
-}
-
func (t *Torrent) logRunHandshookConn(pc *PeerConn, logAll bool, level log.Level) {
err := t.cl.runHandshookConn(pc, t)
if err != nil || logAll {
- t.logger.WithDefaultLevel(level).Printf("error running handshook conn: %v", err)
+ t.logger.WithDefaultLevel(level).Levelf(log.ErrorLevel(err), "error running handshook conn: %v", err)
}
}
t.logRunHandshookConn(pc, false, log.Debug)
}
-func (t *Torrent) startWebsocketAnnouncer(u url.URL) torrentTrackerAnnouncer {
- wtc, release := t.cl.websocketTrackers.Get(u.String())
- go func() {
- <-t.closed.Done()
- release()
- }()
- wst := websocketTrackerStatus{u, wtc}
- go func() {
- err := wtc.Announce(tracker.Started, t.infoHash)
- if err != nil {
- t.logger.WithDefaultLevel(log.Warning).Printf(
- "error in initial announce to %q: %v",
- u.String(), err,
- )
- }
- }()
- return wst
-}
-
func (t *Torrent) startScrapingTracker(_url string) {
if _url == "" {
return
}
sl := func() torrentTrackerAnnouncer {
switch u.Scheme {
- case "ws", "wss":
- if t.cl.config.DisableWebtorrent {
- return nil
- }
- return t.startWebsocketAnnouncer(*u)
case "udp4":
if t.cl.config.DisableIPv4Peers || t.cl.config.DisableIPv4 {
return nil
Event: event,
NumWant: func() int32 {
if t.wantPeers() && len(t.cl.dialers) > 0 {
- return -1
+ return 200 // Win has UDP packet limit. See: https://github.com/anacrolix/torrent/issues/764
} else {
return 0
}
}
func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) {
- t.logger.Log(log.Fstr("hashed piece %d (passed=%t)", piece, passed).SetLevel(log.Debug))
+ t.logger.LazyLog(log.Debug, func() log.Msg {
+ return log.Fstr("hashed piece %d (passed=%t)", piece, passed)
+ })
p := t.piece(piece)
p.numVerifies++
t.cl.event.Broadcast()
} else {
log.Fmsg(
"piece %d failed hash: %d connections contributed", piece, len(p.dirtiers),
- ).AddValues(t, p).SetLevel(log.Debug).Log(t.logger)
+ ).AddValues(t, p).LogLevel(
+
+ log.Debug, t.logger)
+
pieceHashedNotCorrect.Add(1)
}
}
c._stats.incrementPiecesDirtiedGood()
}
t.clearPieceTouchers(piece)
+ hasDirty := p.hasDirtyChunks()
t.cl.unlock()
+ if hasDirty {
+ p.Flush() // You can be synchronous here!
+ }
err := p.Storage().MarkComplete()
if err != nil {
t.logger.Printf("%T: error marking piece complete %d: %s", t.storage, piece, err)
if len(bannableTouchers) >= 1 {
c := bannableTouchers[0]
- log.Printf("would have banned %v for touching piece %v after failed piece check", c.remoteIp(), piece)
- if false {
- t.cl.banPeerIP(c.remoteIp())
- c.drop()
+ if len(bannableTouchers) != 1 {
+ t.logger.Levelf(log.Warning, "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
+ // will never smart-ban them. Discovered in
+ // https://github.com/anacrolix/torrent/issues/715.
+ t.logger.Levelf(log.Warning, "banning %v for being sole dirtier of piece %v after failed piece check", c, piece)
+ c.ban()
}
}
}
}
func (t *Torrent) cancelRequestsForPiece(piece pieceIndex) {
- for ri := t.pieceRequestIndexOffset(piece); ri < t.pieceRequestIndexOffset(piece+1); ri++ {
+ start := t.pieceRequestIndexOffset(piece)
+ end := start + t.pieceNumChunks(piece)
+ for ri := start; ri < end; ri++ {
t.cancelRequest(ri)
}
}
t.iterPeers(func(p *Peer) {
remoteIp := p.remoteIp()
if remoteIp == nil {
- if p.bannableAddr.Ok() {
- log.Printf("can't get remote ip for peer %v", p)
+ if p.bannableAddr.Ok {
+ t.logger.WithDefaultLevel(log.Debug).Printf("can't get remote ip for peer %v", p)
}
return
}
netipAddr := netip.MustParseAddr(remoteIp.String())
- if option.Some(netipAddr) != p.bannableAddr {
- log.Printf(
+ if Some(netipAddr) != p.bannableAddr {
+ t.logger.WithDefaultLevel(log.Debug).Printf(
"peer remote ip does not match its bannable addr [peer=%v, remote ip=%v, bannable addr=%v]",
p, remoteIp, p.bannableAddr)
}
if _, ok := t.cl.badPeerIPs[netipAddr]; ok {
+ // Should this be a close?
p.drop()
- log.Printf("dropped %v for banned remote IP %v", p, netipAddr)
+ t.logger.WithDefaultLevel(log.Debug).Printf("dropped %v for banned remote IP %v", p, netipAddr)
}
})
}
if correct {
for peer := range failedPeers {
t.cl.banPeerIP(peer.AsSlice())
- log.Printf("smart banned %v for piece %v", peer, index)
+ t.logger.WithDefaultLevel(log.Debug).Printf("smart banned %v for piece %v", peer, index)
}
t.dropBannedPeers()
for ri := t.pieceRequestIndexOffset(index); ri < t.pieceRequestIndexOffset(index+1); ri++ {
return &t.cl.config.Callbacks
}
-func (t *Torrent) addWebSeed(url string) {
+type AddWebSeedsOpt func(*webseed.Client)
+
+// Sets the WebSeed trailing path escaper for a webseed.Client.
+func WebSeedPathEscaper(custom webseed.PathEscaper) AddWebSeedsOpt {
+ return func(c *webseed.Client) {
+ c.PathEscaper = custom
+ }
+}
+
+func (t *Torrent) AddWebSeeds(urls []string, opts ...AddWebSeedsOpt) {
+ t.cl.lock()
+ defer t.cl.unlock()
+ for _, u := range urls {
+ t.addWebSeed(u, opts...)
+ }
+}
+
+func (t *Torrent) addWebSeed(url string, opts ...AddWebSeedsOpt) {
if t.cl.config.DisableWebseeds {
return
}
callbacks: t.callbacks(),
},
client: webseed.Client{
- HttpClient: t.cl.webseedHttpClient,
+ HttpClient: t.cl.httpClient,
Url: url,
ResponseBodyWrapper: func(r io.Reader) io.Reader {
return &rateLimitedReader{
},
},
activeRequests: make(map[Request]webseed.Request, maxRequests),
- maxRequests: maxRequests,
+ }
+ ws.peer.initRequestState()
+ for _, opt := range opts {
+ opt(&ws.client)
}
ws.peer.initUpdateRequestsTimer()
ws.requesterCond.L = t.cl.locker()
}
func (t *Torrent) requestIndexToRequest(ri RequestIndex) Request {
- index := ri / t.chunksPerRegularPiece()
+ index := t.pieceIndexOfRequestIndex(ri)
return Request{
pp.Integer(index),
- t.piece(int(index)).chunkIndexSpec(ri % t.chunksPerRegularPiece()),
+ t.piece(index).chunkIndexSpec(ri % t.chunksPerRegularPiece()),
}
}
func (t *Torrent) requestIndexFromRequest(r Request) RequestIndex {
- return t.pieceRequestIndexOffset(pieceIndex(r.Index)) + uint32(r.Begin/t.chunkSize)
+ return t.pieceRequestIndexOffset(pieceIndex(r.Index)) + RequestIndex(r.Begin/t.chunkSize)
}
func (t *Torrent) pieceRequestIndexOffset(piece pieceIndex) RequestIndex {
}
func (t *Torrent) cancelRequest(r RequestIndex) *Peer {
- p := t.pendingRequests[r]
+ p := t.requestingPeer(r)
if p != nil {
p.cancel(r)
}
- delete(t.pendingRequests, r)
+ // TODO: This is a check that an old invariant holds. It can be removed after some testing.
+ //delete(t.pendingRequests, r)
+ if _, ok := t.requestState[r]; ok {
+ panic("expected request state to be gone")
+ }
return p
}
func (t *Torrent) requestingPeer(r RequestIndex) *Peer {
- return t.pendingRequests[r]
+ return t.requestState[r].peer
}
func (t *Torrent) addConnWithAllPieces(p *Peer) {
_, ok := (*f)()
return ok
}
+
+func (t *Torrent) pieceIndexOfRequestIndex(ri RequestIndex) pieceIndex {
+ return pieceIndex(ri / t.chunksPerRegularPiece())
+}
+
+func (t *Torrent) iterUndirtiedRequestIndexesInPiece(
+ reuseIter *typedRoaring.Iterator[RequestIndex],
+ piece pieceIndex,
+ f func(RequestIndex),
+) {
+ reuseIter.Initialize(&t.dirtyChunks)
+ pieceRequestIndexOffset := t.pieceRequestIndexOffset(piece)
+ iterBitmapUnsetInRange(
+ reuseIter,
+ pieceRequestIndexOffset, pieceRequestIndexOffset+t.pieceNumChunks(piece),
+ f,
+ )
+}
+
+type requestState struct {
+ peer *Peer
+ when time.Time
+}
+
+// Returns an error if a received chunk is out of bounds in someway.
+func (t *Torrent) checkValidReceiveChunk(r Request) error {
+ if !t.haveInfo() {
+ return errors.New("torrent missing info")
+ }
+ if int(r.Index) >= t.numPieces() {
+ return fmt.Errorf("chunk index %v, torrent num pieces %v", r.Index, t.numPieces())
+ }
+ pieceLength := t.pieceLength(pieceIndex(r.Index))
+ if r.Begin >= pieceLength {
+ return fmt.Errorf("chunk begins beyond end of piece (%v >= %v)", r.Begin, pieceLength)
+ }
+ // We could check chunk lengths here, but chunk request size is not changed often, and tricky
+ // for peers to manipulate as they need to send potentially large buffers to begin with. There
+ // should be considerable checks elsewhere for this case due to the network overhead. We should
+ // catch most of the overflow manipulation stuff by checking index and begin above.
+ return nil
+}