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

WIP: wdpost listener

Browse Source
This commit is contained in:
Shrenuj Bansal
2023-08-29 12:28:44 -04:00
parent 5b6daa2c6c
commit 6b95bd24ba
4 changed files with 40 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder_miner.go
+4 -4
View File
@@ -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)),
node/modules/storageminer.go
+4 -3
View File
@@ -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
storage/wdpost/wdpost_changehandler.go
+16 -2
View File
@@ -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
storage/wdpost/wdpost_sched.go
+16 -2
View File
@@ -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