cerc-io/lotus
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

workers: Make LocalWorker public

Browse Source
This commit is contained in:
Łukasz Magiera 2020-03-11 22:23:16 +01:00
parent 71afcb0333
commit 56968d858c
4 changed files with 22 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
cmd/lotus-seal-worker/rpc.go
View File

@ -1,36 +1,13 @@
package main package main
import ( import (
"context"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-storage/storage" "github.com/filecoin-project/specs-storage/storage"
"github.com/filecoin-project/go-sectorbuilder"
) )
type worker struct { type worker struct { // TODO: use advmgr.LocalWorker here
spt abi.RegisteredProof sectorbuilder.Basic
}
func (w *worker) SealPreCommit1(ctx context.Context, sectorNum abi.SectorNumber, ticket abi.SealRandomness, pieces []abi.PieceInfo) (storage.PreCommit1Out, error) {
panic("implement me")
}
func (w *worker) SealPreCommit2(context.Context, abi.SectorNumber, storage.PreCommit1Out) (sealedCID cid.Cid, unsealedCID cid.Cid, err error) {
panic("implement me")
}
func (w *worker) SealCommit1(ctx context.Context, sectorNum abi.SectorNumber, ticket abi.SealRandomness, seed abi.InteractiveSealRandomness, pieces []abi.PieceInfo, sealedCID cid.Cid, unsealedCID cid.Cid) (storage.Commit1Out, error) {
panic("implement me")
}
func (w *worker) SealCommit2(context.Context, abi.SectorNumber, storage.Commit1Out) (storage.Proof, error) {
panic("implement me")
}
func (w *worker) FinalizeSector(context.Context, abi.SectorNumber) error {
panic("implement me")
} }
var _ storage.Sealer = &worker{} var _ storage.Sealer = &worker{}

7
cmd/lotus-seal-worker/storage.go
View File

