diff --git a/node/builder_miner.go b/node/builder_miner.go index 23dc9c542..e31cbe584 100644 --- a/node/builder_miner.go +++ b/node/builder_miner.go @@ -103,6 +103,10 @@ func ConfigStorageMiner(c interface{}) Option { If(cfg.Subsystems.EnableSealing, Error(xerrors.Errorf("sealing can only be enabled on a mining node"))), If(cfg.Subsystems.EnableSectorStorage, Error(xerrors.Errorf("sealing can only be enabled on a mining node"))), ), + + Override(new(*harmonydb.DB), func(cfg config.HarmonyDB, id harmonydb.ITestID) (*harmonydb.DB, error) { + return harmonydb.NewFromConfigWithITestID(cfg)(id) + }), If(cfg.Subsystems.EnableMining, If(!cfg.Subsystems.EnableSealing, Error(xerrors.Errorf("sealing can't be disabled on a mining node yet"))), If(!cfg.Subsystems.EnableSectorStorage, Error(xerrors.Errorf("sealing can't be disabled on a mining node yet"))), @@ -126,10 +130,6 @@ func ConfigStorageMiner(c interface{}) Option { Override(new(sectorblocks.SectorBuilder), From(new(*sealing.Sealing))), ), - Override(new(*harmonydb.DB), func(cfg config.HarmonyDB, id harmonydb.ITestID) (*harmonydb.DB, error) { - return harmonydb.NewFromConfigWithITestID(cfg)(id) - }), - If(cfg.Subsystems.EnableSectorStorage, // Sector storage Override(new(*paths.IndexProxy), paths.NewIndexProxyHelper(cfg.Subsystems.EnableSectorIndexDB)), diff --git a/node/modules/storageminer.go b/node/modules/storageminer.go index 74251e21d..533ef6c9b 100644 --- a/node/modules/storageminer.go +++ b/node/modules/storageminer.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "github.com/filecoin-project/lotus/lib/harmony/harmonydb" "net/http" "os" "path/filepath" @@ -300,8 +301,8 @@ func SealingPipeline(fc config.MinerFeeConfig) func(params SealingPipelineParams } } -func WindowPostScheduler(fc config.MinerFeeConfig, pc config.ProvingConfig) func(params SealingPipelineParams) (*wdpost.WindowPoStScheduler, error) { - return func(params SealingPipelineParams) (*wdpost.WindowPoStScheduler, error) { +func WindowPostScheduler(fc config.MinerFeeConfig, pc config.ProvingConfig) func(params SealingPipelineParams, db *harmonydb.DB) (*wdpost.WindowPoStScheduler, error) { + return func(params SealingPipelineParams, db *harmonydb.DB) (*wdpost.WindowPoStScheduler, error) { var ( mctx = params.MetricsCtx lc = params.Lifecycle @@ -315,7 +316,7 @@ func WindowPostScheduler(fc config.MinerFeeConfig, pc config.ProvingConfig) func ctx := helpers.LifecycleCtx(mctx, lc) - fps, err := wdpost.NewWindowedPoStScheduler(api, fc, pc, as, sealer, verif, sealer, j, maddr) + fps, err := wdpost.NewWindowedPoStScheduler(api, fc, pc, as, sealer, verif, sealer, j, maddr, db) if err != nil { return nil, err diff --git a/storage/wdpost/wdpost_changehandler.go b/storage/wdpost/wdpost_changehandler.go index 266b8b042..32baa6165 100644 --- a/storage/wdpost/wdpost_changehandler.go +++ b/storage/wdpost/wdpost_changehandler.go @@ -2,6 +2,7 @@ package wdpost import ( "context" + "github.com/filecoin-project/lotus/lib/harmony/harmonydb" "sync" "github.com/filecoin-project/go-address" @@ -36,11 +37,13 @@ type changeHandler struct { actor address.Address proveHdlr *proveHandler submitHdlr *submitHandler + + db *harmonydb.DB } -func newChangeHandler(api wdPoStCommands, actor address.Address) *changeHandler { +func newChangeHandler(api wdPoStCommands, actor address.Address, db *harmonydb.DB) *changeHandler { posts := newPostsCache() - p := newProver(api, posts) + p := newProver(api, posts, db) s := newSubmitter(api, posts) return &changeHandler{api: api, actor: actor, proveHdlr: p, submitHdlr: s} } @@ -162,11 +165,14 @@ type proveHandler struct { // Used for testing processedHeadChanges chan *headChange processedPostResults chan *postResult + + wdPostTask *WdPostTask } func newProver( api wdPoStCommands, posts *postsCache, + db *harmonydb.DB, ) *proveHandler { ctx, cancel := context.WithCancel(context.Background()) return &proveHandler{ @@ -176,6 +182,7 @@ func newProver( hcs: make(chan *headChange), shutdownCtx: ctx, shutdown: cancel, + wdPostTask: NewWdPostTask(db), } } @@ -211,6 +218,8 @@ func (p *proveHandler) run() { func (p *proveHandler) processHeadChange(ctx context.Context, newTS *types.TipSet, di *dline.Info) { // If the post window has expired, abort the current proof + //log.Errorf("--------------------WINDOW POST CHANGE HANDLER PROCESS HC----------------------") + if p.current != nil && newTS.Height() >= p.current.di.Close { // Cancel the context on the current proof p.current.abort() @@ -234,6 +243,11 @@ func (p *proveHandler) processHeadChange(ctx context.Context, newTS *types.TipSe _, complete = p.posts.get(di) } + //err := p.wdPostTask.AddTask(ctx, newTS, di) + //if err != nil { + // log.Errorf("AddTask failed: %v", err) + //} + // Check if the chain is above the Challenge height for the post window if newTS.Height() < di.Challenge+ChallengeConfidence { return diff --git a/storage/wdpost/wdpost_sched.go b/storage/wdpost/wdpost_sched.go index 29c39ad9e..0a8b5b6c7 100644 --- a/storage/wdpost/wdpost_sched.go +++ b/storage/wdpost/wdpost_sched.go @@ -2,6 +2,7 @@ package wdpost import ( "context" + "github.com/filecoin-project/lotus/lib/harmony/harmonydb" "time" "github.com/ipfs/go-cid" @@ -77,6 +78,7 @@ type WindowPoStScheduler struct { maxPartitionsPerRecoveryMessage int singleRecoveringPartitionPerPostMessage bool ch *changeHandler + ch2 *changeHandler2 actor address.Address @@ -85,6 +87,7 @@ type WindowPoStScheduler struct { // failed abi.ChainEpoch // eps // failLk sync.Mutex + db *harmonydb.DB } // NewWindowedPoStScheduler creates a new WindowPoStScheduler scheduler. @@ -96,7 +99,8 @@ func NewWindowedPoStScheduler(api NodeAPI, verif storiface.Verifier, ft sealer.FaultTracker, j journal.Journal, - actor address.Address) (*WindowPoStScheduler, error) { + actor address.Address, + db *harmonydb.DB) (*WindowPoStScheduler, error) { mi, err := api.StateMinerInfo(context.TODO(), actor, types.EmptyTSK) if err != nil { return nil, xerrors.Errorf("getting sector size: %w", err) @@ -123,6 +127,7 @@ func NewWindowedPoStScheduler(api NodeAPI, evtTypeWdPoStFaults: j.RegisterEventType("wdpost", "faults_processed"), }, journal: j, + db: db, }, nil } @@ -135,10 +140,14 @@ func (s *WindowPoStScheduler) Run(ctx context.Context) { *WindowPoStScheduler }{s.api, s} - s.ch = newChangeHandler(callbacks, s.actor) + s.ch = newChangeHandler(callbacks, s.actor, s.db) defer s.ch.shutdown() s.ch.start() + s.ch2 = newChangeHandler2(callbacks, s.actor, s.db) + defer s.ch2.shutdown() + s.ch2.start() + var ( notifs <-chan []*api.HeadChange err error @@ -222,6 +231,11 @@ func (s *WindowPoStScheduler) update(ctx context.Context, revert, apply *types.T if err != nil { log.Errorf("handling head updates in window post sched: %+v", err) } + + err = s.ch2.update(ctx, revert, apply) + if err != nil { + log.Errorf("handling head updates in window post sched: %+v", err) + } } // onAbort is called when generating proofs or submitting proofs is aborted