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

Storage Manager refactor

Browse Source
This commit is contained in:
Łukasz Magiera
2020-03-03 23:19:22 +01:00
parent 3abb59a550
commit a0dbb6bdd6
27 changed files with 459 additions and 584 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+3 -21
View File
@@ -253,7 +253,6 @@ func Online() Option {
// Storage miner
ApplyIf(func(s *Settings) bool { return s.nodeType == repo.StorageMiner },
Override(new(sectorbuilder.Interface), modules.SectorBuilder),
Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
Override(new(sealing.TicketFn), modules.SealTicketGen),
Override(new(*storage.Miner), modules.StorageMiner),
@@ -346,30 +345,13 @@ func ConfigFullNode(c interface{}) Option {
)
}
func ConfigStorageMiner(c interface{}, lr repo.LockedRepo) Option {
func ConfigStorageMiner(c interface{}) Option {
cfg, ok := c.(*config.StorageMiner)
if !ok {
return Error(xerrors.Errorf("invalid config from repo, got: %T", c))
}
scfg := sectorbuilder.SimplePath(lr.Path())
if cfg.SectorBuilder.Path == "" {
if len(cfg.SectorBuilder.Storage) > 0 {
scfg = cfg.SectorBuilder.Storage
}
} else {
scfg = sectorbuilder.SimplePath(cfg.SectorBuilder.Path)
log.Warn("LEGACY SectorBuilder.Path FOUND IN CONFIG. Please use the new storage config")
}
return Options(
ConfigCommon(&cfg.Common),
Override(new(*sectorbuilder.Config), modules.SectorBuilderConfig(scfg,
cfg.SectorBuilder.WorkerCount,
cfg.SectorBuilder.DisableLocalPreCommit,
cfg.SectorBuilder.DisableLocalCommit)),
)
return Options(ConfigCommon(&cfg.Common))
}
func Repo(r repo.Repo) Option {
@@ -387,7 +369,7 @@ func Repo(r repo.Repo) Option {
Override(new(repo.LockedRepo), modules.LockedRepo(lr)), // module handles closing
ApplyIf(isType(repo.FullNode), ConfigFullNode(c)),
ApplyIf(isType(repo.StorageMiner), ConfigStorageMiner(c, lr)),
ApplyIf(isType(repo.StorageMiner), ConfigStorageMiner(c)),
Override(new(dtypes.MetadataDS), modules.Datastore),
Override(new(dtypes.ChainBlockstore), modules.ChainBlockstore),
node/config/def.go
+3 -13
View File
@@ -3,8 +3,6 @@ package config
import (
"encoding"
"time"
"github.com/filecoin-project/go-sectorbuilder/fs"
)
// Common is common config between full node and miner
@@ -25,7 +23,7 @@ type FullNode struct {
type StorageMiner struct {
Common
SectorBuilder SectorBuilder
Storage Storage
}
// API contains configs for API endpoint
@@ -54,14 +52,8 @@ type Metrics struct {
}
// // Storage Miner
type Storage struct {
type SectorBuilder struct {
Path string // TODO: remove // FORK (-ish)
Storage []fs.PathConfig
WorkerCount uint
DisableLocalPreCommit bool
DisableLocalCommit bool
}
func defCommon() Common {
@@ -95,9 +87,7 @@ func DefaultStorageMiner() *StorageMiner {
cfg := &StorageMiner{
Common: defCommon(),
SectorBuilder: SectorBuilder{
WorkerCount: 5,
},
Storage: Storage{},
}
cfg.Common.API.ListenAddress = "/ip4/127.0.0.1/tcp/2345/http"
return cfg
node/config/storage.go
+68
View File
@@ -0,0 +1,68 @@
package config
import (
"encoding/json"
"io"
"io/ioutil"
"os"
"golang.org/x/xerrors"
)
type LocalPath struct {
Path string
}
// .lotusstorage/storage.json
type StorageConfig struct {
StoragePaths []LocalPath
}
// [path]/metadata.json
type StorageMeta struct {
ID string
Weight int // 0 = readonly
CanCommit bool
CanStore bool
}
func StorageFromFile(path string, def *StorageConfig) (*StorageConfig, error) {
file, err := os.Open(path)
switch {
case os.IsNotExist(err):
if def == nil {
return nil, xerrors.Errorf("couldn't load storage config: %w", err)
}
return def, nil
case err != nil:
return nil, err
}
defer file.Close() //nolint:errcheck // The file is RO
return StorageFromReader(file, *def)
}
func StorageFromReader(reader io.Reader, def StorageConfig) (*StorageConfig, error) {
cfg := def
err := json.NewDecoder(reader).Decode(&cfg)
if err != nil {
return nil, err
}
return &cfg, nil
}
func WriteStorageFile(path string, config StorageConfig) error {
b, err := json.Marshal(config)
if err != nil {
return xerrors.Errorf("marshaling storage config: %w", err)
}
if err := ioutil.WriteFile(path, b, 0644); err != nil {
return xerrors.Errorf("persisting storage config (%s): %w", path, err)
}
return nil
}
node/impl/storminer.go
+13 -18
View File
@@ -3,25 +3,18 @@ package impl
import (
"context"
"encoding/json"
"io"
"mime"
"net/http"
"os"
"strconv"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/gorilla/mux"
files "github.com/ipfs/go-ipfs-files"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/go-sectorbuilder/fs"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/gorilla/mux"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/apistruct"
"github.com/filecoin-project/lotus/lib/tarutil"
"github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/storage"
"github.com/filecoin-project/lotus/storage/sectorblocks"
@@ -31,7 +24,7 @@ type StorageMinerAPI struct {
CommonAPI
SectorBuilderConfig *sectorbuilder.Config
SectorBuilder sectorbuilder.Interface
//SectorBuilder sectorbuilder.Interface
SectorBlocks *sectorblocks.SectorBlocks
Miner *storage.Miner
@@ -57,7 +50,8 @@ func (sm *StorageMinerAPI) ServeRemote(w http.ResponseWriter, r *http.Request) {
}
func (sm *StorageMinerAPI) remoteGetSector(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
panic("todo")
/* vars := mux.Vars(r)
id, err := strconv.ParseUint(vars["id"], 10, 64)
if err != nil {
@@ -98,11 +92,12 @@ func (sm *StorageMinerAPI) remoteGetSector(w http.ResponseWriter, r *http.Reques
if _, err := io.Copy(w, rd); err != nil {
log.Error(err)
return
}
}*/
}
func (sm *StorageMinerAPI) remotePutSector(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
panic("todo")
/* vars := mux.Vars(r)
id, err := strconv.ParseUint(vars["id"], 10, 64)
if err != nil {
@@ -159,13 +154,13 @@ func (sm *StorageMinerAPI) remotePutSector(w http.ResponseWriter, r *http.Reques
w.WriteHeader(200)
log.Infof("received %s sector (%s): %d bytes", vars["type"], vars["sname"], r.ContentLength)
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()
return stat, nil
}
}*/
func (sm *StorageMinerAPI) ActorAddress(context.Context) (address.Address, error) {
return sm.SectorBuilderConfig.Miner, nil
@@ -252,7 +247,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)
}
@@ -260,5 +255,5 @@ func (sm *StorageMinerAPI) WorkerQueue(ctx context.Context, cfg sectorbuilder.Wo
func (sm *StorageMinerAPI) WorkerDone(ctx context.Context, task uint64, res sectorbuilder.SealRes) error {
return sm.SectorBuilder.TaskDone(ctx, task, res)
}
*/
var _ api.StorageMiner = &StorageMinerAPI{}
node/modules/storageminer.go
+30 -60
View File
@@ -2,11 +2,8 @@ package modules
import (
"context"
"math"
"reflect"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/go-address"
dtgraphsync "github.com/filecoin-project/go-data-transfer/impl/graphsync"
piecefilestore "github.com/filecoin-project/go-fil-markets/filestore"
@@ -20,7 +17,6 @@ import (
smnet "github.com/filecoin-project/go-fil-markets/storagemarket/network"
paramfetch "github.com/filecoin-project/go-paramfetch"
"github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/go-sectorbuilder/fs"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-actors/actors/crypto"
@@ -37,10 +33,12 @@ import (
"github.com/ipfs/go-merkledag"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/routing"
"github.com/mitchellh/go-homedir"
"go.uber.org/fx"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/storage/sealmgr"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/gen"
@@ -75,51 +73,32 @@ func GetParams(sbc *sectorbuilder.Config) error {
return nil
}
func SectorBuilderConfig(storage []fs.PathConfig, threads uint, noprecommit, nocommit bool) func(dtypes.MetadataDS, api.FullNode) (*sectorbuilder.Config, error) {
return func(ds dtypes.MetadataDS, fnapi api.FullNode) (*sectorbuilder.Config, error) {
minerAddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
}
ssize, err := fnapi.StateMinerSectorSize(context.TODO(), minerAddr, types.EmptyTSK)
if err != nil {
return nil, err
}
for i := range storage {
storage[i].Path, err = homedir.Expand(storage[i].Path)
if err != nil {
return nil, err
}
}
if threads > math.MaxUint8 {
return nil, xerrors.Errorf("too many sectorbuilder threads specified: %d, max allowed: %d", threads, math.MaxUint8)
}
ppt, spt, err := api.ProofTypeFromSectorSize(ssize)
if err != nil {
return nil, xerrors.Errorf("bad sector size: %w", err)
}
sb := &sectorbuilder.Config{
Miner: minerAddr,
SealProofType: spt,
PoStProofType: ppt,
WorkerThreads: uint8(threads),
NoPreCommit: noprecommit,
NoCommit: nocommit,
Paths: storage,
}
return sb, nil
func SectorBuilderConfig(ds dtypes.MetadataDS, fnapi api.FullNode) (*sectorbuilder.Config, error) {
minerAddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
}
ssize, err := fnapi.StateMinerSectorSize(context.TODO(), minerAddr, types.EmptyTSK)
if err != nil {
return nil, err
}
ppt, spt, err := api.ProofTypeFromSectorSize(ssize)
if err != nil {
return nil, xerrors.Errorf("bad sector size: %w", err)
}
sb := &sectorbuilder.Config{
Miner: minerAddr,
SealProofType: spt,
PoStProofType: ppt,
}
return sb, nil
}
func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h host.Host, ds dtypes.MetadataDS, sb sectorbuilder.Interface, tktFn sealing.TicketFn) (*storage.Miner, error) {
func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h host.Host, ds dtypes.MetadataDS, sealer sealmgr.Manager, tktFn sealing.TicketFn) (*storage.Miner, error) {
maddr, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
@@ -132,9 +111,9 @@ func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h
return nil, err
}
fps := storage.NewFPoStScheduler(api, sb, maddr, worker)
fps := storage.NewFPoStScheduler(api, sealer, maddr, worker)
sm, err := storage.NewMiner(api, maddr, worker, h, ds, sb, tktFn)
sm, err := storage.NewMiner(api, maddr, worker, h, ds, sealer, tktFn)
if err != nil {
return nil, err
}
@@ -271,15 +250,6 @@ func SetupBlockProducer(lc fx.Lifecycle, ds dtypes.MetadataDS, api api.FullNode,
return m, nil
}
func SectorBuilder(cfg *sectorbuilder.Config, ds dtypes.MetadataDS) (*sectorbuilder.SectorBuilder, error) {
sb, err := sectorbuilder.New(cfg, namespace.Wrap(ds, datastore.NewKey("/sectorbuilder")))
if err != nil {
return nil, err
}
return sb, nil
}
func SealTicketGen(fapi api.FullNode) sealing.TicketFn {
return func(ctx context.Context) (*api.SealTicket, error) {
ts, err := fapi.ChainHead(ctx)
@@ -333,8 +303,8 @@ func StorageProvider(ctx helpers.MetricsCtx, fapi api.FullNode, h host.Host, ds
}
// RetrievalProvider creates a new retrieval provider attached to the provider blockstore
func RetrievalProvider(h host.Host, miner *storage.Miner, sb sectorbuilder.Interface, full api.FullNode, ds dtypes.MetadataDS, pieceStore dtypes.ProviderPieceStore, ibs dtypes.StagingBlockstore) (retrievalmarket.RetrievalProvider, error) {
adapter := retrievaladapter.NewRetrievalProviderNode(miner, sb, full)
func RetrievalProvider(h host.Host, miner *storage.Miner, sealer sealmgr.Manager, full api.FullNode, ds dtypes.MetadataDS, pieceStore dtypes.ProviderPieceStore, ibs dtypes.StagingBlockstore) (retrievalmarket.RetrievalProvider, error) {
adapter := retrievaladapter.NewRetrievalProviderNode(miner, sealer, full)
address, err := minerAddrFromDS(ds)
if err != nil {
return nil, err
node/repo/fsrepo.go
+14 -1
View File
@@ -28,6 +28,7 @@ const (
fsAPI = "api"
fsAPIToken = "token"
fsConfig = "config.toml"
fsStorageConfig = "storage.json"
fsDatastore = "datastore"
fsLock = "repo.lock"
fsKeystore = "keystore"
@@ -87,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 {
@@ -276,6 +277,18 @@ func (fsr *fsLockedRepo) Config() (interface{}, error) {
return config.FromFile(fsr.join(fsConfig), defConfForType(fsr.repoType))
}
func (fsr *fsLockedRepo) GetStorage() (config.StorageConfig, error) {
c, err := config.StorageFromFile(fsr.join(fsStorageConfig), nil)
if err != nil {
return config.StorageConfig{}, err
}
return *c, nil
}
func (fsr *fsLockedRepo) SetStorage(c config.StorageConfig) error {
return config.WriteStorageFile(fsr.join(fsStorageConfig), c)
}
func (fsr *fsLockedRepo) SetAPIEndpoint(ma multiaddr.Multiaddr) error {
if err := fsr.stillValid(); err != nil {
return err
node/repo/interface.go
+4
View File
@@ -7,6 +7,7 @@ import (
"github.com/multiformats/go-multiaddr"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/config"
)
var (
@@ -37,6 +38,9 @@ type LockedRepo interface {
// Returns config in this repo
Config() (interface{}, error)
GetStorage() (config.StorageConfig, error)
SetStorage(config.StorageConfig) error
// SetAPIEndpoint sets the endpoint of the current API
// so it can be read by API clients
SetAPIEndpoint(multiaddr.Multiaddr) error
node/repo/memrepo.go
+13
View File
@@ -12,6 +12,7 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/node/config"
)
type MemRepo struct {
@@ -38,6 +39,14 @@ type lockedMemRepo struct {
token *byte
}
func (lmem *lockedMemRepo) GetStorage() (config.StorageConfig, error) {
panic("implement me")
}
func (lmem *lockedMemRepo) SetStorage(config.StorageConfig) error {
panic("implement me")
}
func (lmem *lockedMemRepo) Path() string {
t, err := ioutil.TempDir(os.TempDir(), "lotus-memrepo-temp-")
if err != nil {
@@ -169,6 +178,10 @@ func (lmem *lockedMemRepo) Config() (interface{}, error) {
return lmem.mem.configF(lmem.t), nil
}
func (lmem *lockedMemRepo) Storage() (config.StorageConfig, error) {
panic("implement me")
}
func (lmem *lockedMemRepo) SetAPIEndpoint(ma multiaddr.Multiaddr) error {
if err := lmem.checkToken(); err != nil {
return err