Drop support for go 1.20

[btrtrc.git] / reader.go

diff --git a/reader.go b/reader.go
index c8af14640bb813e98eabeca5305aa02532ffaba1..4b20206cbf2ecc675ccde147cecc278ebc688040 100644 (file)
--- a/reader.go
+++ b/reader.go
@@ -12,7 +12,7 @@ import (
 )
 
 // Accesses Torrent data via a Client. Reads block until the data is available. Seeks and readahead
-// also drive Client behaviour.
+// also drive Client behaviour. Not safe for concurrent use.
 type Reader interface {
        io.ReadSeekCloser
        missinggo.ReadContexter
@@ -22,7 +22,7 @@ type Reader interface {
        // If non-nil, the provided function is called when the implementation needs to know the
        // readahead for the current reader. Calls occur during Reads and Seeks, and while the Client is
        // locked.
-       SetReadaheadFunc(func() int64)
+       SetReadaheadFunc(ReadaheadFunc)
        // Don't wait for pieces to complete and be verified. Read calls return as soon as they can when
        // the underlying chunks become available.
        SetResponsive()
@@ -33,13 +33,21 @@ type pieceRange struct {
        begin, end pieceIndex
 }
 
+type ReadaheadContext struct {
+       ContiguousReadStartPos int64
+       CurrentPos             int64
+}
+
+// Returns the desired readahead for a Reader.
+type ReadaheadFunc func(ReadaheadContext) int64
+
 type reader struct {
        t *Torrent
        // Adjust the read/seek window to handle Readers locked to File extents and the like.
        offset, length int64
 
        // Function to dynamically calculate readahead. If nil, readahead is static.
-       readaheadFunc func() int64
+       readaheadFunc ReadaheadFunc
 
        // Required when modifying pos and readahead.
        mu sync.Locker
@@ -79,7 +87,7 @@ func (r *reader) SetReadahead(readahead int64) {
        r.mu.Unlock()
 }
 
-func (r *reader) SetReadaheadFunc(f func() int64) {
+func (r *reader) SetReadaheadFunc(f ReadaheadFunc) {
        r.mu.Lock()
        r.readaheadFunc = f
        r.posChanged()
@@ -116,7 +124,10 @@ func (r *reader) available(off, max int64) (ret int64) {
 func (r *reader) piecesUncached() (ret pieceRange) {
        ra := r.readahead
        if r.readaheadFunc != nil {
-               ra = r.readaheadFunc()
+               ra = r.readaheadFunc(ReadaheadContext{
+                       ContiguousReadStartPos: r.contiguousReadStartPos,
+                       CurrentPos:             r.pos,
+               })
        }
        if ra < 1 {
                // Needs to be at least 1, because [x, x) means we don't want
@@ -234,24 +245,37 @@ func (r *reader) readOnceAt(ctx context.Context, b []byte, pos int64) (n int, er
                        err = nil
                        return
                }
-               r.t.cl.lock()
-               // TODO: Just reset pieces in the readahead window. This might help
-               // prevent thrashing with small caches and file and piece priorities.
-               r.log(log.Fstr("error reading torrent %s piece %d offset %d, %d bytes: %v",
-                       r.t.infoHash.HexString(), firstPieceIndex, firstPieceOffset, len(b1), err))
-               if !r.t.updatePieceCompletion(firstPieceIndex) {
-                       r.log(log.Fstr("piece %d completion unchanged", firstPieceIndex))
-               }
-               // Update the rest of the piece completions in the readahead window, without alerting to
-               // changes (since only the first piece, the one above, could have generated the read error
-               // we're currently handling).
-               if r.pieces.begin != firstPieceIndex {
-                       panic(fmt.Sprint(r.pieces.begin, firstPieceIndex))
-               }
-               for index := r.pieces.begin + 1; index < r.pieces.end; index++ {
-                       r.t.updatePieceCompletion(index)
+               if r.t.closed.IsSet() {
+                       err = fmt.Errorf("reading from closed torrent: %w", err)
+                       return
                }
-               r.t.cl.unlock()
+               r.t.cl.lock()
+               // I think there's a panic here caused by the Client being closed before obtaining this
+               // lock. TestDropTorrentWithMmapStorageWhileHashing seems to tickle occasionally in CI.
+               func() {
+                       // Just add exceptions already.
+                       defer r.t.cl.unlock()
+                       if r.t.closed.IsSet() {
+                               // Can't update because Torrent's piece order is removed from Client.
+                               return
+                       }
+                       // TODO: Just reset pieces in the readahead window. This might help
+                       // prevent thrashing with small caches and file and piece priorities.
+                       r.log(log.Fstr("error reading torrent %s piece %d offset %d, %d bytes: %v",
+                               r.t.infoHash.HexString(), firstPieceIndex, firstPieceOffset, len(b1), err))
+                       if !r.t.updatePieceCompletion(firstPieceIndex) {
+                               r.log(log.Fstr("piece %d completion unchanged", firstPieceIndex))
+                       }
+                       // Update the rest of the piece completions in the readahead window, without alerting to
+                       // changes (since only the first piece, the one above, could have generated the read error
+                       // we're currently handling).
+                       if r.pieces.begin != firstPieceIndex {
+                               panic(fmt.Sprint(r.pieces.begin, firstPieceIndex))
+                       }
+                       for index := r.pieces.begin + 1; index < r.pieces.end; index++ {
+                               r.t.updatePieceCompletion(index)
+                       }
+               }()
        }
 }
 
@@ -299,10 +323,10 @@ func (r *reader) Seek(off int64, whence int) (newPos int64, err error) {
 }
 
 func (r *reader) log(m log.Msg) {
-       r.t.logger.Log(m.Skip(1))
+       r.t.logger.LogLevel(log.Debug, m.Skip(1))
 }
 
 // Implementation inspired by https://news.ycombinator.com/item?id=27019613.
-func (r *reader) defaultReadaheadFunc() int64 {
-       return r.pos - r.contiguousReadStartPos
+func defaultReadaheadFunc(r ReadaheadContext) int64 {
+       return r.CurrentPos - r.ContiguousReadStartPos
 }