lotus/ffiwrapper/sealer_cgo.go

557 lines
15 KiB
Go
Raw Normal View History

2020-03-26 02:50:56 +00:00
//+build cgo
package ffiwrapper
import (
2020-05-29 15:21:10 +00:00
"bufio"
2020-06-09 09:13:23 +00:00
"bytes"
2020-03-26 02:50:56 +00:00
"context"
"io"
"math/bits"
"os"
2020-05-29 15:21:10 +00:00
"runtime"
2020-03-26 02:50:56 +00:00
"github.com/ipfs/go-cid"
"golang.org/x/xerrors"
ffi "github.com/filecoin-project/filecoin-ffi"
2020-06-09 09:13:23 +00:00
commcid "github.com/filecoin-project/go-fil-commcid"
2020-03-26 02:50:56 +00:00
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-storage/storage"
"github.com/filecoin-project/sector-storage/fr32"
2020-03-27 23:21:36 +00:00
"github.com/filecoin-project/sector-storage/stores"
2020-05-18 22:08:11 +00:00
"github.com/filecoin-project/sector-storage/storiface"
2020-03-27 23:21:36 +00:00
"github.com/filecoin-project/sector-storage/zerocomm"
2020-03-26 02:50:56 +00:00
)
2020-03-26 19:34:38 +00:00
var _ Storage = &Sealer{}
2020-03-26 02:50:56 +00:00
2020-03-26 19:34:38 +00:00
func New(sectors SectorProvider, cfg *Config) (*Sealer, error) {
2020-03-26 02:50:56 +00:00
sectorSize, err := sizeFromConfig(*cfg)
if err != nil {
return nil, err
}
2020-03-26 19:34:38 +00:00
sb := &Sealer{
2020-03-26 02:50:56 +00:00
sealProofType: cfg.SealProofType,
ssize: sectorSize,
sectors: sectors,
stopping: make(chan struct{}),
}
return sb, nil
}
2020-03-26 19:34:38 +00:00
func (sb *Sealer) NewSector(ctx context.Context, sector abi.SectorID) error {
2020-03-26 02:50:56 +00:00
// TODO: Allocate the sector here instead of in addpiece
return nil
}
2020-03-26 19:34:38 +00:00
func (sb *Sealer) AddPiece(ctx context.Context, sector abi.SectorID, existingPieceSizes []abi.UnpaddedPieceSize, pieceSize abi.UnpaddedPieceSize, file storage.Data) (abi.PieceInfo, error) {
2020-05-14 15:35:38 +00:00
var offset abi.UnpaddedPieceSize
for _, size := range existingPieceSizes {
offset += size
}
maxPieceSize := abi.PaddedPieceSize(sb.ssize)
2020-05-14 15:35:38 +00:00
if offset.Padded()+pieceSize.Padded() > maxPieceSize {
2020-05-14 15:35:38 +00:00
return abi.PieceInfo{}, xerrors.Errorf("can't add %d byte piece to sector %v with %d bytes of existing pieces", pieceSize, sector, offset)
2020-03-26 02:50:56 +00:00
}
2020-05-14 15:35:38 +00:00
var err error
2020-03-26 02:50:56 +00:00
var done func()
2020-05-14 15:35:38 +00:00
var stagedFile *partialFile
2020-03-26 02:50:56 +00:00
defer func() {
if done != nil {
done()
}
if stagedFile != nil {
if err := stagedFile.Close(); err != nil {
log.Errorf("closing staged file: %+v", err)
}
}
}()
var stagedPath stores.SectorPaths
if len(existingPieceSizes) == 0 {
stagedPath, done, err = sb.sectors.AcquireSector(ctx, sector, 0, stores.FTUnsealed, true)
if err != nil {
return abi.PieceInfo{}, xerrors.Errorf("acquire unsealed sector: %w", err)
}
2020-05-14 15:35:38 +00:00
stagedFile, err = createPartialFile(maxPieceSize, stagedPath.Unsealed)
2020-03-26 02:50:56 +00:00
if err != nil {
2020-05-14 15:35:38 +00:00
return abi.PieceInfo{}, xerrors.Errorf("creating unsealed sector file: %w", err)
2020-03-26 02:50:56 +00:00
}
} else {
stagedPath, done, err = sb.sectors.AcquireSector(ctx, sector, stores.FTUnsealed, 0, true)
if err != nil {
return abi.PieceInfo{}, xerrors.Errorf("acquire unsealed sector: %w", err)
}
2020-05-14 15:35:38 +00:00
stagedFile, err = openPartialFile(maxPieceSize, stagedPath.Unsealed)
2020-03-26 02:50:56 +00:00
if err != nil {
2020-05-14 15:35:38 +00:00
return abi.PieceInfo{}, xerrors.Errorf("opening unsealed sector file: %w", err)
2020-03-26 02:50:56 +00:00
}
2020-05-14 15:35:38 +00:00
}
2020-03-26 02:50:56 +00:00
w, err := stagedFile.Writer(storiface.UnpaddedByteIndex(offset).Padded(), pieceSize.Padded())
2020-05-14 15:35:38 +00:00
if err != nil {
return abi.PieceInfo{}, xerrors.Errorf("getting partial file writer: %w", err)
}
2020-06-09 09:13:23 +00:00
pw := fr32.NewPadWriter(w)
2020-05-29 15:21:10 +00:00
pr := io.TeeReader(io.LimitReader(file, int64(pieceSize)), pw)
2020-03-26 02:50:56 +00:00
2020-06-09 09:13:23 +00:00
chunk := abi.PaddedPieceSize(4 << 20)
2020-06-09 10:06:21 +00:00
buf := make([]byte, chunk.Unpadded())
2020-06-09 09:13:23 +00:00
var pieceCids []abi.PieceInfo
for {
var read int
for rbuf := buf; len(rbuf) > 0; {
n, err := pr.Read(rbuf)
if err != nil && err != io.EOF {
return abi.PieceInfo{}, xerrors.Errorf("pr read error: %w", err)
}
rbuf = rbuf[n:]
read += n
if err == io.EOF {
break
}
2020-06-09 09:13:23 +00:00
}
if read == 0 {
2020-06-09 09:13:23 +00:00
break
}
c, err := sb.pieceCid(buf[:read])
2020-06-09 10:06:21 +00:00
if err != nil {
return abi.PieceInfo{}, xerrors.Errorf("pieceCid error: %w", err)
2020-06-09 09:13:23 +00:00
}
2020-06-09 10:06:21 +00:00
pieceCids = append(pieceCids, abi.PieceInfo{
Size: abi.UnpaddedPieceSize(len(buf[:read])).Padded(),
2020-06-09 10:06:21 +00:00
PieceCID: c,
})
2020-03-26 02:50:56 +00:00
}
2020-05-29 15:21:10 +00:00
if err := pw.Close(); err != nil {
return abi.PieceInfo{}, xerrors.Errorf("closing padded writer: %w", err)
}
if err := stagedFile.MarkAllocated(storiface.UnpaddedByteIndex(offset).Padded(), pieceSize.Padded()); err != nil {
2020-05-14 15:35:38 +00:00
return abi.PieceInfo{}, xerrors.Errorf("marking data range as allocated: %w", err)
}
if err := stagedFile.Close(); err != nil {
2020-03-26 02:50:56 +00:00
return abi.PieceInfo{}, err
}
2020-05-14 15:35:38 +00:00
stagedFile = nil
2020-03-26 02:50:56 +00:00
2020-06-09 09:13:23 +00:00
if len(pieceCids) == 1 {
return pieceCids[0], nil
}
pieceCID, err := ffi.GenerateUnsealedCID(sb.sealProofType, pieceCids)
if err != nil {
return abi.PieceInfo{}, xerrors.Errorf("generate unsealed CID: %w", err)
}
commp, err := commcid.CIDToDataCommitmentV1(pieceCID)
if err != nil {
return abi.PieceInfo{}, err
}
2020-03-26 02:50:56 +00:00
return abi.PieceInfo{
Size: pieceSize.Padded(),
2020-06-09 09:13:23 +00:00
PieceCID: commcid.PieceCommitmentV1ToCID(commp),
}, nil
2020-03-26 02:50:56 +00:00
}
2020-06-09 09:13:23 +00:00
func (sb *Sealer) pieceCid(in []byte) (cid.Cid, error) {
prf, werr, err := ToReadableFile(bytes.NewReader(in), int64(len(in)))
if err != nil {
return cid.Undef, xerrors.Errorf("getting tee reader pipe: %w", err)
}
pieceCID, err := ffi.GeneratePieceCIDFromFile(sb.sealProofType, prf, abi.UnpaddedPieceSize(len(in)))
if err != nil {
return cid.Undef, xerrors.Errorf("generating piece commitment: %w", err)
}
prf.Close()
2020-05-07 22:22:58 +00:00
2020-06-09 09:13:23 +00:00
return pieceCID, werr()
2020-05-07 22:22:58 +00:00
}
func (sb *Sealer) UnsealPiece(ctx context.Context, sector abi.SectorID, offset storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize, randomness abi.SealRandomness, commd cid.Cid) error {
maxPieceSize := abi.PaddedPieceSize(sb.ssize)
2020-05-18 22:08:11 +00:00
// try finding existing
unsealedPath, done, err := sb.sectors.AcquireSector(ctx, sector, stores.FTUnsealed, stores.FTNone, false)
var pf *partialFile
switch {
case xerrors.Is(err, storiface.ErrSectorNotFound):
unsealedPath, done, err = sb.sectors.AcquireSector(ctx, sector, stores.FTNone, stores.FTUnsealed, false)
if err != nil {
return xerrors.Errorf("acquire unsealed sector path (allocate): %w", err)
}
defer done()
pf, err = createPartialFile(maxPieceSize, unsealedPath.Unsealed)
if err != nil {
return xerrors.Errorf("create unsealed file: %w", err)
}
case err == nil:
defer done()
pf, err = openPartialFile(maxPieceSize, unsealedPath.Unsealed)
if err != nil {
return xerrors.Errorf("opening partial file: %w", err)
}
default:
return xerrors.Errorf("acquire unsealed sector path (existing): %w", err)
}
defer pf.Close()
allocated, err := pf.Allocated()
if err != nil {
return xerrors.Errorf("getting bitruns of allocated data: %w", err)
}
toUnseal, err := computeUnsealRanges(allocated, offset, size)
if err != nil {
return xerrors.Errorf("computing unseal ranges: %w", err)
}
if !toUnseal.HasNext() {
return nil
}
srcPaths, srcDone, err := sb.sectors.AcquireSector(ctx, sector, stores.FTCache|stores.FTSealed, stores.FTNone, false)
if err != nil {
return xerrors.Errorf("acquire sealed sector paths: %w", err)
}
defer srcDone()
2020-06-08 21:53:31 +00:00
sealed, err := os.OpenFile(srcPaths.Sealed, os.O_RDONLY, 0644)
if err != nil {
return xerrors.Errorf("opening sealed file: %w", err)
}
defer sealed.Close()
var at, nextat abi.PaddedPieceSize
2020-05-18 22:08:11 +00:00
for {
piece, err := toUnseal.NextRun()
if err != nil {
return xerrors.Errorf("getting next range to unseal: %w", err)
}
at = nextat
nextat += abi.PaddedPieceSize(piece.Len)
2020-05-18 22:08:11 +00:00
if !piece.Val {
continue
}
out, err := pf.Writer(offset.Padded(), size.Padded())
2020-05-18 22:08:11 +00:00
if err != nil {
return xerrors.Errorf("getting partial file writer: %w", err)
}
// <eww>
2020-06-08 21:53:31 +00:00
opr, opw, err := os.Pipe()
2020-05-18 22:08:11 +00:00
if err != nil {
2020-06-08 21:53:31 +00:00
return xerrors.Errorf("creating out pipe: %w", err)
2020-05-18 22:08:11 +00:00
}
2020-06-08 21:53:31 +00:00
2020-05-18 22:08:11 +00:00
var perr error
outWait := make(chan struct{})
{
go func() {
defer close(outWait)
2020-06-08 21:53:31 +00:00
defer opr.Close()
2020-05-18 23:03:42 +00:00
2020-06-09 09:13:23 +00:00
padwriter := fr32.NewPadWriter(out)
if err != nil {
2020-06-09 09:13:23 +00:00
perr = xerrors.Errorf("creating new padded writer: %w", err)
return
}
2020-05-29 15:21:10 +00:00
bsize := uint64(size.Padded())
if bsize > uint64(runtime.NumCPU())*fr32.MTTresh {
bsize = uint64(runtime.NumCPU()) * fr32.MTTresh
}
2020-06-09 10:06:21 +00:00
bw := bufio.NewWriterSize(padwriter, int(abi.PaddedPieceSize(bsize).Unpadded()))
2020-06-09 09:13:23 +00:00
2020-06-09 10:06:21 +00:00
_, err = io.CopyN(bw, opr, int64(size))
2020-06-09 09:13:23 +00:00
if err != nil {
perr = xerrors.Errorf("copying data: %w", err)
return
}
2020-05-29 15:21:10 +00:00
2020-06-09 09:13:23 +00:00
if err := bw.Flush(); err != nil {
perr = xerrors.Errorf("flushing unpadded data: %w", err)
return
}
if err := padwriter.Close(); err != nil {
perr = xerrors.Errorf("closing padwriter: %w", err)
return
}
2020-05-18 22:08:11 +00:00
}()
}
// </eww>
// TODO: This may be possible to do in parallel
err = ffi.UnsealRange(sb.sealProofType,
srcPaths.Cache,
2020-06-08 21:53:31 +00:00
sealed,
opw,
2020-05-18 22:08:11 +00:00
sector.Number,
sector.Miner,
randomness,
2020-05-18 23:03:42 +00:00
commd,
uint64(at.Unpadded()),
uint64(abi.PaddedPieceSize(piece.Len).Unpadded()))
2020-06-08 21:53:31 +00:00
_ = opr.Close()
2020-05-18 22:08:11 +00:00
if err != nil {
return xerrors.Errorf("unseal range: %w", err)
}
select {
case <-outWait:
case <-ctx.Done():
return ctx.Err()
}
if perr != nil {
return xerrors.Errorf("piping output to unsealed file: %w", perr)
}
if err := pf.MarkAllocated(storiface.PaddedByteIndex(at), abi.PaddedPieceSize(piece.Len)); err != nil {
2020-05-18 22:08:11 +00:00
return xerrors.Errorf("marking unsealed range as allocated: %w", err)
}
if !toUnseal.HasNext() {
break
}
}
return nil
}
func (sb *Sealer) ReadPiece(ctx context.Context, writer io.Writer, sector abi.SectorID, offset storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize) error {
2020-05-18 22:08:11 +00:00
path, done, err := sb.sectors.AcquireSector(ctx, sector, stores.FTUnsealed, stores.FTNone, false)
if err != nil {
return xerrors.Errorf("acquire unsealed sector path: %w", err)
}
defer done()
maxPieceSize := abi.PaddedPieceSize(sb.ssize)
2020-05-18 22:08:11 +00:00
pf, err := openPartialFile(maxPieceSize, path.Unsealed)
if xerrors.Is(err, os.ErrNotExist) {
return xerrors.Errorf("opening partial file: %w", err)
}
f, err := pf.Reader(offset.Padded(), size.Padded())
2020-05-18 22:08:11 +00:00
if err != nil {
pf.Close()
return xerrors.Errorf("getting partial file reader: %w", err)
}
upr, err := fr32.NewUnpadReader(f, size.Padded())
if err != nil {
return xerrors.Errorf("creating unpadded reader: %w", err)
}
if _, err := io.CopyN(writer, upr, int64(size)); err != nil {
2020-05-18 22:08:11 +00:00
pf.Close()
return xerrors.Errorf("reading unsealed file: %w", err)
}
if err := pf.Close(); err != nil {
return xerrors.Errorf("closing partial file: %w", err)
}
return nil
}
2020-03-26 19:34:38 +00:00
func (sb *Sealer) SealPreCommit1(ctx context.Context, sector abi.SectorID, ticket abi.SealRandomness, pieces []abi.PieceInfo) (out storage.PreCommit1Out, err error) {
2020-03-26 02:50:56 +00:00
paths, done, err := sb.sectors.AcquireSector(ctx, sector, stores.FTUnsealed, stores.FTSealed|stores.FTCache, true)
if err != nil {
return nil, xerrors.Errorf("acquiring sector paths: %w", err)
}
defer done()
e, err := os.OpenFile(paths.Sealed, os.O_RDWR|os.O_CREATE, 0644)
if err != nil {
return nil, xerrors.Errorf("ensuring sealed file exists: %w", err)
}
if err := e.Close(); err != nil {
return nil, err
}
if err := os.Mkdir(paths.Cache, 0755); err != nil {
if os.IsExist(err) {
log.Warnf("existing cache in %s; removing", paths.Cache)
if err := os.RemoveAll(paths.Cache); err != nil {
return nil, xerrors.Errorf("remove existing sector cache from %s (sector %d): %w", paths.Cache, sector, err)
}
if err := os.Mkdir(paths.Cache, 0755); err != nil {
return nil, xerrors.Errorf("mkdir cache path after cleanup: %w", err)
}
} else {
return nil, err
}
}
var sum abi.UnpaddedPieceSize
for _, piece := range pieces {
sum += piece.Size.Unpadded()
}
ussize := abi.PaddedPieceSize(sb.ssize).Unpadded()
if sum != ussize {
return nil, xerrors.Errorf("aggregated piece sizes don't match sector size: %d != %d (%d)", sum, ussize, int64(ussize-sum))
}
// TODO: context cancellation respect
p1o, err := ffi.SealPreCommitPhase1(
sb.sealProofType,
paths.Cache,
paths.Unsealed,
2020-03-26 02:50:56 +00:00
paths.Sealed,
sector.Number,
sector.Miner,
ticket,
pieces,
)
if err != nil {
return nil, xerrors.Errorf("presealing sector %d (%s): %w", sector.Number, paths.Unsealed, err)
}
return p1o, nil
}
2020-03-26 19:34:38 +00:00
func (sb *Sealer) SealPreCommit2(ctx context.Context, sector abi.SectorID, phase1Out storage.PreCommit1Out) (storage.SectorCids, error) {
2020-03-26 02:50:56 +00:00
paths, done, err := sb.sectors.AcquireSector(ctx, sector, stores.FTSealed|stores.FTCache, 0, true)
if err != nil {
return storage.SectorCids{}, xerrors.Errorf("acquiring sector paths: %w", err)
}
defer done()
sealedCID, unsealedCID, err := ffi.SealPreCommitPhase2(phase1Out, paths.Cache, paths.Sealed)
if err != nil {
return storage.SectorCids{}, xerrors.Errorf("presealing sector %d (%s): %w", sector.Number, paths.Unsealed, err)
}
return storage.SectorCids{
Unsealed: unsealedCID,
Sealed: sealedCID,
}, nil
}
2020-03-26 19:34:38 +00:00
func (sb *Sealer) SealCommit1(ctx context.Context, sector abi.SectorID, ticket abi.SealRandomness, seed abi.InteractiveSealRandomness, pieces []abi.PieceInfo, cids storage.SectorCids) (storage.Commit1Out, error) {
2020-03-26 02:50:56 +00:00
paths, done, err := sb.sectors.AcquireSector(ctx, sector, stores.FTSealed|stores.FTCache, 0, true)
if err != nil {
return nil, xerrors.Errorf("acquire sector paths: %w", err)
}
defer done()
output, err := ffi.SealCommitPhase1(
sb.sealProofType,
cids.Sealed,
cids.Unsealed,
paths.Cache,
paths.Sealed,
sector.Number,
sector.Miner,
ticket,
seed,
pieces,
)
if err != nil {
log.Warn("StandaloneSealCommit error: ", err)
log.Warnf("num:%d tkt:%v seed:%v, pi:%v sealedCID:%v, unsealedCID:%v", sector.Number, ticket, seed, pieces, cids.Sealed, cids.Unsealed)
return nil, xerrors.Errorf("StandaloneSealCommit: %w", err)
}
return output, nil
}
2020-03-26 19:34:38 +00:00
func (sb *Sealer) SealCommit2(ctx context.Context, sector abi.SectorID, phase1Out storage.Commit1Out) (storage.Proof, error) {
2020-03-26 02:50:56 +00:00
return ffi.SealCommitPhase2(phase1Out, sector.Number, sector.Miner)
}
2020-03-26 19:34:38 +00:00
func (sb *Sealer) FinalizeSector(ctx context.Context, sector abi.SectorID) error {
2020-03-26 02:50:56 +00:00
paths, done, err := sb.sectors.AcquireSector(ctx, sector, stores.FTCache, 0, false)
if err != nil {
return xerrors.Errorf("acquiring sector cache path: %w", err)
}
defer done()
2020-04-10 18:41:59 +00:00
return ffi.ClearCache(uint64(sb.ssize), paths.Cache)
2020-03-26 02:50:56 +00:00
}
func GeneratePieceCIDFromFile(proofType abi.RegisteredProof, piece io.Reader, pieceSize abi.UnpaddedPieceSize) (cid.Cid, error) {
2020-05-28 17:15:15 +00:00
f, werr, err := ToReadableFile(piece, int64(pieceSize))
2020-03-26 02:50:56 +00:00
if err != nil {
return cid.Undef, err
}
pieceCID, err := ffi.GeneratePieceCIDFromFile(proofType, f, pieceSize)
if err != nil {
return cid.Undef, err
}
return pieceCID, werr()
}
func GenerateUnsealedCID(proofType abi.RegisteredProof, pieces []abi.PieceInfo) (cid.Cid, error) {
var sum abi.PaddedPieceSize
for _, p := range pieces {
sum += p.Size
}
2020-04-10 21:01:35 +00:00
ssize, err := proofType.SectorSize()
2020-03-26 02:50:56 +00:00
if err != nil {
return cid.Undef, err
}
{
// pad remaining space with 0 CommPs
toFill := uint64(abi.PaddedPieceSize(ssize) - sum)
n := bits.OnesCount64(toFill)
for i := 0; i < n; i++ {
next := bits.TrailingZeros64(toFill)
psize := uint64(1) << uint(next)
toFill ^= psize
unpadded := abi.PaddedPieceSize(psize).Unpadded()
pieces = append(pieces, abi.PieceInfo{
Size: unpadded.Padded(),
PieceCID: zerocomm.ZeroPieceCommitment(unpadded),
})
}
}
return ffi.GenerateUnsealedCID(proofType, pieces)
}