lotus/extern/sector-storage/manager.go

508 lines
15 KiB
Go
Raw Normal View History

2020-03-23 11:40:02 +00:00
package sectorstorage
import (
"context"
2020-03-24 18:00:08 +00:00
"errors"
2020-03-23 11:40:02 +00:00
"io"
"net/http"
"github.com/hashicorp/go-multierror"
2020-03-23 11:40:02 +00:00
"github.com/ipfs/go-cid"
logging "github.com/ipfs/go-log/v2"
"github.com/mitchellh/go-homedir"
"golang.org/x/xerrors"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-storage/storage"
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
"github.com/filecoin-project/lotus/extern/sector-storage/fsutil"
"github.com/filecoin-project/lotus/extern/sector-storage/sealtasks"
"github.com/filecoin-project/lotus/extern/sector-storage/stores"
"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
2020-03-23 11:40:02 +00:00
)
var log = logging.Logger("advmgr")
2020-03-24 18:00:08 +00:00
var ErrNoWorkers = errors.New("no suitable workers found")
2020-03-23 11:40:02 +00:00
type URLs []string
type Worker interface {
2020-03-26 19:34:38 +00:00
ffiwrapper.StorageSealer
2020-05-14 01:01:38 +00:00
MoveStorage(ctx context.Context, sector abi.SectorID, types stores.SectorFileType) error
2020-06-03 21:44:59 +00:00
2020-06-04 21:30:20 +00:00
Fetch(ctx context.Context, s abi.SectorID, ft stores.SectorFileType, ptype stores.PathType, am stores.AcquireMode) error
UnsealPiece(context.Context, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) error
ReadPiece(context.Context, io.Writer, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize) (bool, error)
2020-03-23 11:40:02 +00:00
TaskTypes(context.Context) (map[sealtasks.TaskType]struct{}, error)
// Returns paths accessible to the worker
Paths(context.Context) ([]stores.StoragePath, error)
Info(context.Context) (storiface.WorkerInfo, error)
2020-03-24 23:49:45 +00:00
2020-05-01 18:00:17 +00:00
// returns channel signalling worker shutdown
Closing(context.Context) (<-chan struct{}, error)
2020-03-24 23:49:45 +00:00
Close() error
2020-03-23 11:40:02 +00:00
}
type SectorManager interface {
SectorSize() abi.SectorSize
ReadPiece(context.Context, io.Writer, abi.SectorID, storiface.UnpaddedByteIndex, abi.UnpaddedPieceSize, abi.SealRandomness, cid.Cid) error
2020-03-23 11:40:02 +00:00
2020-03-26 19:34:38 +00:00
ffiwrapper.StorageSealer
2020-03-23 11:40:02 +00:00
storage.Prover
2020-05-16 21:03:29 +00:00
FaultTracker
2020-03-23 11:40:02 +00:00
}
2020-03-23 14:56:22 +00:00
type WorkerID uint64
2020-03-23 11:40:02 +00:00
type Manager struct {
2020-03-26 02:50:56 +00:00
scfg *ffiwrapper.Config
2020-03-23 11:40:02 +00:00
ls stores.LocalStorage
storage *stores.Remote
localStore *stores.Local
remoteHnd *stores.FetchHandler
index stores.SectorIndex
sched *scheduler
2020-03-23 11:40:02 +00:00
storage.Prover
2020-03-23 11:40:02 +00:00
}
type SealerConfig struct {
2020-07-24 14:43:41 +00:00
ParallelFetchLimit int
// Local worker config
2020-08-17 09:39:29 +00:00
AllowAddPiece bool
AllowPreCommit1 bool
AllowPreCommit2 bool
AllowCommit bool
2020-05-14 01:01:38 +00:00
AllowUnseal bool
}
type StorageAuth http.Header
func New(ctx context.Context, ls stores.LocalStorage, si stores.SectorIndex, cfg *ffiwrapper.Config, sc SealerConfig, urls URLs, sa StorageAuth) (*Manager, error) {
2020-03-23 11:40:02 +00:00
lstor, err := stores.NewLocal(ctx, ls, si, urls)
if err != nil {
return nil, err
}
2020-06-03 20:00:34 +00:00
prover, err := ffiwrapper.New(&readonlyProvider{stor: lstor, index: si}, cfg)
2020-03-23 11:40:02 +00:00
if err != nil {
return nil, xerrors.Errorf("creating prover instance: %w", err)
}
2020-07-24 14:43:41 +00:00
stor := stores.NewRemote(lstor, si, http.Header(sa), sc.ParallelFetchLimit)
2020-03-23 11:40:02 +00:00
m := &Manager{
scfg: cfg,
ls: ls,
storage: stor,
localStore: lstor,
remoteHnd: &stores.FetchHandler{Local: lstor},
index: si,
sched: newScheduler(cfg.SealProofType),
2020-03-23 11:40:02 +00:00
Prover: prover,
}
go m.sched.runSched()
2020-03-23 11:40:02 +00:00
localTasks := []sealtasks.TaskType{
2020-08-17 09:39:29 +00:00
sealtasks.TTCommit1, sealtasks.TTFinalize, sealtasks.TTFetch, sealtasks.TTReadUnsealed,
}
if sc.AllowAddPiece {
localTasks = append(localTasks, sealtasks.TTAddPiece)
}
2020-03-25 20:20:24 +00:00
if sc.AllowPreCommit1 {
localTasks = append(localTasks, sealtasks.TTPreCommit1)
}
if sc.AllowPreCommit2 {
localTasks = append(localTasks, sealtasks.TTPreCommit2)
}
if sc.AllowCommit {
localTasks = append(localTasks, sealtasks.TTCommit2)
}
2020-05-14 01:01:38 +00:00
if sc.AllowUnseal {
localTasks = append(localTasks, sealtasks.TTUnseal)
}
2020-03-23 11:40:02 +00:00
err = m.AddWorker(ctx, NewLocalWorker(WorkerConfig{
SealProof: cfg.SealProofType,
TaskTypes: localTasks,
2020-03-23 11:40:02 +00:00
}, stor, lstor, si))
if err != nil {
return nil, xerrors.Errorf("adding local worker: %w", err)
}
return m, nil
}
func (m *Manager) AddLocalStorage(ctx context.Context, path string) error {
path, err := homedir.Expand(path)
if err != nil {
return xerrors.Errorf("expanding local path: %w", err)
}
if err := m.localStore.OpenPath(ctx, path); err != nil {
return xerrors.Errorf("opening local path: %w", err)
}
if err := m.ls.SetStorage(func(sc *stores.StorageConfig) {
sc.StoragePaths = append(sc.StoragePaths, stores.LocalPath{Path: path})
2020-03-23 11:40:02 +00:00
}); err != nil {
return xerrors.Errorf("get storage config: %w", err)
}
return nil
}
func (m *Manager) AddWorker(ctx context.Context, w Worker) error {
info, err := w.Info(ctx)
if err != nil {
return xerrors.Errorf("getting worker info: %w", err)
}
m.sched.newWorkers <- &workerHandle{
2020-07-21 18:01:25 +00:00
w: w,
wt: &workTracker{
running: map[uint64]storiface.WorkerJob{},
},
info: info,
preparing: &activeResources{},
active: &activeResources{},
2020-03-23 11:40:02 +00:00
}
return nil
}
func (m *Manager) ServeHTTP(w http.ResponseWriter, r *http.Request) {
m.remoteHnd.ServeHTTP(w, r)
}
func (m *Manager) SectorSize() abi.SectorSize {
sz, _ := m.scfg.SealProofType.SectorSize()
return sz
}
func schedNop(context.Context, Worker) error {
return nil
}
2020-06-04 21:30:20 +00:00
func schedFetch(sector abi.SectorID, ft stores.SectorFileType, ptype stores.PathType, am stores.AcquireMode) func(context.Context, Worker) error {
return func(ctx context.Context, worker Worker) error {
2020-06-04 21:30:20 +00:00
return worker.Fetch(ctx, sector, ft, ptype, am)
}
2020-03-23 11:40:02 +00:00
}
func (m *Manager) ReadPiece(ctx context.Context, sink io.Writer, sector abi.SectorID, offset storiface.UnpaddedByteIndex, size abi.UnpaddedPieceSize, ticket abi.SealRandomness, unsealed cid.Cid) error {
2020-06-03 20:00:34 +00:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
if err := m.index.StorageLock(ctx, sector, stores.FTSealed|stores.FTCache, stores.FTUnsealed); err != nil {
return xerrors.Errorf("acquiring sector lock: %w", err)
}
2020-08-11 07:27:03 +00:00
// passing 0 spt because we only need it when allowFetch is true
best, err := m.index.StorageFindSector(ctx, sector, stores.FTUnsealed, 0, false)
2020-05-14 01:01:38 +00:00
if err != nil {
return xerrors.Errorf("read piece: checking for already existing unsealed sector: %w", err)
}
var selector WorkerSelector
if len(best) == 0 { // new
2020-08-05 12:36:49 +00:00
selector = newAllocSelector(m.index, stores.FTUnsealed, stores.PathSealing)
2020-05-14 01:01:38 +00:00
} else { // append to existing
selector = newExistingSelector(m.index, sector, stores.FTUnsealed, false)
}
var readOk bool
if len(best) > 0 {
// There is unsealed sector, see if we can read from it
2020-08-05 12:36:49 +00:00
selector = newExistingSelector(m.index, sector, stores.FTUnsealed, false)
err = m.sched.Schedule(ctx, sector, sealtasks.TTReadUnsealed, selector, schedFetch(sector, stores.FTUnsealed, stores.PathSealing, stores.AcquireMove), func(ctx context.Context, w Worker) error {
readOk, err = w.ReadPiece(ctx, sink, sector, offset, size)
return err
})
if err != nil {
return xerrors.Errorf("reading piece from sealed sector: %w", err)
}
if readOk {
return nil
}
}
2020-05-14 01:01:38 +00:00
unsealFetch := func(ctx context.Context, worker Worker) error {
2020-07-06 14:13:42 +00:00
if err := worker.Fetch(ctx, sector, stores.FTSealed|stores.FTCache, stores.PathSealing, stores.AcquireCopy); err != nil {
return xerrors.Errorf("copy sealed/cache sector data: %w", err)
}
2020-05-29 16:57:44 +00:00
if len(best) > 0 {
2020-07-06 14:13:42 +00:00
if err := worker.Fetch(ctx, sector, stores.FTUnsealed, stores.PathSealing, stores.AcquireMove); err != nil {
2020-05-29 16:57:44 +00:00
return xerrors.Errorf("copy unsealed sector data: %w", err)
}
}
return nil
}
2020-05-14 01:01:38 +00:00
err = m.sched.Schedule(ctx, sector, sealtasks.TTUnseal, selector, unsealFetch, func(ctx context.Context, w Worker) error {
2020-05-14 01:01:38 +00:00
return w.UnsealPiece(ctx, sector, offset, size, ticket, unsealed)
})
if err != nil {
return err
}
2020-08-05 12:36:49 +00:00
selector = newExistingSelector(m.index, sector, stores.FTUnsealed, false)
2020-05-14 01:01:38 +00:00
2020-06-03 21:44:59 +00:00
err = m.sched.Schedule(ctx, sector, sealtasks.TTReadUnsealed, selector, schedFetch(sector, stores.FTUnsealed, stores.PathSealing, stores.AcquireMove), func(ctx context.Context, w Worker) error {
readOk, err = w.ReadPiece(ctx, sink, sector, offset, size)
return err
2020-05-14 01:01:38 +00:00
})
if err != nil {
return xerrors.Errorf("reading piece from sealed sector: %w", err)
}
if readOk {
return xerrors.Errorf("failed to read unsealed piece")
}
2020-05-14 01:01:38 +00:00
return nil
}
2020-03-23 11:40:02 +00:00
func (m *Manager) NewSector(ctx context.Context, sector abi.SectorID) error {
log.Warnf("stub NewSector")
return nil
}
func (m *Manager) AddPiece(ctx context.Context, sector abi.SectorID, existingPieces []abi.UnpaddedPieceSize, sz abi.UnpaddedPieceSize, r io.Reader) (abi.PieceInfo, error) {
2020-06-03 20:00:34 +00:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
if err := m.index.StorageLock(ctx, sector, stores.FTNone, stores.FTUnsealed); err != nil {
return abi.PieceInfo{}, xerrors.Errorf("acquiring sector lock: %w", err)
}
var selector WorkerSelector
2020-03-23 11:40:02 +00:00
var err error
if len(existingPieces) == 0 { // new
2020-08-05 12:36:49 +00:00
selector = newAllocSelector(m.index, stores.FTUnsealed, stores.PathSealing)
2020-05-14 01:01:38 +00:00
} else { // use existing
2020-08-05 12:36:49 +00:00
selector = newExistingSelector(m.index, sector, stores.FTUnsealed, false)
2020-03-23 11:40:02 +00:00
}
var out abi.PieceInfo
err = m.sched.Schedule(ctx, sector, sealtasks.TTAddPiece, selector, schedNop, func(ctx context.Context, w Worker) error {
p, err := w.AddPiece(ctx, sector, existingPieces, sz, r)
if err != nil {
return err
}
out = p
return nil
})
2020-03-23 11:40:02 +00:00
return out, err
2020-03-23 11:40:02 +00:00
}
func (m *Manager) SealPreCommit1(ctx context.Context, sector abi.SectorID, ticket abi.SealRandomness, pieces []abi.PieceInfo) (out storage.PreCommit1Out, err error) {
2020-06-03 20:00:34 +00:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
if err := m.index.StorageLock(ctx, sector, stores.FTUnsealed, stores.FTSealed|stores.FTCache); err != nil {
return nil, xerrors.Errorf("acquiring sector lock: %w", err)
}
2020-03-23 11:40:02 +00:00
// TODO: also consider where the unsealed data sits
2020-08-05 12:36:49 +00:00
selector := newAllocSelector(m.index, stores.FTCache|stores.FTSealed, stores.PathSealing)
2020-03-23 11:40:02 +00:00
2020-06-03 21:44:59 +00:00
err = m.sched.Schedule(ctx, sector, sealtasks.TTPreCommit1, selector, schedFetch(sector, stores.FTUnsealed, stores.PathSealing, stores.AcquireMove), func(ctx context.Context, w Worker) error {
p, err := w.SealPreCommit1(ctx, sector, ticket, pieces)
if err != nil {
return err
}
out = p
return nil
})
2020-03-23 11:40:02 +00:00
return out, err
2020-03-23 11:40:02 +00:00
}
func (m *Manager) SealPreCommit2(ctx context.Context, sector abi.SectorID, phase1Out storage.PreCommit1Out) (out storage.SectorCids, err error) {
2020-06-03 20:00:34 +00:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
if err := m.index.StorageLock(ctx, sector, stores.FTSealed, stores.FTCache); err != nil {
return storage.SectorCids{}, xerrors.Errorf("acquiring sector lock: %w", err)
}
2020-08-05 12:36:49 +00:00
selector := newExistingSelector(m.index, sector, stores.FTCache|stores.FTSealed, true)
2020-03-23 11:40:02 +00:00
2020-06-03 21:44:59 +00:00
err = m.sched.Schedule(ctx, sector, sealtasks.TTPreCommit2, selector, schedFetch(sector, stores.FTCache|stores.FTSealed, stores.PathSealing, stores.AcquireMove), func(ctx context.Context, w Worker) error {
p, err := w.SealPreCommit2(ctx, sector, phase1Out)
if err != nil {
return err
}
out = p
return nil
})
return out, err
2020-03-23 11:40:02 +00:00
}
func (m *Manager) SealCommit1(ctx context.Context, sector abi.SectorID, ticket abi.SealRandomness, seed abi.InteractiveSealRandomness, pieces []abi.PieceInfo, cids storage.SectorCids) (out storage.Commit1Out, err error) {
2020-06-03 20:00:34 +00:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
if err := m.index.StorageLock(ctx, sector, stores.FTSealed, stores.FTCache); err != nil {
return storage.Commit1Out{}, xerrors.Errorf("acquiring sector lock: %w", err)
}
// NOTE: We set allowFetch to false in so that we always execute on a worker
// with direct access to the data. We want to do that because this step is
// generally very cheap / fast, and transferring data is not worth the effort
2020-08-05 12:36:49 +00:00
selector := newExistingSelector(m.index, sector, stores.FTCache|stores.FTSealed, false)
2020-03-23 11:40:02 +00:00
2020-06-03 21:44:59 +00:00
err = m.sched.Schedule(ctx, sector, sealtasks.TTCommit1, selector, schedFetch(sector, stores.FTCache|stores.FTSealed, stores.PathSealing, stores.AcquireMove), func(ctx context.Context, w Worker) error {
p, err := w.SealCommit1(ctx, sector, ticket, seed, pieces, cids)
if err != nil {
return err
}
out = p
return nil
})
return out, err
2020-03-23 11:40:02 +00:00
}
func (m *Manager) SealCommit2(ctx context.Context, sector abi.SectorID, phase1Out storage.Commit1Out) (out storage.Proof, err error) {
selector := newTaskSelector()
2020-03-23 11:40:02 +00:00
err = m.sched.Schedule(ctx, sector, sealtasks.TTCommit2, selector, schedNop, func(ctx context.Context, w Worker) error {
p, err := w.SealCommit2(ctx, sector, phase1Out)
2020-03-23 11:40:02 +00:00
if err != nil {
return err
2020-03-23 11:40:02 +00:00
}
out = p
return nil
})
2020-03-23 11:40:02 +00:00
return out, err
2020-03-23 11:40:02 +00:00
}
func (m *Manager) FinalizeSector(ctx context.Context, sector abi.SectorID, keepUnsealed []storage.Range) error {
2020-06-03 20:00:34 +00:00
ctx, cancel := context.WithCancel(ctx)
defer cancel()
2020-06-03 21:44:59 +00:00
if err := m.index.StorageLock(ctx, sector, stores.FTNone, stores.FTSealed|stores.FTUnsealed|stores.FTCache); err != nil {
2020-06-03 20:00:34 +00:00
return xerrors.Errorf("acquiring sector lock: %w", err)
}
unsealed := stores.FTUnsealed
{
2020-08-11 07:27:03 +00:00
unsealedStores, err := m.index.StorageFindSector(ctx, sector, stores.FTUnsealed, 0, false)
if err != nil {
return xerrors.Errorf("finding unsealed sector: %w", err)
}
if len(unsealedStores) == 0 { // Is some edge-cases unsealed sector may not exist already, that's fine
unsealed = stores.FTNone
}
}
2020-08-05 12:36:49 +00:00
selector := newExistingSelector(m.index, sector, stores.FTCache|stores.FTSealed, false)
2020-03-23 11:40:02 +00:00
2020-08-05 12:36:49 +00:00
err := m.sched.Schedule(ctx, sector, sealtasks.TTFinalize, selector,
schedFetch(sector, stores.FTCache|stores.FTSealed|unsealed, stores.PathSealing, stores.AcquireMove),
func(ctx context.Context, w Worker) error {
return w.FinalizeSector(ctx, sector, keepUnsealed)
})
2020-06-03 21:44:59 +00:00
if err != nil {
return err
}
2020-08-05 12:36:49 +00:00
fetchSel := newAllocSelector(m.index, stores.FTCache|stores.FTSealed, stores.PathStorage)
moveUnsealed := unsealed
{
if len(keepUnsealed) == 0 {
moveUnsealed = stores.FTNone
}
}
2020-06-03 21:44:59 +00:00
err = m.sched.Schedule(ctx, sector, sealtasks.TTFetch, fetchSel,
schedFetch(sector, stores.FTCache|stores.FTSealed|moveUnsealed, stores.PathStorage, stores.AcquireMove),
2020-06-03 21:44:59 +00:00
func(ctx context.Context, w Worker) error {
return w.MoveStorage(ctx, sector, stores.FTCache|stores.FTSealed|moveUnsealed)
2020-06-03 21:44:59 +00:00
})
if err != nil {
return xerrors.Errorf("moving sector to storage: %w", err)
}
return nil
2020-03-23 11:40:02 +00:00
}
func (m *Manager) ReleaseUnsealed(ctx context.Context, sector abi.SectorID, safeToFree []storage.Range) error {
2020-07-03 20:23:36 +00:00
log.Warnw("ReleaseUnsealed todo")
return nil
}
func (m *Manager) Remove(ctx context.Context, sector abi.SectorID) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
if err := m.index.StorageLock(ctx, sector, stores.FTNone, stores.FTSealed|stores.FTUnsealed|stores.FTCache); err != nil {
return xerrors.Errorf("acquiring sector lock: %w", err)
}
var err error
if rerr := m.storage.Remove(ctx, sector, stores.FTSealed, true); rerr != nil {
err = multierror.Append(err, xerrors.Errorf("removing sector (sealed): %w", rerr))
}
if rerr := m.storage.Remove(ctx, sector, stores.FTCache, true); rerr != nil {
err = multierror.Append(err, xerrors.Errorf("removing sector (cache): %w", rerr))
}
if rerr := m.storage.Remove(ctx, sector, stores.FTUnsealed, true); rerr != nil {
err = multierror.Append(err, xerrors.Errorf("removing sector (unsealed): %w", rerr))
}
return err
}
2020-03-23 11:40:02 +00:00
func (m *Manager) StorageLocal(ctx context.Context) (map[stores.ID]string, error) {
l, err := m.localStore.Local(ctx)
if err != nil {
return nil, err
}
out := map[stores.ID]string{}
for _, st := range l {
out[st.ID] = st.LocalPath
}
return out, nil
}
2020-07-08 14:58:09 +00:00
func (m *Manager) FsStat(ctx context.Context, id stores.ID) (fsutil.FsStat, error) {
return m.storage.FsStat(ctx, id)
}
func (m *Manager) SchedDiag(ctx context.Context) (interface{}, error) {
return m.sched.Info(ctx)
}
2020-07-17 10:59:12 +00:00
func (m *Manager) Close(ctx context.Context) error {
return m.sched.Close(ctx)
2020-03-24 23:49:45 +00:00
}
2020-03-23 11:40:02 +00:00
var _ SectorManager = &Manager{}