storage: Weave end offset into unsealed reader code
This commit is contained in:
parent
d1d4b35adf
commit
8b2ef40f4e
44
lib/readerutil/readerutil.go
Normal file
44
lib/readerutil/readerutil.go
Normal file
@ -0,0 +1,44 @@
|
|||||||
|
package readerutil
|
||||||
|
|
||||||
|
import (
|
||||||
|
"io"
|
||||||
|
"os"
|
||||||
|
)
|
||||||
|
|
||||||
|
// NewReadSeekerFromReaderAt returns a new io.ReadSeeker from a io.ReaderAt.
|
||||||
|
// The returned io.ReadSeeker will read from the io.ReaderAt starting at the
|
||||||
|
// given base offset.
|
||||||
|
func NewReadSeekerFromReaderAt(readerAt io.ReaderAt, base int64) io.ReadSeeker {
|
||||||
|
return &readSeekerFromReaderAt{
|
||||||
|
readerAt: readerAt,
|
||||||
|
base: base,
|
||||||
|
pos: 0,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type readSeekerFromReaderAt struct {
|
||||||
|
readerAt io.ReaderAt
|
||||||
|
base int64
|
||||||
|
pos int64
|
||||||
|
}
|
||||||
|
|
||||||
|
func (rs *readSeekerFromReaderAt) Read(p []byte) (n int, err error) {
|
||||||
|
n, err = rs.readerAt.ReadAt(p, rs.pos+rs.base)
|
||||||
|
rs.pos += int64(n)
|
||||||
|
return n, err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (rs *readSeekerFromReaderAt) Seek(offset int64, whence int) (int64, error) {
|
||||||
|
switch whence {
|
||||||
|
case io.SeekStart:
|
||||||
|
rs.pos = offset
|
||||||
|
case io.SeekCurrent:
|
||||||
|
rs.pos += offset
|
||||||
|
case io.SeekEnd:
|
||||||
|
return 0, io.ErrUnexpectedEOF
|
||||||
|
default:
|
||||||
|
return 0, os.ErrInvalid
|
||||||
|
}
|
||||||
|
|
||||||
|
return rs.pos, nil
|
||||||
|
}
|
@ -3,6 +3,7 @@ package paths
|
|||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
"os"
|
||||||
"strconv"
|
"strconv"
|
||||||
@ -35,7 +36,7 @@ func (d *DefaultPartialFileHandler) HasAllocated(pf *partialfile.PartialFile, of
|
|||||||
return pf.HasAllocated(offset, size)
|
return pf.HasAllocated(offset, size)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *DefaultPartialFileHandler) Reader(pf *partialfile.PartialFile, offset storiface.PaddedByteIndex, size abi.PaddedPieceSize) (*os.File, error) {
|
func (d *DefaultPartialFileHandler) Reader(pf *partialfile.PartialFile, offset storiface.PaddedByteIndex, size abi.PaddedPieceSize) (io.Reader, error) {
|
||||||
return pf.Reader(offset, size)
|
return pf.Reader(offset, size)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -2,9 +2,8 @@ package paths
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"os"
|
|
||||||
|
|
||||||
"github.com/filecoin-project/go-state-types/abi"
|
"github.com/filecoin-project/go-state-types/abi"
|
||||||
|
"io"
|
||||||
|
|
||||||
"github.com/filecoin-project/lotus/storage/sealer/fsutil"
|
"github.com/filecoin-project/lotus/storage/sealer/fsutil"
|
||||||
"github.com/filecoin-project/lotus/storage/sealer/partialfile"
|
"github.com/filecoin-project/lotus/storage/sealer/partialfile"
|
||||||
@ -24,7 +23,7 @@ type PartialFileHandler interface {
|
|||||||
HasAllocated(pf *partialfile.PartialFile, offset storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize) (bool, error)
|
HasAllocated(pf *partialfile.PartialFile, offset storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize) (bool, error)
|
||||||
|
|
||||||
// Reader returns a file from which we can read the unsealed piece in the partial file.
|
// Reader returns a file from which we can read the unsealed piece in the partial file.
|
||||||
Reader(pf *partialfile.PartialFile, offset storiface.PaddedByteIndex, size abi.PaddedPieceSize) (*os.File, error)
|
Reader(pf *partialfile.PartialFile, offset storiface.PaddedByteIndex, size abi.PaddedPieceSize) (io.Reader, error)
|
||||||
|
|
||||||
// Close closes the partial file
|
// Close closes the partial file
|
||||||
Close(pf *partialfile.PartialFile) error
|
Close(pf *partialfile.PartialFile) error
|
||||||
|
@ -559,7 +559,7 @@ func (r *Remote) CheckIsUnsealed(ctx context.Context, s storiface.SectorRef, off
|
|||||||
// 1. no worker(local worker included) has an unsealed file for the given sector OR
|
// 1. no worker(local worker included) has an unsealed file for the given sector OR
|
||||||
// 2. no worker(local worker included) has the unsealed piece in their unsealed sector file.
|
// 2. no worker(local worker included) has the unsealed piece in their unsealed sector file.
|
||||||
// Will return a nil reader and a nil error in such a case.
|
// Will return a nil reader and a nil error in such a case.
|
||||||
func (r *Remote) Reader(ctx context.Context, s storiface.SectorRef, offset, size abi.PaddedPieceSize) (func(startOffsetAligned storiface.PaddedByteIndex) (io.ReadCloser, error), error) {
|
func (r *Remote) Reader(ctx context.Context, s storiface.SectorRef, offset, size abi.PaddedPieceSize) (func(startOffsetAligned, endOffsetAligned storiface.PaddedByteIndex) (io.ReadCloser, error), error) {
|
||||||
ft := storiface.FTUnsealed
|
ft := storiface.FTUnsealed
|
||||||
|
|
||||||
// check if we have the unsealed sector file locally
|
// check if we have the unsealed sector file locally
|
||||||
@ -598,20 +598,8 @@ func (r *Remote) Reader(ctx context.Context, s storiface.SectorRef, offset, size
|
|||||||
if has {
|
if has {
|
||||||
log.Infof("returning piece reader for local unsealed piece sector=%+v, (offset=%d, size=%d)", s.ID, offset, size)
|
log.Infof("returning piece reader for local unsealed piece sector=%+v, (offset=%d, size=%d)", s.ID, offset, size)
|
||||||
|
|
||||||
return func(startOffsetAligned storiface.PaddedByteIndex) (io.ReadCloser, error) {
|
return func(startOffsetAligned, endOffsetAligned storiface.PaddedByteIndex) (io.ReadCloser, error) {
|
||||||
// don't reuse between readers unless closed
|
r, err := r.pfHandler.Reader(pf, storiface.PaddedByteIndex(offset)+startOffsetAligned, abi.PaddedPieceSize(endOffsetAligned-startOffsetAligned))
|
||||||
f := pf
|
|
||||||
pf = nil
|
|
||||||
|
|
||||||
if f == nil {
|
|
||||||
f, err = r.pfHandler.OpenPartialFile(abi.PaddedPieceSize(ssize), path)
|
|
||||||
if err != nil {
|
|
||||||
return nil, xerrors.Errorf("opening partial file: %w", err)
|
|
||||||
}
|
|
||||||
log.Debugf("local partial file (re)opened %s (+%d,%d)", path, offset, size)
|
|
||||||
}
|
|
||||||
|
|
||||||
r, err := r.pfHandler.Reader(f, storiface.PaddedByteIndex(offset)+startOffsetAligned, size-abi.PaddedPieceSize(startOffsetAligned))
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -622,22 +610,7 @@ func (r *Remote) Reader(ctx context.Context, s storiface.SectorRef, offset, size
|
|||||||
}{
|
}{
|
||||||
Reader: r,
|
Reader: r,
|
||||||
Closer: funcCloser(func() error {
|
Closer: funcCloser(func() error {
|
||||||
// if we already have a reader cached, close this one
|
// todo keep some refcount, close pf (or push to some lru) when it hits 0
|
||||||
if pf != nil {
|
|
||||||
if f == nil {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if pf == f {
|
|
||||||
pf = nil
|
|
||||||
}
|
|
||||||
|
|
||||||
tmp := f
|
|
||||||
f = nil
|
|
||||||
return tmp.Close()
|
|
||||||
}
|
|
||||||
|
|
||||||
// otherwise stash it away for reuse
|
|
||||||
pf = f
|
|
||||||
return nil
|
return nil
|
||||||
}),
|
}),
|
||||||
}, nil
|
}, nil
|
||||||
@ -685,10 +658,10 @@ func (r *Remote) Reader(ctx context.Context, s storiface.SectorRef, offset, size
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
return func(startOffsetAligned storiface.PaddedByteIndex) (io.ReadCloser, error) {
|
return func(startOffsetAligned, endOffsetAligned storiface.PaddedByteIndex) (io.ReadCloser, error) {
|
||||||
// readRemote fetches a reader that we can use to read the unsealed piece from the remote worker.
|
// readRemote fetches a reader that we can use to read the unsealed piece from the remote worker.
|
||||||
// It uses a ranged HTTP query to ensure we ONLY read the unsealed piece and not the entire unsealed file.
|
// It uses a ranged HTTP query to ensure we ONLY read the unsealed piece and not the entire unsealed file.
|
||||||
rd, err := r.readRemote(ctx, url, offset+abi.PaddedPieceSize(startOffsetAligned), size)
|
rd, err := r.readRemote(ctx, url, offset+abi.PaddedPieceSize(startOffsetAligned), offset+abi.PaddedPieceSize(endOffsetAligned))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warnw("reading from remote", "url", url, "error", err)
|
log.Warnw("reading from remote", "url", url, "error", err)
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -2,6 +2,7 @@ package partialfile
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/binary"
|
"encoding/binary"
|
||||||
|
"github.com/filecoin-project/lotus/lib/readerutil"
|
||||||
"io"
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"syscall"
|
"syscall"
|
||||||
@ -249,7 +250,10 @@ func (pf *PartialFile) Free(offset storiface.PaddedByteIndex, size abi.PaddedPie
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pf *PartialFile) Reader(offset storiface.PaddedByteIndex, size abi.PaddedPieceSize) (*os.File, error) {
|
// Reader forks off a new reader from the underlying file, and returns a reader
|
||||||
|
// starting at the given offset and reading the given size. Safe for concurrent
|
||||||
|
// use.
|
||||||
|
func (pf *PartialFile) Reader(offset storiface.PaddedByteIndex, size abi.PaddedPieceSize) (io.Reader, error) {
|
||||||
if _, err := pf.file.Seek(int64(offset), io.SeekStart); err != nil {
|
if _, err := pf.file.Seek(int64(offset), io.SeekStart); err != nil {
|
||||||
return nil, xerrors.Errorf("seek piece start: %w", err)
|
return nil, xerrors.Errorf("seek piece start: %w", err)
|
||||||
}
|
}
|
||||||
@ -275,7 +279,7 @@ func (pf *PartialFile) Reader(offset storiface.PaddedByteIndex, size abi.PaddedP
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return pf.file, nil
|
return io.LimitReader(readerutil.NewReadSeekerFromReaderAt(pf.file, int64(offset)), int64(size)), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pf *PartialFile) Allocated() (rlepluslazy.RunIterator, error) {
|
func (pf *PartialFile) Allocated() (rlepluslazy.RunIterator, error) {
|
||||||
|
@ -71,7 +71,7 @@ func (p *pieceProvider) IsUnsealed(ctx context.Context, sector storiface.SectorR
|
|||||||
// It will NOT try to schedule an Unseal of a sealed sector file for the read.
|
// It will NOT try to schedule an Unseal of a sealed sector file for the read.
|
||||||
//
|
//
|
||||||
// Returns a nil reader if the piece does NOT exist in any unsealed file or there is no unsealed file for the given sector on any of the workers.
|
// Returns a nil reader if the piece does NOT exist in any unsealed file or there is no unsealed file for the given sector on any of the workers.
|
||||||
func (p *pieceProvider) tryReadUnsealedPiece(ctx context.Context, pc cid.Cid, sector storiface.SectorRef, pieceOffset storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize) (mount.Reader, error) {
|
func (p *pieceProvider) tryReadUnsealedPiece(ctx context.Context, pc cid.Cid, sector storiface.SectorRef, pieceOffset storiface.UnpaddedByteIndex, pieceSize abi.UnpaddedPieceSize) (mount.Reader, error) {
|
||||||
// acquire a lock purely for reading unsealed sectors
|
// acquire a lock purely for reading unsealed sectors
|
||||||
ctx, cancel := context.WithCancel(ctx)
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
if err := p.index.StorageLock(ctx, sector.ID, storiface.FTUnsealed, storiface.FTNone); err != nil {
|
if err := p.index.StorageLock(ctx, sector.ID, storiface.FTUnsealed, storiface.FTNone); err != nil {
|
||||||
@ -82,7 +82,7 @@ func (p *pieceProvider) tryReadUnsealedPiece(ctx context.Context, pc cid.Cid, se
|
|||||||
// Reader returns a reader getter for an unsealed piece at the given offset in the given sector.
|
// Reader returns a reader getter for an unsealed piece at the given offset in the given sector.
|
||||||
// The returned reader will be nil if none of the workers has an unsealed sector file containing
|
// The returned reader will be nil if none of the workers has an unsealed sector file containing
|
||||||
// the unsealed piece.
|
// the unsealed piece.
|
||||||
rg, err := p.storage.Reader(ctx, sector, abi.PaddedPieceSize(pieceOffset.Padded()), size.Padded())
|
rg, err := p.storage.Reader(ctx, sector, abi.PaddedPieceSize(pieceOffset.Padded()), pieceSize.Padded())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cancel()
|
cancel()
|
||||||
log.Debugf("did not get storage reader;sector=%+v, err:%s", sector.ID, err)
|
log.Debugf("did not get storage reader;sector=%+v, err:%s", sector.ID, err)
|
||||||
@ -93,19 +93,23 @@ func (p *pieceProvider) tryReadUnsealedPiece(ctx context.Context, pc cid.Cid, se
|
|||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
buf := make([]byte, fr32.BufSize(size.Padded()))
|
buf := make([]byte, fr32.BufSize(pieceSize.Padded()))
|
||||||
|
|
||||||
pr, err := (&pieceReader{
|
pr, err := (&pieceReader{
|
||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
getReader: func(ctx context.Context, startOffset uint64) (io.ReadCloser, error) {
|
getReader: func(ctx context.Context, startOffset, readSize uint64) (io.ReadCloser, error) {
|
||||||
startOffsetAligned := storiface.UnpaddedByteIndex(startOffset / 127 * 127) // floor to multiple of 127
|
startOffsetAligned := storiface.UnpaddedByteIndex(startOffset / 127 * 127) // floor to multiple of 127
|
||||||
|
startOffsetDiff := int(startOffset - uint64(startOffsetAligned))
|
||||||
|
|
||||||
r, err := rg(startOffsetAligned.Padded())
|
endOffset := startOffset + readSize
|
||||||
|
endOffsetAligned := storiface.UnpaddedByteIndex((endOffset + 126) / 127 * 127) // ceil to multiple of 127
|
||||||
|
|
||||||
|
r, err := rg(startOffsetAligned.Padded(), endOffsetAligned.Padded())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, xerrors.Errorf("getting reader at +%d: %w", startOffsetAligned, err)
|
return nil, xerrors.Errorf("getting reader at +%d: %w", startOffsetAligned, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
upr, err := fr32.NewUnpadReaderBuf(r, size.Padded(), buf)
|
upr, err := fr32.NewUnpadReaderBuf(r, pieceSize.Padded(), buf)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
r.Close() // nolint
|
r.Close() // nolint
|
||||||
return nil, xerrors.Errorf("creating unpadded reader: %w", err)
|
return nil, xerrors.Errorf("creating unpadded reader: %w", err)
|
||||||
@ -113,7 +117,7 @@ func (p *pieceProvider) tryReadUnsealedPiece(ctx context.Context, pc cid.Cid, se
|
|||||||
|
|
||||||
bir := bufio.NewReaderSize(upr, 127)
|
bir := bufio.NewReaderSize(upr, 127)
|
||||||
if startOffset > uint64(startOffsetAligned) {
|
if startOffset > uint64(startOffsetAligned) {
|
||||||
if _, err := bir.Discard(int(startOffset - uint64(startOffsetAligned))); err != nil {
|
if _, err := bir.Discard(startOffsetDiff); err != nil {
|
||||||
r.Close() // nolint
|
r.Close() // nolint
|
||||||
return nil, xerrors.Errorf("discarding bytes for startOffset: %w", err)
|
return nil, xerrors.Errorf("discarding bytes for startOffset: %w", err)
|
||||||
}
|
}
|
||||||
@ -129,7 +133,7 @@ func (p *pieceProvider) tryReadUnsealedPiece(ctx context.Context, pc cid.Cid, se
|
|||||||
}),
|
}),
|
||||||
}, nil
|
}, nil
|
||||||
},
|
},
|
||||||
len: size,
|
len: pieceSize,
|
||||||
onClose: cancel,
|
onClose: cancel,
|
||||||
pieceCid: pc,
|
pieceCid: pc,
|
||||||
}).init()
|
}).init()
|
||||||
|
@ -21,7 +21,7 @@ import (
|
|||||||
var MaxPieceReaderBurnBytes int64 = 1 << 20 // 1M
|
var MaxPieceReaderBurnBytes int64 = 1 << 20 // 1M
|
||||||
var ReadBuf = 128 * (127 * 8) // unpadded(128k)
|
var ReadBuf = 128 * (127 * 8) // unpadded(128k)
|
||||||
|
|
||||||
type pieceGetter func(ctx context.Context, offset uint64) (io.ReadCloser, error)
|
type pieceGetter func(ctx context.Context, offset, size uint64) (io.ReadCloser, error)
|
||||||
|
|
||||||
type pieceReader struct {
|
type pieceReader struct {
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
@ -43,7 +43,7 @@ func (p *pieceReader) init() (_ *pieceReader, err error) {
|
|||||||
stats.Record(p.ctx, metrics.DagStorePRInitCount.M(1))
|
stats.Record(p.ctx, metrics.DagStorePRInitCount.M(1))
|
||||||
|
|
||||||
p.rAt = 0
|
p.rAt = 0
|
||||||
p.r, err = p.getReader(p.ctx, uint64(p.rAt))
|
p.r, err = p.getReader(p.ctx, uint64(p.rAt), uint64(p.len))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -160,7 +160,7 @@ func (p *pieceReader) readAtUnlocked(b []byte, off int64) (n int, err error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
p.rAt = off
|
p.rAt = off
|
||||||
p.r, err = p.getReader(p.ctx, uint64(p.rAt))
|
p.r, err = p.getReader(p.ctx, uint64(p.rAt), uint64(p.len))
|
||||||
p.br = bufio.NewReaderSize(p.r, ReadBuf)
|
p.br = bufio.NewReaderSize(p.r, ReadBuf)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, xerrors.Errorf("getting backing reader: %w", err)
|
return 0, xerrors.Errorf("getting backing reader: %w", err)
|
||||||
|
Loading…
Reference in New Issue
Block a user