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

sealing pipeline: Cleanup journal plumbing

Browse Source
This commit is contained in:
Łukasz Magiera 2022-08-09 12:42:35 +02:00
parent ada6a143ad
commit 4b6a9e0387
3 changed files with 35 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
storage/miner.go
View File

@ -61,20 +61,9 @@ type Miner struct {
getSealConfig dtypes.GetSealingConfigFunc getSealConfig dtypes.GetSealingConfigFunc
sealing *pipeline.Sealing sealing *pipeline.Sealing
sealingEvtType journal.EventType
journal journal.Journal journal journal.Journal
} }
// SealingStateEvt is a journal event that records a sector state transition.
type SealingStateEvt struct {
SectorNumber abi.SectorNumber
SectorType abi.RegisteredSealProof
From pipeline.SectorState
After pipeline.SectorState
Error string
}
// fullNodeFilteredAPI is the subset of the full node API the Miner needs from // fullNodeFilteredAPI is the subset of the full node API the Miner needs from
// a Lotus full node. // a Lotus full node.
type fullNodeFilteredAPI interface { type fullNodeFilteredAPI interface {
@ -151,10 +140,9 @@ func NewMiner(api fullNodeFilteredAPI,
prover: prover, prover: prover,
addrSel: as, addrSel: as,
maddr: maddr, maddr: maddr,
getSealConfig: gsd, getSealConfig: gsd,
journal: journal, journal: journal,
sealingEvtType: journal.RegisterEventType("storage", "sealing_states"),
} }
return m, nil return m, nil
@ -189,7 +177,7 @@ func (m *Miner) Run(ctx context.Context) error {
} }
// Instantiate the sealing FSM. // Instantiate the sealing FSM.
m.sealing = pipeline.New(ctx, m.api, m.feeCfg, evts, m.maddr, m.ds, m.sealer, m.sc, m.verif, m.prover, &pcp, cfg, m.handleSealingNotifications, as) m.sealing = pipeline.New(ctx, m.api, m.feeCfg, evts, m.maddr, m.ds, m.sealer, m.sc, m.verif, m.prover, &pcp, cfg, m.journal, as)
// Run the sealing FSM. // Run the sealing FSM.
go m.sealing.Run(ctx) //nolint:errcheck // logged intside the function go m.sealing.Run(ctx) //nolint:errcheck // logged intside the function
@ -197,18 +185,6 @@ func (m *Miner) Run(ctx context.Context) error {
return nil return nil
} }
func (m *Miner) handleSealingNotifications(before, after pipeline.SectorInfo) {
m.journal.RecordEvent(m.sealingEvtType, func() interface{} {
return SealingStateEvt{
SectorNumber: before.SectorNumber,
SectorType: before.SectorType,
From: before.State,
After: after.State,
Error: after.LastErr,
}
})
}
func (m *Miner) Stop(ctx context.Context) error { func (m *Miner) Stop(ctx context.Context) error {
return m.sealing.Stop(ctx) return m.sealing.Stop(ctx)
} }

26
storage/pipeline/sealing.go
View File

@ -2,6 +2,7 @@ package sealing
import ( import (
"context" "context"
"github.com/filecoin-project/lotus/journal"
"sync" "sync"
"time" "time"
@ -99,8 +100,10 @@ type Sealing struct {
available map[abi.SectorID]struct{} available map[abi.SectorID]struct{}
notifee SectorStateNotifee journal journal.Journal
addrSel AddrSel sealingEvtType journal.EventType
notifee SectorStateNotifee
addrSel AddrSel
stats SectorStats stats SectorStats
@ -149,7 +152,7 @@ type pendingPiece struct {
accepted func(abi.SectorNumber, abi.UnpaddedPieceSize, error) accepted func(abi.SectorNumber, abi.UnpaddedPieceSize, error)
} }
func New(mctx context.Context, api SealingAPI, fc config.MinerFeeConfig, events Events, maddr address.Address, ds datastore.Batching, sealer sealer.SectorManager, sc SectorIDCounter, verif storiface.Verifier, prov storiface.Prover, pcp PreCommitPolicy, gc GetSealingConfigFunc, notifee SectorStateNotifee, as AddrSel) *Sealing { func New(mctx context.Context, api SealingAPI, fc config.MinerFeeConfig, events Events, maddr address.Address, ds datastore.Batching, sealer sealer.SectorManager, sc SectorIDCounter, verif storiface.Verifier, prov storiface.Prover, pcp PreCommitPolicy, gc GetSealingConfigFunc, journal journal.Journal, as AddrSel) *Sealing {
s := &Sealing{ s := &Sealing{
Api: api, Api: api,
DealInfo: &CurrentDealInfoManager{api}, DealInfo: &CurrentDealInfoManager{api},
@ -170,7 +173,9 @@ func New(mctx context.Context, api SealingAPI, fc config.MinerFeeConfig, events
available: map[abi.SectorID]struct{}{}, available: map[abi.SectorID]struct{}{},
notifee: notifee, journal: journal,
sealingEvtType: journal.RegisterEventType("storage", "sealing_states"),
addrSel: as, addrSel: as,
terminator: NewTerminationBatcher(mctx, maddr, api, as, fc, gc), terminator: NewTerminationBatcher(mctx, maddr, api, as, fc, gc),
@ -184,6 +189,19 @@ func New(mctx context.Context, api SealingAPI, fc config.MinerFeeConfig, events
byState: map[SectorState]int64{}, byState: map[SectorState]int64{},
}, },
} }
s.notifee = func(before, after SectorInfo) {
s.journal.RecordEvent(s.sealingEvtType, func() interface{} {
return SealingStateEvt{
SectorNumber: before.SectorNumber,
SectorType: before.SectorType,
From: before.State,
After: after.State,
Error: after.LastErr,
}
})
}
s.startupWait.Add(1) s.startupWait.Add(1)
s.sectors = statemachine.New(namespace.Wrap(ds, datastore.NewKey(SectorStorePrefix)), s, SectorInfo{}) s.sectors = statemachine.New(namespace.Wrap(ds, datastore.NewKey(SectorStorePrefix)), s, SectorInfo{})

9
storage/pipeline/types.go
View File

@ -196,3 +196,12 @@ type SectorIDCounter interface {
} }
type GetSealingConfigFunc func() (sealiface.Config, error) type GetSealingConfigFunc func() (sealiface.Config, error)
// SealingStateEvt is a journal event that records a sector state transition.
type SealingStateEvt struct {
SectorNumber abi.SectorNumber
SectorType abi.RegisteredSealProof
From SectorState
After SectorState
Error string
}