X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=reader.go;h=4b20206cbf2ecc675ccde147cecc278ebc688040;hb=HEAD;hp=a5a865710f7256703fb3099652aca500e0b78d67;hpb=4691b0b8d1943bdcd50088ceed8fcad9ff2a73ab;p=btrtrc.git diff --git a/reader.go b/reader.go index a5a86571..4b20206c 100644 --- a/reader.go +++ b/reader.go @@ -12,13 +12,17 @@ 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 // Configure the number of bytes ahead of a read that should also be prioritized in preparation - // for further reads. + // for further reads. Overridden by non-nil readahead func, see SetReadaheadFunc. SetReadahead(int64) + // 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(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() @@ -29,27 +33,37 @@ 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 - responsive bool + t *Torrent // Adjust the read/seek window to handle Readers locked to File extents and the like. offset, length int64 - // Required when modifying pos and readahead, or reading them without opMu. - mu sync.Locker - pos int64 - // Reads have been initiated since the last seek. This is used to prevent readahead occuring - // after a seek or with a new reader at the starting position. - reading bool - readahead 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 + + readahead, pos int64 // Position that reads have continued contiguously from. contiguousReadStartPos int64 // The cached piece range this reader wants downloaded. The zero value corresponds to nothing. // We cache this so that changes can be detected, and bubbled up to the Torrent only as // required. pieces pieceRange + + // Reads have been initiated since the last seek. This is used to prevent readaheads occurring + // after a seek or with a new reader at the starting position. + reading bool + responsive bool } var _ io.ReadSeekCloser = (*reader)(nil) @@ -67,10 +81,17 @@ func (r *reader) SetNonResponsive() { func (r *reader) SetReadahead(readahead int64) { r.mu.Lock() - defer r.mu.Unlock() r.readahead = readahead r.readaheadFunc = nil r.posChanged() + r.mu.Unlock() +} + +func (r *reader) SetReadaheadFunc(f ReadaheadFunc) { + r.mu.Lock() + r.readaheadFunc = f + r.posChanged() + r.mu.Unlock() } // How many bytes are available to read. Max is the most we could require. @@ -103,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 @@ -221,32 +245,45 @@ 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) + } + }() } } // Hodor func (r *reader) Close() error { r.t.cl.lock() - defer r.t.cl.unlock() r.t.deleteReader(r) + r.t.cl.unlock() return nil } @@ -262,39 +299,34 @@ func (r *reader) posChanged() { } func (r *reader) Seek(off int64, whence int) (newPos int64, err error) { - r.mu.Lock() - defer r.mu.Unlock() - newPos, err = func() (int64, error) { - switch whence { - case io.SeekStart: - return off, err - case io.SeekCurrent: - return r.pos + off, nil - case io.SeekEnd: - return r.length + off, nil - default: - return r.pos, errors.New("bad whence") - } - }() - if err != nil { - return + switch whence { + case io.SeekStart: + newPos = off + r.mu.Lock() + case io.SeekCurrent: + r.mu.Lock() + newPos = r.pos + off + case io.SeekEnd: + newPos = r.length + off + r.mu.Lock() + default: + return 0, errors.New("bad whence") } - if newPos == r.pos { - return + if newPos != r.pos { + r.reading = false + r.pos = newPos + r.contiguousReadStartPos = newPos + r.posChanged() } - r.reading = false - r.pos = newPos - r.contiguousReadStartPos = newPos - - r.posChanged() + r.mu.Unlock() return } 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 }