lotus/storage/fpost_sched.go

177 lines
3.7 KiB
Go
Raw Normal View History

package storage
import (
"context"
2019-12-08 19:48:17 +00:00
"sync"
"go.opencensus.io/trace"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
2020-03-06 05:30:47 +00:00
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-storage/storage"
2020-01-13 20:47:27 +00:00
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
)
const Inactive = 0
const StartConfidence = 4 // TODO: config
type FPoStScheduler struct {
api storageMinerApi
sb storage.Prover
proofType abi.RegisteredProof
actor address.Address
worker address.Address
cur *types.TipSet
// if a post is in progress, this indicates for which ElectionPeriodStart
2020-02-08 02:18:32 +00:00
activeEPS abi.ChainEpoch
abort context.CancelFunc
2019-12-08 19:48:17 +00:00
2020-02-08 02:18:32 +00:00
failed abi.ChainEpoch // eps
2019-12-08 19:48:17 +00:00
failLk sync.Mutex
}
func NewFPoStScheduler(api storageMinerApi, sb storage.Prover, actor address.Address, worker address.Address, rt abi.RegisteredProof) *FPoStScheduler {
return &FPoStScheduler{api: api, sb: sb, actor: actor, worker: worker, proofType: rt}
}
func (s *FPoStScheduler) Run(ctx context.Context) {
notifs, err := s.api.ChainNotify(ctx)
if err != nil {
return
}
current := <-notifs
if len(current) != 1 {
panic("expected first notif to have len = 1")
}
2019-11-28 18:08:10 +00:00
if current[0].Type != store.HCCurrent {
panic("expected first notif to tell current ts")
}
if err := s.update(ctx, current[0].Val); err != nil {
panic(err)
}
defer s.abortActivePoSt()
// not fine to panic after this point
for {
select {
case changes, ok := <-notifs:
if !ok {
log.Warn("FPoStScheduler notifs channel closed")
return
}
ctx, span := trace.StartSpan(ctx, "FPoStScheduler.headChange")
var lowest, highest *types.TipSet = s.cur, 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
}
}
if err := s.revert(ctx, lowest); err != nil {
log.Error("handling head reverts in fallbackPost sched: %+v", err)
}
if err := s.update(ctx, highest); err != nil {
log.Error("handling head updates in fallbackPost sched: %+v", err)
}
span.End()
case <-ctx.Done():
return
}
}
}
func (s *FPoStScheduler) revert(ctx context.Context, newLowest *types.TipSet) error {
if s.cur == newLowest {
return nil
}
s.cur = newLowest
newEPS, _, err := s.shouldFallbackPost(ctx, newLowest)
if err != nil {
return err
}
if newEPS != s.activeEPS {
s.abortActivePoSt()
}
return nil
}
func (s *FPoStScheduler) update(ctx context.Context, new *types.TipSet) error {
2020-01-18 15:00:22 +00:00
if new == nil {
return xerrors.Errorf("no new tipset in FPoStScheduler.update")
2020-01-18 15:00:22 +00:00
}
newEPS, start, err := s.shouldFallbackPost(ctx, new)
if err != nil {
return err
}
2019-12-08 19:48:17 +00:00
s.failLk.Lock()
if s.failed > 0 {
s.failed = 0
s.activeEPS = 0
}
s.failLk.Unlock()
if newEPS == s.activeEPS {
return nil
}
s.abortActivePoSt()
if newEPS != Inactive && start {
s.doPost(ctx, newEPS, new)
}
return nil
}
func (s *FPoStScheduler) abortActivePoSt() {
if s.activeEPS == Inactive {
return // noop
}
if s.abort != nil {
s.abort()
}
log.Warnf("Aborting Fallback PoSt (EPS: %d)", s.activeEPS)
s.activeEPS = Inactive
s.abort = nil
}
2020-02-08 02:18:32 +00:00
func (s *FPoStScheduler) shouldFallbackPost(ctx context.Context, ts *types.TipSet) (abi.ChainEpoch, bool, error) {
ps, err := s.api.StateMinerPostState(ctx, s.actor, ts.Key())
if err != nil {
return 0, false, xerrors.Errorf("getting ElectionPeriodStart: %w", err)
}
if ts.Height() >= ps.ProvingPeriodStart {
return ps.ProvingPeriodStart, ts.Height() >= ps.ProvingPeriodStart+build.FallbackPoStConfidence, nil
}
return 0, false, nil
}