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

refactor RepoType

Browse Source
This commit is contained in:
Anton Evangelatov
2022-03-03 15:37:23 +01:00
parent 6b06c1b5d3
commit e3edab66e3
50 changed files with 248 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+11 -7
View File
@@ -224,12 +224,16 @@ var LibP2P = Options(
)
func IsType(t repo.RepoType) func(s *Settings) bool {
return func(s *Settings) bool { return s.nodeType == t }
return func(s *Settings) bool { return s.nodeType.Type() == t.Type() }
}
func isFullOrLiteNode(s *Settings) bool { return s.nodeType == repo.FullNode }
func isFullNode(s *Settings) bool { return s.nodeType == repo.FullNode && !s.Lite }
func isLiteNode(s *Settings) bool { return s.nodeType == repo.FullNode && s.Lite }
func isFullOrLiteNode(s *Settings) bool { return s.nodeType.Type() == repo.FullNodeRepoType{}.Type() }
func isFullNode(s *Settings) bool {
return s.nodeType.Type() == repo.FullNodeRepoType{}.Type() && !s.Lite
}
func isLiteNode(s *Settings) bool {
return s.nodeType.Type() == repo.FullNodeRepoType{}.Type() && s.Lite
}
func Base() Option {
return Options(
@@ -241,7 +245,7 @@ func Base() Option {
LibP2P,
),
ApplyIf(isFullOrLiteNode, ChainNode),
ApplyIf(IsType(repo.StorageMiner), MinerNode),
ApplyIf(IsType(repo.StorageMinerRepoType{}), MinerNode),
)
}
@@ -314,8 +318,8 @@ func Repo(r repo.Repo) Option {
Override(new(*dtypes.APIAlg), modules.APISecret),
ApplyIf(IsType(repo.FullNode), ConfigFullNode(c)),
ApplyIf(IsType(repo.StorageMiner), ConfigStorageMiner(c)),
ApplyIf(IsType(repo.FullNodeRepoType{}), ConfigFullNode(c)),
ApplyIf(IsType(repo.StorageMinerRepoType{}), ConfigStorageMiner(c)),
)(settings)
}
}
node/builder_chain.go
+1 -1
View File
@@ -248,7 +248,7 @@ func Lite(enable bool) FullOption {
func FullAPI(out *api.FullNode, fopts ...FullOption) Option {
return Options(
func(s *Settings) error {
s.nodeType = repo.FullNode
s.nodeType = repo.FullNodeRepoType{}
s.enableLibp2pNode = true
return nil
},
node/builder_miner.go
+3 -3
View File
@@ -151,8 +151,8 @@ func ConfigStorageMiner(c interface{}) Option {
Override(new(dtypes.RetrievalPricingFunc), modules.RetrievalPricingFunc(cfg.Dealmaking)),
// DAG Store
Override(new(dagstore.MinerAPI), modules.NewMinerAPI),
Override(DAGStoreKey, modules.DAGStore),
Override(new(dagstore.MinerAPI), modules.NewMinerAPI(cfg.DAGStore)),
Override(DAGStoreKey, modules.DAGStore(cfg.DAGStore)),
// Markets (retrieval)
Override(new(dagstore.SectorAccessor), sectoraccessor.NewSectorAccessor),
@@ -221,7 +221,7 @@ func StorageMiner(out *api.StorageMiner, subsystemsCfg config.MinerSubsystemConf
),
func(s *Settings) error {
s.nodeType = repo.StorageMiner
s.nodeType = repo.StorageMinerRepoType{}
s.enableLibp2pNode = subsystemsCfg.EnableMarkets
return nil
},
node/modules/storageminer_dagstore.go
+57 -75
View File
@@ -23,88 +23,70 @@ const (
)
// NewMinerAPI creates a new MinerAPI adaptor for the dagstore mounts.
func NewMinerAPI(lc fx.Lifecycle, r repo.LockedRepo, pieceStore dtypes.ProviderPieceStore, sa mdagstore.SectorAccessor) (mdagstore.MinerAPI, error) {
cfg, err := extractDAGStoreConfig(r)
if err != nil {
return nil, err
}
// caps the amount of concurrent calls to the storage, so that we don't
// spam it during heavy processes like bulk migration.
if v, ok := os.LookupEnv("LOTUS_DAGSTORE_MOUNT_CONCURRENCY"); ok {
concurrency, err := strconv.Atoi(v)
if err == nil {
cfg.MaxConcurrencyStorageCalls = concurrency
}
}
mountApi := mdagstore.NewMinerAPI(pieceStore, sa, cfg.MaxConcurrencyStorageCalls, cfg.MaxConcurrentUnseals)
ready := make(chan error, 1)
pieceStore.OnReady(func(err error) {
ready <- err
})
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
if err := <-ready; err != nil {
return fmt.Errorf("aborting dagstore start; piecestore failed to start: %s", err)
func NewMinerAPI(cfg config.DAGStoreConfig) func(fx.Lifecycle, repo.LockedRepo, dtypes.ProviderPieceStore, mdagstore.SectorAccessor) (mdagstore.MinerAPI, error) {
return func(lc fx.Lifecycle, r repo.LockedRepo, pieceStore dtypes.ProviderPieceStore, sa mdagstore.SectorAccessor) (mdagstore.MinerAPI, error) {
// caps the amount of concurrent calls to the storage, so that we don't
// spam it during heavy processes like bulk migration.
if v, ok := os.LookupEnv("LOTUS_DAGSTORE_MOUNT_CONCURRENCY"); ok {
concurrency, err := strconv.Atoi(v)
if err == nil {
cfg.MaxConcurrencyStorageCalls = concurrency
}
return mountApi.Start(ctx)
},
OnStop: func(context.Context) error {
return nil
},
})
}
return mountApi, nil
mountApi := mdagstore.NewMinerAPI(pieceStore, sa, cfg.MaxConcurrencyStorageCalls, cfg.MaxConcurrentUnseals)
ready := make(chan error, 1)
pieceStore.OnReady(func(err error) {
ready <- err
})
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
if err := <-ready; err != nil {
return fmt.Errorf("aborting dagstore start; piecestore failed to start: %s", err)
}
return mountApi.Start(ctx)
},
OnStop: func(context.Context) error {
return nil
},
})
return mountApi, nil
}
}
// DAGStore constructs a DAG store using the supplied minerAPI, and the
// user configuration. It returns both the DAGStore and the Wrapper suitable for
// passing to markets.
func DAGStore(lc fx.Lifecycle, r repo.LockedRepo, minerAPI mdagstore.MinerAPI) (*dagstore.DAGStore, *mdagstore.Wrapper, error) {
cfg, err := extractDAGStoreConfig(r)
if err != nil {
return nil, nil, err
}
// fall back to default root directory if not explicitly set in the config.
if cfg.RootDir == "" {
cfg.RootDir = filepath.Join(r.Path(), DefaultDAGStoreDir)
}
v, ok := os.LookupEnv(EnvDAGStoreCopyConcurrency)
if ok {
concurrency, err := strconv.Atoi(v)
if err == nil {
cfg.MaxConcurrentReadyFetches = concurrency
func DAGStore(cfg config.DAGStoreConfig) func(fx.Lifecycle, repo.LockedRepo, mdagstore.MinerAPI) (*dagstore.DAGStore, *mdagstore.Wrapper, error) {
return func(lc fx.Lifecycle, r repo.LockedRepo, minerAPI mdagstore.MinerAPI) (*dagstore.DAGStore, *mdagstore.Wrapper, error) {
// fall back to default root directory if not explicitly set in the config.
if cfg.RootDir == "" {
cfg.RootDir = filepath.Join(r.Path(), DefaultDAGStoreDir)
}
v, ok := os.LookupEnv(EnvDAGStoreCopyConcurrency)
if ok {
concurrency, err := strconv.Atoi(v)
if err == nil {
cfg.MaxConcurrentReadyFetches = concurrency
}
}
dagst, w, err := mdagstore.NewDAGStore(cfg, minerAPI)
if err != nil {
return nil, nil, xerrors.Errorf("failed to create DAG store: %w", err)
}
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
return w.Start(ctx)
},
OnStop: func(context.Context) error {
return w.Close()
},
})
return dagst, w, nil
}
dagst, w, err := mdagstore.NewDAGStore(cfg, minerAPI)
if err != nil {
return nil, nil, xerrors.Errorf("failed to create DAG store: %w", err)
}
lc.Append(fx.Hook{
OnStart: func(ctx context.Context) error {
return w.Start(ctx)
},
OnStop: func(context.Context) error {
return w.Close()
},
})
return dagst, w, nil
}
func extractDAGStoreConfig(r repo.LockedRepo) (config.DAGStoreConfig, error) {
cfg, err := r.Config()
if err != nil {
return config.DAGStoreConfig{}, xerrors.Errorf("could not load config: %w", err)
}
mcfg, ok := cfg.(*config.StorageMiner)
if !ok {
return config.DAGStoreConfig{}, xerrors.Errorf("config not expected type; expected config.StorageMiner, got: %T", cfg)
}
return mcfg.DAGStore, nil
}
node/repo/fsrepo.go
+73 -37
View File
@@ -41,47 +41,83 @@ const (
fsKeystore = "keystore"
)
type RepoType int
const (
_ = iota // Default is invalid
FullNode RepoType = iota
StorageMiner
Worker
Wallet
Markets
)
func (t RepoType) String() string {
s := [...]string{
"__invalid__",
"FullNode",
"StorageMiner",
"Worker",
"Wallet",
"Markets",
func NewRepoTypeFromString(t string) RepoType {
switch t {
case "FullNode":
return FullNodeRepoType{}
case "StorageMiner":
return StorageMinerRepoType{}
case "Worker":
return WorkerRepoType{}
case "Wallet":
return WalletRepoType{}
default:
panic("unknown RepoType")
}
if t < 0 || int(t) > len(s) {
return "__invalid__"
}
return s[t]
}
type RepoType interface {
Type() string
Config() interface{}
}
type FullNodeRepoType struct {
}
func (f FullNodeRepoType) Type() string {
return "FullNode"
}
func (f FullNodeRepoType) Config() interface{} {
return config.DefaultFullNode()
}
type StorageMinerRepoType struct {
}
func (f StorageMinerRepoType) Type() string {
return "StorageMiner"
}
func (f StorageMinerRepoType) Config() interface{} {
return config.DefaultStorageMiner()
}
type MarketsRepoType struct {
}
func (f MarketsRepoType) Type() string {
return "Markets"
}
func (f MarketsRepoType) Config() interface{} {
return config.DefaultStorageMiner()
}
type WorkerRepoType struct {
}
func (f WorkerRepoType) Type() string {
return "Worker"
}
func (f WorkerRepoType) Config() interface{} {
return &struct{}{}
}
type WalletRepoType struct {
}
func (f WalletRepoType) Type() string {
return "Wallet"
}
func (f WalletRepoType) Config() interface{} {
return &struct{}{}
}
func defConfForType(t RepoType) interface{} {
switch t {
case FullNode:
return config.DefaultFullNode()
case StorageMiner, Markets:
// markets is a specialised miner service
// this taxonomy needs to be cleaned up
return config.DefaultStorageMiner()
case Worker:
return &struct{}{}
case Wallet:
return &struct{}{}
default:
panic(fmt.Sprintf("unknown RepoType(%d)", int(t)))
}
return t.Config()
}
var log = logging.Logger("repo")
node/repo/fsrepo_test.go
+1 -1
View File
@@ -18,7 +18,7 @@ func genFsRepo(t *testing.T) (*FsRepo, func()) {
t.Fatal(err)
}
err = repo.Init(FullNode)
err = repo.Init(FullNodeRepoType{})
if err != ErrRepoExists && err != nil {
t.Fatal(err)
}
node/repo/memrepo.go
+1 -1
View File
@@ -106,7 +106,7 @@ func (lmem *lockedMemRepo) Path() string {
panic(err) // only used in tests, probably fine
}
if lmem.t == StorageMiner {
if lmem.t.Type() == "StorageMiner" || lmem.t.Type() == "Boost" {
// this is required due to the method makeDealStaging from cmd/lotus-storage-miner/init.go
// deal-staging is the directory deal files are staged in before being sealed into sectors
// for offline deal flow.
node/repo/repo_test.go
+4 -4
View File
@@ -21,12 +21,12 @@ func basicTest(t *testing.T, repo Repo) {
}
assert.Nil(t, apima, "with no api endpoint, return should be nil")
lrepo, err := repo.Lock(FullNode)
lrepo, err := repo.Lock(FullNodeRepoType{})
assert.NoError(t, err, "should be able to lock once")
assert.NotNil(t, lrepo, "locked repo shouldn't be nil")
{
lrepo2, err := repo.Lock(FullNode)
lrepo2, err := repo.Lock(FullNodeRepoType{})
if assert.Error(t, err) {
assert.Equal(t, ErrRepoAlreadyLocked, err)
}
@@ -36,7 +36,7 @@ func basicTest(t *testing.T, repo Repo) {
err = lrepo.Close()
assert.NoError(t, err, "should be able to unlock")
lrepo, err = repo.Lock(FullNode)
lrepo, err = repo.Lock(FullNodeRepoType{})
assert.NoError(t, err, "should be able to relock")
assert.NotNil(t, lrepo, "locked repo shouldn't be nil")
@@ -80,7 +80,7 @@ func basicTest(t *testing.T, repo Repo) {
k1 := types.KeyInfo{Type: "foo"}
k2 := types.KeyInfo{Type: "bar"}
lrepo, err = repo.Lock(FullNode)
lrepo, err = repo.Lock(FullNodeRepoType{})
assert.NoError(t, err, "should be able to relock")
assert.NotNil(t, lrepo, "locked repo shouldn't be nil")