lotus/storage/wdpost_sched.go

193 lines
4.9 KiB
Go
Raw Normal View History

package storage
import (
"context"
"time"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
2020-09-07 03:49:10 +00:00
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/dline"
2020-03-06 05:30:47 +00:00
"github.com/filecoin-project/specs-storage/storage"
2020-01-13 20:47:27 +00:00
"github.com/filecoin-project/lotus/api"
2020-07-10 14:43:14 +00:00
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
2020-08-17 13:39:33 +00:00
sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
2020-07-20 13:45:17 +00:00
"github.com/filecoin-project/lotus/journal"
"github.com/filecoin-project/lotus/node/config"
"go.opencensus.io/trace"
)
2020-04-07 17:41:41 +00:00
type WindowPoStScheduler struct {
2020-04-29 18:06:05 +00:00
api storageMinerApi
2020-08-12 17:47:00 +00:00
feeCfg config.MinerFeeConfig
addrSel *AddressSelector
2020-04-29 18:06:05 +00:00
prover storage.Prover
verifier ffiwrapper.Verifier
faultTracker sectorstorage.FaultTracker
2020-06-15 16:30:49 +00:00
proofType abi.RegisteredPoStProof
2020-04-29 18:06:05 +00:00
partitionSectors uint64
ch *changeHandler
2020-10-21 03:35:18 +00:00
actor address.Address
evtTypes [4]journal.EventType
journal journal.Journal
2019-12-08 19:48:17 +00:00
2020-07-20 13:45:17 +00:00
// failed abi.ChainEpoch // eps
// failLk sync.Mutex
}
func NewWindowedPoStScheduler(api storageMinerApi, fc config.MinerFeeConfig, as *AddressSelector, sb storage.Prover, verif ffiwrapper.Verifier, ft sectorstorage.FaultTracker, j journal.Journal, actor address.Address) (*WindowPoStScheduler, error) {
mi, err := api.StateMinerInfo(context.TODO(), actor, types.EmptyTSK)
if err != nil {
return nil, xerrors.Errorf("getting sector size: %w", err)
}
2020-04-29 18:06:05 +00:00
return &WindowPoStScheduler{
api: api,
2020-08-12 17:47:00 +00:00
feeCfg: fc,
addrSel: as,
2020-04-29 18:06:05 +00:00
prover: sb,
verifier: verif,
2020-05-16 21:50:50 +00:00
faultTracker: ft,
proofType: mi.WindowPoStProofType,
2020-04-29 18:06:05 +00:00
partitionSectors: mi.WindowPoStPartitionSectors,
2020-10-21 03:35:18 +00:00
actor: actor,
evtTypes: [...]journal.EventType{
evtTypeWdPoStScheduler: j.RegisterEventType("wdpost", "scheduler"),
evtTypeWdPoStProofs: j.RegisterEventType("wdpost", "proofs_processed"),
evtTypeWdPoStRecoveries: j.RegisterEventType("wdpost", "recoveries_processed"),
evtTypeWdPoStFaults: j.RegisterEventType("wdpost", "faults_processed"),
},
journal: j,
2020-04-29 18:06:05 +00:00
}, nil
2020-04-07 19:55:34 +00:00
}
type changeHandlerAPIImpl struct {
storageMinerApi
*WindowPoStScheduler
}
2020-04-07 17:41:41 +00:00
func (s *WindowPoStScheduler) Run(ctx context.Context) {
// Initialize change handler
chImpl := &changeHandlerAPIImpl{storageMinerApi: s.api, WindowPoStScheduler: s}
s.ch = newChangeHandler(chImpl, s.actor)
defer s.ch.shutdown()
s.ch.start()
var notifs <-chan []*api.HeadChange
var err error
var gotCur bool
// not fine to panic after this point
for {
if notifs == nil {
notifs, err = s.api.ChainNotify(ctx)
if err != nil {
2020-08-18 19:08:20 +00:00
log.Errorf("ChainNotify error: %+v", err)
2020-07-10 14:43:14 +00:00
build.Clock.Sleep(10 * time.Second)
continue
}
gotCur = false
}
select {
case changes, ok := <-notifs:
if !ok {
log.Warn("window post scheduler notifs channel closed")
notifs = nil
continue
}
if !gotCur {
if len(changes) != 1 {
log.Errorf("expected first notif to have len = 1")
continue
}
chg := changes[0]
if chg.Type != store.HCCurrent {
log.Errorf("expected first notif to tell current ts")
continue
}
2020-09-25 12:31:07 +00:00
ctx, span := trace.StartSpan(ctx, "WindowPoStScheduler.headChange")
s.update(ctx, nil, chg.Val)
2020-09-25 12:31:07 +00:00
span.End()
gotCur = true
continue
}
2020-04-07 17:41:41 +00:00
ctx, span := trace.StartSpan(ctx, "WindowPoStScheduler.headChange")
var lowest, highest *types.TipSet = nil, nil
for _, change := range changes {
2020-01-18 15:00:22 +00:00
if change.Val == nil {
log.Errorf("change.Val was nil")
}
switch change.Type {
case store.HCRevert:
lowest = change.Val
case store.HCApply:
highest = change.Val
}
}
s.update(ctx, lowest, highest)
span.End()
case <-ctx.Done():
return
}
}
}
func (s *WindowPoStScheduler) update(ctx context.Context, revert, apply *types.TipSet) {
if apply == nil {
log.Error("no new tipset in window post WindowPoStScheduler.update")
return
}
err := s.ch.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
func (s *WindowPoStScheduler) onAbort(ts *types.TipSet, deadline *dline.Info) {
s.journal.RecordEvent(s.evtTypes[evtTypeWdPoStScheduler], func() interface{} {
c := evtCommon{}
if ts != nil {
c.Deadline = deadline
c.Height = ts.Height()
c.TipSet = ts.Cids()
}
return WdPoStSchedulerEvt{
evtCommon: c,
State: SchedulerStateAborted,
}
})
}
func (s *WindowPoStScheduler) getEvtCommon(err error) evtCommon {
c := evtCommon{Error: err}
currentTS, currentDeadline := s.ch.currentTSDI()
if currentTS != nil {
c.Deadline = currentDeadline
c.Height = currentTS.Height()
c.TipSet = currentTS.Cids()
}
return c
}