@ -6,16 +6,18 @@ import (
"sort" "sort"
"github.com/filecoin-project/specs-actors/actors/abi" "github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/go-sectorbuilder" "github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/lotus/storage/sealmgr/stores"
"github.com/filecoin-project/lotus/api" "github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/storage/sealmgr/sectorutil" "github.com/filecoin-project/lotus/storage/sealmgr/sectorutil"
) )
type workerStorage struct { type workerStorage struct {
path string // TODO: multi-path support path string // TODO: multi-path support
mid abi.ActorID mid abi.ActorID // ewwhh TODO: passthru in sectobuilder/ffi
local *stores.Local
auth http.Header auth http.Header
api api.StorageMiner api api.StorageMiner
} }
@ -28,6 +30,7 @@ func (w *workerStorage) AcquireSector(ctx context.Context, id abi.SectorNumber,
// extract local storage; prefer // extract local storage; prefer
si, err := w.api.WorkerFindSector(ctx, asid, existing) si, err := w.api.WorkerFindSector(ctx, asid, existing)
if err != nil { if err != nil {
return sectorbuilder.SectorPaths{}, nil, err return sectorbuilder.SectorPaths{}, nil, err

24
storage/sealmgr/advmgr/local.go
View File

@ -18,13 +18,13 @@ import (
"github.com/filecoin-project/lotus/storage/sealmgr" "github.com/filecoin-project/lotus/storage/sealmgr"
) )
type localWorker struct { type LocalWorker struct {
scfg *sectorbuilder.Config scfg *sectorbuilder.Config
storage *stores.Local storage *stores.Local
} }
type localWorkerPathProvider struct { type localWorkerPathProvider struct {
w *localWorker w *LocalWorker
} }
func (l *localWorkerPathProvider) AcquireSector(ctx context.Context, id abi.SectorNumber, existing sectorbuilder.SectorFileType, allocate sectorbuilder.SectorFileType, sealing bool) (sectorbuilder.SectorPaths, func(), error) { func (l *localWorkerPathProvider) AcquireSector(ctx context.Context, id abi.SectorNumber, existing sectorbuilder.SectorFileType, allocate sectorbuilder.SectorFileType, sealing bool) (sectorbuilder.SectorPaths, func(), error) {
@ -39,11 +39,11 @@ func (l *localWorkerPathProvider) AcquireSector(ctx context.Context, id abi.Sect
}, existing, allocate, sealing) }, existing, allocate, sealing)
} }
func (l *localWorker) sb() (sectorbuilder.Basic, error) { func (l *LocalWorker) sb() (sectorbuilder.Basic, error) {
return sectorbuilder.New(&localWorkerPathProvider{w: l}, l.scfg) return sectorbuilder.New(&localWorkerPathProvider{w: l}, l.scfg)
} }
func (l *localWorker) AddPiece(ctx context.Context, sn abi.SectorNumber, epcs []abi.UnpaddedPieceSize, sz abi.UnpaddedPieceSize, r io.Reader) (abi.PieceInfo, error) { func (l *LocalWorker) AddPiece(ctx context.Context, sn abi.SectorNumber, epcs []abi.UnpaddedPieceSize, sz abi.UnpaddedPieceSize, r io.Reader) (abi.PieceInfo, error) {
sb, err := l.sb() sb, err := l.sb()
if err != nil { if err != nil {
return abi.PieceInfo{}, err return abi.PieceInfo{}, err
@ -52,7 +52,7 @@ func (l *localWorker) AddPiece(ctx context.Context, sn abi.SectorNumber, epcs []
return sb.AddPiece(ctx, sn, epcs, sz, r) return sb.AddPiece(ctx, sn, epcs, sz, r)
} }
func (l *localWorker) SealPreCommit1(ctx context.Context, sectorNum abi.SectorNumber, ticket abi.SealRandomness, pieces []abi.PieceInfo) (out storage2.PreCommit1Out, err error) { func (l *LocalWorker) SealPreCommit1(ctx context.Context, sectorNum abi.SectorNumber, ticket abi.SealRandomness, pieces []abi.PieceInfo) (out storage2.PreCommit1Out, err error) {
sb, err := l.sb() sb, err := l.sb()
if err != nil { if err != nil {
return nil, err return nil, err
@ -61,7 +61,7 @@ func (l *localWorker) SealPreCommit1(ctx context.Context, sectorNum abi.SectorNu
return sb.SealPreCommit1(ctx, sectorNum, ticket, pieces) return sb.SealPreCommit1(ctx, sectorNum, ticket, pieces)
} }
func (l *localWorker) SealPreCommit2(ctx context.Context, sectorNum abi.SectorNumber, phase1Out storage2.PreCommit1Out) (sealedCID cid.Cid, unsealedCID cid.Cid, err error) { func (l *LocalWorker) SealPreCommit2(ctx context.Context, sectorNum abi.SectorNumber, phase1Out storage2.PreCommit1Out) (sealedCID cid.Cid, unsealedCID cid.Cid, err error) {
sb, err := l.sb() sb, err := l.sb()
if err != nil { if err != nil {
return cid.Undef, cid.Undef, err return cid.Undef, cid.Undef, err
@ -70,7 +70,7 @@ func (l *localWorker) SealPreCommit2(ctx context.Context, sectorNum abi.SectorNu
return sb.SealPreCommit2(ctx, sectorNum, phase1Out) return sb.SealPreCommit2(ctx, sectorNum, phase1Out)
} }
func (l *localWorker) SealCommit1(ctx context.Context, sectorNum abi.SectorNumber, ticket abi.SealRandomness, seed abi.InteractiveSealRandomness, pieces []abi.PieceInfo, sealedCID cid.Cid, unsealedCID cid.Cid) (output storage2.Commit1Out, err error) { func (l *LocalWorker) SealCommit1(ctx context.Context, sectorNum abi.SectorNumber, ticket abi.SealRandomness, seed abi.InteractiveSealRandomness, pieces []abi.PieceInfo, sealedCID cid.Cid, unsealedCID cid.Cid) (output storage2.Commit1Out, err error) {
sb, err := l.sb() sb, err := l.sb()
if err != nil { if err != nil {
return nil, err return nil, err
@ -79,7 +79,7 @@ func (l *localWorker) SealCommit1(ctx context.Context, sectorNum abi.SectorNumbe
return sb.SealCommit1(ctx, sectorNum, ticket, seed, pieces, sealedCID, unsealedCID) return sb.SealCommit1(ctx, sectorNum, ticket, seed, pieces, sealedCID, unsealedCID)
} }
func (l *localWorker) SealCommit2(ctx context.Context, sectorNum abi.SectorNumber, phase1Out storage2.Commit1Out) (proof storage2.Proof, err error) { func (l *LocalWorker) SealCommit2(ctx context.Context, sectorNum abi.SectorNumber, phase1Out storage2.Commit1Out) (proof storage2.Proof, err error) {
sb, err := l.sb() sb, err := l.sb()
if err != nil { if err != nil {
return nil, err return nil, err
@ -88,7 +88,7 @@ func (l *localWorker) SealCommit2(ctx context.Context, sectorNum abi.SectorNumbe
return sb.SealCommit2(ctx, sectorNum, phase1Out) return sb.SealCommit2(ctx, sectorNum, phase1Out)
} }
func (l *localWorker) FinalizeSector(ctx context.Context, sectorNum abi.SectorNumber) error { func (l *LocalWorker) FinalizeSector(ctx context.Context, sectorNum abi.SectorNumber) error {
sb, err := l.sb() sb, err := l.sb()
if err != nil { if err != nil {
return err return err
@ -97,7 +97,7 @@ func (l *localWorker) FinalizeSector(ctx context.Context, sectorNum abi.SectorNu
return sb.FinalizeSector(ctx, sectorNum) return sb.FinalizeSector(ctx, sectorNum)
} }
func (l *localWorker) TaskTypes(context.Context) (map[sealmgr.TaskType]struct{}, error) { func (l *LocalWorker) TaskTypes(context.Context) (map[sealmgr.TaskType]struct{}, error) {
return map[sealmgr.TaskType]struct{}{ return map[sealmgr.TaskType]struct{}{
sealmgr.TTAddPiece: {}, sealmgr.TTAddPiece: {},
sealmgr.TTPreCommit1: {}, sealmgr.TTPreCommit1: {},
@ -106,8 +106,8 @@ func (l *localWorker) TaskTypes(context.Context) (map[sealmgr.TaskType]struct{},
}, nil }, nil
} }
func (l *localWorker) Paths(context.Context) ([]api.StoragePath, error) { func (l *LocalWorker) Paths(context.Context) ([]api.StoragePath, error) {
return l.storage.Local(), nil return l.storage.Local(), nil
} }
var _ Worker = &localWorker{} var _ Worker = &LocalWorker{}

2
storage/sealmgr/advmgr/manager.go
View File

@ -66,7 +66,7 @@ func New(ls stores.LocalStorage, cfg *sectorbuilder.Config, sc SectorIDCounter)
m := &Manager{ m := &Manager{
workers: []Worker{ workers: []Worker{
&localWorker{scfg: cfg, storage: stor}, &LocalWorker{scfg: cfg, storage: stor},
}, },
scfg: cfg, scfg: cfg,
sc: sc, sc: sc,