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

mod tidy; gofmt

Browse Source
This commit is contained in:
Łukasz Magiera 2020-03-05 20:21:06 +01:00
parent fa5bc3a5a9
commit a5e5918fc5
14 changed files with 142 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api/api_storage.go
View File

@ -116,7 +116,7 @@ type StorageMiner interface {
WorkerQueue(context.Context, sectorbuilder.WorkerCfg) (<-chan sectorbuilder.WorkerTask, error)
WorkerDone(ctx context.Context, task uint64, res sectorbuilder.SealRes) error
*/
*/
MarketImportDealData(ctx context.Context, propcid cid.Cid, path string) error
MarketListDeals(ctx context.Context) ([]storagemarket.StorageDeal, error)
MarketListIncompleteDeals(ctx context.Context) ([]storagemarket.MinerDeal, error)

8
api/apistruct/struct.go
View File

@ -171,11 +171,11 @@ type StorageMinerStruct struct {
SectorsRefs func(context.Context) (map[string][]api.SealedRef, error) `perm:"read"`
SectorsUpdate func(context.Context, abi.SectorNumber, api.SectorState) error `perm:"write"`
/* WorkerStats func(context.Context) (sectorbuilder.WorkerStats, error) `perm:"read"`
/* WorkerStats func(context.Context) (sectorbuilder.WorkerStats, error) `perm:"read"`
WorkerQueue func(ctx context.Context, cfg sectorbuilder.WorkerCfg) (<-chan sectorbuilder.WorkerTask, error) `perm:"admin"` // TODO: worker perm
WorkerDone func(ctx context.Context, task uint64, res sectorbuilder.SealRes) error `perm:"admin"`
*/
WorkerQueue func(ctx context.Context, cfg sectorbuilder.WorkerCfg) (<-chan sectorbuilder.WorkerTask, error) `perm:"admin"` // TODO: worker perm
WorkerDone func(ctx context.Context, task uint64, res sectorbuilder.SealRes) error `perm:"admin"`
*/
SetPrice func(context.Context, types.BigInt) error `perm:"admin"`
DealsImportData func(ctx context.Context, dealPropCid cid.Cid, file string) error `perm:"write"`

25
cmd/lotus-storage-miner/init.go
View File

@ -186,19 +186,19 @@ var initCmd = &cli.Command{
}
if !cctx.Bool("no-local-storage") {
b, err := json.MarshalIndent(&config.StorageMeta{
ID: uuid.New().String(),
Weight: 10,
CanSeal: true,
CanStore: true,
}, "", " ")
if err != nil {
return xerrors.Errorf("marshaling storage config: %w", err)
}
b, err := json.MarshalIndent(&config.StorageMeta{
ID: uuid.New().String(),
Weight: 10,
CanSeal: true,
CanStore: true,
}, "", " ")
if err != nil {
return xerrors.Errorf("marshaling storage config: %w", err)
}
if err := ioutil.WriteFile(filepath.Join(lr.Path(), "sectorstore.json"), b, 0644); err != nil {
return xerrors.Errorf("persisting storage metadata (%s): %w", filepath.Join(lr.Path(), "storage.json"), err)
}
if err := ioutil.WriteFile(filepath.Join(lr.Path(), "sectorstore.json"), b, 0644); err != nil {
return xerrors.Errorf("persisting storage metadata (%s): %w", filepath.Join(lr.Path(), "storage.json"), err)
}
}
sc.StoragePaths = append(sc.StoragePaths, config.LocalPath{
@ -214,7 +214,6 @@ var initCmd = &cli.Command{
}
}
if err := storageMinerInit(ctx, cctx, api, r, ssize); err != nil {
log.Errorf("Failed to initialize lotus-storage-miner: %+v", err)
path, err := homedir.Expand(repoPath)

10
go.sum
View File

@ -119,8 +119,6 @@ github.com/filecoin-project/go-padreader v0.0.0-20200210211231-548257017ca6/go.m
github.com/filecoin-project/go-paramfetch v0.0.2-0.20200218225740-47c639bab663 h1:eYxi6vI5CyeXD15X1bB3bledDXbqKxqf0wQzTLgwYwA=
github.com/filecoin-project/go-paramfetch v0.0.2-0.20200218225740-47c639bab663/go.mod h1:fZzmf4tftbwf9S37XRifoJlz7nCjRdIrMGLR07dKLCc=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200226210935-4739f8749f56/go.mod h1:tzTc9BxxSbjlIzhFwm5h9oBkXKkRuLxeiWspntwnKyw=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200229022239-442fe78a3168 h1:6fBZezecouqHNRi4Q4YlH7QqgJ3yZZ7adMBpUBAx4F8=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200229022239-442fe78a3168/go.mod h1:gEQJVRVqQX8Vx02IosTdC2UA4l2MgHfG3POJEI2GIpc=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200304050010-2cfac00a93e7 h1:pGvvPZgIfBXAda+OGc943zFcM8Wt4GG9UVymVeyfx6c=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200304050010-2cfac00a93e7/go.mod h1:gEQJVRVqQX8Vx02IosTdC2UA4l2MgHfG3POJEI2GIpc=
github.com/filecoin-project/go-statemachine v0.0.0-20200226041606-2074af6d51d9 h1:k9qVR9ItcziSB2rxtlkN/MDWNlbsI6yzec+zjUatLW0=
@ -129,14 +127,6 @@ github.com/filecoin-project/go-statestore v0.1.0 h1:t56reH59843TwXHkMcwyuayStBIi
github.com/filecoin-project/go-statestore v0.1.0/go.mod h1:LFc9hD+fRxPqiHiaqUEZOinUJB4WARkRfNl10O7kTnI=
github.com/filecoin-project/specs-actors v0.0.0-20200210130641-2d1fbd8672cf/go.mod h1:xtDZUB6pe4Pksa/bAJbJ693OilaC5Wbot9jMhLm3cZA=
github.com/filecoin-project/specs-actors v0.0.0-20200226200336-94c9b92b2775/go.mod h1:0HAWYrvajFHDgRaKbF0rl+IybVLZL5z4gQ8koCMPhoU=
github.com/filecoin-project/specs-actors v0.0.0-20200229011003-1d726e3afd04 h1:O343OeQLkLWLj5ZqQ5nhevAGBTeB5LioiA53ddScqdY=
github.com/filecoin-project/specs-actors v0.0.0-20200229011003-1d726e3afd04/go.mod h1:0HAWYrvajFHDgRaKbF0rl+IybVLZL5z4gQ8koCMPhoU=
github.com/filecoin-project/specs-actors v0.0.0-20200302223606-0eaf97b10aaf h1:3IojVqJAD5IXMxvZ+WYx+LRbfSB/rOXpYBuHh6o3XkY=
github.com/filecoin-project/specs-actors v0.0.0-20200302223606-0eaf97b10aaf h1:3IojVqJAD5IXMxvZ+WYx+LRbfSB/rOXpYBuHh6o3XkY=
github.com/filecoin-project/specs-actors v0.0.0-20200302223606-0eaf97b10aaf/go.mod h1:0HAWYrvajFHDgRaKbF0rl+IybVLZL5z4gQ8koCMPhoU=
github.com/filecoin-project/specs-actors v0.0.0-20200302223606-0eaf97b10aaf/go.mod h1:0HAWYrvajFHDgRaKbF0rl+IybVLZL5z4gQ8koCMPhoU=
github.com/filecoin-project/specs-storage v0.0.0-20200303230804-16c9a38030eb h1:oHB9hKaD7g75NFulnfh+SCYS5bSl8hB6Eanf8A6l5tw=
github.com/filecoin-project/specs-storage v0.0.0-20200303230804-16c9a38030eb/go.mod h1:sC2Ck2l1G8hXI5Do/3sp0yxbMRMnukbFwP9KF1CRFLw=
github.com/filecoin-project/specs-actors v0.0.0-20200304210626-21ee86aadcb9 h1:5/XkV9N7Zlidi2RYY/04BToD/XeQrudUseI7Gx6owl8=
github.com/filecoin-project/specs-actors v0.0.0-20200304210626-21ee86aadcb9/go.mod h1:0HAWYrvajFHDgRaKbF0rl+IybVLZL5z4gQ8koCMPhoU=
github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=

1
node/config/def.go
View File

@ -53,7 +53,6 @@ type Metrics struct {
// // Storage Miner
type Storage struct {
}
func defCommon() Common {

3
node/config/storage.go
View File

@ -20,7 +20,7 @@ type StorageConfig struct {
// [path]/metadata.json
type StorageMeta struct {
ID string
ID string
Weight uint64 // 0 = readonly
CanSeal bool
@ -65,4 +65,3 @@ func WriteStorageFile(path string, config StorageConfig) error {
return nil
}

172
node/impl/storminer.go
View File

@ -29,7 +29,7 @@ type StorageMinerAPI struct {
SectorBuilderConfig *sectorbuilder.Config
//SectorBuilder sectorbuilder.Interface
SectorBlocks *sectorblocks.SectorBlocks
SectorBlocks *sectorblocks.SectorBlocks
StorageProvider storagemarket.StorageProvider
Miner *storage.Miner
@ -56,111 +56,112 @@ func (sm *StorageMinerAPI) ServeRemote(w http.ResponseWriter, r *http.Request) {
func (sm *StorageMinerAPI) remoteGetSector(w http.ResponseWriter, r *http.Request) {
panic("todo")
/* vars := mux.Vars(r)
/* vars := mux.Vars(r)
id, err := strconv.ParseUint(vars["id"], 10, 64)
if err != nil {
log.Error("parsing sector id: ", err)
w.WriteHeader(500)
return
}
path, err := sm.SectorBuilder.SectorPath(fs.DataType(vars["type"]), abi.SectorNumber(id))
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
stat, err := os.Stat(string(path))
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
var rd io.Reader
if stat.IsDir() {
rd, err = tarutil.TarDirectory(string(path))
w.Header().Set("Content-Type", "application/x-tar")
} else {
rd, err = os.OpenFile(string(path), os.O_RDONLY, 0644)
w.Header().Set("Content-Type", "application/octet-stream")
}
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
w.WriteHeader(200)
if _, err := io.Copy(w, rd); err != nil {
log.Error(err)
return
}*/
}
func (sm *StorageMinerAPI) remotePutSector(w http.ResponseWriter, r *http.Request) {
panic("todo")
/* vars := mux.Vars(r)
id, err := strconv.ParseUint(vars["id"], 10, 64)
if err != nil {
log.Error("parsing sector id: ", err)
w.WriteHeader(500)
return
}
// This is going to get better with worker-to-worker transfers
path, err := sm.SectorBuilder.SectorPath(fs.DataType(vars["type"]), abi.SectorNumber(id))
if err != nil {
if err != fs.ErrNotFound {
log.Error(err)
id, err := strconv.ParseUint(vars["id"], 10, 64)
if err != nil {
log.Error("parsing sector id: ", err)
w.WriteHeader(500)
return
}
path, err = sm.SectorBuilder.AllocSectorPath(fs.DataType(vars["type"]), abi.SectorNumber(id), true)
path, err := sm.SectorBuilder.SectorPath(fs.DataType(vars["type"]), abi.SectorNumber(id))
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
}
mediatype, _, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
if err := os.RemoveAll(string(path)); err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
switch mediatype {
case "application/x-tar":
if err := tarutil.ExtractTar(r.Body, string(path)); err != nil {
stat, err := os.Stat(string(path))
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
default:
if err := files.WriteTo(files.NewReaderFile(r.Body), string(path)); err != nil {
var rd io.Reader
if stat.IsDir() {
rd, err = tarutil.TarDirectory(string(path))
w.Header().Set("Content-Type", "application/x-tar")
} else {
rd, err = os.OpenFile(string(path), os.O_RDONLY, 0644)
w.Header().Set("Content-Type", "application/octet-stream")
}
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
}
w.WriteHeader(200)
log.Infof("received %s sector (%s): %d bytes", vars["type"], vars["sname"], r.ContentLength)*/
w.WriteHeader(200)
if _, err := io.Copy(w, rd); err != nil {
log.Error(err)
return
}*/
}
func (sm *StorageMinerAPI) remotePutSector(w http.ResponseWriter, r *http.Request) {
panic("todo")
/* vars := mux.Vars(r)
id, err := strconv.ParseUint(vars["id"], 10, 64)
if err != nil {
log.Error("parsing sector id: ", err)
w.WriteHeader(500)
return
}
// This is going to get better with worker-to-worker transfers
path, err := sm.SectorBuilder.SectorPath(fs.DataType(vars["type"]), abi.SectorNumber(id))
if err != nil {
if err != fs.ErrNotFound {
log.Error(err)
w.WriteHeader(500)
return
}
path, err = sm.SectorBuilder.AllocSectorPath(fs.DataType(vars["type"]), abi.SectorNumber(id), true)
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
}
mediatype, _, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
if err := os.RemoveAll(string(path)); err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
switch mediatype {
case "application/x-tar":
if err := tarutil.ExtractTar(r.Body, string(path)); err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
default:
if err := files.WriteTo(files.NewReaderFile(r.Body), string(path)); err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
}
w.WriteHeader(200)
log.Infof("received %s sector (%s): %d bytes", vars["type"], vars["sname"], r.ContentLength)*/
}
/*
func (sm *StorageMinerAPI) WorkerStats(context.Context) (sectorbuilder.WorkerStats, error) {
stat := sm.SectorBuilder.WorkerStats()
@ -252,6 +253,7 @@ func (sm *StorageMinerAPI) SectorsRefs(context.Context) (map[string][]api.Sealed
func (sm *StorageMinerAPI) SectorsUpdate(ctx context.Context, id abi.SectorNumber, state api.SectorState) error {
return sm.Miner.ForceSectorState(ctx, id, state)
}
/*
func (sm *StorageMinerAPI) WorkerQueue(ctx context.Context, cfg sectorbuilder.WorkerCfg) (<-chan sectorbuilder.WorkerTask, error) {
return sm.SectorBuilder.AddWorker(ctx, cfg)

14
node/repo/fsrepo.go
View File

@ -25,13 +25,13 @@ import (
)
const (
fsAPI = "api"
fsAPIToken = "token"
fsConfig = "config.toml"
fsAPI = "api"
fsAPIToken = "token"
fsConfig = "config.toml"
fsStorageConfig = "storage.json"
fsDatastore = "datastore"
fsLock = "repo.lock"
fsKeystore = "keystore"
fsDatastore = "datastore"
fsLock = "repo.lock"
fsKeystore = "keystore"
)
type RepoType int
@ -88,7 +88,7 @@ func (fsr *FsRepo) Exists() (bool, error) {
func (fsr *FsRepo) Init(t RepoType) error {
exist, err := fsr.Exists()
if err != nil{
if err != nil {
return err
}
if exist {

5
storage/fpost_run.go
View File

@ -92,7 +92,10 @@ func (s *FPoStScheduler) declareFaults(ctx context.Context, fc uint64, params *m
func (s *FPoStScheduler) checkFaults(ctx context.Context, ssi []abi.SectorNumber) ([]abi.SectorNumber, error) {
//faults := s.sb.Scrub(ssi)
log.Warnf("Stub checkFaults")
var faults []struct{SectorNum abi.SectorNumber; Err error}
var faults []struct {
SectorNum abi.SectorNumber
Err error
}
declaredFaults := map[abi.SectorNumber]struct{}{}

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

@ -15,7 +15,7 @@ import (
)
type localWorker struct {
scfg *sectorbuilder.Config
scfg *sectorbuilder.Config
storage *storage
}
@ -41,7 +41,7 @@ func (l *localWorkerPathProvider) AcquireSector(id abi.SectorNumber, existing se
}
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, sz abi.UnpaddedPieceSize, sn abi.SectorNumber, r io.Reader, epcs []abi.UnpaddedPieceSize) (abi.PieceInfo, error) {
@ -100,10 +100,10 @@ func (l *localWorker) FinalizeSector(ctx context.Context, sectorNum abi.SectorNu
func (l *localWorker) TaskTypes() map[sealmgr.TaskType]struct{} {
return map[sealmgr.TaskType]struct{}{
sealmgr.TTAddPiece: {},
sealmgr.TTAddPiece: {},
sealmgr.TTPreCommit1: {},
sealmgr.TTPreCommit2: {},
sealmgr.TTCommit2: {},
sealmgr.TTCommit2: {},
}
}

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

@ -25,12 +25,12 @@ type LocalStorage interface {
}
type Path struct {
ID string
ID string
Weight uint64
LocalPath string
CanSeal bool
CanSeal bool
CanStore bool
}
@ -43,8 +43,8 @@ type Worker interface {
type Manager struct {
workers []Worker
scfg *sectorbuilder.Config
sc SectorIDCounter
scfg *sectorbuilder.Config
sc SectorIDCounter
storage *storage
@ -70,11 +70,11 @@ func New(ls LocalStorage, cfg *sectorbuilder.Config, sc SectorIDCounter) (*Manag
}
m := &Manager{
workers: []Worker{
workers: []Worker{
&localWorker{scfg: cfg, storage: stor},
},
scfg: cfg,
sc: sc,
sc: sc,
storage: stor,
@ -159,7 +159,7 @@ func (m *Manager) AddPiece(ctx context.Context, sz abi.UnpaddedPieceSize, sn abi
func (m *Manager) SealPreCommit1(ctx context.Context, sectorNum abi.SectorNumber, ticket abi.SealRandomness, pieces []abi.PieceInfo) (out []byte, err error) {
// TODO: also consider where the unsealed data sits
best, err := m.storage.findBestAllocStorage(sectorbuilder.FTCache | sectorbuilder.FTSealed, true)
best, err := m.storage.findBestAllocStorage(sectorbuilder.FTCache|sectorbuilder.FTSealed, true)
if err != nil {
return nil, xerrors.Errorf("finding path for sector sealing: %w", err)
}
@ -174,7 +174,7 @@ func (m *Manager) SealPreCommit1(ctx context.Context, sectorNum abi.SectorNumber
func (m *Manager) SealPreCommit2(ctx context.Context, sectorNum abi.SectorNumber, phase1Out []byte) (sealedCID cid.Cid, unsealedCID cid.Cid, err error) {
// TODO: allow workers to fetch the sectors
best, err := m.storage.findSector(m.minerID(), sectorNum, sectorbuilder.FTCache | sectorbuilder.FTSealed)
best, err := m.storage.findSector(m.minerID(), sectorNum, sectorbuilder.FTCache|sectorbuilder.FTSealed)
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("finding path for sector sealing: %w", err)
}
@ -187,7 +187,7 @@ func (m *Manager) SealPreCommit2(ctx context.Context, sectorNum abi.SectorNumber
}
func (m *Manager) SealCommit1(ctx context.Context, sectorNum abi.SectorNumber, ticket abi.SealRandomness, seed abi.InteractiveSealRandomness, pieces []abi.PieceInfo, sealedCID cid.Cid, unsealedCID cid.Cid) (output []byte, err error) {
best, err := m.storage.findSector(m.minerID(), sectorNum, sectorbuilder.FTCache | sectorbuilder.FTSealed)
best, err := m.storage.findSector(m.minerID(), sectorNum, sectorbuilder.FTCache|sectorbuilder.FTSealed)
if err != nil {
return nil, xerrors.Errorf("finding path for sector sealing: %w", err)
}
@ -212,7 +212,7 @@ func (m *Manager) SealCommit2(ctx context.Context, sectorNum abi.SectorNumber, p
}
func (m *Manager) FinalizeSector(ctx context.Context, sectorNum abi.SectorNumber) error {
best, err := m.storage.findSector(m.minerID(), sectorNum, sectorbuilder.FTCache | sectorbuilder.FTSealed | sectorbuilder.FTUnsealed)
best, err := m.storage.findSector(m.minerID(), sectorNum, sectorbuilder.FTCache|sectorbuilder.FTSealed|sectorbuilder.FTUnsealed)
if err != nil {
return xerrors.Errorf("finding sealed sector: %w", err)
}

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

@ -8,7 +8,7 @@ import (
type readonlyProvider struct {
miner abi.ActorID
stor *storage
stor *storage
}
func (l *readonlyProvider) AcquireSectorNumber() (abi.SectorNumber, error) {

15
storage/sealmgr/advmgr/storage.go
View File

@ -17,17 +17,18 @@ import (
)
const metaFile = "sectorstore.json"
var pathTypes = []sectorbuilder.SectorFileType{sectorbuilder.FTUnsealed, sectorbuilder.FTSealed, sectorbuilder.FTCache}
type storage struct {
localLk sync.RWMutex
localLk sync.RWMutex
localStorage LocalStorage
paths []path
}
type path struct {
meta config.StorageMeta
meta config.StorageMeta
local string
sectors map[abi.SectorID]sectorbuilder.SectorFileType
@ -102,7 +103,7 @@ func (st *storage) open() error {
}
func (st *storage) acquireSector(mid abi.ActorID, id abi.SectorNumber, existing sectorbuilder.SectorFileType, allocate sectorbuilder.SectorFileType, sealing bool) (sectorbuilder.SectorPaths, func(), error) {
if existing | allocate != existing ^ allocate {
if existing|allocate != existing^allocate {
return sectorbuilder.SectorPaths{}, nil, xerrors.New("can't both find and allocate a sector")
}
@ -111,7 +112,7 @@ func (st *storage) acquireSector(mid abi.ActorID, id abi.SectorNumber, existing
var out sectorbuilder.SectorPaths
for _, fileType := range pathTypes {
if fileType & existing == 0 {
if fileType&existing == 0 {
continue
}
@ -123,7 +124,7 @@ func (st *storage) acquireSector(mid abi.ActorID, id abi.SectorNumber, existing
if !ok {
continue
}
if s & fileType == 0 {
if s&fileType == 0 {
continue
}
@ -143,7 +144,7 @@ func (st *storage) acquireSector(mid abi.ActorID, id abi.SectorNumber, existing
}
for _, fileType := range pathTypes {
if fileType & allocate == 0 {
if fileType&allocate == 0 {
continue
}
@ -219,7 +220,7 @@ func (st *storage) findSector(mid abi.ActorID, sn abi.SectorNumber, typ sectorbu
Miner: mid,
Number: sn,
}]
if t | typ == 0 {
if t|typ == 0 {
continue
}
out = append(out, p.meta)

1
storage/sealmgr/simple.go
View File

@ -16,7 +16,6 @@ import (
ffi "github.com/filecoin-project/filecoin-ffi"
)
type LocalWorker struct {
sectorbuilder.Basic
}