-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add SeekStart method to inflator reader (#656)
* feat: add SeekStart method to inflator reader, so we can seek back to the start when retrying AddPiece * fix: storage provider restart in publish stage (#657)
- Loading branch information
Showing
10 changed files
with
317 additions
and
52 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| package shared | ||
|
|
||
| import ( | ||
| "io" | ||
|
|
||
| "github.com/ipfs/go-cid" | ||
| "golang.org/x/xerrors" | ||
|
|
||
| "github.com/filecoin-project/go-commp-utils/writer" | ||
| commcid "github.com/filecoin-project/go-fil-commcid" | ||
| commp "github.com/filecoin-project/go-fil-commp-hashhash" | ||
| ) | ||
|
|
||
| func GenerateCommp(reader io.Reader, payloadSize uint64, targetSize uint64) (cid.Cid, error) { | ||
| // dump the CARv1 payload of the CARv2 file to the Commp Writer and get back the CommP. | ||
| w := &writer.Writer{} | ||
| written, err := io.Copy(w, reader) | ||
| if err != nil { | ||
| return cid.Undef, xerrors.Errorf("failed to write to CommP writer: %w", err) | ||
| } | ||
| if written != int64(payloadSize) { | ||
| return cid.Undef, xerrors.Errorf("number of bytes written to CommP writer %d not equal to the CARv1 payload size %d", written, payloadSize) | ||
| } | ||
|
|
||
| cidAndSize, err := w.Sum() | ||
| if err != nil { | ||
| return cid.Undef, xerrors.Errorf("failed to get CommP: %w", err) | ||
| } | ||
|
|
||
| if uint64(cidAndSize.PieceSize) < targetSize { | ||
| // need to pad up! | ||
| rawPaddedCommp, err := commp.PadCommP( | ||
| // we know how long a pieceCid "hash" is, just blindly extract the trailing 32 bytes | ||
| cidAndSize.PieceCID.Hash()[len(cidAndSize.PieceCID.Hash())-32:], | ||
| uint64(cidAndSize.PieceSize), | ||
| uint64(targetSize), | ||
| ) | ||
| if err != nil { | ||
| return cid.Undef, err | ||
| } | ||
| cidAndSize.PieceCID, _ = commcid.DataCommitmentV1ToCID(rawPaddedCommp) | ||
| } | ||
|
|
||
| return cidAndSize.PieceCID, err | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package shared | ||
|
|
||
| import ( | ||
| "io" | ||
| "sync" | ||
|
|
||
| "github.com/filecoin-project/go-padreader" | ||
| "github.com/filecoin-project/go-state-types/abi" | ||
| ) | ||
|
|
||
| // ReadSeekStarter implements io.Reader and allows the caller to seek to | ||
| // the start of the reader | ||
| type ReadSeekStarter interface { | ||
| io.Reader | ||
| SeekStart() error | ||
| } | ||
|
|
||
| // inflatorReader wraps the MultiReader returned by padreader so that we can | ||
| // add a SeekStart method. It's used for example when there is an error | ||
| // reading from the reader and we need to return to the start. | ||
| type inflatorReader struct { | ||
| readSeeker io.ReadSeeker | ||
| payloadSize uint64 | ||
| targetSize abi.UnpaddedPieceSize | ||
|
|
||
| lk sync.RWMutex | ||
| paddedReader io.Reader | ||
| } | ||
|
|
||
| var _ ReadSeekStarter = (*inflatorReader)(nil) | ||
|
|
||
| func NewInflatorReader(readSeeker io.ReadSeeker, payloadSize uint64, targetSize abi.UnpaddedPieceSize) (*inflatorReader, error) { | ||
| paddedReader, err := padreader.NewInflator(readSeeker, payloadSize, targetSize) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &inflatorReader{ | ||
| readSeeker: readSeeker, | ||
| paddedReader: paddedReader, | ||
| payloadSize: payloadSize, | ||
| targetSize: targetSize, | ||
| }, nil | ||
| } | ||
|
|
||
| func (r *inflatorReader) Read(p []byte) (n int, err error) { | ||
| r.lk.RLock() | ||
| defer r.lk.RUnlock() | ||
|
|
||
| return r.paddedReader.Read(p) | ||
| } | ||
|
|
||
| func (r *inflatorReader) SeekStart() error { | ||
| r.lk.Lock() | ||
| defer r.lk.Unlock() | ||
|
|
||
| // Seek to the start of the underlying reader | ||
| _, err := r.readSeeker.Seek(0, io.SeekStart) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // Recreate the padded reader | ||
| paddedReader, err := padreader.NewInflator(r.readSeeker, r.payloadSize, r.targetSize) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| r.paddedReader = paddedReader | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| package shared | ||
|
|
||
| import ( | ||
| "io" | ||
| "os" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/filecoin-project/go-state-types/abi" | ||
| ) | ||
|
|
||
| func TestInflatorReader(t *testing.T) { | ||
| req := require.New(t) | ||
|
|
||
| // Create a temp file | ||
| f, err := os.CreateTemp(t.TempDir(), "buff") | ||
| req.NoError(err) | ||
| defer f.Close() // nolint | ||
|
|
||
| // Store a sample string to the temp file | ||
| sampleString := "Testing 123" | ||
| n, err := f.WriteString(sampleString) | ||
| req.NoError(err) | ||
| req.Len(sampleString, n) | ||
|
|
||
| // Seek to the start of the file | ||
| _, err = f.Seek(0, io.SeekStart) | ||
| req.NoError(err) | ||
|
|
||
| // Create an inflator reader over the file | ||
| paddedSize := 1024 | ||
| padded := abi.PaddedPieceSize(paddedSize) | ||
| ir, err := NewInflatorReader(f, uint64(n), padded.Unpadded()) | ||
| req.NoError(err) | ||
|
|
||
| // Read all bytes into a buffer | ||
| buff := make([]byte, paddedSize) | ||
| _, err = ir.Read(buff) | ||
| req.NoError(err) | ||
|
|
||
| // Check that the correct number of bytes was read | ||
| req.Len(buff, paddedSize) | ||
| // Check that the first part of the buffer matches the sample string | ||
| req.Equal([]byte(sampleString), buff[:len(sampleString)]) | ||
| // Check that the rest of the buffer is zeros | ||
| for _, b := range buff[len(sampleString):] { | ||
| req.EqualValues(0, b) | ||
| } | ||
|
|
||
| // Seek to the start of the reader | ||
| err = ir.SeekStart() | ||
| req.NoError(err) | ||
|
|
||
| // Verify that the reader returns the correct bytes, as above | ||
| buff = make([]byte, paddedSize) | ||
| _, err = ir.Read(buff) | ||
| req.NoError(err) | ||
| req.Len(buff, paddedSize) | ||
| req.Equal([]byte(sampleString), buff[:len(sampleString)]) | ||
| for _, b := range buff[len(sampleString):] { | ||
| req.EqualValues(0, b) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.