lotus/storage/wdpost_sched.go

227 lines
4.8 KiB
Go
Raw Normal View History

package storage
import (
"context"
"time"
"go.opencensus.io/trace"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/sector-storage/ffiwrapper"
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/chain/store"
"github.com/filecoin-project/lotus/chain/types"
)
const StartConfidence = 4 // TODO: config
2020-04-07 17:41:41 +00:00
type WindowPoStScheduler struct {
api storageMinerApi
2020-04-07 19:55:34 +00:00
prover 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-04-07 19:55:34 +00:00
activeDeadline *Deadline
2020-04-10 21:07:18 +00:00
abort context.CancelFunc
2019-12-08 19:48:17 +00:00
2020-04-07 19:55:34 +00:00
//failed abi.ChainEpoch // eps
//failLk sync.Mutex
}
func NewWindowedPoStScheduler(api storageMinerApi, sb storage.Prover, actor address.Address, worker address.Address) (*WindowPoStScheduler, error) {
mss, err := api.StateMinerSectorSize(context.TODO(), actor, types.EmptyTSK)
if err != nil {
return nil, xerrors.Errorf("getting sector size: %w", err)
}
spt, err := ffiwrapper.SealProofTypeFromSectorSize(mss)
if err != nil {
return nil, err
}
rt, err := spt.RegisteredWindowPoStProof()
if err != nil {
return nil, err
}
return &WindowPoStScheduler{api: api, prover: sb, actor: actor, worker: worker, proofType: rt}, nil
2020-04-07 19:55:34 +00:00
}
type Deadline struct {
// ID
start abi.ChainEpoch
}
func (Deadline) Equals(other Deadline) bool {
panic("maybe equal")
}
type abiPartition uint64
func (s *WindowPoStScheduler) getCurrentDeadline(ts *types.TipSet) (Deadline, error) {
return Deadline{}, nil
}
func (s *WindowPoStScheduler) getDeadlinePartitions(ts *types.TipSet, d Deadline) ([]abiPartition, error) {
return nil, nil
}
func (s *WindowPoStScheduler) getPartitionSectors(ts *types.TipSet, partition []abiPartition) ([]abi.SectorInfo, error) {
// TODO: maybe make this per partition
return nil, nil
}
2020-04-07 17:41:41 +00:00
func (s *WindowPoStScheduler) Run(ctx context.Context) {
defer s.abortActivePoSt()
var notifs <-chan []*store.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 {
log.Errorf("ChainNotify error: %+v")
time.Sleep(10 * time.Second)
continue
}
gotCur = false
}
select {
case changes, ok := <-notifs:
if !ok {
2020-04-07 17:41:41 +00:00
log.Warn("WindowPoStScheduler notifs channel closed")
notifs = nil
continue
}
if !gotCur {
if len(changes) != 1 {
log.Errorf("expected first notif to have len = 1")
continue
}
if changes[0].Type != store.HCCurrent {
log.Errorf("expected first notif to tell current ts")
continue
}
if err := s.update(ctx, changes[0].Val); err != nil {
log.Errorf("%+v", err)
}
gotCur = true
continue
}
2020-04-07 17:41:41 +00:00
ctx, span := trace.StartSpan(ctx, "WindowPoStScheduler.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
}
}
}
2020-04-07 17:41:41 +00:00
func (s *WindowPoStScheduler) revert(ctx context.Context, newLowest *types.TipSet) error {
if s.cur == newLowest {
return nil
}
s.cur = newLowest
2020-04-07 19:55:34 +00:00
newDeadline, err := s.getCurrentDeadline(newLowest)
if err != nil {
return err
}
2020-04-07 19:55:34 +00:00
if !s.activeDeadline.Equals(newDeadline) {
s.abortActivePoSt()
}
return nil
}
2020-04-07 17:41:41 +00:00
func (s *WindowPoStScheduler) update(ctx context.Context, new *types.TipSet) error {
2020-01-18 15:00:22 +00:00
if new == nil {
2020-04-07 17:41:41 +00:00
return xerrors.Errorf("no new tipset in WindowPoStScheduler.update")
2020-01-18 15:00:22 +00:00
}
2020-04-07 19:55:34 +00:00
shouldPost, newDeadline, err := s.shouldPost(ctx, new)
if err != nil {
return err
}
2020-04-07 19:55:34 +00:00
if !shouldPost {
return nil
}
2020-04-07 19:55:34 +00:00
/*s.failLk.Lock()
2019-12-08 19:48:17 +00:00
if s.failed > 0 {
s.failed = 0
s.activeEPS = 0
}
2020-04-07 19:55:34 +00:00
s.failLk.Unlock()*/
s.abortActivePoSt()
2020-04-07 19:55:34 +00:00
if newDeadline != nil {
s.doPost(ctx, newDeadline, new)
}
return nil
}
2020-04-07 17:41:41 +00:00
func (s *WindowPoStScheduler) abortActivePoSt() {
2020-04-07 19:55:34 +00:00
if s.activeDeadline == nil {
return // noop
}
if s.abort != nil {
s.abort()
}
2020-04-07 19:55:34 +00:00
log.Warnf("Aborting Fallback PoSt (Deadline: %+v)", s.activeDeadline)
2020-04-07 19:55:34 +00:00
s.activeDeadline = nil
s.abort = nil
}
2020-04-07 19:55:34 +00:00
func (s *WindowPoStScheduler) shouldPost(ctx context.Context, ts *types.TipSet) (bool, *Deadline, error) {
2020-04-16 17:17:56 +00:00
2020-04-07 19:55:34 +00:00
// call getCurrentDeadline, set activeDeadline if needed
panic("todo check actor state for post in the deadline")
return true, nil, nil
}